From 3a454e493ff6562be4fca4a5c1b42fb33835aa7e Mon Sep 17 00:00:00 2001
From: Dongjoon Hyun <dongjoon@apache.org>
Date: Tue, 3 Jan 2023 02:42:42 -0800
Subject: [PATCH] [SPARK-41858][SQL] Fix ORC reader perf regression due to
 DEFAULT value feature

---
 .../datasources/orc/OrcDeserializer.scala     | 21 ++++++++++---------
 .../spark/sql/sources/InsertSuite.scala       |  9 ++++++--
 2 files changed, 18 insertions(+), 12 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
index 5276f5c6d7b98..5b207a04adad6 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/orc/OrcDeserializer.scala
@@ -57,14 +57,7 @@ class OrcDeserializer(
             } else {
               new RowUpdater(resultRow)
             }
-          val writer: (Int, WritableComparable[_]) => Unit =
-            (ordinal, value) =>
-              if (value == null) {
-                rowUpdater.setNullAt(ordinal)
-              } else {
-                val writerFunc = newWriter(f.dataType, rowUpdater)
-                writerFunc(ordinal, value)
-              }
+          val writer = newWriter(f.dataType, rowUpdater)
           (value: WritableComparable[_]) => writer(index, value)
         }
       }.toArray
@@ -75,7 +68,11 @@ class OrcDeserializer(
     while (targetColumnIndex < fieldWriters.length) {
       if (fieldWriters(targetColumnIndex) != null) {
         val value = orcStruct.getFieldValue(requestedColIds(targetColumnIndex))
-        fieldWriters(targetColumnIndex)(value)
+        if (value == null) {
+          resultRow.setNullAt(targetColumnIndex)
+        } else {
+          fieldWriters(targetColumnIndex)(value)
+        }
       }
       targetColumnIndex += 1
     }
@@ -88,7 +85,11 @@ class OrcDeserializer(
     while (targetColumnIndex < fieldWriters.length) {
       if (fieldWriters(targetColumnIndex) != null) {
         val value = orcValues(requestedColIds(targetColumnIndex))
-        fieldWriters(targetColumnIndex)(value)
+        if (value == null) {
+          resultRow.setNullAt(targetColumnIndex)
+        } else {
+          fieldWriters(targetColumnIndex)(value)
+        }
       }
       targetColumnIndex += 1
     }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
index dd37c93871e08..7c4a39d6ff462 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/sources/InsertSuite.scala
@@ -1679,7 +1679,8 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
           Config(
             None),
           Config(
-            Some(SQLConf.ORC_VECTORIZED_READER_ENABLED.key -> "false")))),
+            Some(SQLConf.ORC_VECTORIZED_READER_ENABLED.key -> "false"),
+            insertNullsToStorage = false))),
       TestCase(
         dataSource = "parquet",
         Seq(
@@ -1943,7 +1944,11 @@ class InsertSuite extends DataSourceTest with SharedSparkSession {
               Row(Seq(Row(1, 2)), Seq(Map(false -> "def", true -> "jkl"))),
               Seq(Map(true -> "xyz"))),
             Row(2,
-              null,
+              if (config.dataSource != "orc") {
+                null
+              } else {
+                Row(Seq(Row(1, 2)), Seq(Map(false -> "def", true -> "jkl")))
+              },
               Seq(Map(true -> "xyz"))),
             Row(3,
               Row(Seq(Row(3, 4)), Seq(Map(false -> "mno", true -> "pqr"))),