feat: hook native Parquet writes into Spark's WriteFilesExec seam on Spark 4.0+

[andygrove]

Which issue does this PR close?

Part of #2967 and #1625. Restructures the native write path on Spark 4.0+ 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.

Supersedes #5293, which made the same change but removed the Spark 3.x writer along
the way. That regression is what held #5293 back, so this version is purely additive:
Spark 3.4/3.5 keep the existing native writer, unchanged.

Rationale for this change

Native writes replace the whole DataWritingCommandExec, which means
InsertIntoHadoopFsRelationCommand.run never runs. Everything that method does has to
be re-implemented inside CometNativeWriteExec: a hardcoded
SQLHadoopMapReduceCommitProtocol (so spark.sql.sources.commitProtocolClass is
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 4.0 added the right seam. V1WritesUtils.getWriteFilesOpt matches the
WriteFilesExecBase trait there (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 this is additive rather than a replacement. 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. So CometDataWritingCommand and CometNativeWriteExec stay
exactly as they are and remain the 3.4/3.5 path. CometExecRule picks the path by
Spark version and the two never both fire. The 3.x path goes away with 3.x support.

What changes are included in this PR?

Spark 4.0+:

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

Spark 3.4/3.5 is unchanged:

CometNativeWrite                            <- Comet: the whole write
+- 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.
• Nothing is deleted. CometNativeWriteExec, CometDataWritingCommand,
  CometMetricNode.reportNativeWriteOutputMetrics and the
  EliminateRedundantTransitions rule for native writes all remain and serve 3.x.
• 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. The 3.x writer keeps
  inventing its own names.
• The output path is re-escaped through Path before being parsed as a URI. Both
  serdes receive it as Path.toString, which decodes percent escapes, so a directory
  containing a space or a literal % produced a string URI.create rejects - a query
  failure on 4.0+, and a silent fallback to Spark's writer on 3.x. Covered by a
  version-independent test.
• HDFS destinations whose path the native writer would spell differently are declined at
  planning, in both serdes. create_hdfs_object_store hands the Rust-escaped
  url.path() to object_store::path::Path::parse, so the native writer would create
  dir%20with%20space or caf%C3%A9 while Spark's committer commits the unescaped
  name, and job commit would succeed with the data somewhere else. The gate covers the
  destination directory and the effective mapreduce.output.basename, which reaches
  every committed file name: a basename holding ? or # is truncated by the native
  URL parser, so every task would write the same name and they would overwrite each
  other at commit. The basename is checked by running the same predicate over the path
  it produces rather than over the name alone, so it also catches a literal % and the
  other characters java.net.URI escapes while the url crate keeps them ([, ^,
  |). A matching check runs again in executeTask against the actual
  newTaskTempFile path, which a custom commit protocol owns and planning cannot
  predict; that one fails the task rather than falling back. Sharing one predicate is
  what keeps planning from admitting a name the task guard then aborts on, and the test
  asserts that the two agree for every basename it covers. The character set the JVM
  guard mirrors is pinned by a Rust test, so a url crate upgrade cannot reopen the
  hole with a green build.
• Column names, nullability and Parquet field IDs come from
  WriteJobDescription.dataColumns rather than the query output. Spark's
  castAndRenameQueryOutput normally makes the two agree, so this is not by itself what
  fixes [COMET NATIVE WRITER] INSERT INTO TABLE - complex type but different names #3426 - see the test note below - but dataColumns is what Spark treats as
  authoritative and the only source that stays correct once partitioned writes land,
  since FileFormatWriter excludes the partition columns from it.
• The compression codec is resolved case-insensitively, matching Spark's
  ParquetOptions, and the shared helpers live in NativeWriteUtils so the Spark 3.x
  serde gets the same fix. On 4.0+ the writer's codec is additionally re-derived per
  task from CodecConfig.from(taskAttemptContext) - the same CodecConfig the file
  extension comes from - so a file's name and its contents cannot disagree.
• 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 reports 0 there, so CometWriteFiles publishes no
  files_written/bytes_written/rows_written of its own on this path; the numbers on
  the enclosing command are the authoritative ones.
• executeTask and the native write loop use Spark's
  Utils.tryWithSafeFinallyAndFailureCallbacks / tryWithSafeFinally, as
  FileFormatWriter.executeTask does, so a failure while aborting the task or closing
  the Comet iterator is attached to the original failure as a suppressed exception
  rather than replacing it. The stats trackers are constructed inside that guard, so a
  throwing newTaskInstance still reaches abortTask.
• Proto: ParquetWriter.work_dir becomes genuinely optional. When it is set (3.x) the
  native writer derives the file name from it as before; when it is unset (4.0+),
  output_path is the exact file to write and is used verbatim. output_path was
  already unused on the 3.x path, so no field changes meaning for an existing plan.
• On 4.0+ the opt-in moves to spark.comet.operator.WriteFilesExec.allowIncompatible,
  with the old DataWritingCommandExec key kept as a deprecated alternative.
  CometConf.isOperatorAllowIncompat now resolves alternatives; the planner's by-name
  lookup previously bypassed the ConfigEntry, so an old key would have read true
  from the entry while the planner saw false.
• WriteFilesExec declines dynamic partition overwrite (it is always a partitioned
  write) and spark.sql.files.maxRecordsPerFile, which Spark's own writer uses to roll
  a new file every N rows.

Fixed along the way

AQE re-plans the write command's child and re-inserts a WriteFilesExec above the
node Comet already converted. On the 3.x path that needs an explicit guard in
CometExecRule (still present). Leaving DataWritingCommandExec in place on 4.0+
means the situation cannot arise there.

How are these changes tested?

• CometParquetWriterSuite: 49/49 on Spark 4.0, 4.1 and 4.2 - the 33 existing tests
  plus 16 new ones. Version-gated to 4.0+: _SUCCESS (Comet writer doesn't create _SUCCESS file #2985), Spark-compatible file
  naming, a custom mapreduce.output.basename being honored on local storage, INSERT INTO ... SELECT visibility ([Native Writer] INSERT INTO ... SELECT fails due to stale catalog cache after write #3521), INSERT INTO ... SELECT writing the target's
  column names, dynamic-overwrite fallback, the maxRecordsPerFile fallback (both the
  write option and the conf, verifying Spark's writer rolls 10 files), the schema-only
  empty-input write (SPARK-23271), an empty non-zero partition writing no file and still
  committing, task abort through an injected failing commit protocol, and the deprecated
  opt-in key. Version-independent: output paths needing URI escaping, the HDFS
  destination and basename guards plus the execution-time check and the assertion that
  the two agree, a mixed-case compression option being honored, a mixed-case
  unsupported codec falling back, and the third-party WriteTaskStatsTracker warning.
• CometParquetWriterSuite on Spark 3.4 (36/36 + 13 cancelled) and 3.5 (38/38 +
  11 cancelled): every pre-existing test still passes on the 3.x writer, which is the
  point of keeping it.
• CometEmptyRelationParquetWriterSuite (1/1 on 4.0, 4.1 and 4.2), which landed in
  feat: support Spark 4 EmptyRelationExec as a native input #5821 while this branch was open. A native empty relation is a zero-partition RDD,
  which is exactly why CometDataWritingCommand declines it on 3.x: that writer only
  maps existing partitions, so no task runs and no file carries the schema. On the 4.0+
  seam CometWriteFilesExec swaps in a dummy single-partition RDD, as Spark's own
  WriteFilesExec does, so partition 0 still writes the schema-only file. The suite's
  readback assertions pass unchanged; it now asserts a native write when the relation is
  native and a fallback when it is not.
• CometTaskMetricsSuite: 15/15 on both 3.5 and 4.1. The suite's native-write test now
  picks the version-appropriate opt-in key, so it genuinely exercises the native path on
  both. It also confirms that dropping the node's own write metrics costs nothing: its
  outputMetrics come from Spark's tracker.
• Regression sweep on 4.1: CometExecSuite (144), CometFallbackInvarianceSuite (6),
  CometPublicApiSuite (1).
• Native: a parquet_writer unit test asserting the writer uses a
  commit-protocol-chosen path verbatim (and that a non-zero partition id does not leak
  into the name), and url_path_rewritten_characters, which pins the exact set of
  characters the url crate rewrites inside a path so the JVM guard that mirrors it
  cannot silently drift. cargo test -p datafusion-comet parquet_writer,
  cargo test -p datafusion-comet url_path_rewritten_characters and
  cargo clippy --all-targets --workspace -- -D warnings pass.
• Compiles and test-compiles against Spark 3.4, 3.5, 4.0, 4.1 and 4.2.

A note on #3426, because the test came out differently from what I expected. The
scenario - Spark's own INSERT INTO TABLE - complex type but different names, with a
top-level rename added - passes on 4.0+ and fails on 3.5, but it fails on 3.5 by
returning no rows at all (#3521), not by writing the wrong names. Reverting
description.dataColumns to child.output does not change the outcome on 4.0+, because
Spark's castAndRenameQueryOutput has already aliased the query's output to the
target's names and cast the struct to the target's nested field names before Comet sees
anything. So what closes #3426 is leaving Spark's command in the plan, not the
dataColumns line; the line is still the correct source to read, for the reasons above.

Not covered, and not claimed: an end-to-end HDFS write, speculative concurrent attempts,
and any measurement of recordRows.

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), and the four footer metadata keys (#3427: legacyINT96 and
timeZone drive rebase decisions on read, so omitting them is a correctness risk).
Then partitioned (#3193) → bucketed (#3194) → object stores.

[rich7420]

Could we use new Path(outputPath).toUri here? Paths containing spaces or a literal % currently fail in URI.create, while the base successfully falls back to Spark. I tested this change and both cases write natively. Could you also add regression tests for these paths?

[andygrove]

Done — switched to new Path(outputPath).toUri, and made the same change in CometDataWritingCommand, which had the same URI.create call on the 3.x path. There the serde's catch turned it into a silent fallback rather than a failure, so the native writer was simply never used for those paths.

The regression test covers a directory with spaces, one with a literal %, and one with both, and it isn't version-gated so it runs against both writers. I checked it fails without the fix: IllegalArgumentException: Illegal character in path on 4.1, and zero native write operators in the plan on 3.5.

[rich7420]

@andygrove thanks for the patch!

[sunchao]

Correctness

Reviewed df968953 against 7f1e0018. The prior writer replaced the entire DataWritingCommandExec, bypassing Spark's normal insertion, job-commit and catalog-refresh lifecycle. On Spark 4.0+, the new WriteFilesExecBase implementation leaves that lifecycle with Spark and supplies the per-task native write through doExecuteWrite. I compared this with the maintained Spark 3.5 and 4.0 branches: 3.5 discovers the concrete WriteFilesExec, whereas 4.0 discovers the base trait. Retaining the old path on 3.4/3.5 is therefore justified.

The task path follows Spark's job/task/attempt identifiers, requests the committer's exact filename, finishes the native writer before committing, and aborts/rethrows on write or commit failure. Partition zero retains a schema-only file for empty input; other empty partitions produce no file. Target names, nested types, nullability and field IDs come from WriteJobDescription.dataColumns. Partitioned/bucketed writes remain excluded, and the record-count rollover guard matches Spark's option-over-configuration precedence. The deprecated opt-in is respected, with an explicitly set new key taking precedence.

[P2] Preserve Hadoop path escaping

The existing path-parsing finding remains unresolved. The output-path tag contains Hadoop Path.toString, and URI.create(outputPath) throws for spaces or a literal %. The new serde has no enclosing fallback catch; the base catches that conversion error and retains Spark's writer. A local Java 17/Hadoop component check reproduced the exception for both local and HDFS path strings, with ordinary paths as controls; Path.toUri handled all six cases. This supports the existing comment, so I have not added a duplicate inline. This local check exercised path conversion only, not a complete Spark write.

Validation and limits

The inspected CI jobs checked out merge 2e35d315, whose parents are the reviewed base/head and whose entire tree equals the reviewed head. CometParquetWriterSuite passed 41/41 on Spark 4.0 and 41/41 on Spark 4.1; Spark 3.5 passed 33 with eight version-gated cancellations, and 3.4 passed 32 with nine cancellations. Spark 4.0's metrics suite passed 15/15, including native-write output metrics. The Rust job passed 1,184 tests, including exact-path writing, with four skipped. The JVM consumers downloaded the same native artifact digest uploaded after the producer rebuilt this source.

At the reviewed snapshot, 63 checks passed, nine were skipped, and one Iceberg check failed downloading Gradle with HTTP 504 before its tests started. Maintained 3.4, 4.1 and 4.2 source branches were unavailable; CI coverage does not replace those source comparisons. No local Spark/native integration build or benchmark was run. The injected failure test checks abort and a subsequent whole-write retry; concurrent speculative attempts remain untested. The INSERT visibility test verifies readback but does not independently assert the native write plan. I found no additional verified P1/P2 issue in the authored changes.

Performance

The write stays columnar and preserves the existing native Parquet implementation. Resolving NativeWriteTask on the driver avoids capturing the whole execution-plan object in the task closure, and the small writer plan is rebuilt per task because its destination is task-specific. File-size accounting now uses Spark's filesystem-aware tracker, which also supplies the task output counters.

recordRows adds one callback per written row after native writing completes. Keeping the tracker loop outside the row loop limits dispatch overhead, but the work still scales with row count. A focused large-batch, narrow-row write benchmark with BasicWriteTaskStatsTracker would quantify this cost before making a throughput claim. The functional tests establish correctness coverage, not a measured speedup; I have no verified performance blocker.

Design

Using Spark's existing task-write seam gives job commit, SaveMode handling, _SUCCESS, cache refresh and catalog statistics one owner. Passing the exact filename back to native code also removes the old assumption that the committer only cares about a staging directory. The filesystem admission gate still limits this path to local/HDFS output; it does not establish support for object-store committers.

The conservative command/native-child/partition/bucket/rollover guards keep the supported case understandable. The maintained InsertIntoHadoopFsRelationCommand supplies the basic count-only stats tracker, which fits the current implementation. Row-inspecting trackers need a separate compatibility solution before broadening that scope; a warning cannot supply their missing row contents. Experimental opt-in remains appropriate while the disclosed timestamp/footer and Parquet writer-property gaps remain.

Abstraction & complexity

The two small version shims isolate the actual Spark API difference. NativeWriteTask has a clear serialization purpose, and the static task runner keeps executor work separate from driver plan state. Preserving the write node as originalPlan, together with restarting child native-block serialization, also makes the AQE ownership boundary explicit.

The optional work_dir remains the discriminator between the old directory-based path and the new exact-file path without changing protobuf field numbers. Both Scala execution paths are necessary while Spark 3.x is supported; the compatibility branch is localized enough to remove with that support. Beyond the remaining path-parsing correction, I found no actionable abstraction or complexity blocker.

[andygrove]

Path escaping is fixed in d36ad9f, in both serdes — the 3.x one had the same URI.create call, where the enclosing catch turned it into a silent fallback instead of a query failure. The new test isn't version-gated and I confirmed it fails both ways without the fix.

I also took your point on the INSERT visibility test: it now captures the executed plan and asserts a native write, so a fallback can't make the read-back pass for the wrong reason.

On recordRows, agreed there's no measured number here and I'm not claiming one. Speculative concurrent attempts are still untested — the attempt path is Spark's own newTaskTempFile, but I'd rather say that's unverified than imply otherwise.

[sunchao]

Correctness

Rechecked d36ad9ff against 7f1e0018, including the three-file increment from the previous review. The original Java URI-parsing finding is fixed: Spark 3.x uses the command's Path.toUri, and Spark 4.0+ reconstructs the tagged Hadoop path before extracting configuration. The new local-path test checks native-plan presence and readback for spaces, literal percent signs, and their combination. The INSERT visibility test now asserts its native write as well.

[P2] Preserve fallback for escaped HDFS destinations

The Spark 3.x change also admits HDFS paths that previously fell back to Spark, exposing a separate native destination mismatch. With a native child reading local input and a cold HDFS object-store cache, a destination containing dir with space reaches create_hdfs_object_store, which uses Path::parse(url.path()). The URL path contains dir%20with%20space, and that constructor preserves the percent encoding. The native writer passes this different directory through OpenDAL and libhdfs. Spark's committer still owns the original directory containing spaces. This can leave data outside the committer's staging tree while job commit reports success. The same destination mismatch affects the newly admitted Spark 4.0+ case.

Please keep these HDFS destinations on Spark during planning until the native path conversion preserves Hadoop's filename semantics. Fixing the native conversion should cover both cold and warm cache paths, which currently use different path constructors. The new local-filesystem tests cannot detect this HDFS branch. This is a downstream issue exposed by the fallback removal, separate from the fixed URI.create exception.

I verified the path flow against the locked URL, object-store, OpenDAL and hdrs sources. A Java 17/Hadoop 3.4.1 component check passed eight planning cases and confirmed that the committer and traced native paths are distinct. A second check used the real Hadoop FileOutputCommitter on local storage with that destination mismatch injected: both algorithms 1 and 2 created _SUCCESS without moving the misplaced data file into the intended output. These are component checks plus source analysis, not a Spark/native/HDFS end-to-end reproduction.

The maintained Spark 4.0 comparison still supports the unchanged write lifecycle: eligibility declines happen during planning. Spark owns insertion and job commit/abort. Native writing finishes before task commit. Task failures abort and rethrow. Partitioned, bucketed and record-rollover writes remain excluded. An execution-time fallback would be too late once overwrite deletion or task output has occurred.

Validation and limits

Current CI's Rust job passed 1,208 tests with five skipped, including the exact-file-path unit test. It executed merge 45074b00 with the reviewed head and newer main 8e684685, not the authoritative base/head tree. Fourteen of the fifteen contributed files are identical to the reviewed head. The differing planner file has an identical Parquet-writer arm. At the 2026-09-08T17:47:28Z refresh, 40 checks passed, seven were skipped, 19 were running and one was queued. Current JVM writer-suite outcomes have not been independently verified in this review. The previous head's passing writer suites and the author's updated test counts are not current independent validation. Maintained Spark 3.4/4.1/4.2 source gaps remain. No local Spark/native build, HDFS cluster run or speculative-attempt test was performed.

Performance

The URI correction adds one Hadoop Path reconstruction during Spark 4.0+ planning and reuses the existing path on 3.x. It adds no per-batch or per-row work. The task writer and its per-row statistics callbacks are unchanged. The author now explicitly acknowledges that their cost has not been measured. No throughput conclusion follows from this update.

Design

The write framework remains Spark-owned, and the corrected local path reaches the existing task writer without changing commit ordering. The remaining HDFS issue belongs at the boundary between Hadoop filenames and native URL paths. A planning-time restriction is sufficient to preserve the prior safe behavior while that boundary is fixed. Catching errors after native writing starts would not recover files written outside the committer's directory.

Abstraction & complexity

Reusing cmd.outputPath.toUri on 3.x avoids an unnecessary round trip. The Spark 4.0+ conversion stays localized where the string tag is consumed. The by-name captureWritePlan overload reuses the listener setup for INSERT and improves the test without adding production machinery. No additional abstraction change is needed beyond correcting or restricting the HDFS path boundary.

[sunchao]

Correctness

[P2] Keep escaped HDFS destinations on Spark until native path decoding is fixed

This removes the Spark 3.x fallback for HDFS destinations containing spaces, but the native HDFS path is still different from the committer's path. With local native input and a cold HDFS object-store cache, create_hdfs_object_store passes url.path() to object_store::path::Path::parse, retaining dir%20with%20space. ParquetWriter then sends that path through OpenDAL/libhdfs, while Spark commits the original dir with space staging directory. Data can therefore remain outside the committed output even though job commit succeeds. The local-path regression test does not exercise this branch. Please decline such HDFS writes during planning until native path handling preserves Hadoop filenames on both cold and warm cache paths. This is a source-confirmed destination mismatch. A Hadoop committer component probe confirmed the commit consequence, but I did not run an end-to-end HDFS write.

[andygrove]

Fixed in 4095ea6 — declined at planning, as you asked.

You're right that the previous change fixed the wrong half for HDFS. Making URI.create stop throwing removed the accidental fallback, but the native and committer paths still name different directories, so all that did was convert a benign fallback into a silent data-placement bug. A write that commits successfully with the data somewhere else is a worse outcome than not accelerating it.

NativeWriteUtils.escapedHdfsDestination is the new gate, called from getSupportLevel in both serdes — CometWriteFiles for 4.0+ and CometDataWritingCommand for 3.x — since the mismatch is in the native object store and does not care which seam the write came through. It compares uri.getRawPath against uri.getPath, which differ exactly when the path contained something the URI form had to escape, and it is scoped to hdfs: so local writes keep the behaviour the earlier fix gave them.

The fallback reason names the consequence rather than the rule, so the EXPLAIN output is actionable:

HDFS output paths needing URI escaping are not supported: the native writer would write to the escaped path while Spark commits the unescaped one (/dir with space/output.parquet)

The regression test is HDFS output paths needing URI escaping are declined at planning. It exercises the predicate directly rather than through a write, since I have no HDFS to write to here — it covers a space, a literal %, and a nested escaped segment as declined, and asserts the two things that would make the gate too broad: an ordinary hdfs: path still passes, and file: paths with either shape still pass. The existing local-path test still writes natively.

I have not tried to fix the native path handling here. When that lands, this gate is one call to delete, and the reason string points at what has to change.

[sunchao]

Correctness

Rechecked 4095ea6f against 7f1e0018, including the four-file update since d36ad9ff. The new Unsupported checks run during planning in both serdes and cannot be bypassed by allowIncompatible. They fix the reported space/percent examples while preserving local native writes. The original Java URI-parsing fix remains intact.

[P2] Also decline Unicode HDFS destinations

The existing destination-mismatch finding remains for paths such as hdfs://ns/café/output. Hadoop's URI returns /café/output from both getRawPath and getPath, so the new predicate returns None. The locked Rust URL parser percent-encodes non-ASCII UTF-8 bytes, producing caf%C3%A9. With a native child reading local input and a cold HDFS object-store cache, Path::parse(url.path()) preserves that encoding and the native writer still creates files outside Spark's Unicode staging directory. This is the same unresolved P2, narrowed to a case the new guard misses.

Please also decline non-ASCII HDFS paths, or use an escape check aligned with the native URL parser, and add Unicode inputs alongside the current space/percent cases. The ordinary-HDFS and local-path controls should continue to pass. A Java getRawPath != getPath comparison alone does not cover the native encoding boundary.

I executed the exact current Scala method, copied byte-for-byte into a small Scala 2.13.17/Hadoop 3.4.1 component: all six existing controls behaved as expected, while accented, CJK, emoji and combining-character paths were admitted. Source inspection of locked url 2.5.8, percent-encoding 2.3.2 and the downstream object-store/HDFS dependencies confirms the different native filename. A real Hadoop FileOutputCommitter check with that Unicode destination mismatch injected produced _SUCCESS without the data file in the intended output under both commit algorithms. These are component checks plus source analysis. I did not run a Spark/native/HDFS write.

The maintained Spark 4.0 comparison confirms that its commit protocol returns Hadoop filenames and Spark owns job commit/abort. Native task output must land under that same staging directory before task commit. Spark 3.5 retains the concrete WriteFilesExec route and this PR's legacy writer selection. The new guard is correctly placed before write side effects, but its incomplete predicate leaves the Spark 4.0+ seam exposed. I am not claiming that Unicode handling newly regresses the pre-existing Spark 3.x writer.

Validation and limits

Current writer-suite logs show 43 passed on Spark 4.0, 4.1 and 4.2. Spark 3.4 has 34 passed/9 canceled and 3.5 has 35 passed/8 canceled. The new guard test passes in all five jobs, but contains no Unicode case. The Rust job passed 1,259 tests with five skipped, including the exact-file-path writer test.

These jobs checked out merge fc17a3b3 with this head and newer main 424c31aa, not the assigned base/head tree. Twelve of sixteen authored files match the head. The guard, both serdes, task writer and writer suite are identical, as are the checked writer regions in differing files. At 2026-09-09T18:21:01.391690+00:00, 51 checks passed, ten were skipped, one failed and one was queued. The failed Spark 4.1 SQL build reports runner shutdown/cancellation. It does not establish a source compilation failure. Maintained Spark 3.4/4.1/4.2 source gaps remain. No full local Comet build, HDFS cluster or speculative-attempt run was performed.

Performance

The helper adds path parsing and comparison during HDFS planning. Local paths return before that work, and the update adds no per-row or per-batch processing. The existing per-row statistics callbacks are unchanged. Their cost remains unmeasured. This update supplies no throughput evidence.

Design

A shared planning restriction is a suitable bounded mitigation while native path conversion remains unfixed. Both write paths consult it before execution, which avoids attempting recovery after overwrite deletion or misplaced output. Complete the predicate for native Unicode escaping. Cold and warm object-store paths still use different constructors and must both be considered when lifting the restriction.

Abstraction & complexity

The shared helper keeps one restriction and one diagnostic across the two serdes without changing their execution lifecycle. No additional abstraction is needed. Its comment currently overstates what Java's raw/decoded comparison detects. Correcting that explanation together with the Unicode guard and tests is part of the same P2.

[andygrove]

You were right, and the reason the first guard missed it is worth writing down: percent_encoding's should_percent_encode is !byte.is_ascii() || set.contains(byte), so non-ASCII is escaped regardless of which encode set is in play, while java.net.URI treats non-ASCII path characters as legal and leaves them alone. The raw/decoded comparison can only ever see what Java itself escaped. Fixed in 39e02de.

Rather than special-case non-ASCII, I derived the whole set the native parser rewrites from the locked url 2.5 crate instead of reasoning about it. A throwaway test parsed hdfs://ns/pre<c>post/output for every printable ASCII c and compared url.path() with the input:

KEPT     : !$%&'()*+,-.0123456789:;=@A-Z[\]^_a-z|~
ESCAPED  :  "#<>?`{}
NONASCII "café"   -> "/caf%C3%A9/output"
NONASCII "日本語"  -> "/%E6%97%A5%E6%9C%AC%E8%AA%9E/output"
NONASCII "🙂"     -> "/%F0%9F%99%82/output"
NONASCII "e\u{301}" -> "/e%CC%81/output"

So nine ASCII characters plus every non-ASCII byte, and notably %, [, \, ], ^ and | survive. The guard now declines anything in that set, on the string actually handed to the native writer, which is outputPath itself rather than the URI-decoded form. Deriving the set this way is also what keeps it from over-declining: gating on "not plain ASCII alphanumerics" would have refused dt=2026-09-09/hour=17 and Spark's own _temporary/0/_temporary/attempt_.../part-0.parquet, both of which are now controls in the test.

I kept the Java comparison as a second condition rather than replacing it. The native parser leaves % alone, so 50%off never gets rewritten; it reaches Path::parse as an invalid escape instead, and only Java's raw/decoded difference sees it. The comment says that now, and no longer claims the raw/decoded comparison detects the divergence on its own, which was the overstatement you flagged.

Tests cover accented Latin both precomposed (U+00E9) and as e plus combining acute (U+0301), CJK, an astral-plane emoji, non-ASCII in a nested segment rather than the leaf, and the remaining escaped ASCII characters. They are built from code points via Character.toChars, because scalastyle's NonASCIICharacterChecker rejects non-ASCII source characters, which is also why the scaladoc refers to U+00E9 rather than spelling it.

I checked the new condition is load-bearing rather than assuming it. Disabling only nativeEscaped and leaving the tests in place:

- HDFS output paths needing URI escaping are declined at planning *** FAILED ***
  escapedHdfsDestination(path).isDefined was false
  expected hdfs://ns/café/output.parquet to be declined

With it enabled, all 43 CometParquetWriterSuite tests pass locally on the default profile, which matches the count from the Spark 4.0/4.1/4.2 jobs. spotless:apply and scalastyle are clean.

On the lint failure you spotted in the earlier round: the first run of the suite here failed 43 of 43 for an unrelated reason worth recording, a stale libcomet in spark/target/classes left by another branch's build. Maven skips the copy when the destination is newer, so the suite silently ran against a library without this branch's parquet_writer.rs and proto changes. Removing it and rebuilding gave the clean run above.

[andygrove]

@rich7420 @sunchao Thanks for the reviews. I think I've addressed all feedback so far. Could you take another look?

[sunchao]

One confirmed P2 correctness issue remains. I would fix it before merging.

Reviewed head 39e02de3 against base 7f1e0018.

Finding: HDFS filename prefixes can cause silent data loss

CometWriteFilesExec.scala:210

The new guard checks the output directory, but Spark’s generated filename also includes mapreduce.output.basename.

With a basename such as part?x or part#x, native URL parsing discards everything after ? or #, including Spark’s unique partition/job suffix. Each task therefore writes a file named part. During commit, those files overwrite each other.

I reproduced this with real Spark/native execution and a local HDFS cluster, using 100 rows across two partitions:

Writer	Basename	Rows committed	Output
Spark	part?x / part#x	100	Two data files
Native	part?x / part#x	50	One file named part
Native	part	100	Two data files

All writes returned successfully and created _SUCCESS. The affected plans contained CometWriteFiles.

Suggested fix: extend admission checks to the effective filename prefix and protect the complete committer-returned path, or correct native Hadoop-path conversion. Add a multi-partition HDFS regression test. The older writer’s fixed filename generation did not expose this configuration.

Other review areas

• Performance: the per-row statistics callback adds work whose throughput impact remains unmeasured. No additional performance blocker confirmed.
• Design: retaining Spark’s job commit, SaveMode handling, and catalog refresh is sound.
• Abstraction & complexity: the version shims and separate task state are justified. No further actionable issue found.

Validation

• Local writer suite: 43/43 passed.
• Local task-metrics suite: 15/15 passed.
• Destination guard: 1,128,853 inputs checked, with no mismatch among admitted paths.
• Current CI: 65 passed, nine skipped.

Validation boundary: local native compilation was blocked by the dependency mirror. JVM tests and HDFS reproduction used current-head JVM code with a checksum-verified CI native library from merge d5bbfa54. The writer source, writer protobuf definition, and relevant native planner section match the reviewed head.

[comphead]

Reviewed against the branch-3.5 / branch-4.0 / branch-4.1 Spark sources. The premise checks out: V1WritesUtils.getWriteFilesOpt returns Option[WriteFilesExecBase] on 4.0/4.1 (V1Writes.scala:227) and Option[WriteFilesExec] on 3.5, so the additive split really is the only way in. executeTask is a faithful transcription of FileFormatWriter.executeTask for the parts it keeps, including the literal mapreduce.task.partition = 0, the exact negation of the EmptyDirectoryDataWriter guard, and the closeFile / commitTask / getFinalStats(taskCommitTime) ordering. BasicWriteTaskStatsTracker.newRow is indeed just numRows += 1, and its getFinalStats is what populates outputMetrics, so dropping reportNativeWriteOutputMetrics on 4.0+ loses nothing. TreeNode.withNewChildren calls copyTagsFrom (TreeNode.scala:348), so the WRITE_OUTPUT_PATH tag survives the child conversion.

No blockers. Two things I would want addressed, then some smaller notes.

Major

1. CometWriteFiles.parseCompressionCodec reads write options case-sensitively, Spark does not.

op.options.get("compression").orElse(op.options.get(ParquetOutputFormat.COMPRESSION)) runs against a plain Map. Spark resolves the same two keys through CaseInsensitiveMap (ParquetOptions.scala:33,39-40, reached via prepareWrite(..., caseInsensitiveOptions, ...)), and DataFrameWriter passes optionsWithPath.originalMap (DataFrameWriter.scala:144,273) so the caller's key case reaches WriteFilesExec.options intact.

This mattered less on 3.x because the native writer invented part-00000-00123.parquet with no codec suffix. Now the name comes from Spark (ParquetUtils.scala:533-535, CodecConfig.from(context).getCodec.getExtension) and the content comes from Comet, so they can disagree. df.write.option("Compression", "gzip") gets a file named ...-c000.gz.parquet containing SNAPPY. Worse, option("Compression", "lz4_raw") or "brotli" makes getSupportLevel read the SQLConf default instead, find it in supportedCompressionCodecs, and accept the write, so the unsupported-codec guard is bypassed entirely.

The neighbouring check in the same object already gets this right: CaseInsensitiveMap(op.options).get("maxRecordsPerFile").

Fix is one wrapper, but I would also move parseCompressionCodec, supportedCompressionCodecs, and the string-to-proto mapping into NativeWriteUtils rather than keep the third verbatim copy of the precedence rule. Stronger variant: keep the planning-time check for the fallback decision but re-derive the codec at execution time from CodecConfig.from(taskAttemptContext), which makes the agreement structural.

2. NativeWriteUtils.nativeUrlEscapedAscii mirrors the url crate's encode set with nothing on the Rust side pinning it.

The set is correct today, and the reasoning in the comment holds up against the WHATWG path percent-encode set. The issue is that nothing fails when it stops being correct. The new Scala test asserts the predicate against a hardcoded path list, so it locks in the predicate but not its agreement with the parser it is modelling. A url bump in Cargo.lock, or a change to how prepare_object_store_with_configs derives the object-store path, reintroduces the bug with a green build, and the failure mode is the silent one the guard exists to prevent: commitJob succeeds, _SUCCESS lands, data sits at caf%C3%A9.

Suggest a Rust unit test next to create_hdfs_operator that parses hdfs://ns/pre<c>post/output for every printable ASCII c and asserts exactly that set is rewritten, plus one non-ASCII case, cross-referenced from the Scala comment. Or invert needsNativeUrlEscaping to an allowlist so anything unrecognised is declined rather than admitted.

Minor

• Dead drain. In executeTask, the else branch is reached only when sparkPartitionId != 0 && !batches.hasNext, so batches.foreach(_.close()) closes nothing and the comment describes work that never happens.
• statsTrackers sits outside the try, so a throwing newTaskInstance skips abortTask. Spark constructs its trackers inside the FileFormatDataWriter constructor, which runs inside tryWithSafeFinallyAndFailureCallbacks. Low likelihood, one-line move.
• DataWritingCommandExec is now never tagged with a fallback reason on 4.0+. The guard is unconditional rather than conditional on the write being native, so a write with no WriteFilesExec child (plannedWrite.enabled=false, InsertIntoDataSourceCommand, Hive inserts) silently disappears from the fallback report instead of saying why. if isSpark40Plus && d.child.isInstanceOf[CometWriteFilesExec] would keep the explanation for the cases that genuinely fell back.
• Duplicate write metrics. CometWriteFilesExec still exposes files_written / bytes_written / rows_written from the native side, and bytes_written uses std::fs::metadata, which the PR notes returns 0 on HDFS. On HDFS the UI will show written data: 0 B on the Comet node next to Spark's correct numOutputBytes. Now that BasicWriteJobStatsTracker is authoritative here, consider dropping them.
• installation.md now shows a 4.0+ only example on a version-agnostic page. On 3.4/3.5 the reason is still on the command.

Questions

• Why the TreeNodeTag pre-pass rather than matching the parent? The 3.x arm right below already matches DataWritingCommandExec(_, w: WriteFilesExec). Matching DataWritingCommandExec(cmd: InsertIntoHadoopFsRelationCommand, w: WriteFilesExec) and converting w there would drop the dependence on two invariants Comet does not control: that withNewChildren copies tags, and that nothing ever hands CometExecRule a WriteFilesExec without its enclosing command. Both hold today, neither is covered by a test.
• Why is no EliminateRedundantTransitions arm needed for CometWriteFilesExec? The 3.x node needs case ColumnarToRowExec(nativeWrite: CometNativeWriteExec). The new node is equally columnar and CometExecRule is a preColumnarTransitions rule, so a transition would normally be inserted and executeWrite would hit "has write support mismatch". It works only because of Columnar.scala:530-539, which passes outputsColumnar = write.child.supportsColumnar for a V1WriteCommand under plannedWriteEnabled. That is load-bearing and would fail at runtime rather than at planning. Worth a one-line comment next to the existing originalPlan / AQE note.

Test gaps

• Nothing distinguishes the #3426 fix. DataWritingCommand.logicalPlanOutputWithNames only rewrites top-level names, so the fix is precisely about the top-level name diverging from the query's. The INSERT INTO ... SELECT test uses SELECT id, name into (id, name), and every other write test goes through DataFrameWriter where the names match by construction. Something like INSERT INTO t SELECT id + 1, upper(name) FROM s, asserting the written file's schema, is what would fail if the change were reverted.
• No coverage of a non-lowercase compression option (Major 1).
• No partially empty multi-partition write, so the sparkPartitionId != 0 && !hasNext branch that must write no file but still commitTask is unexercised.
• No coverage of the third-party WriteTaskStatsTracker warning path, which is the documented known limitation.

What is good here

Scoping is right and the additive framing is the right call. Nothing unrelated is refactored, and the two 3.x fixes that came along (URI.create on Path.toString, the HDFS escaping guard) belong here because the defect is shared. The abort test is the best of the new set: injecting through spark.sql.sources.commitProtocolClass both exercises the abort path with a real staging file to clean up and demonstrates the custom-committer claim the 3.x path could not honour, and asserting no surviving _temporary covers abortJob too. nativeWriteAllowIncompatKey and isNativeWriteExec keep the 33 pre-existing tests meaningful on both writers instead of skipping them on one. NativeWriteTask deliberately not capturing this is a real improvement over CometNativeWriteExec, which ships the converted plan tree in every task binary. And asserting that a fully native write leaves no fallback reason on the command is a good guard against exactly the explain regression this design invites.

Two compatibility notes for the record, both pre-existing and both correctly listed as follow-ups: spark.sql.parquet.outputTimestampType still defaults to INT96 while Comet writes INT64 micros (#3425), and the footer keys that drive rebase decisions on read are still missing (#3427). Both stay behind Incompatible plus the opt-in, so no regression, but worth keeping the "highly experimental" wording as loud as it is.

The one remaining per-row JVM loop is recordRows. It is inherent to the WriteTaskStatsTracker API with no bulk alternative, and at roughly a nanosecond per call it should not matter, but it is worth knowing about if a large write ever shows unexplained task time.

[sunchao]

Rechecked 39e02de3 against 7f1e0018. The Unicode HDFS directory counterexample from the previous review is fixed. The exact guard passed the devbox Scala/Hadoop component checks, including Unicode, ASCII and local-path controls, and the locked Rust URL comparison agrees with the new character check.

Two already-reported [P2] issues remain. The HDFS filename-prefix issue is unaffected because the guard checks only the directory, while the committer's basename can still lose the unique suffix during native URL parsing. The compression-option case mismatch also remains: Comet reads case-sensitive keys where Spark accepts mixed-case options. I have not duplicated either finding inline.

The inspected Spark 4.0 writer job passed 43 tests and the metrics job passed 15. These ran on merge d5bbfa54. The checked writer paths match this head, but other source and dependency changes remain. The new runs were component checks, not Spark/native/HDFS execution or a benchmark. The earlier HDFS reproduction is separate evidence, not my run. No additional verified P1/P2 emerged from this re-review.

[peterxcli]

Is it possible that we match the enclosing DataWritingCommandExec and pass cmd.outputPath explicitly when converting its WriteFilesExec child, avoiding this separate tag set and get?

[andygrove]

Yes, and it is better. Done in 268fdb9: the pre-pass is gone and the conversion happens at the command.

case d @ DataWritingCommandExec(cmd: InsertIntoHadoopFsRelationCommand, w: WriteFilesExec)
    if isSpark40Plus =>
  w.setTagValue(CometExecRule.WRITE_OUTPUT_PATH, cmd.outputPath.toString)
  d.withNewChildren(Seq(convertToComet(w, CometWriteFiles).getOrElse(w)))

The tag itself stays, because CometOperatorSerde only ever sees the operator and there is nowhere else to put the path, but it is now set and read inside one expression. That drops both of the invariants @comphead pointed at in the same round: that withNewChildren copies tags, and that nothing hands CometExecRule a WriteFilesExec without its enclosing command. Neither is Comet's to guarantee, and if the second one ever stopped holding the write now just stays on Spark instead of being converted against a missing tag.

WriteFilesExec moves to the unconditional "never tagged as a fallback" list as a result, since on both version paths the conversion decision is made at the command above it.

[peterxcli]

nit, want to separate the comment into branches.

Suggested change

	// Spark 4.0+ hands over the exact file to write, chosen by the JVM commit protocol.
	// Spark 3.x hands over a working directory instead and expects the writer to name the
	// file; that branch goes away with Spark 3.x support.
	let part_file = match &work_dir {
	None => self.output_path.clone(),
	Some(work_dir) => match task_attempt_id {
	Some(attempt_id) => format!(
	"{}/part-{:05}-{:05}.parquet",
	work_dir, self.partition_id, attempt_id
	),
	None => format!("{}/part-{:05}.parquet", work_dir, self.partition_id),
	},
	let part_file = match &work_dir {
	// Spark 4.0+ hands over the exact file to write, chosen by the JVM commit protocol.
	None => self.output_path.clone(),
	// Spark 3.x hands over a working directory instead and expects the writer to name the
	// file; that branch goes away with Spark 3.x support.
	Some(work_dir) => match task_attempt_id {
	Some(attempt_id) => format!(
	"{}/part-{:05}-{:05}.parquet",
	work_dir, self.partition_id, attempt_id
	),
	None => format!("{}/part-{:05}.parquet", work_dir, self.partition_id),
	},

[andygrove]

Applied as suggested in 268fdb9.

[peterxcli]

this test actually performs a second whole write after an injected commit failure; it does not demonstrate speculative attempts or an automatic task retry.

[andygrove]

You are right, and the name was claiming more than the test does. Renamed to a failing task aborts and cleans up its staging file in 268fdb9, and the comment now says what the second write is:

The second write at the end is a fresh write of the same data, not an automatic task retry and not a speculative attempt: it only shows that the failed job left nothing behind that would break the next one. Task-attempt isolation itself is Spark's newTaskTempFile, and is still untested here.

[peterxcli]

writeNatively uses a plain finally:

try {
  while (execIterator.hasNext) {
    execIterator.next().close()
  }
} finally {
  execIterator.close()
}

If execution throws A, and close() throws B, B replaces A. The outer task guard still aborts, but it receives the cleanup error instead of the original execution error.

close() can throw: it propagates teardown failures, including errors from the final native metrics update. The repository already has an injected metrics-failure test demonstrating that behavior. It also attempts every cleanup step and releases the native plan despite teardown errors, so this finding concerns preserving the original failure, rather than a demonstrated skipped plan release.

• Iterator cleanup
  

  datafusion-comet/spark/src/main/scala/org/apache/comet/CometExecIterator.scala

  Line 275 in 39e02de

  def close(): Unit = synchronized {

  ),
• existing lifecycle test
  

  datafusion-comet/spark/src/test/scala/org/apache/spark/CometExecIteratorLifecycleSuite.scala

  Line 176 in 39e02de

  test("releasePlan frees the native context even when the final metrics update fails") {

Spark’s tryWithSafeFinallyAndFailureCallbacks preserves the original throwable and attaches cleanup failures as suppressed exceptions. Using that behavior here would also bring abort-error handling closer to Spark -- Spark helper

[andygrove]

Agreed, and fixed in 268fdb9 using the helper you pointed at, at both levels.

writeNatively now uses Utils.tryWithSafeFinally, so an execution failure survives a throwing close() with the cleanup error attached as suppressed rather than replacing it. executeTask uses Utils.tryWithSafeFinallyAndFailureCallbacks with committer.abortTask as the catch block, which is exactly FileFormatWriter.executeTask's shape — it also means a throwing abortTask is suppressed onto the original instead of being swallowed by the tryLogNonFatalError that was there before, and markTaskFailed now runs so Spark's task failure listeners fire.

While moving that I also took @comphead's related note in the same area: description.statsTrackers.map(_.newTaskInstance()) moved inside the guarded block, so a tracker whose constructor throws still reaches abortTask. Spark gets that for free because its trackers are built by the FileFormatDataWriter constructor, which already runs inside the same block.

[sunchao]

Rechecked unchanged head 39e02de3 against 7f1e0018 after the new discussion. I agree with Peter's failure-preservation concern [P2]: when execution throws A and iterator cleanup throws B, the plain finally replaces A with B. Cleanup still attempts native-plan release. Please preserve A and attach cleanup failures as suppressed exceptions, following Spark's task-write helper. A nine-case Scala component check using the exact control-flow blocks and injected test doubles confirmed the distinction; it did not execute Spark or JNI.

The earlier [P2] HDFS filename-prefix and mixed-case compression-option findings remain unchanged. I have not duplicated these existing comments inline.

Peter's retry-test clarification is also correct: it exercises abort followed by a second whole write, not automatic task retry or speculation. Explicit parent-path passing is a feasible simplification, while the current tag survives Spark's child replacement; I found no additional correctness defect in that mechanism.

CI remains green. Freshly inspected Spark 4.0 logs show 43 writer tests and 15 metrics tests passing on merge d5bbfa54. That merge uses newer main 424c31aa, and 12 of 16 authored files match this head. It is not full-tree validation of the assigned pair. No new native/HDFS integration run or benchmark was performed.

[andygrove]

Everything from this round is addressed in 268fdb9. Thanks @sunchao and @comphead — the two P2s were both real, and the second one had a second half I had not looked for.

HDFS filename prefixes (#5763 (comment))

Fixed, and the fix has two layers because one is not enough.

The guard was checking the destination directory when the thing that has to survive intact is the whole committer-returned path. HadoopMapReduceCommitProtocol.getFilename reads mapreduce.output.basename straight from the task configuration, so caller-controlled text lands in every file name, and ?/# are worse there than anywhere else in the path: the native URL parser does not escape them, it treats them as delimiters and truncates. Every task then writes the same truncated name and they overwrite each other during commit, which is exactly the 50-rows-of-100 result you measured. escapedHdfsDestination now takes the effective basename alongside the directory and declines on either.

Planning can only decline what it can predict, though, and the path actually comes from FileCommitProtocol.newTaskTempFile — a replaceable component. So NativeWriteUtils.checkNativeWriteDestination runs the same predicate against the real path in executeTask, before the writer opens anything. That one throws rather than falling back, which is the right trade at that point: the alternative is a job that commits successfully with the data somewhere else.

I have no HDFS here, so the regression test exercises the predicate directly rather than through a write — space, %, and non-ASCII basenames declined, part / out / data_v2 / part-of-it accepted, a file: destination unaffected by its basename, and the execution-time check accepting a real staging path and rejecting the same path with a ? basename. I confirmed it fails without the fix:

- HDFS destinations needing URI escaping are declined at planning *** FAILED ***
  NativeWriteUtils.escapedHdfsDestination(plainHdfs, basename).isDefined was false
  expected basename 'part?x' to be declined

Mixed-case compression option (@comphead, Major 1)

Fixed, and you were right that the second half is the dangerous one. parseCompressionCodec, supportedCompressionCodecs and the string-to-proto mapping all moved to NativeWriteUtils behind a CaseInsensitiveMap, which also removes the identical bug from the Spark 3.x serde — it mattered less there only because that writer invents a name with no codec suffix.

I took the stronger variant as well. CometWriteFilesExec now re-derives the codec per task from CodecConfig.from(taskAttemptContext), which is the same CodecConfig the file extension comes from (ParquetUtils.prepareWrite resolves the write's options into parquet.compression on the job conf, and getFileExtension reads it back). Name and contents now agree by construction rather than by two copies of the precedence rule agreeing.

Two tests. The load-bearing one is parquet write with a mixed-case unsupported compression codec falls back to Spark: with the case-sensitive read restored it fails, because getSupportLevel finds snappy in the SQLConf, accepts the write, and Comet writes SNAPPY into a file Spark named .lz4raw.parquet. Worth noting that the other one, honors a mixed-case compression option, passes under that same mutation on 4.0+ — the execution-time re-derivation catches it. That is the defense in depth working, but it does mean that test alone would not have caught the bug.

Pinning the url encode set (@comphead, Major 2)

Took the Rust-test option. url_path_rewritten_characters in parquet_support.rs parses hdfs://ns/pre<c>post/output for every printable ASCII c and asserts the rewritten set is exactly "#<>?`{} (nine characters, leading space), plus four non-ASCII names that must all be percent-encoded. The Scala guard's comment now points at it and its comment points back. The failure message says which constant to update.

The smaller notes

• Dead drain. Right — batches.foreach(_.close()) in the else branch only runs when !hasNext already returned false, so it closed nothing. Removed; the comment now says hasNext already ran the child to completion, which is also what Spark's EmptyDirectoryDataWriter relies on.
• DataWritingCommandExec never tagged on 4.0+. Fixed as you suggested — the skip is now conditional on d.child.isInstanceOf[CometWriteFilesExec], so a write with no WriteFilesExec child gets its reason back.
• Duplicate write metrics. Dropped all three. BasicWriteJobStatsTracker is authoritative on this path and reports on the command, and the native bytes_written reads 0 on HDFS, so the node was contributing a second and less trustworthy copy. CometTaskMetricsSuite still passes 15/15, which is the check that matters: its output metrics come from Spark's tracker, not from the node.
• EliminateRedundantTransitions. No arm is needed, and you found the reason before I documented it. There is now a comment on doExecute naming Columnar.scala's outputsColumnar = write.child.supportsColumnar for a V1WriteCommand under plannedWriteEnabled, and saying that it is load-bearing and would fail at execution rather than at planning.
• installation.md. Now says the example is Spark 4.0+ and that on 3.4/3.5 the same message appears on the command.

Test gaps

• Mixed-case compression — two tests, above.
• Partially empty multi-partition write — an empty partition writes no file and still commits. Hash-partitions into eight with AQE coalescing off and keeps one id, so at most one partition has rows and at most two files can appear. Forcing the write branch unconditionally fails it.
• Third-party WriteTaskStatsTracker — nothing Spark ships lets a V1 write install one, so the test drives recordRows directly with a recording tracker and asserts three callbacks, all with empty rows, plus the warning naming the tracker's class. recordRows became public for this; it is the only way to reach the documented limitation.
• [COMET NATIVE WRITER] INSERT INTO TABLE - complex type but different names #3426 — this one came out differently from what we both expected, and it corrects a claim in my own description. I added the scenario from Spark's own INSERT INTO TABLE - complex type but different names with a top-level rename on top, and it does pass on 4.0+ and fail on 3.5. But it fails on 3.5 with zero rows read back, i.e. [Native Writer] INSERT INTO ... SELECT fails due to stale catalog cache after write #3521, not with the wrong names — and reverting description.dataColumns to child.output does not change the outcome on 4.0+. The reason is that Spark's castAndRenameQueryOutput has already aliased the query's output to the target's column names and cast the struct to the target's nested field names before Comet sees anything, so the two sources agree. dataColumns is still the right one to read — it is what Spark treats as authoritative, and it is the only one that stays correct once partitioned writes land, since FileFormatWriter excludes the partition columns from it — but it is not what fixes [COMET NATIVE WRITER] INSERT INTO TABLE - complex type but different names #3426. Restructuring is. I have corrected the code comment and the description accordingly.

Testing

CometParquetWriterSuite is 48/48 on 4.0, 4.1 and 4.2, 36/36 + 12 cancelled on 3.4 and 38/38 + 10 cancelled on 3.5. CometTaskMetricsSuite 15/15, CometExecSuite 144/144, CometFallbackInvarianceSuite 6/6, CometPublicApiSuite 1/1, all on 4.1. Native: cargo test -p datafusion-comet parquet_writer and the new url_path_rewritten_characters pass, cargo clippy --all-targets --workspace -- -D warnings is clean, and spotless, scalastyle and prettier are clean.

Still not covered, and I am not claiming otherwise: an end-to-end HDFS write, speculative concurrent attempts, and any measurement of recordRows.

[sunchao]

Re-reviewed 5f5397f5 against 7f1e0018. One P2 remains in HDFS basename admission, detailed inline.

Correctness

The Spark 4.x integration preserves Spark's job commit, SaveMode handling and catalog refresh while replacing the per-task writer. I rechecked version routing, AQE/logical links, columnar transitions, target schema and field IDs, compression selection, native filename ownership, and cleanup. The Spark 3.x execution path remains separate.

I reran the earlier HDFS filename reproduction using native and JVM code built locally from this exact head. With 100 rows in two input partitions, part?x, part#x and part space now fall back to Spark and preserve all 100 IDs. The ordinary part case still writes natively and preserves all 100 IDs. The previous silent-overwrite reproduction is addressed.

The remaining case is a literal % in the configured basename. Both part%foo and part%25 are accepted during planning, then fail in checkNativeWriteDestination. The Spark-writer controls succeed. This is a predictable configuration that should fall back before task execution, rather than a custom committer path that can only be checked at runtime.

Local validation

• Exact-head cargo build --locked succeeded with the default native features. The native library copied into the JVM resources has the same SHA-256 as the local build.
• Root-reactor Maven tests on Spark 4.1.3 / JDK 17 passed 66/66: writer 48, task metrics 15, and iterator lifecycle 3. Formatting/style checks in that invocation passed.
• Native writer tests: 4 passed, 4 ignored because those fixtures require their own HDFS setup. The URL-character test passed separately.
• A separate real MiniDFS integration ran 12 cases: ten successful Spark/native/fallback controls and the two reproduced percent-prefix failures. Successful cases checked every ID, and executed plans confirmed native writing versus fallback.
• Five separate Spark-helper component checks passed for original-error preservation, throwing close/abort callbacks, tracker-construction failure, self-suppression, and successful write/close/commit ordering. These are component checks, not injected native-writer failures.

CI

At the publication check, CI has 53 successful checks and ten skipped, with no failed or pending checks. I inspected the writer-suite logs: Spark 4.0/4.1/4.2 each passed 48/48, Spark 3.5 passed 38 with ten cancellations, and Spark 3.4 passed 36 with twelve cancellations. The Rust job passed 1,442 tests with five skipped.

Those CI jobs checked out merge 8931be12, combining this head with newer main f69c4c81. Its full tree differs from the reviewed head. The local build, focused suites and HDFS reproduction above provide separate exact-head evidence.

Performance and design

The smaller write boundary and explicit per-task state are justified. I found no additional confirmed ownership or performance regression. recordRows still performs one callback per row, and I have not measured its throughput cost. Concurrent speculative attempts were not tested. Existing experimental timestamp/footer and writer-property limitations remain outside the fixes in this PR.

[sunchao]

[P2] Fall back for percent-bearing HDFS basenames during planning

Could we apply the Java URI-escaping check to fileNamePrefix as well? needsNativeUrlEscaping deliberately excludes %, so mapreduce.output.basename=part%foo or part%25 is accepted here. Once Spark includes it in the task filename, hdfsPathDivergence detects getRawPath != getPath and checkNativeWriteDestination aborts the job instead of falling back.

I reproduced both cases with a native scan over 100 rows in two partitions on Spark 4.1.3 and a real MiniDFS cluster, using native and JVM code built from 5f5397f5. The Spark-writer controls preserve all 100 IDs. With native writing enabled, both plans contain CometWriteFiles and fail with UnsupportedOperationException at the complete-path guard. The ordinary part control writes natively, and the fixed ?/# cases fall back successfully.

The basename is already available during planning. Please reject its Java-escaped forms here too, and add % basename cases to the admission and fallback regression coverage. The existing basename test currently covers ?, #, spaces and Unicode, but no literal percent.

[sunchao]

Rechecked 5f5397f5 against 7f1e0018. The mixed-case codec, ?/# basename fallback, and primary-error-preservation fixes are in place. The existing P2 for percent-bearing HDFS basenames remains: part%foo and part%25 pass planning, then fail the task guard. A fresh component probe of the exact JVM guard with Hadoop Path reproduced this; ordinary HDFS and local controls passed. Please include these prefixes in the planning fallback before merging.

I found no additional P1/P2. CI passed the 48 writer tests on Spark 4.0 and 4.1 using merge 8931be12; the JVM writer and suite match this head. That is separate from the latest merge preview. I did not rerun the native writer end to end, automatic task retries, or speculation locally.

[sunchao]

Rechecked a69b76c1 against 5705a58a. The remaining P2 about percent-bearing HDFS basenames is fixed. Planning now applies the Java and native escaping checks to the basename. A fresh component probe with the exact old/current helpers and real Hadoop Path reproduced the six old ASCII gaps and found none on this head. Ordinary HDFS names and local percent-bearing names still pass.

The update also covers AQE empty relations through the existing single-partition write path. Current Spark 4.1 CI passed all 49 writer tests and the empty-relation test on merge 5faaf694, whose changed files match this head. The Rust URL-escaping regression passed too.

No new or remaining verified P1/P2 finding. At 00:30 UTC, CI has 22 successful checks, 13 skipped and one still running (Spark 4.1 expressions). Local validation was a JVM helper probe, not an end-to-end native HDFS write. Concurrent speculative attempts and the statistics-callback cost remain unmeasured.