feat: Native columnar to row conversion (Phase 2)

[andygrove]

Which issue does this PR close?

Part of #3268

Follow on from #3221

Rationale for this change

Enable CometNativeColumnarToRowExec in Comet tests, but still disabled by default.

What changes are included in this PR?

• Mutable buffer fallback - Fall back to JVM C2R for V2 Parquet scans and native_iceberg_compat scans with partition columns that use mutable buffers
• Int32/Int64 → Decimal128 conversion - Handle small-precision decimals stored as Int32/Int64 when COMET_USE_DECIMAL_128=false
• Dictionary-encoded decimal corruption - Fix 2x value multiplication when casting dictionary-encoded decimals
• FixedSizeBinary support - Add missing support for FixedSizeBinary data type
• NullArray handling - Cast NullArray to expected schema type when all column values are null
• Dictionary unpacking - Auto-unpack dictionary-encoded arrays when schema expects non-dictionary type
• Dictionary type mismatch - Fix "Expected Dictionary type but got Binary" error with BloomFilter columns
• doExecuteBroadcast - Add broadcast support to CometNativeColumnarToRowExec
• refactor to avoid duplicate code

How are these changes tested?

[codecov-commenter]

Codecov Report

❌ Patch coverage is 65.88235% with 29 lines in your changes missing coverage. Please review.
✅ Project coverage is 59.95%. Comparing base (f09f8af) to head (e4b534c).
⚠️ Report is 898 commits behind head on main.

Files with missing lines	Patch %	Lines
...spark/sql/comet/CometNativeColumnarToRowExec.scala	65.27%	20 Missing and 5 partials ⚠️
...he/comet/rules/EliminateRedundantTransitions.scala	75.00%	0 Missing and 3 partials ⚠️
.../scala/org/apache/spark/sql/comet/util/Utils.scala	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #3266      +/-   ##
============================================
+ Coverage     56.12%   59.95%   +3.82%     
- Complexity      976     1473     +497     
============================================
  Files           119      175      +56     
  Lines         11743    16167    +4424     
  Branches       2251     2682     +431     
============================================
+ Hits           6591     9693    +3102     
- Misses         4012     5126    +1114     
- Partials       1140     1348     +208     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[andygrove]

@wForget I've now added the fix for closing the ColumnarBatch after conversion (commit a585707).

The issue was that after calling converter.convert(batch), the batch was not being closed, which could lead to Arrow resource leaks. The fix wraps the row iterator to call batch.close() when hasNext returns false:

new Iterator[InternalRow] {
  override def hasNext: Boolean = {
    val hasMore = result.hasNext
    if (!hasMore) {
      batch.close()
    }
    hasMore
  }
  override def next(): InternalRow = result.next()
}

This fix is applied to both the doExecute() path and the broadcast execution path.

[Copilot]

Pull request overview

Enables and hardens the native (Rust) columnar-to-row conversion path, and turns it on for the Comet test suite while keeping runtime defaults unchanged.

Changes:

• Enable CometNativeColumnarToRowExec in test configs and plan checks, updating affected tests.
• Add broadcast support and additional batch handling to CometNativeColumnarToRowExec.
• Extend native C2R (Rust) to handle more Arrow edge-cases (dictionary unpacking, null arrays, FixedSizeBinary, small-precision decimal representations) and refactor conversion internals.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
spark/src/test/scala/org/apache/spark/sql/comet/CometPlanChecker.scala	Allow native C2R exec in plan validation for tests.
spark/src/test/scala/org/apache/spark/sql/CometTestBase.scala	Enable native columnar-to-row in Comet tests via SparkConf.
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala	Update broadcast/AQE assertions to accept native or JVM C2R.
spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala	Update plan shape expectations for native C2R in tests.
spark/src/main/scala/org/apache/spark/sql/comet/CometNativeColumnarToRowExec.scala	Implement doExecuteBroadcast and improve batch lifecycle handling.
spark/src/main/scala/org/apache/comet/rules/EliminateRedundantTransitions.scala	Select native vs JVM C2R with a mutable-buffer scan fallback.
native/core/src/execution/columnar_to_row.rs	Add missing type/encoding handling and refactor downcasting/writing logic.
common/src/main/scala/org/apache/spark/sql/comet/util/Utils.scala	Accept Arrow NullVector during batch serialization.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[wForget]

Thanks @andygrove , LGTM.

[andygrove]

Thanks for the review @wForget!