rebase

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

@@ -143,7 +143,7 @@ case class ApproximatePercentile(
     if (result.length == 0) {
       null
     } else if (returnPercentileArray) {
-      new GenericArrayData(result)
+      GenericArrayData.allocate(result)
     } else {
       result(0)
     }

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

@@ -39,30 +39,30 @@ class ObjectExpressionsSuite extends SparkFunSuite with ExpressionEvalHelper {
     // test UnsafeRow-backed data
     val structEncoder = ExpressionEncoder[Array[Tuple2[java.lang.Integer, java.lang.Integer]]]
     val structInputRow = InternalRow.fromSeq(Seq(Array((1, 2), (3, 4))))
-    val structExpected = new GenericArrayData(
+    val structExpected = GenericArrayData.allocate(
       Array(InternalRow.fromSeq(Seq(1, 2)), InternalRow.fromSeq(Seq(3, 4))))
     checkEvalutionWithUnsafeProjection(
       structEncoder.serializer.head, structExpected, structInputRow)
 
     // test UnsafeArray-backed data
     val arrayEncoder = ExpressionEncoder[Array[Array[Int]]]
     val arrayInputRow = InternalRow.fromSeq(Seq(Array(Array(1, 2), Array(3, 4))))
-    val arrayExpected = new GenericArrayData(
-      Array(new GenericArrayData(Array(1, 2)), new GenericArrayData(Array(3, 4))))
+    val arrayExpected = GenericArrayData.allocate(
+      Array(GenericArrayData.allocate(Array(1, 2)), GenericArrayData.allocate(Array(3, 4))))
     checkEvalutionWithUnsafeProjection(
       arrayEncoder.serializer.head, arrayExpected, arrayInputRow)
 
     // test UnsafeMap-backed data
     val mapEncoder = ExpressionEncoder[Array[Map[Int, Int]]]
     val mapInputRow = InternalRow.fromSeq(Seq(Array(
       Map(1 -> 100, 2 -> 200), Map(3 -> 300, 4 -> 400))))
-    val mapExpected = new GenericArrayData(Seq(
+    val mapExpected = GenericArrayData.allocate(Seq(
       new ArrayBasedMapData(
-        new GenericArrayData(Array(1, 2)),
-        new GenericArrayData(Array(100, 200))),
+        GenericArrayData.allocate(Array(1, 2)),
+        GenericArrayData.allocate(Array(100, 200))),
       new ArrayBasedMapData(
-        new GenericArrayData(Array(3, 4)),
-        new GenericArrayData(Array(300, 400)))))
+        GenericArrayData.allocate(Array(3, 4)),
+        GenericArrayData.allocate(Array(300, 400)))))
     checkEvalutionWithUnsafeProjection(
       mapEncoder.serializer.head, mapExpected, mapInputRow)
   }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCRDD.scala

@@ -229,137 +229,6 @@ private[jdbc] class JDBCRDD(
     }
   }
 
-  // A `JDBCValueSetter` is responsible for converting and setting a value from `ResultSet`
-  // into a field for `MutableRow`. The last argument `Int` means the index for the
-  // value to be set in the row and also used for the value to retrieve from `ResultSet`.
-  private type JDBCValueSetter = (ResultSet, MutableRow, Int) => Unit
-
-  /**
-   * Creates `JDBCValueSetter`s according to [[StructType]], which can set
-   * each value from `ResultSet` to each field of [[MutableRow]] correctly.
-   */
-  def makeSetters(schema: StructType): Array[JDBCValueSetter] =
-    schema.fields.map(sf => makeSetter(sf.dataType, sf.metadata))
-
-  private def makeSetter(dt: DataType, metadata: Metadata): JDBCValueSetter = dt match {
-    case BooleanType =>
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        row.setBoolean(pos, rs.getBoolean(pos + 1))
-
-    case DateType =>
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        // DateTimeUtils.fromJavaDate does not handle null value, so we need to check it.
-        val dateVal = rs.getDate(pos + 1)
-        if (dateVal != null) {
-          row.setInt(pos, DateTimeUtils.fromJavaDate(dateVal))
-        } else {
-          row.update(pos, null)
-        }
-
-    // When connecting with Oracle DB through JDBC, the precision and scale of BigDecimal
-    // object returned by ResultSet.getBigDecimal is not correctly matched to the table
-    // schema reported by ResultSetMetaData.getPrecision and ResultSetMetaData.getScale.
-    // If inserting values like 19999 into a column with NUMBER(12, 2) type, you get through
-    // a BigDecimal object with scale as 0. But the dataframe schema has correct type as
-    // DecimalType(12, 2). Thus, after saving the dataframe into parquet file and then
-    // retrieve it, you will get wrong result 199.99.
-    // So it is needed to set precision and scale for Decimal based on JDBC metadata.
-    case DecimalType.Fixed(p, s) =>
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        val decimal =
-          nullSafeConvert[java.math.BigDecimal](rs.getBigDecimal(pos + 1), d => Decimal(d, p, s))
-        row.update(pos, decimal)
-
-    case DoubleType =>
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        row.setDouble(pos, rs.getDouble(pos + 1))
-
-    case FloatType =>
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        row.setFloat(pos, rs.getFloat(pos + 1))
-
-    case IntegerType =>
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        row.setInt(pos, rs.getInt(pos + 1))
-
-    case LongType if metadata.contains("binarylong") =>
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        val bytes = rs.getBytes(pos + 1)
-        var ans = 0L
-        var j = 0
-        while (j < bytes.size) {
-          ans = 256 * ans + (255 & bytes(j))
-          j = j + 1
-        }
-        row.setLong(pos, ans)
-
-    case LongType =>
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        row.setLong(pos, rs.getLong(pos + 1))
-
-    case StringType =>
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        // TODO(davies): use getBytes for better performance, if the encoding is UTF-8
-        row.update(pos, UTF8String.fromString(rs.getString(pos + 1)))
-
-    case TimestampType =>
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        val t = rs.getTimestamp(pos + 1)
-        if (t != null) {
-          row.setLong(pos, DateTimeUtils.fromJavaTimestamp(t))
-        } else {
-          row.update(pos, null)
-        }
-
-    case BinaryType =>
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        row.update(pos, rs.getBytes(pos + 1))
-
-    case ArrayType(et, _) =>
-      val elementConversion = et match {
-        case TimestampType =>
-          (array: Object) =>
-            array.asInstanceOf[Array[java.sql.Timestamp]].map { timestamp =>
-              nullSafeConvert(timestamp, DateTimeUtils.fromJavaTimestamp)
-            }
-
-        case StringType =>
-          (array: Object) =>
-            array.asInstanceOf[Array[java.lang.String]]
-              .map(UTF8String.fromString)
-
-        case DateType =>
-          (array: Object) =>
-            array.asInstanceOf[Array[java.sql.Date]].map { date =>
-              nullSafeConvert(date, DateTimeUtils.fromJavaDate)
-            }
-
-        case dt: DecimalType =>
-          (array: Object) =>
-            array.asInstanceOf[Array[java.math.BigDecimal]].map { decimal =>
-              nullSafeConvert[java.math.BigDecimal](
-                decimal, d => Decimal(d, dt.precision, dt.scale))
-            }
-
-        case LongType if metadata.contains("binarylong") =>
-          throw new IllegalArgumentException(s"Unsupported array element " +
-            s"type ${dt.simpleString} based on binary")
-
-        case ArrayType(_, _) =>
-          throw new IllegalArgumentException("Nested arrays unsupported")
-
-        case _ => (array: Object) => array.asInstanceOf[Array[Any]]
-      }
-
-      (rs: ResultSet, row: MutableRow, pos: Int) =>
-        val array = nullSafeConvert[Object](
-          rs.getArray(pos + 1).getArray,
-          array => GenericArrayData.allocate(elementConversion.apply(array)))
-        row.update(pos, array)
-
-    case _ => throw new IllegalArgumentException(s"Unsupported type ${dt.simpleString}")
-  }
-
   /**
    * Runs the SQL query against the JDBC driver.
    *

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala

@@ -425,7 +425,7 @@ object JdbcUtils extends Logging {
       (rs: ResultSet, row: InternalRow, pos: Int) =>
         val array = nullSafeConvert[Object](
           rs.getArray(pos + 1).getArray,
-          array => new GenericArrayData(elementConversion.apply(array)))
+          array => GenericArrayData.allocate(elementConversion.apply(array)))
         row.update(pos, array)
 
     case _ => throw new IllegalArgumentException(s"Unsupported type ${dt.simpleString}")