feat: drive Spark's file writers from Arrow batches, including partitioned and bucketed writes [experiment]

[andygrove]

Which issue does this PR close?

Part of #5625. That issue proposes handing Spark's writers a zero-copy row view instead of a
materialized UnsafeRow, and restricts the idea to unpartitioned, unbucketed writes. #5626
implements it that way. This PR is an alternative that lifts the restriction.

Draft / RFC. Off by default. Opening as a draft to get the approach and the writer-internals
coupling in front of people, not because anything is unfinished.

Rationale for this change

The restriction in #5625 is not arbitrary. A partitioned or bucketed write never reaches
SingleDirectoryDataWriter; it goes through BaseDynamicPartitionDataWriter, which projects
every row a second time before the OutputWriter sees it:

protected val getOutputRow =
  UnsafeProjection.create(description.dataColumns, description.allColumns)  // :266

protected def writeRecord(record: InternalRow): Unit = {
  val outputRow = getOutputRow(record)   // strips partition / bucket columns
  currentWriter.write(outputRow)
  ...
}

So simply planting a row view under the write does not help a partitioned write. Today the cost is
a slow field-by-field build of nested UnsafeRow / UnsafeArrayData in the columnar-to-row
transition, followed by a cheap second projection that hits GenerateUnsafeProjection's
instanceof UnsafeRow / UnsafeArrayData / UnsafeMapData bulk-copy branches
(GenerateUnsafeProjection.scala:72,193,227). Feeding the writer a ColumnarBatchRow would delete
the cheap pass and leave the expensive one, because those branches stop matching.

The projection exists for one reason: to drop the partition and bucket columns. Spark needs a full
row copy to do that because it is row-at-a-time. On a ColumnarBatch the same pruning is
new ColumnarBatch(subsetOfVectors, numRows), which copies nothing. Getting that requires
replacing the writer, not the transition.

Spark 4.0 provides the seam. V1WritesUtils.getWriteFilesOpt matches the WriteFilesExecBase
trait, so a Comet node extending it is driven through
FileFormatWriter.executeWrite -> SparkPlan.executeWrite -> doExecuteWrite, and everything
above the per-task write stays with Spark: SaveMode, the commit protocol, _SUCCESS, dynamic
partition overwrite, stats tracker aggregation, catalog updates. This is the same seam #5293 opens
for native writes, and the two are consistent by design (see below).

Measurements

CometWriteRowViewBenchmark, added here. 1M rows, Spark 4.1, release build, M3 Ultra, timing the
whole scan -> write -> file pipeline. The gain compares the two Comet arms, which run the same
plan and differ only in the write node (uncompressed / snappy):

schema	unpartitioned	partitioned
flat, 10 columns	0.2% / 0.6% (declined by the gate; noise floor)	3.7% / 4.5%
nested (struct + array + map)	20.4% / 13.3%	16.3% / 16.1%
deeply nested (4 levels, array-of-structs, map-of-array-of-structs)	12.3% / 13.3%	16.5% / 16.6%

The flat unpartitioned row is the noise floor: the rule declines it, so both Comet arms run the
identical plan there.

Two caveats on reading this. The unpartitioned nested case moved 15.2% -> 20.4% between two runs
while every other case held within a point, so treat it as "roughly 13-20%" rather than 20%; the
partitioned numbers reproduced tightly across both runs. And this does not touch what actually
makes a write expensive, which is parquet-mr encoding. Only the native writer changes that. What it
does is make the interim path cheaper while native writes remain experimental, and unlike the
native writer it stays Spark-compatible by construction because Spark's own writer still produces
the bytes.

Relationship to #5293 and #5626

• refactor: hook native Parquet writes into Spark's WriteFilesExec seam #5293 (refactor: hook native Parquet writes into Spark's WriteFilesExec seam) puts a
  CometWriteFilesExec at this seam that writes Parquet with the native Rust writer, currently
  unpartitioned only. This PR reuses that PR's ShimCometWriteFilesExec traits verbatim, so
  whichever lands second is a trivial rebase. The node here is named
  CometRowViewWriteFilesExec to avoid a file-level collision.
• feat: experimental zero-copy row views for writes of complex-typed data [experiment] #5626 implements Feed Spark's write path zero-copy row views instead of materializing UnsafeRow #5625 the other way, as a CometColumnarToRowViewExec transition planted
  under the write. If this PR is the direction people prefer, feat: experimental zero-copy row views for writes of complex-typed data [experiment] #5626 should be closed in favour of
  it: this one subsumes it, covering the unpartitioned case too, with a stronger safety argument
  and larger measured gains.

On that argument: #5626's main caveat is that a reused mutable row becomes an operator's output
and could reach a consumer that retains it. Here the reused rows never leave
CometRowViewWriteFilesExec.doExecuteWrite; they are created and consumed inside one method whose
only consumer is Spark's OutputWriter. Extending WriteFilesExecBase is also what stops AQE from
re-inserting a second WriteFilesExec above the node, which is the hazard #5293 documents.

What changes are included in this PR?

CometRowViewWriteFilesExec replaces WriteFilesExec and drives Spark's own writers from
child.executeColumnar(). Its executeTask is a direct port of FileFormatWriter.executeTask,
differing in two places:

• the partitioned and bucketed case builds a CometRowViewDynamicPartitionWriter, a subclass of
  DynamicPartitionDataSingleWriter that overrides only writeRecord to substitute a pruned view
  for getOutputRow(record). Partition-change detection, writer renewal and maxRecordsPerFile
  are inherited unchanged and still see the full row.
• rows come from two iterators advanced in lockstep over the same batch: one over all columns,
  which Spark's writer uses to compute partition values and bucket ids, and one over the pruned
  batch, which is what reaches the OutputWriter. Both are views over the same Arrow vectors, so
  the pair costs one object per batch rather than a copy per row.

The unpartitioned case needs no subclass: SingleDirectoryDataWriter hands the row straight to the
OutputWriter, so the batch row goes in as-is.

EliminateRedundantTransitions plants the node, gated on:

• Spark 4.0+. On 3.4 / 3.5 getWriteFilesOpt matches the concrete WriteFilesExec case class,
  so a replacement node is invisible and the write would silently take a path that calls
  doExecute on it. Same reasoning and same shim as refactor: hook native Parquet writes into Spark's WriteFilesExec seam #5293.
• spark.sql.maxConcurrentOutputFileWriters at its default of 0. Above 0,
  V1WritesUtils.getSortOrder plants no sort and FileFormatWriter picks
  DynamicPartitionDataConcurrentWriter, which spills through UnsafeKVExternalSorter. This gate
  is load-bearing in both directions: it keeps that writer out, and it is what guarantees
  DynamicPartitionDataSingleWriter gets the sorted input it requires.
• one of Spark's own FileFormats, whose OutputWriters encode each row on the spot. A
  third-party format is free to buffer the InternalRow it is handed.
• a complex data column, for unpartitioned writes only. The gate asks how many projections the
  write removes. A partitioned or bucketed write removes two, the transition and getOutputRow, so
  it qualifies at any schema: the flat partitioned row in the table above is 3.7-4.5% against a
  0.2-0.6% noise floor. An unpartitioned write removes only the transition, and on a flat schema
  that is a generated fixed-width copy which measures inside the noise, so it still needs a struct,
  array or map to be worth putting a reused mutable row in front of Spark's writer. Partition and
  bucket columns never count towards that test; they are stripped before the OutputWriter sees
  the row.

No check for an intervening SortExec is needed. The rule only rewrites w.child, so a write
whose required ordering was satisfied by a Spark SortExec rather than a Comet one simply does not
match. This is worth stating because #5625 and #5626 both cite UnsafeExternalSorter as a reason
to exclude partitioned writes; that exclusion was already enforced structurally.

Behind spark.comet.exec.write.rowView.enabled, default false.

How are these changes tested?

New CometWriteRowViewSuite, 15 tests, registered in both PR build workflows and gated on
isSpark40Plus. The bar for each is that enabling the config changes nothing observable but the
plan, with the baseline written by the same Comet plan with the config off:

• same data and row count for: unpartitioned; one dynamic partition column; two partition columns;
  a flat partitioned schema; deeply nested types (four levels, nulls at every level) with
  partitions; null and empty-string partition values, which exercise __HIVE_DEFAULT_PARTITION__
  and the Empty2Null projection; maxRecordsPerFile set and unset across partition boundaries;
  ORC as well as Parquet
• identical partition directory layout
• the node is present exactly once for unpartitioned, dynamically partitioned, flat partitioned and
  bucketed writes, absent by default, absent for a flat unpartitioned schema, and absent when
  maxConcurrentOutputFileWriters is raised

Ran alongside CometParquetWriterSuite: 48 tests pass on Spark 4.1. CometExecSuite passes as a
regression sweep (144 tests). Compiles against Spark 3.4, 3.5, 4.0 and 4.1.

CometWriteRowViewBenchmark asserts through a QueryExecutionListener that each arm planned the
write node its label names, and warns into the results file when it did not, so a silently declined
arm cannot masquerade as a measurement. The runs behind the table above produced no such warnings.

Caveats I would want a second opinion on

1. The coupling. writeRecord, currentWriter, statsTrackers and recordsInFile are
   protected members of Spark's writer, and this depends on their current structure across
   4.0 / 4.1 / 4.2. A future refactor there would be a compile error rather than silent corruption,
   which is the good case, but the dependency is real.
2. Stats trackers see a reused row. WriteTaskStatsTracker.newRow now receives a view rather
   than a materialized row. That is correct for BasicWriteTaskStatsTracker, which ignores it, and
   strictly better than refactor: hook native Parquet writes into Spark's WriteFilesExec seam #5293's native path, which passes InternalRow.empty. A third-party
   tracker that retains rows would still be wrong. There is a format gate but no tracker gate;
   statsTrackers only exists at execution time.
3. The format gate is a package-name prefix check, inherited from feat: experimental zero-copy row views for writes of complex-typed data [experiment] #5626's approach and equally
   crude here. A defensive option would be to restrict it to Parquet and ORC by class.