fix

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala

@@ -38,13 +38,14 @@ private[orc] class OrcDeserializer(
 
   private[this] val mutableRow = new SpecificInternalRow(requiredSchema.map(_.dataType))
 
-  private[this] val unwrappers = requiredSchema.fields.map(f => unwrapperFor(f.dataType))
+  private[this] val length = requiredSchema.length
+
+  private[this] val unwrappers = requiredSchema.map(_.dataType).map(unwrapperFor).toArray
 
   def deserialize(orcStruct: OrcStruct): InternalRow = {
     var i = 0
-    val len = requiredSchema.length
     val names = orcStruct.getSchema.getFieldNames
-    while (i < len) {
+    while (i < length) {
       val name = requiredSchema(i).name
       val writable = if (missingColumnNames.contains(name)) {
         null
@@ -65,6 +66,46 @@ private[orc] class OrcDeserializer(
     mutableRow
   }
 
+  private[this] def unwrapperFor(dataType: DataType): (Any, InternalRow, Int) => Unit =
+    dataType match {
+      case NullType =>
+        (value: Any, row: InternalRow, ordinal: Int) =>
+          row.setNullAt(ordinal)
+
+      case BooleanType =>
+        (value: Any, row: InternalRow, ordinal: Int) =>
+          row.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
+
+      case ByteType =>
+        (value: Any, row: InternalRow, ordinal: Int) =>
+          row.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
+
+      case ShortType =>
+        (value: Any, row: InternalRow, ordinal: Int) =>
+          row.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
+
+      case IntegerType =>
+        (value: Any, row: InternalRow, ordinal: Int) =>
+          row.setInt(ordinal, value.asInstanceOf[IntWritable].get)
+
+      case LongType =>
+        (value: Any, row: InternalRow, ordinal: Int) =>
+          row.setLong(ordinal, value.asInstanceOf[LongWritable].get)
+
+      case FloatType =>
+        (value: Any, row: InternalRow, ordinal: Int) =>
+          row.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
+
+      case DoubleType =>
+        (value: Any, row: InternalRow, ordinal: Int) =>
+          row.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
+
+      case _ =>
+        val unwrapper = getValueUnwrapper(dataType)
+        (value: Any, row: InternalRow, ordinal: Int) =>
+          row(ordinal) = unwrapper(value)
+    }
+
   /**
    * Convert Apache ORC OrcStruct to Apache Spark InternalRow.
    * If internalRow is not None, fill into it. Otherwise, create a SpecificInternalRow and use it.
@@ -173,43 +214,4 @@ private[orc] class OrcDeserializer(
     case _ =>
       throw new UnsupportedOperationException(s"$dataType is not supported yet.")
   }
-
-  private[this] def unwrapperFor(dataType: DataType): (Any, InternalRow, Int) => Unit =
-    dataType match {
-      case NullType =>
-        (value: Any, row: InternalRow, ordinal: Int) => row.setNullAt(ordinal)
-
-      case BooleanType =>
-        (value: Any, row: InternalRow, ordinal: Int) =>
-          row.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
-
-      case ByteType =>
-        (value: Any, row: InternalRow, ordinal: Int) =>
-          row.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
-
-      case ShortType =>
-        (value: Any, row: InternalRow, ordinal: Int) =>
-          row.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
-
-      case IntegerType =>
-        (value: Any, row: InternalRow, ordinal: Int) =>
-          row.setInt(ordinal, value.asInstanceOf[IntWritable].get)
-
-      case LongType =>
-        (value: Any, row: InternalRow, ordinal: Int) =>
-          row.setLong(ordinal, value.asInstanceOf[LongWritable].get)
-
-      case FloatType =>
-        (value: Any, row: InternalRow, ordinal: Int) =>
-          row.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
-
-      case DoubleType =>
-        (value: Any, row: InternalRow, ordinal: Int) =>
-          row.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
-
-      case _ =>
-        val unwrapper = getValueUnwrapper(dataType)
-        (value: Any, row: InternalRow, ordinal: Int) =>
-          row(ordinal) = unwrapper(value)
-    }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcSerializer.scala

@@ -23,22 +23,70 @@ import org.apache.orc.storage.common.`type`.HiveDecimal
 import org.apache.orc.storage.serde2.io.{DateWritable, HiveDecimalWritable}
 
 import org.apache.spark.sql.catalyst.InternalRow
-import org.apache.spark.sql.catalyst.expressions.SpecificInternalRow
+import org.apache.spark.sql.catalyst.expressions.{SpecializedGetters, SpecificInternalRow}
 import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.execution.datasources.orc.OrcUtils.{getTypeDescription, withNullSafe}
 import org.apache.spark.sql.types._
 import org.apache.spark.unsafe.types.UTF8String
 
 private[orc] class OrcSerializer(dataSchema: StructType) {
 
-  private[this] lazy val orcStruct: OrcStruct =
-    createOrcValue(dataSchema).asInstanceOf[OrcStruct]
+  private[this] lazy val orcStruct: OrcStruct = createOrcValue(dataSchema).asInstanceOf[OrcStruct]
 
-  private[this] val writableWrappers =
-    dataSchema.fields.map(f => getWritableWrapper(f.dataType))
+  private[this] lazy val length = dataSchema.length
+
+  private[this] val writers = dataSchema.map(_.dataType).map(makeWriter).toArray
 
   def serialize(row: InternalRow): OrcStruct = {
-    convertInternalRowToOrcStruct(row, dataSchema, Some(writableWrappers), Some(orcStruct))
+    var i = 0
+    while (i < length) {
+      if (row.isNullAt(i)) {
+        orcStruct.setFieldValue(i, null)
+      } else {
+        writers(i)(row, i)
+      }
+      i += 1
+    }
+    orcStruct
+  }
+
+  private[this] def makeWriter(dataType: DataType): (SpecializedGetters, Int) => Unit = {
+    dataType match {
+      case BooleanType =>
+        (row: SpecializedGetters, ordinal: Int) =>
+          orcStruct.setFieldValue(ordinal, new BooleanWritable(row.getBoolean(ordinal)))
+
+      case ByteType =>
+        (row: SpecializedGetters, ordinal: Int) =>
+          orcStruct.setFieldValue(ordinal, new ByteWritable(row.getByte(ordinal)))
+
+      case ShortType =>
+        (row: SpecializedGetters, ordinal: Int) =>
+          orcStruct.setFieldValue(ordinal, new ShortWritable(row.getShort(ordinal)))
+
+      case IntegerType =>
+        (row: SpecializedGetters, ordinal: Int) =>
+          orcStruct.setFieldValue(ordinal, new IntWritable(row.getInt(ordinal)))
+
+      case LongType =>
+        (row: SpecializedGetters, ordinal: Int) =>
+          orcStruct.setFieldValue(ordinal, new LongWritable(row.getLong(ordinal)))
+
+      case FloatType =>
+        (row: SpecializedGetters, ordinal: Int) =>
+          orcStruct.setFieldValue(ordinal, new FloatWritable(row.getFloat(ordinal)))
+
+      case DoubleType =>
+        (row: SpecializedGetters, ordinal: Int) =>
+          orcStruct.setFieldValue(ordinal, new DoubleWritable(row.getDouble(ordinal)))
+
+      case _ =>
+        val wrapper = getWritableWrapper(dataType)
+        (row: SpecializedGetters, ordinal: Int) => {
+          val value = wrapper(row.get(ordinal, dataType)).asInstanceOf[WritableComparable[_]]
+          orcStruct.setFieldValue(ordinal, value)
+        }
+    }
   }
 
   /**
@@ -50,24 +98,22 @@ private[orc] class OrcSerializer(dataSchema: StructType) {
   /**
    * Convert Apache Spark InternalRow to Apache ORC OrcStruct.
    */
-  private[this] def convertInternalRowToOrcStruct(
-      row: InternalRow,
-      schema: StructType,
-      valueWrappers: Option[Seq[Any => Any]] = None,
-      struct: Option[OrcStruct] = None): OrcStruct = {
-    val wrappers =
-      valueWrappers.getOrElse(schema.fields.map(_.dataType).map(getWritableWrapper).toSeq)
-    val orcStruct = struct.getOrElse(createOrcValue(schema).asInstanceOf[OrcStruct])
-
-    for (schemaIndex <- 0 until schema.length) {
-      val fieldType = schema(schemaIndex).dataType
-      if (row.isNullAt(schemaIndex)) {
-        orcStruct.setFieldValue(schemaIndex, null)
+  private[this] def convertInternalRowToOrcStruct(row: InternalRow, schema: StructType) = {
+    val wrappers = schema.map(_.dataType).map(getWritableWrapper).toArray
+    val orcStruct = createOrcValue(schema).asInstanceOf[OrcStruct]
+
+    var i = 0
+    val length = schema.length
+    while (i < length) {
+      val fieldType = schema(i).dataType
+      if (row.isNullAt(i)) {
+        orcStruct.setFieldValue(i, null)
       } else {
-        val field = row.get(schemaIndex, fieldType)
-        val fieldValue = wrappers(schemaIndex)(field).asInstanceOf[WritableComparable[_]]
-        orcStruct.setFieldValue(schemaIndex, fieldValue)
+        val field = row.get(i, fieldType)
+        val fieldValue = wrappers(i)(field).asInstanceOf[WritableComparable[_]]
+        orcStruct.setFieldValue(i, fieldValue)
       }
+      i += 1
     }
     orcStruct
   }