refactor: hook native Parquet writes into Spark's WriteFilesExec seam

[andygrove]

Which issue does this PR close?

Part of #2967 and #1625. Restructures the native write path so the following can be
fixed at all, and closes the ones that were purely symptoms of the old design:

Closes #2985 (no _SUCCESS file)
Closes #3521 (INSERT INTO ... SELECT invisible to subsequent reads)
Closes #3426 (complex type with different names)

Unblocks (not fixed here, but no longer require re-implementing Spark's write
framework inside Comet): #2957, #2970, #3015, #3041, #3193, #3194, #3417, #3428.

Rationale for this change

Native writes replaced the whole DataWritingCommandExec, which means
InsertIntoHadoopFsRelationCommand.run never ran. Everything that method does had to
be re-implemented inside CometNativeWriteExec: a hardcoded
SQLHadoopMapReduceCommitProtocol (so spark.sql.sources.commitProtocolClass was
ignored), dynamicPartitionOverwrite pinned to false, a hand-ported copy of the
SaveMode logic, a bespoke commit-message accumulator, and its own commitJob call.

Most of the open native-writer issues are symptoms of that one decision rather than
independent defects. Fixing them one at a time against the old design means writing a
second, worse FileFormatWriter inside Comet.

Spark already has the right seam. On Spark 4.0+, V1WritesUtils.getWriteFilesOpt
matches the WriteFilesExecBase trait (introduced in 4.0 precisely for this), so a
Comet node that extends it gets driven through
FileFormatWriter.executeWrite → SparkPlan.executeWrite → doExecuteWrite, and
Spark keeps ownership of everything above the per-task write.

Why Spark 4.0+ only. On 3.4/3.5 getWriteFilesOpt matches the concrete
WriteFilesExec case class. A Comet node there would not be found, writeFilesOpt
would be None, and Spark would silently take FileFormatWriter's non-planned,
row-based branch — ignoring doExecuteWrite entirely. The only way in on 3.x is to
inherit from a case class, which brings copy/equals hazards; that isn't worth
carrying, so native writes now require 4.0+ and report a fallback reason on 3.x.

What changes are included in this PR?

Execute InsertIntoHadoopFsRelationCommand   <- Spark: SaveMode, catalog, commitJob, _SUCCESS
+- CometWriteFiles                          <- Comet: native per-task write only
   +- CometNativeScan ...

• New CometWriteFilesExec overriding doExecuteWrite, mirroring
  FileFormatWriter.executeTask for the parts Comet must do itself: build the
  TaskAttemptContext, ask the commit protocol for a path, run the native writer,
  drive the stats trackers, commit or abort. Plus the CometWriteFiles serde and a
  two-line ShimCometWriteFilesExec in spark-4.x / spark-3.x.
• Deleted CometNativeWriteExec.scala and CometDataWritingCommand.scala (594
  lines). Net −549 lines.
• File paths come from FileCommitProtocol.newTaskTempFile and are used verbatim,
  so names match Spark's part-<id>-<uuid>-c000.<codec>.parquet and committers that
  track individual files (S3A magic, streaming manifest) work. Previously only
  .getParent was kept and the native writer invented its own names.
• Column names come from WriteJobDescription.dataColumns rather than the query
  output, so INSERT INTO t SELECT a+1 writes the target column's name ([COMET NATIVE WRITER] INSERT INTO TABLE - complex type but different names #3426).
• Byte/row counts come from BasicWriteTaskStatsTracker, which stats files through the
  FileSystem API and is therefore correct on HDFS. The native writer's
  std::fs::metadata call reported 0 there. CometMetricNode's now-redundant
  reportNativeWriteOutputMetrics is removed.
• Proto: work_dir / job_id / task_attempt_id removed (reserved); output_path is
  now the exact file to write, set per task.
• Opt-in moves to spark.comet.operator.WriteFilesExec.allowIncompatible, with the old
  DataWritingCommandExec key kept via withAlternative.

Fixed along the way

AQE re-plans the write command's child and re-inserts a WriteFilesExec above the
node Comet already converted. Without a guard that produced nested native writes — the
data written twice, and the inner node's empty output read as a zero-column schema.
CometExecRule now collapses the redundant node. The
basic parquet write with repartition test catches this.

How are these changes tested?

• CometParquetWriterSuite: 33/33 on Spark 4.1 — the 30 existing tests plus three
  new regression tests for _SUCCESS (Comet writer doesn't create _SUCCESS file #2985), Spark-compatible file naming, and
  INSERT INTO ... SELECT visibility ([Native Writer] INSERT INTO ... SELECT fails due to stale catalog cache after write #3521). Suite-level assume(isSpark40Plus).
• CometTaskMetricsSuite: 6/6. Note this suite's native-write test was pinned to the
  old operator's config key, so it would have silently fallen back to Spark's writer
  and still passed; it is repointed and now genuinely exercises the native path.
• Regression sweep: CometExecSuite (143), CometExpressionSuite +
  CometAggregateSuite + CometFuzzTestSuite (273) — all pass.
• Compiles against Spark 3.4, 3.5, 4.0 and 4.1. cargo check and the native
  parquet_writer unit tests pass.

Known limitation

WriteTaskStatsTracker.newRow(filePath, row) is a per-row callback. Comet has columnar
batches, so rather than materializing every row just to hand it straight back,
recordRows passes InternalRow.empty and feeds only the count. That is exactly right
for BasicWriteTaskStatsTracker, which ignores the row argument, but a third-party
tracker inspecting row contents would see empty rows — so that case logs a warning
rather than silently reporting wrong statistics. A plan-time guard isn't possible
because statsTrackers only exists at execution time.

Follow-ups

Independent of this change and the next highest-value work, since the Spark default is
affected: full WriterProperties (block/page size, dictionary, writer version), INT96
timestamps (#3425 — spark.sql.parquet.outputTimestampType defaults to INT96 and we
write INT64 micros), the four footer metadata keys (#3427 — legacyINT96 and timeZone
drive rebase decisions on read, so omitting them is a correctness risk), field IDs, and
Catalyst nullability. Then partitioned (#3193) → bucketed (#3194) → object stores.

[andygrove]

Relationship to #4658

@jordepic you may be interested in this one. It reaches the same shape as #4658 from the opposite direction: both separate per-task data-file writing from the commit so the write runs inside AQE. #4658 had to build that split because Spark's Iceberg V2 write is one operator; for V1 Parquet Spark already provides it (DataWritingCommandExec over WriteFilesExec), so this PR mostly deletes Comet's hand-rolled commit code.

Relevant to writing Iceberg data files natively via iceberg-rust: that would inherit this writer's output, which is not yet Spark-comparable — INT96 (#3425), footer metadata keys (#3427), Catalyst nullability and field IDs (#5305), writer properties (#5304).

Also borrowed from #4658: IcebergWriteExec has the SPARK-23271 zero-partition guard and CometWriteFilesExec does not — filed as #5303.

No merge conflicts with apache/main. One coordination item: spark.comet.write.iceberg.* vs spark.comet.parquet.write.* are two namespaces for one feature family (#5306), cheap to settle while both are still experimental and off by default.

[comphead]

we prob need to include a test with dynamic overwrite.
better to fallback to spark for dynamicPartitionOverwrite

[andygrove]

Added in 43bce31: a dynamic-overwrite test that asserts the untouched partitions survive, alongside tests for the abort-and-retry path (injected failing commit protocol), the maxRecordsPerFile fallback (both the write option and the conf, verifying Spark's writer rolls 10 files), and the schema-only empty-input write.

On the fallback itself: dynamicPartitionOverwrite is only set when staticPartitions.size < partitionColumns.length, which implies partition columns, so it always lands on the existing partitionColumns.nonEmpty || staticPartitions.nonEmpty guard and declines.

[comphead]

@andygrove wondering if you can trigger https://github.com/apache/datafusion-comet/actions/workflows/spark_sql_writer_tests.yml for your local branch?

[comphead]

so Spark 3.x falls back to JVM writes?

[andygrove]

Yes. On 3.x getSupportLevel returns Unsupported("Native Parquet writes require Spark 4.0 or later") and the write runs through Spark's own writer unchanged. It is reported as a fallback reason so users on 3.x can see why, rather than silently getting a different path.

[comphead]

we need also fallback for dynamicPartitionOverwrite

[andygrove]

Covered in 43bce31. InsertIntoHadoopFsRelationCommand only sets dynamicPartitionOverwrite when staticPartitions.size < partitionColumns.length, which implies partition columns, so a dynamic overwrite always hits the existing partitionColumns.nonEmpty || staticPartitions.nonEmpty guard and is declined. I made that explicit in a comment rather than adding a redundant condition.

[comphead]

P1 — No test for task failure / commit abort path

Description: CometWriteFilesExec.executeTask wraps the native write in try { ... } catch { case t: Throwable => committer.abortTask(...); throw t } (spark/src/main/scala/org/apache/spark/sql/comet/CometWriteFilesExec.scala). No test forces a failure inside the native writer to exercise the abort branch.
This leaves the entire failure lifecycle uncovered: whether abortTask is actually called, whether the staging file for the failing attempt is cleaned up, whether the exception surfaces unwrapped, and whether a subsequent retry produces correct output.

Reason: The try/catch is new code that was previously handled by an ad-hoc handleTaskEnd block. Without coverage a broken abortTask would ship silently.

Example tests:

• Write to a read-only local path (chmod 555 outputPath) and assert (i) an IOException/SparkException propagates, (ii) _temporary under outputPath is empty, (iii) the output directory contains no part-*.parquet files.
• Include a raise_error('boom') expression in the projected columns and assert the write fails and leaves nothing behind. Then remove the expression and assert the retry succeeds and produces correct data.

P2 — Zero-partition RDD no longer produces the SPARK-23271 schema-only metadata file

Description: Spark's WriteFilesExec.doExecuteWrite (Spark 4.0 WriteFiles.scala:80-92) inserts a dummy single-partition RDD when rdd.partitions.length == 0 so a schema-only Parquet file is written for downstream readers (SPARK-23271). CometWriteFilesExec.doExecuteWrite calls child.executeColumnar()
and immediately does childRDD.mapPartitionsInternal { ... } with no such fallback.

Reason: If the child produces zero partitions, no task is spawned and no schema-only file is written, so spark.read.parquet(<empty-write>) sees no schema. This diverges from Spark 4.0's own writer path.

P2 — description.maxRecordsPerFile is silently ignored

Description: Spark's SingleDirectoryDataWriter.write (FileFormatDataWriter.scala:185-197) rolls a new file every description.maxRecordsPerFile rows and increments the -c$fileCounter%03d suffix. CometWriteFilesExec.executeTask reads no maxRecordsPerFile, hard-codes -c000, and calls
newTaskTempFile once per task. Neither CometWriteFiles.getSupportLevel nor executeTask inspects the field.

Reason: Any user with spark.sql.files.maxRecordsPerFile>0 gets a different file layout than Spark's own writer with no warning or fallback, quietly violating the contract even though the write itself is only labeled Incompatible for other reasons.

Interesting finding on spark.sql.files.maxRecordsPerFile we should fallback if it is set

[comphead]

Another thing, would this PR supersede #5143 ?

[andygrove]

Another thing, would this PR supersede #5143 ?

I think you'd need to rebase 5143 once this merges, or we can merge that one first and then rebase this PR

[andygrove]

Thanks for the review @comphead. I have addresses feedback so far.

[comphead]

Thanks @andygrove the improvement makes sense to me, however we may want to wait with merge as this would block writer testing on top of Spark3. Maybe this PR can be merged closer to 1.1.0 release where Comet rescinds the support for Spark 3.4.3

[peterxcli]

Is it possible to use git rename for (DataWritingCommand -> WriteFiles) and (NativeWriteExec -> WriteFilesExec) instead? because the renamed looks similar.
IIRC, git can still detect this as a move as long as git diff sees that more than 50% of the file content is the same.
Perhaps we could still view the diff between the original file and renamed file on the review page.

[peterxcli]

IIUC, we can just remove them, as the native parquet writer is experimental, and it's merely impossible to have different version of comet nodes in same cluster.

[peterxcli]

also the number can be rearrange to be continuous

[andygrove]

Done in 551c604 - dropped the reserved block and renumbered so the tags are contiguous:

message ParquetWriter {
  string output_path = 1;
  CompressionCodec compression = 2;
  repeated string column_names = 3;
  map<string, string> object_store_options = 4;
}

Agreed on the reasoning: plans are serialized on the driver and deserialized on executors within a single jar version and are never persisted, so there is no wire-compatibility surface the reserved tags were protecting. Generated accessors key off field names rather than tags, so no Rust or Scala call site changed. cargo check and a full build pass, and CometParquetWriterSuite is 37/37 on Spark 4.1.

One consistency note for a possible follow-up: HashAggregate still carries reserved 3, 8 from #4507 for the same removed-field situation, so the file now does this two ways.

[andygrove]

@peterxcli on the rename question - I looked at whether this is achievable and it isn't, for two reasons.

Git doesn't record renames in commits. It stores trees, and rename detection is a diff-time heuristic run over file content. Doing git mv first and editing afterwards produces a byte-identical tree and therefore a byte-identical diff, so it wouldn't change what GitHub renders.

The content is also well under the threshold you mention. Measured similarity index:

$ git diff -M05% --summary main...comet-write-files-seam -- spark/src/main/scala
rename .../serde/operator/{CometDataWritingCommand.scala => CometWriteFiles.scala} (27%)
rename .../sql/comet/{CometNativeWriteExec.scala => CometWriteFilesExec.scala} (14%)

27% and 14%, against git's 50% default. The exec file in particular is a rewrite rather than a move: the old one owned the commit protocol, the SaveMode logic and commitJob, and the new one only does the per-task write.

What does work is lowering the threshold locally, which pairs both files and gives the side-by-side view you're after:

git diff -M10% main...comet-write-files-seam -- spark/src/main/scala

Your proto suggestion is applied in 551c604.

[andygrove]

Thanks @andygrove the improvement makes sense to me, however we may want to wait with merge as this would block writer testing on top of Spark3. Maybe this PR can be merged closer to 1.1.0 release where Comet rescinds the support for Spark 3.4.3

There is already a PR open to drop support for Spark 3.4, so I think this is going to happen pretty quickly.

[comphead]

Related to #5358

[sunchao]

Thanks for moving job-level write handling back into Spark. I found one compatibility issue with the legacy native-write opt-in. The inline comment includes the behavior reproduced against Spark 4.0.4.

[sunchao]

[P2] Could the planner read the registered COMET_OPERATOR_WRITE_FILES_ALLOW_INCOMPAT entry? CometExecRule.isOperatorEnabled calls isOperatorAllowIncompat("WriteFilesExec"), which reads only the new key through SQLConf.getConfString. That bypasses the alternative declared here.

On Spark 4, an existing job with spark.comet.parquet.write.enabled=true and only spark.comet.operator.DataWritingCommandExec.allowIncompatible=true now falls back to Spark's writer. I reproduced this with the exact PR's CometConf and Spark 4.0.4: the registered config entry returns true, while the planner's check returns false.

Could you preserve the old-key-only behavior and add a regression test that also checks that an explicitly set new key takes precedence?

[sunchao]

LGTM. So the plan is to merge this once we drop the support of Spark 3.4?

[andygrove]

Closing in favour of #5763, which makes this change purely additive.

@comphead's concern was the one that mattered: this PR deleted the Spark 3.x native
writer, so native writes became 4.0+ only and writer testing on Spark 3 (the 3.5
option in the manual Spark SQL Writer Tests workflow) would have had nothing to test
until 3.4/3.5 support is dropped. Rather than park the change until then, #5763 keeps
CometDataWritingCommand and CometNativeWriteExec exactly as they are on main and
adds the WriteFilesExec seam alongside them. CometExecRule picks the path by Spark
version and the two never both fire. Nothing is deleted, and the 3.x path goes away
with 3.x support rather than ahead of it.

Every review item here is carried over:

• P1, no test for the task failure / commit abort path — carried over: a failing task aborts, cleans up its staging file, and the retry succeeds, using a commit
  protocol that fails commitTask on demand so the staging file already exists and the
  cleanup is observable.
• P2, zero-partition RDD and SPARK-23271 — carried over: the dummy single-partition
  swap in doExecuteWrite, plus the empty input still writes a schema-only file test.
• P2, maxRecordsPerFile silently ignored — carried over: rollsFilesByRecordCount
  declines the write, with a test covering both the write option and the conf and
  asserting Spark's writer rolls the expected 10 files.
• Dynamic partition overwrite — carried over: the fallback test that asserts the
  untouched partitions survive.
• @sunchao's P2 on the opt-in key — fixed properly rather than only at the
  ConfigEntry. CometConf.isOperatorAllowIncompat now resolves the deprecated
  alternative, which is what the planner's by-name SQLConf lookup was bypassing. The
  regression test covers both halves you asked for: old key alone still enables native
  writes on Spark 4, and an explicitly set new key takes precedence.
• @peterxcli on the reserved proto tags — moot in feat: hook native Parquet writes into Spark's WriteFilesExec seam on Spark 4.0+ #5763. Nothing is removed from
  ParquetWriter, so there is no reserved block to drop. work_dir simply becomes
  genuinely optional and acts as the discriminator: set (3.x) means the native writer
  derives the file name as before, unset (4.0+) means output_path is the exact file
  and is used verbatim. output_path was already unused on the 3.x native path, so no
  field changes meaning for an existing plan.
• @peterxcli on git rename detection — also moot, and this is the nicest consequence
  of the additive approach: the old and new files coexist, so the diff is additions
  rather than a pair of low-similarity renames that GitHub renders as unrelated files.

One thing worth flagging for #5143: it adds an elapsed_compute metric to
CometNativeWriteExec. That file is untouched by #5763, so there is no conflict there,
but whichever lands second should add the same metric to CometWriteFilesExec or Spark
4 loses it. The parquet_writer.rs changes in the two PRs are in different regions but
close enough that a rebase will want a look.