fix a bug

Author: cloud-fan

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

@@ -37,7 +37,7 @@
  * element in `values` region. We can get the length of this element by subtracting next offset.
  * Note that offset can by negative which means this element is null.
  *
- * In ghe `values` region, we store the content of elements. As we can get length info, so elements
+ * In the `values` region, we store the content of elements. As we can get length info, so elements
  * can be variable-length.
  *
  * Note that when we write out this array, we should write out the `numElements` at first 4 bytes,
@@ -315,4 +315,19 @@ public void writeToMemory(Object target, long targetOffset) {
       sizeInBytes
     );
   }
+
+  @Override
+  public UnsafeArrayData copy() {
+    UnsafeArrayData arrayCopy = new UnsafeArrayData();
+    final byte[] arrayDataCopy = new byte[sizeInBytes];
+    PlatformDependent.copyMemory(
+      baseObject,
+      baseOffset,
+      arrayDataCopy,
+      PlatformDependent.BYTE_ARRAY_OFFSET,
+      sizeInBytes
+    );
+    arrayCopy.pointTo(arrayDataCopy, PlatformDependent.BYTE_ARRAY_OFFSET, numElements, sizeInBytes);
+    return arrayCopy;
+  }
 }

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

@@ -58,4 +58,9 @@ public ArrayData keyArray() {
   public ArrayData valueArray() {
     return values;
   }
+
+  @Override
+  public UnsafeMapData copy() {
+    return new UnsafeMapData(keys.copy(), values.copy());
+  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/GenerateUnsafeProjection.scala

@@ -379,13 +379,18 @@ object GenerateUnsafeProjection extends CodeGenerator[Seq[Expression], UnsafePro
           case _: MapType => "UnsafeMapData"
           case _ => ctx.javaType(elementType)
         }
+        val copy = elementType match {
+          // We reuse the buffer during conversion, need copy it before process next element.
+          case _: StructType | _: ArrayType | _: MapType => ".copy()"
+          case _ => ""
+        }
 
         s"""
           final $unsafeType[] $elements = new $unsafeType[$numElements];
           for (int $index = 0; $index < $numElements; $index++) {
             ${convertedElement.code}
             if (!${convertedElement.isNull}) {
-              $elements[$index] = ${convertedElement.primitive};
+              $elements[$index] = ${convertedElement.primitive}$copy;
               $numBytes += $elementSize;
             }
           }

sql/catalyst/src/main/scala/org/apache/spark/sql/types/ArrayBasedMapData.scala

@@ -22,6 +22,8 @@ class ArrayBasedMapData(val keyArray: ArrayData, val valueArray: ArrayData) exte
 
   override def numElements(): Int = keyArray.numElements()
 
+  override def copy(): MapData = new ArrayBasedMapData(keyArray.copy(), valueArray.copy())
+
   // We need to check equality of map type in tests.
   override def equals(o: Any): Boolean = {
     if (!o.isInstanceOf[ArrayBasedMapData]) {

sql/catalyst/src/main/scala/org/apache/spark/sql/types/ArrayData.scala

@@ -24,6 +24,8 @@ import org.apache.spark.sql.catalyst.expressions.SpecializedGetters
 abstract class ArrayData extends SpecializedGetters with Serializable {
   def numElements(): Int
 
+  def copy(): ArrayData
+
   def toBooleanArray(): Array[Boolean] = {
     val size = numElements()
     val values = new Array[Boolean](size)

sql/catalyst/src/main/scala/org/apache/spark/sql/types/GenericArrayData.scala

@@ -26,6 +26,8 @@ class GenericArrayData(private[sql] val array: Array[Any])
 
   override def genericGet(ordinal: Int): Any = array(ordinal)
 
+  override def copy(): ArrayData = new GenericArrayData(array.clone())
+
   // todo: Array is invariant in scala, maybe use toSeq instead?
   override def toArray[T: ClassTag](elementType: DataType): Array[T] = array.map(_.asInstanceOf[T])
 

sql/catalyst/src/main/scala/org/apache/spark/sql/types/MapData.scala

@@ -25,6 +25,8 @@ abstract class MapData extends Serializable {
 
   def valueArray(): ArrayData
 
+  def copy(): MapData
+
   def foreach(keyType: DataType, valueType: DataType, f: (Any, Any) => Unit): Unit = {
     val length = numElements()
     val keys = keyArray()