You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2020/08/04 20:55:24 UTC

[GitHub] [spark] msamirkhan opened a new pull request #29353: [Spark-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

msamirkhan opened a new pull request #29353:
URL: https://github.com/apache/spark/pull/29353


   <!--
   Thanks for sending a pull request!  Here are some tips for you:
     1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
     2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
     3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
     4. Be sure to keep the PR description updated to reflect all changes.
     5. Please write your PR title to summarize what this PR proposes.
     6. If possible, provide a concise example to reproduce the issue for a faster review.
     7. If you want to add a new configuration, please read the guideline first for naming configurations in
        'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
   -->
   
   ### What changes were proposed in this pull request?
   Changes to Serializing/Deserializing ORC file format. There are 3 commits. Attached PDF has info on improvements in each commit. Essentially pushing a lot of processing, which can be done solely based on the data schema, to the initialization stage so that it is not repeated for each row of data. Eg, before it would convert catalyst schema to orc schema and recursively create orc structs, for each row of data. Now, it only has to do the conversion once at the schema level.
   <!--
   Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue. 
   If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
     1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
     2. If you fix some SQL features, you can provide some references of other DBMSes.
     3. If there is design documentation, please add the link.
     4. If there is a discussion in the mailing list, please add the link.
   -->
   
   
   ### Why are the changes needed?
   Spark performance is really slow for these cases. Attached pdf has info on improvements, but summarily the times (in seconds) go down as follows in each case:
   Read:
   Nested Structs: 184 -> 44
   Array of Struct: 66 -> 15
   
   Write
   Nested Structs: 543 -> 39
   Array of Struct: 330 -> 37
   <!--
   Please clarify why the changes are needed. For instance,
     1. If you propose a new API, clarify the use case for a new API.
     2. If you fix a bug, you can clarify why it is a bug.
   -->
   
   
   ### Does this PR introduce _any_ user-facing change?
   No
   <!--
   Note that it means *any* user-facing change including all aspects such as the documentation fix.
   If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
   If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
   If no, write 'No'.
   -->
   
   
   ### How was this patch tested?
   Ran benchmarks (please see the PR: https://github.com/apache/spark/pull/29352 for [SPARK-32532](https://issues.apache.org/jira/browse/SPARK-32532) for additional benchmarks) and tests.
   <!--
   If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
   If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
   If tests were not added, please describe why they were not added and/or why it was difficult to add.
   -->
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r466144939



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -73,135 +75,157 @@ class OrcDeserializer(
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
   private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
+      dataType: DataType, reuseObj: Boolean)
+  : (CatalystDataUpdater, Int, WritableComparable[_]) => Unit =
     dataType match {
-      case NullType => (ordinal, _) =>
+      case NullType => (updater, ordinal, _) =>

Review comment:
       Only the top level field writers are currently being created once. For each level below, they are created for each data point. Eg, the case for struct did the following:
   
   `case st: StructType => (ordinal, value) =>`
   `...`
   `    val fieldConverters = st.map(_.dataType).map { dt => newWriter(dt, fieldUpdater) }.toArray`
   `...`
   
   i.e., it returns a function, which when called will create relevant fieldConverters. Instead, the field converters should be created outside the function so that they are created only once.
   
   Now, the updaters are not always reused. Eg, under an array or a map (i.e., whenever reuseObj turns to false).




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669461208






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA removed a comment on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669465889


   **[Test build #127104 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127104/testReport)** for PR 29353 at commit [`b360fb6`](https://github.com/apache/spark/commit/b360fb66f1e3b45561af7ec9d5dea352545e7f33).


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] github-actions[bot] commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-740272998


   We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
   If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669609253






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668986802






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins removed a comment on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668986802






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r465860164



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcSerializer.scala
##########
@@ -150,69 +156,110 @@ class OrcSerializer(dataSchema: StructType) {
     case DecimalType.Fixed(precision, scale) =>
       OrcShimUtils.getHiveDecimalWritable(precision, scale)
 
-    case st: StructType => (getter, ordinal) =>
-      val result = createOrcValue(st).asInstanceOf[OrcStruct]
-      val fieldConverters = st.map(_.dataType).map(newConverter(_))
-      val numFields = st.length
-      val struct = getter.getStruct(ordinal, numFields)
-      var i = 0
-      while (i < numFields) {
-        if (struct.isNullAt(i)) {
-          result.setFieldValue(i, null)
-        } else {
-          result.setFieldValue(i, fieldConverters(i)(struct, i))
+    case st: StructType =>
+      val orcSchema = schemaConverter(st)
+      val fieldConverters = st.map(_.dataType).map(newConverter(_, reuseObj))
+      val baseConverter: (OrcStruct, SpecializedGetters, Int) => OrcStruct =
+        (result, getter, ordinal) => {
+          val numFields = st.length
+          val struct = getter.getStruct(ordinal, numFields)
+          var i = 0
+          while (i < numFields) {
+            if (struct.isNullAt(i)) {
+              result.setFieldValue(i, null)
+            } else {
+              result.setFieldValue(i, fieldConverters(i)(struct, i))
+            }
+            i += 1
+          }
+          result
         }
-        i += 1
-      }
-      result
 
-    case ArrayType(elementType, _) => (getter, ordinal) =>
-      val result = createOrcValue(dataType).asInstanceOf[OrcList[WritableComparable[_]]]
+//      if (reuseObj) {
+//        val result = new OrcStruct(orcSchema)
+//        (getter, ordinal) => baseConverter.apply(result, getter, ordinal)
+//      } else {
+        (getter, ordinal) =>
+          val result = new OrcStruct(orcSchema)
+          baseConverter.apply(result, getter, ordinal)
+//      }
+
+    case ArrayType(elementType, _) =>
+      val orcSchema = schemaConverter(dataType)
       // Need to put all converted values to a list, can't reuse object.
-      val elementConverter = newConverter(elementType, reuseObj = false)
-      val array = getter.getArray(ordinal)
-      var i = 0
-      while (i < array.numElements()) {
-        if (array.isNullAt(i)) {
-          result.add(null)
-        } else {
-          result.add(elementConverter(array, i))
+      val elementConverter = newConverter(elementType, false)
+      val baseArrayConverter: (OrcList[WritableComparable[_]], SpecializedGetters, Int) => Unit = {
+        (result, getter, ordinal) => {
+          val array = getter.getArray(ordinal)
+          var i = 0
+          while (i < array.numElements()) {
+            if (array.isNullAt(i)) {
+              result.add(null)
+            } else {
+              result.add(elementConverter(array, i))
+            }
+            i += 1
+          }
         }
-        i += 1
       }
-      result
 
-    case MapType(keyType, valueType, _) => (getter, ordinal) =>
-      val result = createOrcValue(dataType)
-        .asInstanceOf[OrcMap[WritableComparable[_], WritableComparable[_]]]
+//      if (reuseObj) {
+//        val result = new OrcList[WritableComparable[_]](orcSchema)
+//        (getter, ordinal) => {
+//          result.clear()
+//          baseArrayConverter.apply(result, getter, ordinal)
+//          result
+//        }
+//      } else {
+        (getter, ordinal) => {
+          val result = new OrcList[WritableComparable[_]](orcSchema)
+          baseArrayConverter.apply(result, getter, ordinal)
+          result
+        }
+//      }
+
+    case MapType(keyType, valueType, _) =>
+      val orcSchema = schemaConverter(dataType)
       // Need to put all converted values to a list, can't reuse object.
-      val keyConverter = newConverter(keyType, reuseObj = false)
-      val valueConverter = newConverter(valueType, reuseObj = false)
-      val map = getter.getMap(ordinal)
-      val keyArray = map.keyArray()
-      val valueArray = map.valueArray()
-      var i = 0
-      while (i < map.numElements()) {
-        val key = keyConverter(keyArray, i)
-        if (valueArray.isNullAt(i)) {
-          result.put(key, null)
-        } else {
-          result.put(key, valueConverter(valueArray, i))
+      val keyConverter = newConverter(keyType, false)
+      val valueConverter = newConverter(valueType, false)
+      val baseMapConverter:
+        (OrcMap[WritableComparable[_], WritableComparable[_]], SpecializedGetters, Int) => Unit = {
+        (result, getter, ordinal) => {
+          val map = getter.getMap(ordinal)
+          val keyArray = map.keyArray()
+          val valueArray = map.valueArray()
+          var i = 0
+          while (i < map.numElements()) {
+            val key = keyConverter(keyArray, i)
+            if (valueArray.isNullAt(i)) {
+              result.put(key, null)
+            } else {
+              result.put(key, valueConverter(valueArray, i))
+            }
+            i += 1
+          }
         }
-        i += 1
       }
-      result
 
-    case udt: UserDefinedType[_] => newConverter(udt.sqlType)
+//      if (reuseObj) {
+//        val result = new OrcMap[WritableComparable[_], WritableComparable[_]](orcSchema)
+//        (getter, ordinal) => {
+//          result.clear()
+//          baseMapConverter.apply(result, getter, ordinal)
+//          result
+//        }
+//      } else {
+        (getter, ordinal) => {
+          val result = new OrcMap[WritableComparable[_], WritableComparable[_]](orcSchema)
+          baseMapConverter.apply(result, getter, ordinal)
+          result
+        }
+//      }
+
+    case udt: UserDefinedType[_] => newConverter(udt.sqlType, reuseObj)
 
     case _ =>
       throw new UnsupportedOperationException(s"$dataType is not supported yet.")
   }
-
-  /**
-   * Return a Orc value object for the given Spark schema.
-   */
-  private def createOrcValue(dataType: DataType) = {
-    OrcStruct.createValue(TypeDescription.fromString(OrcFileFormat.getQuotedSchemaString(dataType)))
-  }

Review comment:
       From the profiler, this was a prominent hotspot. There are two problems here.
   1) Converts dataType => String => orc type description for each data point. Instead can be done only once at the schema level with the new method schemaConverter
   2) OrcStruct.createValue (https://github.com/apache/orc/blob/master/java/mapreduce/src/java/org/apache/orc/mapred/OrcStruct.java) for a struct type would recursively go down the fields and create relevant objects until it hits either a map, array, union, or a primitive type. We weren't actually using these child objects here, so now with the change it will only create the one OrcStruct object that is needed.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan edited a comment on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan edited a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669459288


   Created https://github.com/apache/spark/pull/29366 for the changes referred to in https://github.com/apache/spark/pull/29353#discussion_r465955120
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r465466392



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -72,137 +74,191 @@ class OrcDeserializer(
   /**
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
-  private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
-    dataType match {
-      case NullType => (ordinal, _) =>
-        updater.setNullAt(ordinal)
-
-      case BooleanType => (ordinal, value) =>
-        updater.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
-
-      case ByteType => (ordinal, value) =>
-        updater.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
-
-      case ShortType => (ordinal, value) =>
-        updater.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
-
-      case IntegerType => (ordinal, value) =>
-        updater.setInt(ordinal, value.asInstanceOf[IntWritable].get)
-
-      case LongType => (ordinal, value) =>
-        updater.setLong(ordinal, value.asInstanceOf[LongWritable].get)
-
-      case FloatType => (ordinal, value) =>
-        updater.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
-
-      case DoubleType => (ordinal, value) =>
-        updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
-
-      case StringType => (ordinal, value) =>
-        updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
-
-      case BinaryType => (ordinal, value) =>
-        val binary = value.asInstanceOf[BytesWritable]
-        val bytes = new Array[Byte](binary.getLength)
-        System.arraycopy(binary.getBytes, 0, bytes, 0, binary.getLength)
-        updater.set(ordinal, bytes)
-
-      case DateType => (ordinal, value) =>
-        updater.setInt(ordinal, OrcShimUtils.getGregorianDays(value))
-
-      case TimestampType => (ordinal, value) =>
-        updater.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
-
-      case DecimalType.Fixed(precision, scale) => (ordinal, value) =>
-        val v = OrcShimUtils.getDecimal(value)
-        v.changePrecision(precision, scale)
-        updater.set(ordinal, v)
-
-      case st: StructType => (ordinal, value) =>
-        val result = new SpecificInternalRow(st)
-        val fieldUpdater = new RowUpdater(result)
-        val fieldConverters = st.map(_.dataType).map { dt =>
-          newWriter(dt, fieldUpdater)
-        }.toArray
-        val orcStruct = value.asInstanceOf[OrcStruct]
-
-        var i = 0
-        while (i < st.length) {
-          val value = orcStruct.getFieldValue(i)
-          if (value == null) {
-            result.setNullAt(i)
-          } else {
-            fieldConverters(i)(i, value)
+  private def newWriter(dataType: DataType, reuseObj: Boolean):
+  (CatalystDataUpdater, Int, WritableComparable[_]) => Unit = dataType match {
+    case NullType => (updater, ordinal, _) =>
+      updater.setNullAt(ordinal)
+
+    case BooleanType => (updater, ordinal, value) =>
+      updater.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
+
+    case ByteType => (updater, ordinal, value) =>
+      updater.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
+
+    case ShortType => (updater, ordinal, value) =>
+      updater.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
+
+    case IntegerType => (updater, ordinal, value) =>
+      updater.setInt(ordinal, value.asInstanceOf[IntWritable].get)
+
+    case LongType => (updater, ordinal, value) =>
+      updater.setLong(ordinal, value.asInstanceOf[LongWritable].get)
+
+    case FloatType => (updater, ordinal, value) =>
+      updater.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
+
+    case DoubleType => (updater, ordinal, value) =>
+      updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
+
+    case StringType => (updater, ordinal, value) =>
+      updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
+
+    case BinaryType => (updater, ordinal, value) =>
+      val binary = value.asInstanceOf[BytesWritable]
+      val bytes = new Array[Byte](binary.getLength)
+      System.arraycopy(binary.getBytes, 0, bytes, 0, binary.getLength)
+      updater.set(ordinal, bytes)
+
+    case DateType => (updater, ordinal, value) =>
+      updater.setInt(ordinal, OrcShimUtils.getGregorianDays(value))
+
+    case TimestampType => (updater, ordinal, value) =>
+      updater.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
+
+    case DecimalType.Fixed(precision, scale) => (updater, ordinal, value) =>
+      val v = OrcShimUtils.getDecimal(value)
+      v.changePrecision(precision, scale)
+      updater.set(ordinal, v)
+
+    case st: StructType =>
+      val createRow: () => InternalRow = getRowCreator(st)
+      val fieldConverters = st.map(_.dataType).map { dt => newWriter(dt, reuseObj)}.toArray
+      val baseConverter:
+          (InternalRow, RowUpdater, CatalystDataUpdater, Int, WritableComparable[_]) => Unit = {
+        (row, rowUpdater, updater, ordinal, value) => {
+          val orcStruct = value.asInstanceOf[OrcStruct]
+
+          var i = 0
+          while (i < st.length) {
+            val value = orcStruct.getFieldValue(i)
+            if (value == null) {
+              row.setNullAt(i)
+            } else {
+              fieldConverters(i)(rowUpdater, i, value)
+            }
+            i += 1
           }
-          i += 1
+
+          updater.set(ordinal, row)
         }
+      }
 
-        updater.set(ordinal, result)
+      if (reuseObj) {
+        val row = createRow.apply()
+        val rowUpdater = new RowUpdater(row)
+        (updater, ordinal, value) => baseConverter.apply(row, rowUpdater, updater, ordinal, value)
+      } else {
+        (updater, ordinal, value) => {
+          val row = createRow.apply()
+          val rowUpdater = new RowUpdater(row)
+          baseConverter.apply(row, rowUpdater, updater, ordinal, value)
+        }
+      }
 
-      case ArrayType(elementType, _) => (ordinal, value) =>
+    case ArrayType(elementType, _) =>
+      val elementConverter = newWriter(elementType, false)
+      val createArray = getArrayDataCreator(elementType)
+      (updater, ordinal, value) => {
         val orcArray = value.asInstanceOf[OrcList[WritableComparable[_]]]
         val length = orcArray.size()
-        val result = createArrayData(elementType, length)
+        val result = createArray(length)
         val elementUpdater = new ArrayDataUpdater(result)
-        val elementConverter = newWriter(elementType, elementUpdater)
 
         var i = 0
         while (i < length) {
           val value = orcArray.get(i)
           if (value == null) {
             result.setNullAt(i)
           } else {
-            elementConverter(i, value)
+            elementConverter(elementUpdater, i, value)
           }
           i += 1
         }
 
         updater.set(ordinal, result)
+      }
 
-      case MapType(keyType, valueType, _) => (ordinal, value) =>
+    case MapType(keyType, valueType, _) =>
+      val keyConverter = newWriter(keyType, false)
+      val createKeyArray = getArrayDataCreator(keyType)
+      val valueConverter = newWriter(valueType, false)
+      val createValueArray = getArrayDataCreator(valueType)
+      (updater, ordinal, value) => {
         val orcMap = value.asInstanceOf[OrcMap[WritableComparable[_], WritableComparable[_]]]
         val length = orcMap.size()
-        val keyArray = createArrayData(keyType, length)
+        val keyArray = createKeyArray(length)
         val keyUpdater = new ArrayDataUpdater(keyArray)
-        val keyConverter = newWriter(keyType, keyUpdater)
-        val valueArray = createArrayData(valueType, length)
+        val valueArray = createValueArray(length)
         val valueUpdater = new ArrayDataUpdater(valueArray)
-        val valueConverter = newWriter(valueType, valueUpdater)
 
         var i = 0
         val it = orcMap.entrySet().iterator()
         while (it.hasNext) {
           val entry = it.next()
-          keyConverter(i, entry.getKey)
+          keyConverter(keyUpdater, i, entry.getKey)
           val value = entry.getValue
           if (value == null) {
             valueArray.setNullAt(i)
           } else {
-            valueConverter(i, value)
+            valueConverter(valueUpdater, i, value)
           }
           i += 1
         }
 
         // The ORC map will never have null or duplicated map keys, it's safe to create a
         // ArrayBasedMapData directly here.
         updater.set(ordinal, new ArrayBasedMapData(keyArray, valueArray))
+      }
 
-      case udt: UserDefinedType[_] => newWriter(udt.sqlType, updater)
+    case udt: UserDefinedType[_] =>
+      val converter = newWriter(udt.sqlType, reuseObj)
+      (updater, ordinal, value) => {
+        converter(updater, ordinal, value)
+      }
+
+    case _ =>
+      throw new UnsupportedOperationException(s"$dataType is not supported yet.")
+  }
+
+  private def getArrayDataCreator(elementType: DataType): Int => ArrayData = elementType match {
+    case BooleanType => length => UnsafeArrayData.createFreshArray(length, 1)
+    case ByteType => length => UnsafeArrayData.createFreshArray(length, 1)
+    case ShortType => length => UnsafeArrayData.createFreshArray(length, 2)
+    case IntegerType => length => UnsafeArrayData.createFreshArray(length, 4)
+    case LongType => length => UnsafeArrayData.createFreshArray(length, 8)
+    case FloatType => length => UnsafeArrayData.createFreshArray(length, 4)
+    case DoubleType => length => UnsafeArrayData.createFreshArray(length, 8)
+    case _ => length => new GenericArrayData(new Array[Any](length))
+  }
 
-      case _ =>
-        throw new UnsupportedOperationException(s"$dataType is not supported yet.")
+  private def getRowCreator(st: StructType): () => InternalRow = {
+    val constructorsArray = new Array[() => MutableValue](st.fields.length)
+    var i = 0
+    while (i < st.fields.length) {
+      st.fields(i).dataType match {
+        case BooleanType => constructorsArray(i) = () => new MutableBoolean
+        case ByteType => constructorsArray(i) = () => new MutableByte
+        case ShortType => constructorsArray(i) = () => new MutableShort
+        // We use INT for DATE internally
+        case IntegerType | DateType => constructorsArray(i) = () => new MutableInt
+        // We use Long for Timestamp internally
+        case LongType | TimestampType => constructorsArray(i) = () => new MutableLong
+        case FloatType => constructorsArray(i) = () => new MutableFloat
+        case DoubleType => constructorsArray(i) = () => new MutableDouble
+        case _ => constructorsArray(i) = () => new MutableAny
+      }
+      i += 1
     }
 
-  private def createArrayData(elementType: DataType, length: Int): ArrayData = elementType match {

Review comment:
       Here as well. If we can get the length from the type as you did `getArrayDataCreator`, looks we can just set it here and pull the `createArrayData` logic out of the function so it creates once. Presumably, same as `MapType`.

##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -72,137 +74,191 @@ class OrcDeserializer(
   /**
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
-  private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
-    dataType match {
-      case NullType => (ordinal, _) =>
-        updater.setNullAt(ordinal)
-
-      case BooleanType => (ordinal, value) =>
-        updater.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
-
-      case ByteType => (ordinal, value) =>
-        updater.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
-
-      case ShortType => (ordinal, value) =>
-        updater.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
-
-      case IntegerType => (ordinal, value) =>
-        updater.setInt(ordinal, value.asInstanceOf[IntWritable].get)
-
-      case LongType => (ordinal, value) =>
-        updater.setLong(ordinal, value.asInstanceOf[LongWritable].get)
-
-      case FloatType => (ordinal, value) =>
-        updater.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
-
-      case DoubleType => (ordinal, value) =>
-        updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
-
-      case StringType => (ordinal, value) =>
-        updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
-
-      case BinaryType => (ordinal, value) =>
-        val binary = value.asInstanceOf[BytesWritable]
-        val bytes = new Array[Byte](binary.getLength)
-        System.arraycopy(binary.getBytes, 0, bytes, 0, binary.getLength)
-        updater.set(ordinal, bytes)
-
-      case DateType => (ordinal, value) =>
-        updater.setInt(ordinal, OrcShimUtils.getGregorianDays(value))
-
-      case TimestampType => (ordinal, value) =>
-        updater.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
-
-      case DecimalType.Fixed(precision, scale) => (ordinal, value) =>
-        val v = OrcShimUtils.getDecimal(value)
-        v.changePrecision(precision, scale)
-        updater.set(ordinal, v)
-
-      case st: StructType => (ordinal, value) =>
-        val result = new SpecificInternalRow(st)
-        val fieldUpdater = new RowUpdater(result)
-        val fieldConverters = st.map(_.dataType).map { dt =>
-          newWriter(dt, fieldUpdater)

Review comment:
       So the problem was that this creates the field writer for each record. Why don't we just simply pull this logic itself out of the function so it creates once?




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r465870192



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -73,136 +74,180 @@ class OrcDeserializer(
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
   private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
+      dataType: DataType): (CatalystDataUpdater, Int, WritableComparable[_]) => Unit =
     dataType match {
-      case NullType => (ordinal, _) =>
+      case NullType => (updater, ordinal, _) =>
         updater.setNullAt(ordinal)
 
-      case BooleanType => (ordinal, value) =>
+      case BooleanType => (updater, ordinal, value) =>
         updater.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
 
-      case ByteType => (ordinal, value) =>
+      case ByteType => (updater, ordinal, value) =>
         updater.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
 
-      case ShortType => (ordinal, value) =>
+      case ShortType => (updater, ordinal, value) =>
         updater.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
 
-      case IntegerType => (ordinal, value) =>
+      case IntegerType => (updater, ordinal, value) =>
         updater.setInt(ordinal, value.asInstanceOf[IntWritable].get)
 
-      case LongType => (ordinal, value) =>
+      case LongType => (updater, ordinal, value) =>
         updater.setLong(ordinal, value.asInstanceOf[LongWritable].get)
 
-      case FloatType => (ordinal, value) =>
+      case FloatType => (updater, ordinal, value) =>
         updater.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
 
-      case DoubleType => (ordinal, value) =>
+      case DoubleType => (updater, ordinal, value) =>
         updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
 
-      case StringType => (ordinal, value) =>
+      case StringType => (updater, ordinal, value) =>
         updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
 
-      case BinaryType => (ordinal, value) =>
+      case BinaryType => (updater, ordinal, value) =>
         val binary = value.asInstanceOf[BytesWritable]
         val bytes = new Array[Byte](binary.getLength)
         System.arraycopy(binary.getBytes, 0, bytes, 0, binary.getLength)
         updater.set(ordinal, bytes)
 
-      case DateType => (ordinal, value) =>
+      case DateType => (updater, ordinal, value) =>
         updater.setInt(ordinal, OrcShimUtils.getGregorianDays(value))
 
-      case TimestampType => (ordinal, value) =>
+      case TimestampType => (updater, ordinal, value) =>
         updater.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
 
-      case DecimalType.Fixed(precision, scale) => (ordinal, value) =>
+      case DecimalType.Fixed(precision, scale) => (updater, ordinal, value) =>
         val v = OrcShimUtils.getDecimal(value)
         v.changePrecision(precision, scale)
         updater.set(ordinal, v)
 
-      case st: StructType => (ordinal, value) =>
-        val result = new SpecificInternalRow(st)
-        val fieldUpdater = new RowUpdater(result)
+      case st: StructType =>
+        val createRow: () => InternalRow = getRowCreator(st)
         val fieldConverters = st.map(_.dataType).map { dt =>
-          newWriter(dt, fieldUpdater)
+          newWriter(dt)
         }.toArray
-        val orcStruct = value.asInstanceOf[OrcStruct]
-
-        var i = 0
-        while (i < st.length) {
-          val value = orcStruct.getFieldValue(i)
-          if (value == null) {
-            result.setNullAt(i)
-          } else {
-            fieldConverters(i)(i, value)
+        (updater, ordinal, value) => {
+          val result = createRow.apply()
+          val fieldUpdater = new RowUpdater(result)
+          val orcStruct = value.asInstanceOf[OrcStruct]
+
+          var i = 0
+          while (i < st.length) {
+            val value = orcStruct.getFieldValue(i)
+            if (value == null) {
+              result.setNullAt(i)
+            } else {
+              fieldConverters(i)(fieldUpdater, i, value)
+            }
+            i += 1
           }
-          i += 1
+
+          updater.set(ordinal, result)
         }
 
-        updater.set(ordinal, result)
-
-      case ArrayType(elementType, _) => (ordinal, value) =>
-        val orcArray = value.asInstanceOf[OrcList[WritableComparable[_]]]
-        val length = orcArray.size()
-        val result = createArrayData(elementType, length)
-        val elementUpdater = new ArrayDataUpdater(result)
-        val elementConverter = newWriter(elementType, elementUpdater)
-
-        var i = 0
-        while (i < length) {
-          val value = orcArray.get(i)
-          if (value == null) {
-            result.setNullAt(i)
-          } else {
-            elementConverter(i, value)
+      case ArrayType(elementType, _) =>
+        val elementConverter = newWriter(elementType)
+        val createArray = getArrayDataCreator(elementType)
+        (updater, ordinal, value) => {
+          val orcArray = value.asInstanceOf[OrcList[WritableComparable[_]]]
+          val length = orcArray.size()
+          val result = createArray(length)
+          val elementUpdater = new ArrayDataUpdater(result)
+
+          var i = 0
+          while (i < length) {
+            val value = orcArray.get(i)
+            if (value == null) {
+              result.setNullAt(i)
+            } else {
+              elementConverter(elementUpdater, i, value)
+            }
+            i += 1
           }
-          i += 1
+
+          updater.set(ordinal, result)
         }
 
-        updater.set(ordinal, result)
-
-      case MapType(keyType, valueType, _) => (ordinal, value) =>
-        val orcMap = value.asInstanceOf[OrcMap[WritableComparable[_], WritableComparable[_]]]
-        val length = orcMap.size()
-        val keyArray = createArrayData(keyType, length)
-        val keyUpdater = new ArrayDataUpdater(keyArray)
-        val keyConverter = newWriter(keyType, keyUpdater)
-        val valueArray = createArrayData(valueType, length)
-        val valueUpdater = new ArrayDataUpdater(valueArray)
-        val valueConverter = newWriter(valueType, valueUpdater)
-
-        var i = 0
-        val it = orcMap.entrySet().iterator()
-        while (it.hasNext) {
-          val entry = it.next()
-          keyConverter(i, entry.getKey)
-          val value = entry.getValue
-          if (value == null) {
-            valueArray.setNullAt(i)
-          } else {
-            valueConverter(i, value)
+      case MapType(keyType, valueType, _) =>
+        val keyConverter = newWriter(keyType)
+        val createKeyArray = getArrayDataCreator(keyType)
+        val valueConverter = newWriter(valueType)
+        val createValueArray = getArrayDataCreator(valueType)
+        (updater, ordinal, value) => {
+          val orcMap = value.asInstanceOf[OrcMap[WritableComparable[_], WritableComparable[_]]]
+          val length = orcMap.size()
+          val keyArray = createKeyArray(length)
+          val keyUpdater = new ArrayDataUpdater(keyArray)
+          val valueArray = createValueArray(length)
+          val valueUpdater = new ArrayDataUpdater(valueArray)
+
+          var i = 0
+          val it = orcMap.entrySet().iterator()
+          while (it.hasNext) {
+            val entry = it.next()
+            keyConverter(keyUpdater, i, entry.getKey)
+            val value = entry.getValue
+            if (value == null) {
+              valueArray.setNullAt(i)
+            } else {
+              valueConverter(valueUpdater, i, value)
+            }
+            i += 1
           }
-          i += 1
-        }
 
-        // The ORC map will never have null or duplicated map keys, it's safe to create a
-        // ArrayBasedMapData directly here.
-        updater.set(ordinal, new ArrayBasedMapData(keyArray, valueArray))
+          // The ORC map will never have null or duplicated map keys, it's safe to create a
+          // ArrayBasedMapData directly here.
+          updater.set(ordinal, new ArrayBasedMapData(keyArray, valueArray))
+        }
 
-      case udt: UserDefinedType[_] => newWriter(udt.sqlType, updater)
+      case udt: UserDefinedType[_] =>
+        val converter = newWriter(udt.sqlType)
+        (updater, ordinal, value) => {
+          converter(updater, ordinal, value)
+        }
 
       case _ =>
         throw new UnsupportedOperationException(s"$dataType is not supported yet.")
     }
 
-  private def createArrayData(elementType: DataType, length: Int): ArrayData = elementType match {
-    case BooleanType => UnsafeArrayData.fromPrimitiveArray(new Array[Boolean](length))
-    case ByteType => UnsafeArrayData.fromPrimitiveArray(new Array[Byte](length))
-    case ShortType => UnsafeArrayData.fromPrimitiveArray(new Array[Short](length))
-    case IntegerType => UnsafeArrayData.fromPrimitiveArray(new Array[Int](length))
-    case LongType => UnsafeArrayData.fromPrimitiveArray(new Array[Long](length))
-    case FloatType => UnsafeArrayData.fromPrimitiveArray(new Array[Float](length))
-    case DoubleType => UnsafeArrayData.fromPrimitiveArray(new Array[Double](length))
-    case _ => new GenericArrayData(new Array[Any](length))
+  private def getArrayDataCreator(elementType: DataType): Int => ArrayData = elementType match {
+    case BooleanType => length => UnsafeArrayData.createFreshArray(length, 1)
+    case ByteType => length => UnsafeArrayData.createFreshArray(length, 1)
+    case ShortType => length => UnsafeArrayData.createFreshArray(length, 2)
+    case IntegerType => length => UnsafeArrayData.createFreshArray(length, 4)
+    case LongType => length => UnsafeArrayData.createFreshArray(length, 8)
+    case FloatType => length => UnsafeArrayData.createFreshArray(length, 4)
+    case DoubleType => length => UnsafeArrayData.createFreshArray(length, 8)
+    case _ => length => new GenericArrayData(new Array[Any](length))
+  }
+
+  private def getRowCreator(st: StructType): () => InternalRow = {
+    val constructorsArray = new Array[() => MutableValue](st.fields.length)
+    var i = 0
+    while (i < st.fields.length) {
+      st.fields(i).dataType match {
+        case BooleanType => constructorsArray(i) = () => new MutableBoolean
+        case ByteType => constructorsArray(i) = () => new MutableByte
+        case ShortType => constructorsArray(i) = () => new MutableShort
+        // We use INT for DATE internally
+        case IntegerType | DateType => constructorsArray(i) = () => new MutableInt
+        // We use Long for Timestamp internally
+        case LongType | TimestampType => constructorsArray(i) = () => new MutableLong
+        case FloatType => constructorsArray(i) = () => new MutableFloat
+        case DoubleType => constructorsArray(i) = () => new MutableDouble
+        case _ => constructorsArray(i) = () => new MutableAny
+      }
+      i += 1
+    }
+
+    () => {
+      val array = new Array[MutableValue](constructorsArray.length)
+      var i = 0
+      while (i < constructorsArray.length) {
+        array(i) = constructorsArray(i).apply()
+        i += 1
+      }
+      new SpecificInternalRow(array)
+    }

Review comment:
       While working with avro (https://github.com/apache/spark/pull/29354) the profiler showed some time being spent in SpecificInternalRow constructor, and we saw improvements when moving to this model (please see the pdf in the avro PR) where based on the schema we can fill in a constructors array and for each data point, call these constructors one by one.
   
   In retrospect, I did not check if the cost overhead is because of two Scala map calls in SpecificInternalRow constructors or because of a lot of case/match calls. I will investigate a bit.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r465460837



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -72,137 +74,191 @@ class OrcDeserializer(
   /**
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
-  private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
-    dataType match {
-      case NullType => (ordinal, _) =>
-        updater.setNullAt(ordinal)
-
-      case BooleanType => (ordinal, value) =>
-        updater.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
-
-      case ByteType => (ordinal, value) =>
-        updater.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
-
-      case ShortType => (ordinal, value) =>
-        updater.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
-
-      case IntegerType => (ordinal, value) =>
-        updater.setInt(ordinal, value.asInstanceOf[IntWritable].get)
-
-      case LongType => (ordinal, value) =>
-        updater.setLong(ordinal, value.asInstanceOf[LongWritable].get)
-
-      case FloatType => (ordinal, value) =>
-        updater.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
-
-      case DoubleType => (ordinal, value) =>
-        updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
-
-      case StringType => (ordinal, value) =>
-        updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
-
-      case BinaryType => (ordinal, value) =>
-        val binary = value.asInstanceOf[BytesWritable]
-        val bytes = new Array[Byte](binary.getLength)
-        System.arraycopy(binary.getBytes, 0, bytes, 0, binary.getLength)
-        updater.set(ordinal, bytes)
-
-      case DateType => (ordinal, value) =>
-        updater.setInt(ordinal, OrcShimUtils.getGregorianDays(value))
-
-      case TimestampType => (ordinal, value) =>
-        updater.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
-
-      case DecimalType.Fixed(precision, scale) => (ordinal, value) =>
-        val v = OrcShimUtils.getDecimal(value)
-        v.changePrecision(precision, scale)
-        updater.set(ordinal, v)
-
-      case st: StructType => (ordinal, value) =>
-        val result = new SpecificInternalRow(st)
-        val fieldUpdater = new RowUpdater(result)
-        val fieldConverters = st.map(_.dataType).map { dt =>
-          newWriter(dt, fieldUpdater)
-        }.toArray
-        val orcStruct = value.asInstanceOf[OrcStruct]
-
-        var i = 0
-        while (i < st.length) {
-          val value = orcStruct.getFieldValue(i)
-          if (value == null) {
-            result.setNullAt(i)
-          } else {
-            fieldConverters(i)(i, value)
+  private def newWriter(dataType: DataType, reuseObj: Boolean):
+  (CatalystDataUpdater, Int, WritableComparable[_]) => Unit = dataType match {

Review comment:
       I would keep this style as:
   
   ```scala
   private def newWriter(
       dataType: DataType, reuseObj: Boolean)
   : (CatalystDataUpdater, Int, WritableComparable[_]) => Unit =
     dataType match {
         case NullType => (updater, ordinal, _) =>
         ...
     }
   ```
   
   to reduce the diff.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-683229295






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] dongjoon-hyun commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r466835600



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcSerializer.scala
##########
@@ -27,12 +27,19 @@ import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.types._
 
 /**
- * A serializer to serialize Spark rows to ORC structs.
+ * A serializer to serialize Spark rows to ORC structs. Multiple calls return the same [[OrcStruct]]
+ * object. If needed, caller should copy before making a new call.
  */
 class OrcSerializer(dataSchema: StructType) {
 
-  private val result = createOrcValue(dataSchema).asInstanceOf[OrcStruct]
-  private val converters = dataSchema.map(_.dataType).map(newConverter(_)).toArray
+  // TODO: repeating steps here. Can pass as parameter to newConverter, but only done once at the

Review comment:
       When you use TODO, could you file a JIRA ID and make this as an IDed TODO like `TODO(SPARK-XXXXX)`, please?




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-683181113


   **[Test build #128005 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128005/testReport)** for PR 29353 at commit [`d67ceed`](https://github.com/apache/spark/commit/d67ceed965fce5f56f2096032188c6fa9b3cfa5b).


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins removed a comment on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669609253






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r465963114



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -72,137 +74,191 @@ class OrcDeserializer(
   /**
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
-  private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
-    dataType match {
-      case NullType => (ordinal, _) =>
-        updater.setNullAt(ordinal)
-
-      case BooleanType => (ordinal, value) =>
-        updater.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
-
-      case ByteType => (ordinal, value) =>
-        updater.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
-
-      case ShortType => (ordinal, value) =>
-        updater.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
-
-      case IntegerType => (ordinal, value) =>
-        updater.setInt(ordinal, value.asInstanceOf[IntWritable].get)
-
-      case LongType => (ordinal, value) =>
-        updater.setLong(ordinal, value.asInstanceOf[LongWritable].get)
-
-      case FloatType => (ordinal, value) =>
-        updater.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
-
-      case DoubleType => (ordinal, value) =>
-        updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
-
-      case StringType => (ordinal, value) =>
-        updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
-
-      case BinaryType => (ordinal, value) =>
-        val binary = value.asInstanceOf[BytesWritable]
-        val bytes = new Array[Byte](binary.getLength)
-        System.arraycopy(binary.getBytes, 0, bytes, 0, binary.getLength)
-        updater.set(ordinal, bytes)
-
-      case DateType => (ordinal, value) =>
-        updater.setInt(ordinal, OrcShimUtils.getGregorianDays(value))
-
-      case TimestampType => (ordinal, value) =>
-        updater.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
-
-      case DecimalType.Fixed(precision, scale) => (ordinal, value) =>
-        val v = OrcShimUtils.getDecimal(value)
-        v.changePrecision(precision, scale)
-        updater.set(ordinal, v)
-
-      case st: StructType => (ordinal, value) =>
-        val result = new SpecificInternalRow(st)
-        val fieldUpdater = new RowUpdater(result)
-        val fieldConverters = st.map(_.dataType).map { dt =>
-          newWriter(dt, fieldUpdater)
-        }.toArray
-        val orcStruct = value.asInstanceOf[OrcStruct]
-
-        var i = 0
-        while (i < st.length) {
-          val value = orcStruct.getFieldValue(i)
-          if (value == null) {
-            result.setNullAt(i)
-          } else {
-            fieldConverters(i)(i, value)
+  private def newWriter(dataType: DataType, reuseObj: Boolean):
+  (CatalystDataUpdater, Int, WritableComparable[_]) => Unit = dataType match {

Review comment:
       Fixed.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r466144939



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -73,135 +75,157 @@ class OrcDeserializer(
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
   private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
+      dataType: DataType, reuseObj: Boolean)
+  : (CatalystDataUpdater, Int, WritableComparable[_]) => Unit =
     dataType match {
-      case NullType => (ordinal, _) =>
+      case NullType => (updater, ordinal, _) =>

Review comment:
       Only the top level field writers are currently being created once. For each level below, they are created for each data point. Eg, the case for struct did the following:
   
   `case st: StructType => (ordinal, value) =>`
   `...`
   `    val fieldConverters = st.map(_.dataType).map { dt => newWriter(dt, fieldUpdater) }.toArray`
   `...`
   
   i.e., it returns a function, which when called will create relevant fieldConverters. Instead, the field converters should be created outside the function so that they are created only once.
   
   Now, the updaters can not always be reused. Eg, under an array or a map (i.e., whenever reuseObj turns to false).




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins removed a comment on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-683181444






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins removed a comment on pull request #29353: [Spark-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668914789






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA removed a comment on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669460508


   **[Test build #127103 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127103/testReport)** for PR 29353 at commit [`182d6dc`](https://github.com/apache/spark/commit/182d6dc872c057f96b64ecfcfc6ecc65f5a00802).


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] github-actions[bot] closed pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
github-actions[bot] closed pull request #29353:
URL: https://github.com/apache/spark/pull/29353


   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r465964580



##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/SpecificInternalRow.scala
##########
@@ -192,24 +192,41 @@ final class MutableAny extends MutableValue {
  */
 final class SpecificInternalRow(val values: Array[MutableValue]) extends BaseGenericInternalRow {
 
-  def this(dataTypes: Seq[DataType]) =
-    this(
-      dataTypes.map {
-        case BooleanType => new MutableBoolean
-        case ByteType => new MutableByte
-        case ShortType => new MutableShort
-        // We use INT for DATE internally
-        case IntegerType | DateType => new MutableInt
-        // We use Long for Timestamp internally
-        case LongType | TimestampType => new MutableLong
-        case FloatType => new MutableFloat
-        case DoubleType => new MutableDouble
-        case _ => new MutableAny
-      }.toArray)
+  private[this] def constructorHelper(dataType: DataType): MutableValue = dataType match {
+    case BooleanType => new MutableBoolean
+    case ByteType => new MutableByte
+    case ShortType => new MutableShort
+    // We use INT for DATE internally
+    case IntegerType | DateType => new MutableInt
+    // We use Long for Timestamp internally
+    case LongType | TimestampType => new MutableLong
+    case FloatType => new MutableFloat
+    case DoubleType => new MutableDouble
+    case _ => new MutableAny
+  }
+
+  def this(dataTypes: Seq[DataType]) = {
+    this(new Array[MutableValue](dataTypes.length))
+    val length = values.length
+    var i = 0
+    while (i < length) {
+      values(i) = constructorHelper(dataTypes(i))
+      i += 1
+    }
+  }
 
   def this() = this(Seq.empty)
 
-  def this(schema: StructType) = this(schema.fields.map(_.dataType))
+  def this(schema: StructType) = {
+    this(new Array[MutableValue](schema.fields.length))
+    val length = values.length
+    val fields = schema.fields
+    var i = 0
+    while (i < length) {
+      values(i) = constructorHelper(fields(i).dataType)
+      i += 1
+    }
+  }
 
   override def numFields: Int = values.length
 

Review comment:
       Accidentally attached, will remove.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins removed a comment on pull request #29353: [Spark-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668823274


   Can one of the admins verify this patch?


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r465863127



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -72,137 +74,191 @@ class OrcDeserializer(
   /**
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
-  private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
-    dataType match {
-      case NullType => (ordinal, _) =>
-        updater.setNullAt(ordinal)
-
-      case BooleanType => (ordinal, value) =>
-        updater.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
-
-      case ByteType => (ordinal, value) =>
-        updater.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
-
-      case ShortType => (ordinal, value) =>
-        updater.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
-
-      case IntegerType => (ordinal, value) =>
-        updater.setInt(ordinal, value.asInstanceOf[IntWritable].get)
-
-      case LongType => (ordinal, value) =>
-        updater.setLong(ordinal, value.asInstanceOf[LongWritable].get)
-
-      case FloatType => (ordinal, value) =>
-        updater.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
-
-      case DoubleType => (ordinal, value) =>
-        updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
-
-      case StringType => (ordinal, value) =>
-        updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
-
-      case BinaryType => (ordinal, value) =>
-        val binary = value.asInstanceOf[BytesWritable]
-        val bytes = new Array[Byte](binary.getLength)
-        System.arraycopy(binary.getBytes, 0, bytes, 0, binary.getLength)
-        updater.set(ordinal, bytes)
-
-      case DateType => (ordinal, value) =>
-        updater.setInt(ordinal, OrcShimUtils.getGregorianDays(value))
-
-      case TimestampType => (ordinal, value) =>
-        updater.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
-
-      case DecimalType.Fixed(precision, scale) => (ordinal, value) =>
-        val v = OrcShimUtils.getDecimal(value)
-        v.changePrecision(precision, scale)
-        updater.set(ordinal, v)
-
-      case st: StructType => (ordinal, value) =>
-        val result = new SpecificInternalRow(st)
-        val fieldUpdater = new RowUpdater(result)
-        val fieldConverters = st.map(_.dataType).map { dt =>
-          newWriter(dt, fieldUpdater)

Review comment:
       Yes, so previously it would only create the top level converter at the initializing stage and the rest would be created via a newWriter call for each data point. Instead, converters at all levels of the schema can be created at the initialization stage because they only depend on the schema.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669608678


   **[Test build #127104 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127104/testReport)** for PR 29353 at commit [`b360fb6`](https://github.com/apache/spark/commit/b360fb66f1e3b45561af7ec9d5dea352545e7f33).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669613513


   **[Test build #127103 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127103/testReport)** for PR 29353 at commit [`182d6dc`](https://github.com/apache/spark/commit/182d6dc872c057f96b64ecfcfc6ecc65f5a00802).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins commented on pull request #29353: [Spark-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668823274


   Can one of the admins verify this patch?


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669460508


   **[Test build #127103 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127103/testReport)** for PR 29353 at commit [`182d6dc`](https://github.com/apache/spark/commit/182d6dc872c057f96b64ecfcfc6ecc65f5a00802).


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r465460837



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -72,137 +74,191 @@ class OrcDeserializer(
   /**
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
-  private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
-    dataType match {
-      case NullType => (ordinal, _) =>
-        updater.setNullAt(ordinal)
-
-      case BooleanType => (ordinal, value) =>
-        updater.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
-
-      case ByteType => (ordinal, value) =>
-        updater.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
-
-      case ShortType => (ordinal, value) =>
-        updater.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
-
-      case IntegerType => (ordinal, value) =>
-        updater.setInt(ordinal, value.asInstanceOf[IntWritable].get)
-
-      case LongType => (ordinal, value) =>
-        updater.setLong(ordinal, value.asInstanceOf[LongWritable].get)
-
-      case FloatType => (ordinal, value) =>
-        updater.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
-
-      case DoubleType => (ordinal, value) =>
-        updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
-
-      case StringType => (ordinal, value) =>
-        updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
-
-      case BinaryType => (ordinal, value) =>
-        val binary = value.asInstanceOf[BytesWritable]
-        val bytes = new Array[Byte](binary.getLength)
-        System.arraycopy(binary.getBytes, 0, bytes, 0, binary.getLength)
-        updater.set(ordinal, bytes)
-
-      case DateType => (ordinal, value) =>
-        updater.setInt(ordinal, OrcShimUtils.getGregorianDays(value))
-
-      case TimestampType => (ordinal, value) =>
-        updater.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
-
-      case DecimalType.Fixed(precision, scale) => (ordinal, value) =>
-        val v = OrcShimUtils.getDecimal(value)
-        v.changePrecision(precision, scale)
-        updater.set(ordinal, v)
-
-      case st: StructType => (ordinal, value) =>
-        val result = new SpecificInternalRow(st)
-        val fieldUpdater = new RowUpdater(result)
-        val fieldConverters = st.map(_.dataType).map { dt =>
-          newWriter(dt, fieldUpdater)
-        }.toArray
-        val orcStruct = value.asInstanceOf[OrcStruct]
-
-        var i = 0
-        while (i < st.length) {
-          val value = orcStruct.getFieldValue(i)
-          if (value == null) {
-            result.setNullAt(i)
-          } else {
-            fieldConverters(i)(i, value)
+  private def newWriter(dataType: DataType, reuseObj: Boolean):
+  (CatalystDataUpdater, Int, WritableComparable[_]) => Unit = dataType match {

Review comment:
       I would keep this style as:
   
   ```scala
   private def newWriter(
       dataType: DataType, reuseObj: Boolean)
   : (CatalystDataUpdater, Int, WritableComparable[_]) => Unit =
     dataType match {
       case NullType => (updater, ordinal, _) =>
         ...
     }
   ```
   
   to reduce the diff.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r465860164



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcSerializer.scala
##########
@@ -150,69 +156,110 @@ class OrcSerializer(dataSchema: StructType) {
     case DecimalType.Fixed(precision, scale) =>
       OrcShimUtils.getHiveDecimalWritable(precision, scale)
 
-    case st: StructType => (getter, ordinal) =>
-      val result = createOrcValue(st).asInstanceOf[OrcStruct]
-      val fieldConverters = st.map(_.dataType).map(newConverter(_))
-      val numFields = st.length
-      val struct = getter.getStruct(ordinal, numFields)
-      var i = 0
-      while (i < numFields) {
-        if (struct.isNullAt(i)) {
-          result.setFieldValue(i, null)
-        } else {
-          result.setFieldValue(i, fieldConverters(i)(struct, i))
+    case st: StructType =>
+      val orcSchema = schemaConverter(st)
+      val fieldConverters = st.map(_.dataType).map(newConverter(_, reuseObj))
+      val baseConverter: (OrcStruct, SpecializedGetters, Int) => OrcStruct =
+        (result, getter, ordinal) => {
+          val numFields = st.length
+          val struct = getter.getStruct(ordinal, numFields)
+          var i = 0
+          while (i < numFields) {
+            if (struct.isNullAt(i)) {
+              result.setFieldValue(i, null)
+            } else {
+              result.setFieldValue(i, fieldConverters(i)(struct, i))
+            }
+            i += 1
+          }
+          result
         }
-        i += 1
-      }
-      result
 
-    case ArrayType(elementType, _) => (getter, ordinal) =>
-      val result = createOrcValue(dataType).asInstanceOf[OrcList[WritableComparable[_]]]
+//      if (reuseObj) {
+//        val result = new OrcStruct(orcSchema)
+//        (getter, ordinal) => baseConverter.apply(result, getter, ordinal)
+//      } else {
+        (getter, ordinal) =>
+          val result = new OrcStruct(orcSchema)
+          baseConverter.apply(result, getter, ordinal)
+//      }
+
+    case ArrayType(elementType, _) =>
+      val orcSchema = schemaConverter(dataType)
       // Need to put all converted values to a list, can't reuse object.
-      val elementConverter = newConverter(elementType, reuseObj = false)
-      val array = getter.getArray(ordinal)
-      var i = 0
-      while (i < array.numElements()) {
-        if (array.isNullAt(i)) {
-          result.add(null)
-        } else {
-          result.add(elementConverter(array, i))
+      val elementConverter = newConverter(elementType, false)
+      val baseArrayConverter: (OrcList[WritableComparable[_]], SpecializedGetters, Int) => Unit = {
+        (result, getter, ordinal) => {
+          val array = getter.getArray(ordinal)
+          var i = 0
+          while (i < array.numElements()) {
+            if (array.isNullAt(i)) {
+              result.add(null)
+            } else {
+              result.add(elementConverter(array, i))
+            }
+            i += 1
+          }
         }
-        i += 1
       }
-      result
 
-    case MapType(keyType, valueType, _) => (getter, ordinal) =>
-      val result = createOrcValue(dataType)
-        .asInstanceOf[OrcMap[WritableComparable[_], WritableComparable[_]]]
+//      if (reuseObj) {
+//        val result = new OrcList[WritableComparable[_]](orcSchema)
+//        (getter, ordinal) => {
+//          result.clear()
+//          baseArrayConverter.apply(result, getter, ordinal)
+//          result
+//        }
+//      } else {
+        (getter, ordinal) => {
+          val result = new OrcList[WritableComparable[_]](orcSchema)
+          baseArrayConverter.apply(result, getter, ordinal)
+          result
+        }
+//      }
+
+    case MapType(keyType, valueType, _) =>
+      val orcSchema = schemaConverter(dataType)
       // Need to put all converted values to a list, can't reuse object.
-      val keyConverter = newConverter(keyType, reuseObj = false)
-      val valueConverter = newConverter(valueType, reuseObj = false)
-      val map = getter.getMap(ordinal)
-      val keyArray = map.keyArray()
-      val valueArray = map.valueArray()
-      var i = 0
-      while (i < map.numElements()) {
-        val key = keyConverter(keyArray, i)
-        if (valueArray.isNullAt(i)) {
-          result.put(key, null)
-        } else {
-          result.put(key, valueConverter(valueArray, i))
+      val keyConverter = newConverter(keyType, false)
+      val valueConverter = newConverter(valueType, false)
+      val baseMapConverter:
+        (OrcMap[WritableComparable[_], WritableComparable[_]], SpecializedGetters, Int) => Unit = {
+        (result, getter, ordinal) => {
+          val map = getter.getMap(ordinal)
+          val keyArray = map.keyArray()
+          val valueArray = map.valueArray()
+          var i = 0
+          while (i < map.numElements()) {
+            val key = keyConverter(keyArray, i)
+            if (valueArray.isNullAt(i)) {
+              result.put(key, null)
+            } else {
+              result.put(key, valueConverter(valueArray, i))
+            }
+            i += 1
+          }
         }
-        i += 1
       }
-      result
 
-    case udt: UserDefinedType[_] => newConverter(udt.sqlType)
+//      if (reuseObj) {
+//        val result = new OrcMap[WritableComparable[_], WritableComparable[_]](orcSchema)
+//        (getter, ordinal) => {
+//          result.clear()
+//          baseMapConverter.apply(result, getter, ordinal)
+//          result
+//        }
+//      } else {
+        (getter, ordinal) => {
+          val result = new OrcMap[WritableComparable[_], WritableComparable[_]](orcSchema)
+          baseMapConverter.apply(result, getter, ordinal)
+          result
+        }
+//      }
+
+    case udt: UserDefinedType[_] => newConverter(udt.sqlType, reuseObj)
 
     case _ =>
       throw new UnsupportedOperationException(s"$dataType is not supported yet.")
   }
-
-  /**
-   * Return a Orc value object for the given Spark schema.
-   */
-  private def createOrcValue(dataType: DataType) = {
-    OrcStruct.createValue(TypeDescription.fromString(OrcFileFormat.getQuotedSchemaString(dataType)))
-  }

Review comment:
       From the profiler, this was a prominent hotspot. There are two problems here.
   1) Converts dataType => String => orc type description for each data point. Instead can be done only once at the schema level (i.e. at initialization stage) with the new method schemaConverter
   2) OrcStruct.createValue (https://github.com/apache/orc/blob/master/java/mapreduce/src/java/org/apache/orc/mapred/OrcStruct.java) for a struct type would recursively go down the fields and create relevant objects until it hits either a map, array, union, or a primitive type. We weren't actually using these child objects here, so now with the change it will only create the one OrcStruct object that is needed.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA removed a comment on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668914387


   **[Test build #127071 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127071/testReport)** for PR 29353 at commit [`13af454`](https://github.com/apache/spark/commit/13af454cf4bb53a1c6b2c5f6aef19b6b92420e26).


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669614039






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669465889


   **[Test build #127104 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127104/testReport)** for PR 29353 at commit [`b360fb6`](https://github.com/apache/spark/commit/b360fb66f1e3b45561af7ec9d5dea352545e7f33).


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r465963022



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -72,137 +74,191 @@ class OrcDeserializer(
   /**
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
-  private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
-    dataType match {
-      case NullType => (ordinal, _) =>
-        updater.setNullAt(ordinal)
-
-      case BooleanType => (ordinal, value) =>
-        updater.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
-
-      case ByteType => (ordinal, value) =>
-        updater.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
-
-      case ShortType => (ordinal, value) =>
-        updater.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
-
-      case IntegerType => (ordinal, value) =>
-        updater.setInt(ordinal, value.asInstanceOf[IntWritable].get)
-
-      case LongType => (ordinal, value) =>
-        updater.setLong(ordinal, value.asInstanceOf[LongWritable].get)
-
-      case FloatType => (ordinal, value) =>
-        updater.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
-
-      case DoubleType => (ordinal, value) =>
-        updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
-
-      case StringType => (ordinal, value) =>
-        updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
-
-      case BinaryType => (ordinal, value) =>
-        val binary = value.asInstanceOf[BytesWritable]
-        val bytes = new Array[Byte](binary.getLength)
-        System.arraycopy(binary.getBytes, 0, bytes, 0, binary.getLength)
-        updater.set(ordinal, bytes)
-
-      case DateType => (ordinal, value) =>
-        updater.setInt(ordinal, OrcShimUtils.getGregorianDays(value))
-
-      case TimestampType => (ordinal, value) =>
-        updater.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
-
-      case DecimalType.Fixed(precision, scale) => (ordinal, value) =>
-        val v = OrcShimUtils.getDecimal(value)
-        v.changePrecision(precision, scale)
-        updater.set(ordinal, v)
-
-      case st: StructType => (ordinal, value) =>
-        val result = new SpecificInternalRow(st)
-        val fieldUpdater = new RowUpdater(result)
-        val fieldConverters = st.map(_.dataType).map { dt =>
-          newWriter(dt, fieldUpdater)
-        }.toArray
-        val orcStruct = value.asInstanceOf[OrcStruct]
-
-        var i = 0
-        while (i < st.length) {
-          val value = orcStruct.getFieldValue(i)
-          if (value == null) {
-            result.setNullAt(i)
-          } else {
-            fieldConverters(i)(i, value)
+  private def newWriter(dataType: DataType, reuseObj: Boolean):
+  (CatalystDataUpdater, Int, WritableComparable[_]) => Unit = dataType match {
+    case NullType => (updater, ordinal, _) =>
+      updater.setNullAt(ordinal)
+
+    case BooleanType => (updater, ordinal, value) =>
+      updater.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
+
+    case ByteType => (updater, ordinal, value) =>
+      updater.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
+
+    case ShortType => (updater, ordinal, value) =>
+      updater.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
+
+    case IntegerType => (updater, ordinal, value) =>
+      updater.setInt(ordinal, value.asInstanceOf[IntWritable].get)
+
+    case LongType => (updater, ordinal, value) =>
+      updater.setLong(ordinal, value.asInstanceOf[LongWritable].get)
+
+    case FloatType => (updater, ordinal, value) =>
+      updater.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
+
+    case DoubleType => (updater, ordinal, value) =>
+      updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
+
+    case StringType => (updater, ordinal, value) =>
+      updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
+
+    case BinaryType => (updater, ordinal, value) =>
+      val binary = value.asInstanceOf[BytesWritable]
+      val bytes = new Array[Byte](binary.getLength)
+      System.arraycopy(binary.getBytes, 0, bytes, 0, binary.getLength)
+      updater.set(ordinal, bytes)
+
+    case DateType => (updater, ordinal, value) =>
+      updater.setInt(ordinal, OrcShimUtils.getGregorianDays(value))
+
+    case TimestampType => (updater, ordinal, value) =>
+      updater.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
+
+    case DecimalType.Fixed(precision, scale) => (updater, ordinal, value) =>
+      val v = OrcShimUtils.getDecimal(value)
+      v.changePrecision(precision, scale)
+      updater.set(ordinal, v)
+
+    case st: StructType =>
+      val createRow: () => InternalRow = getRowCreator(st)
+      val fieldConverters = st.map(_.dataType).map { dt => newWriter(dt, reuseObj)}.toArray
+      val baseConverter:
+          (InternalRow, RowUpdater, CatalystDataUpdater, Int, WritableComparable[_]) => Unit = {
+        (row, rowUpdater, updater, ordinal, value) => {
+          val orcStruct = value.asInstanceOf[OrcStruct]
+
+          var i = 0
+          while (i < st.length) {
+            val value = orcStruct.getFieldValue(i)
+            if (value == null) {
+              row.setNullAt(i)
+            } else {
+              fieldConverters(i)(rowUpdater, i, value)
+            }
+            i += 1
           }
-          i += 1
+
+          updater.set(ordinal, row)
         }
+      }
 
-        updater.set(ordinal, result)
+      if (reuseObj) {
+        val row = createRow.apply()
+        val rowUpdater = new RowUpdater(row)
+        (updater, ordinal, value) => baseConverter.apply(row, rowUpdater, updater, ordinal, value)
+      } else {
+        (updater, ordinal, value) => {
+          val row = createRow.apply()
+          val rowUpdater = new RowUpdater(row)
+          baseConverter.apply(row, rowUpdater, updater, ordinal, value)
+        }
+      }
 
-      case ArrayType(elementType, _) => (ordinal, value) =>
+    case ArrayType(elementType, _) =>
+      val elementConverter = newWriter(elementType, false)
+      val createArray = getArrayDataCreator(elementType)
+      (updater, ordinal, value) => {
         val orcArray = value.asInstanceOf[OrcList[WritableComparable[_]]]
         val length = orcArray.size()
-        val result = createArrayData(elementType, length)
+        val result = createArray(length)
         val elementUpdater = new ArrayDataUpdater(result)
-        val elementConverter = newWriter(elementType, elementUpdater)
 
         var i = 0
         while (i < length) {
           val value = orcArray.get(i)
           if (value == null) {
             result.setNullAt(i)
           } else {
-            elementConverter(i, value)
+            elementConverter(elementUpdater, i, value)
           }
           i += 1
         }
 
         updater.set(ordinal, result)
+      }
 
-      case MapType(keyType, valueType, _) => (ordinal, value) =>
+    case MapType(keyType, valueType, _) =>
+      val keyConverter = newWriter(keyType, false)
+      val createKeyArray = getArrayDataCreator(keyType)
+      val valueConverter = newWriter(valueType, false)
+      val createValueArray = getArrayDataCreator(valueType)
+      (updater, ordinal, value) => {
         val orcMap = value.asInstanceOf[OrcMap[WritableComparable[_], WritableComparable[_]]]
         val length = orcMap.size()
-        val keyArray = createArrayData(keyType, length)
+        val keyArray = createKeyArray(length)
         val keyUpdater = new ArrayDataUpdater(keyArray)
-        val keyConverter = newWriter(keyType, keyUpdater)
-        val valueArray = createArrayData(valueType, length)
+        val valueArray = createValueArray(length)
         val valueUpdater = new ArrayDataUpdater(valueArray)
-        val valueConverter = newWriter(valueType, valueUpdater)
 
         var i = 0
         val it = orcMap.entrySet().iterator()
         while (it.hasNext) {
           val entry = it.next()
-          keyConverter(i, entry.getKey)
+          keyConverter(keyUpdater, i, entry.getKey)
           val value = entry.getValue
           if (value == null) {
             valueArray.setNullAt(i)
           } else {
-            valueConverter(i, value)
+            valueConverter(valueUpdater, i, value)
           }
           i += 1
         }
 
         // The ORC map will never have null or duplicated map keys, it's safe to create a
         // ArrayBasedMapData directly here.
         updater.set(ordinal, new ArrayBasedMapData(keyArray, valueArray))
+      }
 
-      case udt: UserDefinedType[_] => newWriter(udt.sqlType, updater)
+    case udt: UserDefinedType[_] =>
+      val converter = newWriter(udt.sqlType, reuseObj)
+      (updater, ordinal, value) => {
+        converter(updater, ordinal, value)
+      }
+
+    case _ =>
+      throw new UnsupportedOperationException(s"$dataType is not supported yet.")
+  }
+
+  private def getArrayDataCreator(elementType: DataType): Int => ArrayData = elementType match {
+    case BooleanType => length => UnsafeArrayData.createFreshArray(length, 1)
+    case ByteType => length => UnsafeArrayData.createFreshArray(length, 1)
+    case ShortType => length => UnsafeArrayData.createFreshArray(length, 2)
+    case IntegerType => length => UnsafeArrayData.createFreshArray(length, 4)
+    case LongType => length => UnsafeArrayData.createFreshArray(length, 8)
+    case FloatType => length => UnsafeArrayData.createFreshArray(length, 4)
+    case DoubleType => length => UnsafeArrayData.createFreshArray(length, 8)
+    case _ => length => new GenericArrayData(new Array[Any](length))
+  }
 
-      case _ =>
-        throw new UnsupportedOperationException(s"$dataType is not supported yet.")
+  private def getRowCreator(st: StructType): () => InternalRow = {
+    val constructorsArray = new Array[() => MutableValue](st.fields.length)
+    var i = 0
+    while (i < st.fields.length) {
+      st.fields(i).dataType match {
+        case BooleanType => constructorsArray(i) = () => new MutableBoolean
+        case ByteType => constructorsArray(i) = () => new MutableByte
+        case ShortType => constructorsArray(i) = () => new MutableShort
+        // We use INT for DATE internally
+        case IntegerType | DateType => constructorsArray(i) = () => new MutableInt
+        // We use Long for Timestamp internally
+        case LongType | TimestampType => constructorsArray(i) = () => new MutableLong
+        case FloatType => constructorsArray(i) = () => new MutableFloat
+        case DoubleType => constructorsArray(i) = () => new MutableDouble
+        case _ => constructorsArray(i) = () => new MutableAny
+      }
+      i += 1
     }
 
-  private def createArrayData(elementType: DataType, length: Int): ArrayData = elementType match {

Review comment:
       Changed now.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins commented on pull request #29353: [Spark-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668823788


   Can one of the admins verify this patch?


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins removed a comment on pull request #29353: [Spark-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668823788


   Can one of the admins verify this patch?


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA commented on pull request #29353: [Spark-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668914387


   **[Test build #127071 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127071/testReport)** for PR 29353 at commit [`13af454`](https://github.com/apache/spark/commit/13af454cf4bb53a1c6b2c5f6aef19b6b92420e26).


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] viirya commented on pull request #29353: [Spark-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668912282


   ok to test


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins removed a comment on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-683229295






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins commented on pull request #29353: [Spark-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668914789






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r479566417



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcSerializer.scala
##########
@@ -27,12 +27,19 @@ import org.apache.spark.sql.catalyst.util._
 import org.apache.spark.sql.types._
 
 /**
- * A serializer to serialize Spark rows to ORC structs.
+ * A serializer to serialize Spark rows to ORC structs. Multiple calls return the same [[OrcStruct]]
+ * object. If needed, caller should copy before making a new call.
  */
 class OrcSerializer(dataSchema: StructType) {
 
-  private val result = createOrcValue(dataSchema).asInstanceOf[OrcStruct]
-  private val converters = dataSchema.map(_.dataType).map(newConverter(_)).toArray
+  // TODO: repeating steps here. Can pass as parameter to newConverter, but only done once at the

Review comment:
       Apologies for the delay. Created https://issues.apache.org/jira/browse/SPARK-32732 and added to TODO comment.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669459288


   These are the changes referred to in https://github.com/apache/spark/pull/29353#discussion_r465955120 I'll put up a separate PR with this. 
   [SpecificInternalRow.txt](https://github.com/apache/spark/files/5030852/SpecificInternalRow.txt)
   


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins removed a comment on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669461208






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] msamirkhan commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
msamirkhan commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r465955120



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -73,136 +74,180 @@ class OrcDeserializer(
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
   private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
+      dataType: DataType): (CatalystDataUpdater, Int, WritableComparable[_]) => Unit =
     dataType match {
-      case NullType => (ordinal, _) =>
+      case NullType => (updater, ordinal, _) =>
         updater.setNullAt(ordinal)
 
-      case BooleanType => (ordinal, value) =>
+      case BooleanType => (updater, ordinal, value) =>
         updater.setBoolean(ordinal, value.asInstanceOf[BooleanWritable].get)
 
-      case ByteType => (ordinal, value) =>
+      case ByteType => (updater, ordinal, value) =>
         updater.setByte(ordinal, value.asInstanceOf[ByteWritable].get)
 
-      case ShortType => (ordinal, value) =>
+      case ShortType => (updater, ordinal, value) =>
         updater.setShort(ordinal, value.asInstanceOf[ShortWritable].get)
 
-      case IntegerType => (ordinal, value) =>
+      case IntegerType => (updater, ordinal, value) =>
         updater.setInt(ordinal, value.asInstanceOf[IntWritable].get)
 
-      case LongType => (ordinal, value) =>
+      case LongType => (updater, ordinal, value) =>
         updater.setLong(ordinal, value.asInstanceOf[LongWritable].get)
 
-      case FloatType => (ordinal, value) =>
+      case FloatType => (updater, ordinal, value) =>
         updater.setFloat(ordinal, value.asInstanceOf[FloatWritable].get)
 
-      case DoubleType => (ordinal, value) =>
+      case DoubleType => (updater, ordinal, value) =>
         updater.setDouble(ordinal, value.asInstanceOf[DoubleWritable].get)
 
-      case StringType => (ordinal, value) =>
+      case StringType => (updater, ordinal, value) =>
         updater.set(ordinal, UTF8String.fromBytes(value.asInstanceOf[Text].copyBytes))
 
-      case BinaryType => (ordinal, value) =>
+      case BinaryType => (updater, ordinal, value) =>
         val binary = value.asInstanceOf[BytesWritable]
         val bytes = new Array[Byte](binary.getLength)
         System.arraycopy(binary.getBytes, 0, bytes, 0, binary.getLength)
         updater.set(ordinal, bytes)
 
-      case DateType => (ordinal, value) =>
+      case DateType => (updater, ordinal, value) =>
         updater.setInt(ordinal, OrcShimUtils.getGregorianDays(value))
 
-      case TimestampType => (ordinal, value) =>
+      case TimestampType => (updater, ordinal, value) =>
         updater.setLong(ordinal, DateTimeUtils.fromJavaTimestamp(value.asInstanceOf[OrcTimestamp]))
 
-      case DecimalType.Fixed(precision, scale) => (ordinal, value) =>
+      case DecimalType.Fixed(precision, scale) => (updater, ordinal, value) =>
         val v = OrcShimUtils.getDecimal(value)
         v.changePrecision(precision, scale)
         updater.set(ordinal, v)
 
-      case st: StructType => (ordinal, value) =>
-        val result = new SpecificInternalRow(st)
-        val fieldUpdater = new RowUpdater(result)
+      case st: StructType =>
+        val createRow: () => InternalRow = getRowCreator(st)
         val fieldConverters = st.map(_.dataType).map { dt =>
-          newWriter(dt, fieldUpdater)
+          newWriter(dt)
         }.toArray
-        val orcStruct = value.asInstanceOf[OrcStruct]
-
-        var i = 0
-        while (i < st.length) {
-          val value = orcStruct.getFieldValue(i)
-          if (value == null) {
-            result.setNullAt(i)
-          } else {
-            fieldConverters(i)(i, value)
+        (updater, ordinal, value) => {
+          val result = createRow.apply()
+          val fieldUpdater = new RowUpdater(result)
+          val orcStruct = value.asInstanceOf[OrcStruct]
+
+          var i = 0
+          while (i < st.length) {
+            val value = orcStruct.getFieldValue(i)
+            if (value == null) {
+              result.setNullAt(i)
+            } else {
+              fieldConverters(i)(fieldUpdater, i, value)
+            }
+            i += 1
           }
-          i += 1
+
+          updater.set(ordinal, result)
         }
 
-        updater.set(ordinal, result)
-
-      case ArrayType(elementType, _) => (ordinal, value) =>
-        val orcArray = value.asInstanceOf[OrcList[WritableComparable[_]]]
-        val length = orcArray.size()
-        val result = createArrayData(elementType, length)
-        val elementUpdater = new ArrayDataUpdater(result)
-        val elementConverter = newWriter(elementType, elementUpdater)
-
-        var i = 0
-        while (i < length) {
-          val value = orcArray.get(i)
-          if (value == null) {
-            result.setNullAt(i)
-          } else {
-            elementConverter(i, value)
+      case ArrayType(elementType, _) =>
+        val elementConverter = newWriter(elementType)
+        val createArray = getArrayDataCreator(elementType)
+        (updater, ordinal, value) => {
+          val orcArray = value.asInstanceOf[OrcList[WritableComparable[_]]]
+          val length = orcArray.size()
+          val result = createArray(length)
+          val elementUpdater = new ArrayDataUpdater(result)
+
+          var i = 0
+          while (i < length) {
+            val value = orcArray.get(i)
+            if (value == null) {
+              result.setNullAt(i)
+            } else {
+              elementConverter(elementUpdater, i, value)
+            }
+            i += 1
           }
-          i += 1
+
+          updater.set(ordinal, result)
         }
 
-        updater.set(ordinal, result)
-
-      case MapType(keyType, valueType, _) => (ordinal, value) =>
-        val orcMap = value.asInstanceOf[OrcMap[WritableComparable[_], WritableComparable[_]]]
-        val length = orcMap.size()
-        val keyArray = createArrayData(keyType, length)
-        val keyUpdater = new ArrayDataUpdater(keyArray)
-        val keyConverter = newWriter(keyType, keyUpdater)
-        val valueArray = createArrayData(valueType, length)
-        val valueUpdater = new ArrayDataUpdater(valueArray)
-        val valueConverter = newWriter(valueType, valueUpdater)
-
-        var i = 0
-        val it = orcMap.entrySet().iterator()
-        while (it.hasNext) {
-          val entry = it.next()
-          keyConverter(i, entry.getKey)
-          val value = entry.getValue
-          if (value == null) {
-            valueArray.setNullAt(i)
-          } else {
-            valueConverter(i, value)
+      case MapType(keyType, valueType, _) =>
+        val keyConverter = newWriter(keyType)
+        val createKeyArray = getArrayDataCreator(keyType)
+        val valueConverter = newWriter(valueType)
+        val createValueArray = getArrayDataCreator(valueType)
+        (updater, ordinal, value) => {
+          val orcMap = value.asInstanceOf[OrcMap[WritableComparable[_], WritableComparable[_]]]
+          val length = orcMap.size()
+          val keyArray = createKeyArray(length)
+          val keyUpdater = new ArrayDataUpdater(keyArray)
+          val valueArray = createValueArray(length)
+          val valueUpdater = new ArrayDataUpdater(valueArray)
+
+          var i = 0
+          val it = orcMap.entrySet().iterator()
+          while (it.hasNext) {
+            val entry = it.next()
+            keyConverter(keyUpdater, i, entry.getKey)
+            val value = entry.getValue
+            if (value == null) {
+              valueArray.setNullAt(i)
+            } else {
+              valueConverter(valueUpdater, i, value)
+            }
+            i += 1
           }
-          i += 1
-        }
 
-        // The ORC map will never have null or duplicated map keys, it's safe to create a
-        // ArrayBasedMapData directly here.
-        updater.set(ordinal, new ArrayBasedMapData(keyArray, valueArray))
+          // The ORC map will never have null or duplicated map keys, it's safe to create a
+          // ArrayBasedMapData directly here.
+          updater.set(ordinal, new ArrayBasedMapData(keyArray, valueArray))
+        }
 
-      case udt: UserDefinedType[_] => newWriter(udt.sqlType, updater)
+      case udt: UserDefinedType[_] =>
+        val converter = newWriter(udt.sqlType)
+        (updater, ordinal, value) => {
+          converter(updater, ordinal, value)
+        }
 
       case _ =>
         throw new UnsupportedOperationException(s"$dataType is not supported yet.")
     }
 
-  private def createArrayData(elementType: DataType, length: Int): ArrayData = elementType match {
-    case BooleanType => UnsafeArrayData.fromPrimitiveArray(new Array[Boolean](length))
-    case ByteType => UnsafeArrayData.fromPrimitiveArray(new Array[Byte](length))
-    case ShortType => UnsafeArrayData.fromPrimitiveArray(new Array[Short](length))
-    case IntegerType => UnsafeArrayData.fromPrimitiveArray(new Array[Int](length))
-    case LongType => UnsafeArrayData.fromPrimitiveArray(new Array[Long](length))
-    case FloatType => UnsafeArrayData.fromPrimitiveArray(new Array[Float](length))
-    case DoubleType => UnsafeArrayData.fromPrimitiveArray(new Array[Double](length))
-    case _ => new GenericArrayData(new Array[Any](length))
+  private def getArrayDataCreator(elementType: DataType): Int => ArrayData = elementType match {
+    case BooleanType => length => UnsafeArrayData.createFreshArray(length, 1)
+    case ByteType => length => UnsafeArrayData.createFreshArray(length, 1)
+    case ShortType => length => UnsafeArrayData.createFreshArray(length, 2)
+    case IntegerType => length => UnsafeArrayData.createFreshArray(length, 4)
+    case LongType => length => UnsafeArrayData.createFreshArray(length, 8)
+    case FloatType => length => UnsafeArrayData.createFreshArray(length, 4)
+    case DoubleType => length => UnsafeArrayData.createFreshArray(length, 8)
+    case _ => length => new GenericArrayData(new Array[Any](length))
+  }
+
+  private def getRowCreator(st: StructType): () => InternalRow = {
+    val constructorsArray = new Array[() => MutableValue](st.fields.length)
+    var i = 0
+    while (i < st.fields.length) {
+      st.fields(i).dataType match {
+        case BooleanType => constructorsArray(i) = () => new MutableBoolean
+        case ByteType => constructorsArray(i) = () => new MutableByte
+        case ShortType => constructorsArray(i) = () => new MutableShort
+        // We use INT for DATE internally
+        case IntegerType | DateType => constructorsArray(i) = () => new MutableInt
+        // We use Long for Timestamp internally
+        case LongType | TimestampType => constructorsArray(i) = () => new MutableLong
+        case FloatType => constructorsArray(i) = () => new MutableFloat
+        case DoubleType => constructorsArray(i) = () => new MutableDouble
+        case _ => constructorsArray(i) = () => new MutableAny
+      }
+      i += 1
+    }
+
+    () => {
+      val array = new Array[MutableValue](constructorsArray.length)
+      var i = 0
+      while (i < constructorsArray.length) {
+        array(i) = constructorsArray(i).apply()
+        i += 1
+      }
+      new SpecificInternalRow(array)
+    }

Review comment:
       I ran the read benchmark again with changes to the SpecificInternalRow constructor to remove the map calls instead and it ran in the same time. I will update this PR and create a separate one for changes to SpecificInternalRow since it impacts both Orc and Avro (https://github.com/apache/spark/pull/29354).




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-668986200


   **[Test build #127071 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/127071/testReport)** for PR 29353 at commit [`13af454`](https://github.com/apache/spark/commit/13af454cf4bb53a1c6b2c5f6aef19b6b92420e26).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-683229139


   **[Test build #128005 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128005/testReport)** for PR 29353 at commit [`d67ceed`](https://github.com/apache/spark/commit/d67ceed965fce5f56f2096032188c6fa9b3cfa5b).
    * This patch passes all tests.
    * This patch merges cleanly.
    * This patch adds no public classes.


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins removed a comment on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-669614039






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] AmplabJenkins commented on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-683181444






----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] HyukjinKwon commented on a change in pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on a change in pull request #29353:
URL: https://github.com/apache/spark/pull/29353#discussion_r466108888



##########
File path: sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
##########
@@ -73,135 +75,157 @@ class OrcDeserializer(
    * Creates a writer to write ORC values to Catalyst data structure at the given ordinal.
    */
   private def newWriter(
-      dataType: DataType, updater: CatalystDataUpdater): (Int, WritableComparable[_]) => Unit =
+      dataType: DataType, reuseObj: Boolean)
+  : (CatalystDataUpdater, Int, WritableComparable[_]) => Unit =
     dataType match {
-      case NullType => (ordinal, _) =>
+      case NullType => (updater, ordinal, _) =>

Review comment:
       @msamirkhan sorry if I rushed to read but why do we need pull this updater out here? If the field writers are created once, we don't have to bother about reusing the updater for nested types as they are just created once and cheap.




----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] [spark] SparkQA removed a comment on pull request #29353: [SPARK-32532][SQL] Improve ORC read/write performance on nested structs and array of structs

Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #29353:
URL: https://github.com/apache/spark/pull/29353#issuecomment-683181113


   **[Test build #128005 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/128005/testReport)** for PR 29353 at commit [`d67ceed`](https://github.com/apache/spark/commit/d67ceed965fce5f56f2096032188c6fa9b3cfa5b).


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org