parallel execution for columns serialization

apache · Dec 3, 2023 · e1f054e · e1f054e
1 parent b306bb4
commit e1f054e
Showing 1 changed file with 6 additions and 12 deletions.
diff --git a/...kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/schema/RowSet.scala b/...kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/schema/RowSet.scala
@@ -95,19 +95,13 @@ object RowSet {
     var i = 0
     val columnSize = schema.length
     val sw1 = StopWatch.createStarted()
-    sw1.suspend()
     val sw2 = StopWatch.createStarted()
-    sw2.suspend()
-    while (i < columnSize) {
-      sw1.resume()
-      val field = schema(i)
-      sw1.suspend()
-
-      sw2.resume()
-      val tColumn = toTColumn(rows, i, field.dataType)
-      sw2.suspend()
-      tRowSet.addToColumns(tColumn)
-      i += 1
+    (0 until columnSize).par.map { ordinal =>
+      val field = schema(ordinal)
+      val tCol = toTColumn(rows, ordinal, field.dataType)
+      (ordinal, tCol)
+    }.toList.sortBy(_._1).foreach { case (_, tCol) =>
+      tRowSet.addToColumns(tCol)
     }
     sw1.stop()
     sw2.stop()