feat: add experimental native columnar to row conversion

This PR adds an experimental native (Rust-based) implementation of
ColumnarToRowExec that converts Arrow columnar data to Spark UnsafeRow
format.

Benefits over the current Scala implementation:
- Zero-copy for variable-length types: String and Binary data is written
  directly to the output buffer without intermediate Java object allocation
- Vectorized processing: The native implementation processes data in a
  columnar fashion, improving CPU cache utilization
- Reduced GC pressure: All conversion happens in native memory, avoiding
  the creation of temporary Java objects that would need garbage collection
- Buffer reuse: The output buffer is allocated once and reused across
  batches, minimizing memory allocation overhead

The feature is disabled by default and can be enabled by setting:
  spark.comet.exec.columnarToRow.native.enabled=true

Supported data types:
- Primitive types: Boolean, Byte, Short, Int, Long, Float, Double
- Date and Timestamp (microseconds)
- Decimal (both inline precision<=18 and variable-length precision>18)
- String and Binary
- Complex types: Struct, Array, Map (nested)

This is an experimental feature for evaluation and benchmarking purposes.

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: andygrove

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

@@ -286,6 +286,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.