replace 'new GenericArrayData' with 'GenericArrayData.allocate'

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/CatalystTypeConverters.scala

@@ -159,17 +159,17 @@ object CatalystTypeConverters {
     override def toCatalystImpl(scalaValue: Any): ArrayData = {
       scalaValue match {
         case a: Array[_] =>
-          new GenericArrayData(a.map(elementConverter.toCatalyst))
+          GenericArrayData.allocate(a.map(elementConverter.toCatalyst))
         case s: Seq[_] =>
-          new GenericArrayData(s.map(elementConverter.toCatalyst).toArray)
+          GenericArrayData.allocate(s.map(elementConverter.toCatalyst).toArray)
         case i: JavaIterable[_] =>
           val iter = i.iterator
           val convertedIterable = scala.collection.mutable.ArrayBuffer.empty[Any]
           while (iter.hasNext) {
             val item = iter.next()
             convertedIterable += elementConverter.toCatalyst(item)
           }
-          new GenericArrayData(convertedIterable.toArray)
+          GenericArrayData.allocate(convertedIterable.toArray)
       }
     }
 
@@ -410,7 +410,7 @@ object CatalystTypeConverters {
     case t: Timestamp => TimestampConverter.toCatalyst(t)
     case d: BigDecimal => new DecimalConverter(DecimalType(d.precision, d.scale)).toCatalyst(d)
     case d: JavaBigDecimal => new DecimalConverter(DecimalType(d.precision, d.scale)).toCatalyst(d)
-    case seq: Seq[Any] => new GenericArrayData(seq.map(convertToCatalyst).toArray)
+    case seq: Seq[Any] => GenericArrayData.allocate(seq.map(convertToCatalyst).toArray)
     case r: Row => InternalRow(r.toSeq.map(convertToCatalyst): _*)
     case arr: Array[Any] => new GenericArrayData(arr.map(convertToCatalyst))
     case map: Map[_, _] =>

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/ScalaReflection.scala

@@ -20,7 +20,7 @@ package org.apache.spark.sql.catalyst
 import org.apache.spark.sql.catalyst.analysis.{GetColumnByOrdinal, UnresolvedAttribute, UnresolvedExtractValue}
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.objects._
-import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, DateTimeUtils, GenericArrayData}
+import org.apache.spark.sql.catalyst.util.{ArrayBasedMapData, DateTimeUtils, GenericRefArrayData}
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.{CalendarInterval, UTF8String}
 
@@ -459,7 +459,7 @@ object ScalaReflection extends ScalaReflection {
 
         case dt =>
           NewInstance(
-            classOf[GenericArrayData],
+            classOf[GenericRefArrayData],
             input :: Nil,
             dataType = ArrayType(dt, schemaFor(elementType).nullable))
       }

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

@@ -388,7 +388,7 @@ case class Cast(child: Expression, dataType: DataType) extends UnaryExpression w
           values(i) = elementCast(e)
         }
       })
-      new GenericArrayData(values)
+      GenericArrayData.allocate(values)
     })
   }
 
@@ -864,7 +864,7 @@ case class Cast(child: Expression, dataType: DataType) extends UnaryExpression w
             }
           }
         }
-        $evPrim = new $arrayClass($values);
+        $evPrim = $arrayClass.allocate($values);
       """
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/PivotFirst.scala

@@ -131,7 +131,7 @@ case class PivotFirst(
     for (i <- 0 until indexSize) {
       result(i) = input.get(mutableAggBufferOffset + i, valueDataType)
     }
-    new GenericArrayData(result)
+    GenericArrayData.allocate(result)
   }
 
   override def withNewInputAggBufferOffset(newInputAggBufferOffset: Int): ImperativeAggregate =

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/collect.scala

@@ -78,7 +78,7 @@ abstract class Collect extends ImperativeAggregate {
   }
 
   override def eval(input: InternalRow): Any = {
-    new GenericArrayData(buffer.toArray)
+    GenericArrayData.allocate(buffer.toArray)
   }
 }
 

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

@@ -96,7 +96,7 @@ object GenerateSafeProjection extends CodeGenerator[Seq[Expression], Projection]
           $values[$index] = ${elementConverter.value};
         }
       }
-      final ArrayData $output = new $arrayClass($values);
+      final ArrayData $output = $arrayClass.allocate($values);
     """
 
     ExprCode(code, "false", output)

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

@@ -206,7 +206,7 @@ case class SortArray(base: Expression, ascendingOrder: Expression)
     if (elementType != NullType) {
       java.util.Arrays.sort(data, if (ascending.asInstanceOf[Boolean]) lt else gt)
     }
-    new GenericArrayData(data.asInstanceOf[Array[Any]])
+    GenericArrayData.allocate(data.asInstanceOf[Array[Any]])
   }
 
   override def prettyName: String = "sort_array"

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

@@ -52,7 +52,7 @@ case class CreateArray(children: Seq[Expression]) extends Expression {
   override def nullable: Boolean = false
 
   override def eval(input: InternalRow): Any = {
-    new GenericArrayData(children.map(_.eval(input)).toArray)
+    GenericArrayData.allocate(children.map(_.eval(input)).toArray)
   }
 
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
@@ -76,7 +76,7 @@ case class CreateArray(children: Seq[Expression]) extends Expression {
            """
         }) +
       s"""
-        final ArrayData ${ev.value} = new $arrayClass($values);
+        final ArrayData ${ev.value} = $arrayClass.allocate($values);
         this.$values = null;
       """)
   }
@@ -130,7 +130,8 @@ case class CreateMap(children: Seq[Expression]) extends Expression {
       throw new RuntimeException("Cannot use null as map key!")
     }
     val valueArray = values.map(_.eval(input)).toArray
-    new ArrayBasedMapData(new GenericArrayData(keyArray), new GenericArrayData(valueArray))
+    new ArrayBasedMapData(
+      GenericArrayData.allocate(keyArray), GenericArrayData.allocate(valueArray))
   }
 
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
@@ -141,8 +142,8 @@ case class CreateMap(children: Seq[Expression]) extends Expression {
     ctx.addMutableState("Object[]", keyArray, s"this.$keyArray = null;")
     ctx.addMutableState("Object[]", valueArray, s"this.$valueArray = null;")
 
-    val keyData = s"new $arrayClass($keyArray)"
-    val valueData = s"new $arrayClass($valueArray)"
+    val keyData = s"$arrayClass.allocate($keyArray)"
+    val valueData = s"$arrayClass.allocate($valueArray)"
     ev.copy(code = s"""
       final boolean ${ev.isNull} = false;
       $keyArray = new Object[${keys.size}];

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

@@ -176,7 +176,7 @@ case class GetArrayStructFields(
       }
       i += 1
     }
-    new GenericArrayData(result)
+    GenericArrayData.allocate(result)
   }
 
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
@@ -201,7 +201,7 @@ case class GetArrayStructFields(
             }
           }
         }
-        ${ev.value} = new $arrayClass($values);
+        ${ev.value} = $arrayClass.allocate($values);
       """
     })
   }

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

@@ -537,7 +537,7 @@ case class MapObjects private(
           $loopIndex += 1;
         }
 
-        ${ev.value} = new ${classOf[GenericArrayData].getName}($convertedArray);
+        ${ev.value} = ${classOf[GenericArrayData].getName}.allocate($convertedArray);
       }
     """
     ev.copy(code = code, isNull = genInputData.isNull)

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

@@ -191,14 +191,14 @@ case class StringSplit(str: Expression, pattern: Expression)
 
   override def nullSafeEval(string: Any, regex: Any): Any = {
     val strings = string.asInstanceOf[UTF8String].split(regex.asInstanceOf[UTF8String], -1)
-    new GenericArrayData(strings.asInstanceOf[Array[Any]])
+    GenericArrayData.allocate(strings.asInstanceOf[Array[Any]])
   }
 
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     val arrayClass = classOf[GenericArrayData].getName
     nullSafeCodeGen(ctx, ev, (str, pattern) =>
       // Array in java is covariant, so we don't need to cast UTF8String[] to Object[].
-      s"""${ev.value} = new $arrayClass($str.split($pattern, -1));""")
+      s"""${ev.value} = $arrayClass.allocate($str.split($pattern, -1));""")
   }
 
   override def prettyName: String = "split"

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/json/JacksonParser.scala

@@ -404,7 +404,7 @@ class JacksonParser(
       values += fieldConverter.apply(parser)
     }
 
-    new GenericArrayData(values.toArray)
+    GenericArrayData.allocate(values.toArray)
   }
 
   /**

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisErrorSuite.scala

@@ -57,8 +57,8 @@ private[sql] class UngroupableUDT extends UserDefinedType[UngroupableData] {
   override def sqlType: DataType = MapType(IntegerType, IntegerType)
 
   override def serialize(ungroupableData: UngroupableData): MapData = {
-    val keyArray = new GenericArrayData(ungroupableData.data.keys.toSeq)
-    val valueArray = new GenericArrayData(ungroupableData.data.values.toSeq)
+    val keyArray = GenericArrayData.allocate(ungroupableData.data.keys.toSeq)
+    val valueArray = GenericArrayData.allocate(ungroupableData.data.values.toSeq)
     new ArrayBasedMapData(keyArray, valueArray)
   }
 

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/EncoderResolutionSuite.scala

@@ -70,11 +70,11 @@ class EncoderResolutionSuite extends PlanTest {
     val bound = encoder.resolveAndBind(attrs)
 
     // If no null values appear, it should works fine
-    bound.fromRow(InternalRow(new GenericArrayData(Array(1, 2))))
+    bound.fromRow(InternalRow(GenericArrayData.allocate(Array(1, 2))))
 
     // If there is null value, it should throw runtime exception
     val e = intercept[RuntimeException] {
-      bound.fromRow(InternalRow(new GenericArrayData(Array(1, null))))
+      bound.fromRow(InternalRow(GenericArrayData.allocate(Array(1, null))))
     }
     assert(e.getMessage.contains("Null value appeared in non-nullable field"))
   }

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/encoders/RowEncoderSuite.scala

@@ -51,7 +51,7 @@ class ExamplePointUDT extends UserDefinedType[ExamplePoint] {
     val output = new Array[Any](2)
     output(0) = p.x
     output(1) = p.y
-    new GenericArrayData(output)
+    GenericArrayData.allocate(output)
   }
 
   override def deserialize(datum: Any): ExamplePoint = {

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/UnsafeRowConverterSuite.scala

@@ -291,7 +291,7 @@ class UnsafeRowConverterSuite extends SparkFunSuite with Matchers {
     assert(unsafeRow.getSizeInBytes == 8 + 2 * 8 + row1.getSizeInBytes + row2.getSizeInBytes)
   }
 
-  private def createArray(values: Any*): ArrayData = new GenericArrayData(values.toArray)
+  private def createArray(values: Any*): ArrayData = GenericArrayData.allocate(values.toArray)
 
   private def createMap(keys: Any*)(values: Any*): MapData = {
     assert(keys.length == values.length)

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/expressions/codegen/GeneratedProjectionSuite.scala

@@ -86,7 +86,7 @@ class GeneratedProjectionSuite extends SparkFunSuite {
   test("generated unsafe projection with array of binary") {
     val row = InternalRow(
       Array[Byte](1, 2),
-      new GenericArrayData(Array(Array[Byte](1, 2), null, Array[Byte](3, 4))))
+      GenericArrayData.allocate(Array(Array[Byte](1, 2), null, Array[Byte](3, 4))))
     val fields = (BinaryType :: ArrayType(BinaryType) :: Nil).toArray[DataType]
 
     val unsafeProj = UnsafeProjection.create(fields)

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala

@@ -491,7 +491,7 @@ private[parquet] class ParquetRowConverter(
 
     override def getConverter(fieldIndex: Int): Converter = elementConverter
 
-    override def end(): Unit = updater.set(new GenericArrayData(currentArray.toArray))
+    override def end(): Unit = updater.set(GenericArrayData.allocate(currentArray.toArray))
 
     // NOTE: We can't reuse the mutable `ArrayBuffer` here and must instantiate a new buffer for the
     // next value.  `Row.copy()` only copies row cells, it doesn't do deep copy to objects stored
@@ -590,7 +590,7 @@ private[parquet] class ParquetRowConverter(
 
     protected def newArrayUpdater(updater: ParentContainerUpdater) = new ParentContainerUpdater {
       override def start(): Unit = currentArray = ArrayBuffer.empty[Any]
-      override def end(): Unit = updater.set(new GenericArrayData(currentArray.toArray))
+      override def end(): Unit = updater.set(GenericArrayData.allocate(currentArray.toArray))
       override def set(value: Any): Unit = currentArray += value
     }
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/python/EvaluatePython.scala

@@ -119,10 +119,10 @@ object EvaluatePython {
     case (c, BinaryType) if c.getClass.isArray && c.getClass.getComponentType.getName == "byte" => c
 
     case (c: java.util.List[_], ArrayType(elementType, _)) =>
-      new GenericArrayData(c.asScala.map { e => fromJava(e, elementType)}.toArray)
+      GenericArrayData.allocate(c.asScala.map { e => fromJava(e, elementType)}.toArray)
 
     case (c, ArrayType(elementType, _)) if c.getClass.isArray =>
-      new GenericArrayData(c.asInstanceOf[Array[_]].map(e => fromJava(e, elementType)))
+      GenericArrayData.allocate(c.asInstanceOf[Array[_]].map(e => fromJava(e, elementType)))
 
     case (javaMap: java.util.Map[_, _], MapType(keyType, valueType, _)) =>
       ArrayBasedMapData(

sql/core/src/main/scala/org/apache/spark/sql/test/ExamplePointUDT.scala

@@ -49,7 +49,7 @@ private[sql] class ExamplePointUDT extends UserDefinedType[ExamplePoint] {
     val output = new Array[Any](2)
     output(0) = p.x
     output(1) = p.y
-    new GenericArrayData(output)
+    GenericArrayData.allocate(output)
   }
 
   override def deserialize(datum: Any): ExamplePoint = {

sql/core/src/test/scala/org/apache/spark/sql/UnsafeRowSuite.scala

@@ -160,7 +160,7 @@ class UnsafeRowSuite extends SparkFunSuite {
   }
 
   test("calling hashCode on unsafe array returned by getArray(ordinal)") {
-    val row = InternalRow.apply(new GenericArrayData(Array(1L)))
+    val row = InternalRow.apply(GenericArrayData.allocate(Array(1L)))
     val unsafeRow = UnsafeProjection.create(Array[DataType](ArrayType(LongType))).apply(row)
     // Makes sure hashCode on unsafe array won't crash
     unsafeRow.getArray(0).hashCode()

sql/core/src/test/scala/org/apache/spark/sql/UserDefinedTypeSuite.scala

@@ -50,7 +50,7 @@ object UDT {
     override def sqlType: DataType = ArrayType(DoubleType, containsNull = false)
 
     override def serialize(features: MyDenseVector): ArrayData = {
-      new GenericArrayData(features.data.map(_.asInstanceOf[Any]))
+      GenericArrayData.allocate(features.data.map(_.asInstanceOf[Any]))
     }
 
     override def deserialize(datum: Any): MyDenseVector = {

sql/core/src/test/scala/org/apache/spark/sql/execution/columnar/ColumnarTestUtils.scala

@@ -56,7 +56,7 @@ object ColumnarTestUtils {
       case STRUCT(_) =>
         new GenericInternalRow(Array[Any](UTF8String.fromString(Random.nextString(10))))
       case ARRAY(_) =>
-        new GenericArrayData(Array[Any](Random.nextInt(), Random.nextInt()))
+        GenericArrayData.allocate(Array[Any](Random.nextInt(), Random.nextInt()))
       case MAP(_) =>
         ArrayBasedMapData(
           Map(Random.nextInt() -> UTF8String.fromString(Random.nextString(Random.nextInt(32)))))

sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveInspectorSuite.scala

@@ -229,7 +229,7 @@ class HiveInspectorSuite extends SparkFunSuite with HiveInspectors {
   test("wrap / unwrap Array Type") {
     val dt = ArrayType(dataTypes(0))
 
-    val d = new GenericArrayData(Array(row(0), row(0)))
+    val d = GenericArrayData.allocate(Array(row(0), row(0)))
     checkValue(d, unwrap(wrap(d, toInspector(dt), dt), toInspector(dt)))
     checkValue(null, unwrap(wrap(null, toInspector(dt), dt), toInspector(dt)))
     checkValue(d,