Skip to content

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

Draft
andygrove wants to merge 2 commits into
apache:mainfrom
andygrove:comet-write-files-seam
Draft

refactor: hook native Parquet writes into Spark's WriteFilesExec seam#5293
andygrove wants to merge 2 commits into
apache:mainfrom
andygrove:comet-write-files-seam

Conversation

@andygrove

@andygrove andygrove commented Aug 7, 2026

Copy link
Copy Markdown
Member

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
refactor: hook native Parquet writes into Spark's WriteFilesExec seam
0f946b5 
Native Parquet writes previously replaced the whole DataWritingCommandExec,
so InsertIntoHadoopFsRelationCommand.run never ran and Comet had to re-implement
the write framework itself: a hardcoded SQLHadoopMapReduceCommitProtocol, a
hand-ported copy of the SaveMode logic, its own commit-message accumulator, and
its own commitJob call. Most of the open native-writer bugs are symptoms of that
one decision rather than independent defects.

Replace only WriteFilesExec instead. On Spark 4.0+ V1WritesUtils.getWriteFilesOpt
matches the WriteFilesExecBase trait, so extending it is what makes Spark route
the write through CometWriteFilesExec.doExecuteWrite while keeping ownership of
the commit protocol, staging, _SUCCESS, dynamic partition overwrite, stats
tracker aggregation, and catalog refresh. Comet is left owning just the per-task
native encode.

Gated to Spark 4.0+: Spark 3.x matches the concrete WriteFilesExec case class,
so a Comet node would be silently ignored and the write would drop into
FileFormatWriter's non-planned, row-based branch.

Other behaviour changes that fall out of the new seam:

- 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 work. Previously only the parent directory 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 name.
- Byte and row counts come from BasicWriteTaskStatsTracker, which stats files
  through the FileSystem API and so is correct for HDFS. The native writer's
  std::fs::metadata call reported 0 there. CometMetricNode's now-redundant
  reportNativeWriteOutputMetrics is removed.
- AQE re-plans the write command's child and re-inserts a WriteFilesExec above
  the already-converted node; collapse it rather than nesting two native writes,
  which would write the data twice.

Deletes CometNativeWriteExec and CometDataWritingCommand (594 lines).

The operator opt-in moves to spark.comet.operator.WriteFilesExec.allowIncompatible,
with the previous DataWritingCommandExec key kept as an alternative.
@andygrove
refactor: address review feedback on the native write path
bcfef4b 
Reuse:
- Use CometExec.serializeNativePlan and CometExec.getCometIterator instead of
  hand-rolling protobuf serialization and the CometExecIterator constructor.
- Use CometArrowStream.countingIterator instead of a bespoke counting map.

Altitude:
- originalPlan now points at the WriteFilesExec rather than the child. CometExecRule
  copies originalPlan's logical link onto every CometExec, so pointing it at the child
  linked the write node to the child's logical plan; AQE then mistook it for the child's
  query stage and re-wrapped it in a second WriteFilesExec. That was the actual cause of
  the nested-write problem, so the collapse guard in CometExecRule is no longer needed
  and is removed. This also fixes RevertNativeForTransitionHeavyStages, which reverts a
  CometExec via originalPlan.withNewChildren and would otherwise have dropped the write.
- DataWritingCommandExec joins the never-replaced list. It is deliberately left in the
  plan for a fully native write, so tagging it reported an accelerated write as a
  fallback in extended explain and skewed the accelerated-operator count.
- Delete the ColumnarToRowExec(CometWriteFilesExec) arm in EliminateRedundantTransitions.
  Spark's ensureOutputsRowBased already special-cases a DataWritingCommandExec over a
  V1WriteCommand with a columnar child under plannedWrite, so no transition is ever
  inserted and the arm was dead.
- Move the Spark 4.0+ gate from CometExecRule into CometWriteFiles.getSupportLevel,
  where every other operator's version and shape gating lives. This also gives users on
  3.x a reason whether or not they opted in.

Efficiency:
- The write task closure no longer captures the exec node. Per-task work moves to
  object CometWriteFilesExec, taking a NativeWriteTask container built on the driver, so
  the task binary no longer carries nativeOp plus the whole converted child subtree
  (each node of which holds its own non-transient protobuf). Mirrors how Spark's
  WriteFilesExec delegates to the static FileFormatWriter.executeTask, and hoists
  CometMetricNode.fromCometPlan to the driver as CometNativeExec already does.
- recordRows loops per tracker on the outside, so the per-row inner loop has one
  receiver and no per-row closure allocation.

Simplification:
- createOperatorIncompatConfig takes Option[String]; ConfigBuilder mutates in place, so
  the Seq destructuring was rebinding the same object.
- Flatten the writeFile/writtenFile Option, which only fed a debug message.

Also fixes the scalafix CI failure: FileCommitProtocol was imported only for a scaladoc
link, which RemoveUnused does not count as a use.

Adds a regression assertion that a fully native write leaves no fallback reason on
DataWritingCommandExec.
This was referenced Aug 7, 2026
Open
Open
Open
@andygrove

andygrove commented Aug 7, 2026

Copy link
Copy Markdown
Member Author

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.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

1 participant

@andygrove