Merge remote-tracking branch 'upstream/master' into add_date_timestamp

Conflicts:
	sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeRow.java
	sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/rows.scala

sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/expressions/UnsafeRow.java

@@ -30,7 +30,7 @@
 import scala.collection.mutable.ArraySeq;
 
 import org.apache.spark.sql.catalyst.util.DateUtils;
-import org.apache.spark.sql.Row;
+import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.BaseMutableRow;
 import org.apache.spark.sql.types.DataType;
 import org.apache.spark.sql.types.StructType;

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/UnsafeRowConverter.scala

@@ -18,6 +18,7 @@
 package org.apache.spark.sql.catalyst.expressions
 
 import org.apache.spark.sql.catalyst.util.DateUtils
+import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.PlatformDependent
 import org.apache.spark.unsafe.array.ByteArrayMethods
@@ -265,24 +266,24 @@ private class StringUnsafeColumnWriter private() extends UnsafeColumnWriter {
 }
 
 private class DateUnsafeColumnWriter private() extends UnsafeColumnWriter {
-  def getSize(source: Row, column: Int): Int = {
+  def getSize(source: InternalRow, column: Int): Int = {
     0
   }
 
-  override def write(source: Row, target: UnsafeRow, column: Int, appendCursor: Int): Int = {
+  override def write(source: InternalRow, target: UnsafeRow, column: Int, appendCursor: Int): Int = {
     target.setDate(column, source.getDate(column))
     0
   }
 }
 
 private class TimestampUnsafeColumnWriter private() extends UnsafeColumnWriter {
-  def getSize(source: Row, column: Int): Int = {
+  def getSize(source: InternalRow, column: Int): Int = {
     // Although Timestamp is fixed length, it needs 12-byte that is more than 8-byte word per field.
     // So we need to store it at the variable-length section.
     16
   }
 
-  override def write(source: Row, target: UnsafeRow, column: Int, appendCursor: Int): Int = {
+  override def write(source: InternalRow, target: UnsafeRow, column: Int, appendCursor: Int): Int = {
     val value = DateUtils.toJavaTimestamp(source.get(column).asInstanceOf[Long])
     val time = value.getTime()
     val nanos = value.getNanos()

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/rows.scala

@@ -20,7 +20,8 @@ package org.apache.spark.sql.catalyst.expressions
 import java.sql.{Date, Timestamp}
 
 import org.apache.spark.sql.catalyst.util.DateUtils
-import org.apache.spark.sql.types.{UTF8String, DataType, StructType, AtomicType}
+import org.apache.spark.sql.types.{DataType, StructType, AtomicType}
+import org.apache.spark.unsafe.types.UTF8String
 
 /**
  * An extended interface to [[InternalRow]] that allows the values for each column to be updated.
@@ -219,7 +220,7 @@ class GenericMutableRow(v: Array[Any]) extends GenericRow(v) with MutableRow {
   override def setInt(ordinal: Int, value: Int): Unit = { values(ordinal) = value }
   override def setLong(ordinal: Int, value: Long): Unit = { values(ordinal) = value }
   override def setString(ordinal: Int, value: String): Unit = {
-    values(ordinal) = UTF8String(value)
+    values(ordinal) = UTF8String.fromString(value)
   }
 
   override def setDate(ordinal: Int, value: java.sql.Date): Unit = {