feat: Native columnar to row conversion (Phase 1) (#3221)

Author: andygrove

.github/workflows/pr_build_linux.yml

@@ -202,6 +202,7 @@ jobs:
             value: |
               org.apache.comet.exec.CometShuffleSuite
               org.apache.comet.exec.CometShuffle4_0Suite
+              org.apache.comet.exec.CometNativeColumnarToRowSuite
               org.apache.comet.exec.CometNativeShuffleSuite
               org.apache.comet.exec.CometShuffleEncryptionSuite
               org.apache.comet.exec.CometShuffleManagerSuite

.github/workflows/pr_build_macos.yml

@@ -145,6 +145,7 @@ jobs:
             value: |
               org.apache.comet.exec.CometShuffleSuite
               org.apache.comet.exec.CometShuffle4_0Suite
+              org.apache.comet.exec.CometNativeColumnarToRowSuite
               org.apache.comet.exec.CometNativeShuffleSuite
               org.apache.comet.exec.CometShuffleEncryptionSuite
               org.apache.comet.exec.CometShuffleManagerSuite

.gitignore

@@ -1,3 +1,4 @@
+CLAUDE.md
 target
 .idea
 *.iml

common/src/main/scala/org/apache/comet/CometConf.scala

@@ -296,6 +296,17 @@ object CometConf extends ShimCometConf {
   val COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED: ConfigEntry[Boolean] =
     createExecEnabledConfig("localTableScan", defaultValue = false)
 
+  val COMET_NATIVE_COLUMNAR_TO_ROW_ENABLED: ConfigEntry[Boolean] =
+    conf(s"$COMET_EXEC_CONFIG_PREFIX.columnarToRow.native.enabled")
+      .category(CATEGORY_EXEC)
+      .doc(
+        "Whether to enable native columnar to row conversion. When enabled, Comet will use " +
+          "native Rust code to convert Arrow columnar data to Spark UnsafeRow format instead " +
+          "of the JVM implementation. This can improve performance for queries that need to " +
+          "convert between columnar and row formats. This is an experimental feature.")
+      .booleanConf
+      .createWithDefault(false)
+
   val COMET_EXEC_SORT_MERGE_JOIN_WITH_JOIN_FILTER_ENABLED: ConfigEntry[Boolean] =
     conf("spark.comet.exec.sortMergeJoinWithJoinFilter.enabled")
       .category(CATEGORY_ENABLE_EXEC)

common/src/main/scala/org/apache/comet/vector/NativeUtil.scala

@@ -78,6 +78,26 @@ class NativeUtil {
     (arrays, schemas)
   }
 
+  /**
+   * Exports a ColumnarBatch to Arrow FFI and returns the memory addresses.
+   *
+   * This is a convenience method that allocates Arrow structs, exports the batch, and returns
+   * just the memory addresses (without exposing the Arrow types).
+   *
+   * @param batch
+   *   the columnar batch to export
+   * @return
+   *   a tuple of (array addresses, schema addresses, number of rows)
+   */
+  def exportBatchToAddresses(batch: ColumnarBatch): (Array[Long], Array[Long], Int) = {
+    val numCols = batch.numCols()
+    val (arrays, schemas) = allocateArrowStructs(numCols)
+    val arrayAddrs = arrays.map(_.memoryAddress())
+    val schemaAddrs = schemas.map(_.memoryAddress())
+    val numRows = exportBatch(arrayAddrs, schemaAddrs, batch)
+    (arrayAddrs, schemaAddrs, numRows)
+  }
+
   /**
    * Exports a Comet `ColumnarBatch` into a list of memory addresses that can be consumed by the
    * native execution.