feat: experimental zero-copy row views for writes of complex-typed data [experiment]

[andygrove]

Which issue does this PR close?

Closes #5625.

Experimental / RFC. Off by default, and I am not convinced the gain justifies the hazard yet -
see the caveats at the bottom. Opening as a draft to get the measurements and the safety argument
in front of people rather than because I think it is ready to land.

Rationale for this change

Comet has to do a columnar-to-row conversion before a write because native write support is still
experimental. That conversion materialises an UnsafeRow per row, and nothing on Spark's write
path needs one:

• OutputWriter.write(InternalRow), FileFormatDataWriter.write(InternalRow) and
  WriteTaskStatsTracker.newRow(String, InternalRow) are all typed on InternalRow
• ParquetWriteSupport extends WriteSupport<InternalRow> and reads fields through
  SpecializedGetters
• BasicWriteTaskStatsTracker.newRow ignores the row entirely and just increments a counter

Meanwhile CometColumnarToRowExec.doExecute already produces batch.rowIterator() - a reused
ColumnarBatchRow that is a zero-copy view over the Arrow buffers - and then throws it away by
applying an UnsafeProjection. For a write that copy is pure overhead: the writer decodes straight
back out of the row it was just given.

This does not touch what actually makes a write expensive, which is parquet-mr encoding. Only the
native writer changes that. What it does do 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.

What changes are included in this PR?

CometColumnarToRowViewExec returns batch.rowIterator() unprojected. It is deliberately not
CodegenSupport: whole-stage codegen would generate an UnsafeRowWriter loop and reintroduce
exactly the copy this exists to avoid.

EliminateRedundantTransitions plants it under WriteFilesExec (planned writes) or
DataWritingCommandExec (when plannedWrite is off). The row it hands over is reused and mutable,
so it is only correct for a consumer that finishes with a row before pulling the next one. Three
gates keep it there:

• unpartitioned and unbucketed, which is what makes FileFormatWriter choose
  SingleDirectoryDataWriter. The other writers do not qualify: the required ordering on
  partition/bucket columns puts a SortExec in between and UnsafeExternalSorter needs
  UnsafeRow, and DynamicPartitionDataConcurrentWriter spills through
  UnsafeKVExternalSorter.insertKV, which is typed on UnsafeRow.
• one of Spark's own FileFormats, whose OutputWriters encode each row on the spot (Parquet
  through ParquetWriteSupport, ORC through OrcSerializer into a VectorizedRowBatch, the text
  formats directly). A third-party format is free to buffer the InternalRow it is handed.
• a schema containing a struct, array or map. On flat schemas the projection is a generated
  fixed-width copy and the saving is inside the noise, which does not pay for the hazard above.

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

Also adds CometParquetWriteBenchmark. Note it sets spark.shuffle.manager explicitly - without
it isCometLoaded disables Comet and every "Comet" arm silently measures Spark, which is #5624.
Each arm asserts through a QueryExecutionListener that it planned the transition its label names,
and warns into the results file if not.

Measurements

M3 Max, 1M rows, Spark 4.1, release build, best-of-N, versus today's CometColumnarToRowExec:

schema	uncompressed	snappy
fixed width (10 cols)	542 -> 526 ms (declined by the gate; noise)	553 -> 543 ms (noise)
strings	620 -> 618 ms (declined; noise)	-
wide, 50 columns	3166 -> 3096 ms (declined; noise)	-
struct + array + map	1027 -> 920 ms (10%)	1104 -> 934 ms (15%)
struct + array-of-structs + map-of-array-of-structs	1838 -> 1630 ms (11%)	1850 -> 1616 ms (13%)
single struct, depth 1	351 -> 315 ms (10%)	350 -> 323 ms (8%)
single struct, depth 2	431 -> 394 ms (9%)	434 -> 382 ms (12%)
single struct, depth 4	575 -> 538 ms (6%)	573 -> 537 ms (6%)
single struct, depth 8	875 -> 816 ms (7%)	881 -> 815 ms (8%)

The declined rows are both arms running the identical plan, so they double as a noise-floor
estimate: about 0-3%.

Two things worth noting. The gain comes from complex types being present, not from depth - one
level already captures it, and it flattens out after that. And the native C2R
(spark.comet.exec.columnarToRow.native.enabled) is consistently slower than the JVM one on
nested data here, 0.9X against Spark in several groups, which matches its documented per-batch JNI
cost.

How are these changes tested?

New CometWriteRowViewSuite, 11 tests, registered in both PR build workflows. The bar for each is
that turning the config on changes nothing observable but the plan:

• byte-identical round trip against the same Comet plan with the config off, for hand-written
  primitives/strings/nulls/nested, for fuzz-generated flat and nested schemas, and for a
  deliberately deep four-level schema with nulls at every level
• the same for ORC and JSON, not just Parquet
• maxRecordsPerFile set and unset
• the transition is present exactly once when expected, absent by default, absent for a flat
  schema, absent for partitioned and for bucketed writes, and present for a complex column sitting
  alongside flat ones

Also ran CometParquetWriterSuite and CometNativeColumnarToRowSuite alongside it: 67 tests pass.

Caveats I would want a second opinion on

1. Is 6-15% on complex-typed writes worth a reused-mutable-row in the plan at all? The gates are
   argued from Spark's source rather than enforced by anything, and a future Spark change to a
   writer that starts retaining rows would be silent data corruption rather than a test failure. A
   defensive option would be to restrict it to ParquetFileFormat only.
2. The format gate is a package-name prefix check, which is crude. I could not find a better
   signal for "this OutputWriter does not retain the row".
3. Only exercised on Spark 4.1 so far.

[comphead]

interesting idea, would it also work Iceberg writers?

[andygrove]

Not as written, and I think that is the right default for now.

The rule only matches WriteFilesExec and DataWritingCommandExec(InsertIntoHadoopFsRelationCommand), and then gates on the FileFormat class name being under org.apache.spark.sql.execution.datasources.. Iceberg writes are DSv2, so they land on AppendDataExec / OverwriteByExpressionExec / WriteDeltaExec (V2TableWriteExec) instead, and neither arm ever fires. The format prefix check was written precisely to exclude third-party writers, so Iceberg is the case it is keeping out rather than one it happens to miss.

Whether it could work is a more interesting question, and structurally the plain append path looks like it would:

• Iceberg's writer API is row-typed the same way Spark's is: DataWriter<InternalRow>, and the append task in Spark is DataWritingSparkTask, which is just writer.writeAll(iter). Nothing between the iterator and the writer copies or buffers.
• On the Iceberg side an unpartitioned append goes RollingDataWriter -> SparkFileWriterFactory -> SparkParquetWriters, which encodes field by field on the spot.

But the safety argument would have to be redone against a much bigger writer set than Spark's, and it is a set we do not control the version of:

• partitioned tables use the clustered/fanout writers, which key rows into maps
• MERGE/UPDATE go through the position-delta and equality-delete writers via DeltaWritingSparkTask, and the equality-delete path keys inserted rows into a StructLikeMap
• Iceberg's default distribution mode puts a shuffle and a local sort between the scan and the write for partitioned or sorted tables, and both of those need UnsafeRow anyway, so those plans would not qualify even if the writer did

The part that bothers me is that for Spark's own writers the gate is at least argued against source that moves in lockstep with the Spark version we compile against. For Iceberg it would be argued against a dependency on its own release cadence, where a writer that starts retaining a row is a silent-corruption bug for us and a perfectly reasonable change for them. If this lands at all, that pushes me further toward caveat 1 in the description: an explicit allowlist of known-safe consumers rather than a package prefix, and Iceberg only added to it with its own round of round-trip tests.

So: worth a follow-up issue rather than scope for this PR, and only after the core mechanism has convinced people it is worth having.