Conversation
Contributor
|
Thanks @peterxcli please check on your fork https://github.com/apache/datafusion-comet/actions/workflows/spark_sql_writer_tests.yml |
peterxcli
marked this pull request as ready for review
July 13, 2026 18:24
Member
Author
|
@comphead would you like to take a look? Thanks! |
Contributor
|
Thanks @peterxcli there is another angle #5293 |
…Spark 4.0+ 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. 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. Spark 3.x has no such trait: getWriteFilesOpt matches the concrete WriteFilesExec case class, a Comet node there would not be found, and Spark would silently take FileFormatWriter's non-planned, row-based branch. So the new seam is additive. CometDataWritingCommand and CometNativeWriteExec are kept unchanged and remain the 3.4/3.5 path; CometExecRule picks the path by version and the two never both fire. The legacy path goes away with Spark 3.x support. Add: - CometWriteFilesExec, overriding doExecuteWrite and 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. - 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. - Column names, nullability and field IDs come from WriteJobDescription.dataColumns rather than the query output, so INSERT INTO t SELECT a+1 writes the target column's name. - Byte and row counts come from BasicWriteTaskStatsTracker, which stats files through the FileSystem API and is therefore correct on HDFS. - ParquetWriter proto: work_dir is now optional. When set (3.x) the native writer derives the file name as before; when unset (4.0+) output_path is the exact file to write and is used verbatim. - On 4.0+ the opt-in moves to spark.comet.operator.WriteFilesExec .allowIncompatible, with the old DataWritingCommandExec key kept as a deprecated alternative. isOperatorAllowIncompat now resolves alternatives, which the planner's by-name lookup previously bypassed. AQE re-plans the write command's child and re-inserts a WriteFilesExec above the node Comet already converted; leaving DataWritingCommandExec in place means Comet no longer has to guard against the resulting nested native writes.
…configs Matches the reviewed form on apache#5293: ConfigBuilder mutates in place, so the Seq destructuring was rebinding the same object. Only one operator has an alternative and there is no reason to expect more.
The output path reaches both write serdes as `Path.toString`, which decodes percent escapes: a directory containing a space or a literal `%` yields a string that is not a valid URI, and `URI.create` throws on it. On Spark 4.0+ that exception escaped `CometWriteFiles.convert` and failed the query. On 3.x `CometDataWritingCommand.convert` caught it and silently handed the write back to Spark, so the native writer was never used for those paths. Round-trip through `Path` instead, which re-escapes. Only the scheme and authority reach `extractObjectStoreOptions`, but parsing has to succeed to get at them. Also assert that the INSERT INTO visibility test's write actually went native, rather than inferring it from the read-back.
The raw/decoded URI comparison only catches what java.net.URI had to
escape, and java.net.URI leaves non-ASCII path characters alone, so an
hdfs://ns/cafe<U+0301>/output destination was admitted. percent_encoding's
should_percent_encode is !byte.is_ascii() || set.contains(byte), so the
native parser escapes every non-ASCII byte regardless of the encode set
and the writer creates caf%C3%A9 outside Spark's staging directory.
The guard now also declines any character the native parser rewrites. The
ASCII half of that set was determined against the locked url 2.5 crate by
parsing hdfs://ns/pre<c>post/output for every printable ASCII c: space, ",
#, <, >, ?, backtick, { and } are rewritten and the rest survive, so
partition directories and Spark's _temporary attempt paths still qualify.
The comment no longer claims the Java comparison detects the divergence on
its own; both conditions are kept because the Java one still catches a
literal % that the native parser leaves alone.
Tests add accented (precomposed and combining), CJK, emoji and nested
non-ASCII cases plus the remaining escaped ASCII characters, built from
code points since scalastyle forbids non-ASCII source. Disabling the new
condition makes the accented case fail, so the Java comparison alone
demonstrably does not cover it.
peterxcli
marked this pull request as draft
September 12, 2026 17:30
github-actions
Bot
added
enhancement
Correctness:
- Decline HDFS writes whose *file names* would diverge, not just their
directory. `mapreduce.output.basename` is caller-controlled and reaches every
committed name through `HadoopMapReduceCommitProtocol.getFilename`; a basename
holding `?` or `#` makes the native URL parser truncate, so every task writes
the same name and they overwrite each other at commit. Adds an execution-time
backstop over the complete `newTaskTempFile` path, which a custom commit
protocol owns and planning cannot predict.
- Read the compression option case-insensitively, as Spark's `ParquetOptions`
does. `option("Compression", "lz4_raw")` used to fall through to the SQLConf
default, so the unsupported-codec guard was bypassed and Comet wrote SNAPPY
into a file Spark had named `.lz4raw.parquet`. The codec is now also
re-derived per task from `CodecConfig.from(taskAttemptContext)`, the same
place the file extension comes from, so the name and the contents agree by
construction. The shared helpers move to `NativeWriteUtils`, which fixes the
identical bug on the Spark 3.x path.
- Use `Utils.tryWithSafeFinallyAndFailureCallbacks` / `tryWithSafeFinally` in
`executeTask` and `writeNatively`, matching `FileFormatWriter`: a failure
while aborting or closing the iterator is attached as a suppressed exception
instead of replacing the failure that caused it. `statsTrackers` moves inside
the guard so a throwing `newTaskInstance` still reaches `abortTask`.
Planning and reporting:
- Convert `WriteFilesExec` from its enclosing `DataWritingCommandExec` rather
than from a separate tag pre-pass, so the output path comes straight from the
command that owns it and neither `withNewChildren` copying tags nor "nothing
hands us a bare WriteFilesExec" has to hold.
- Only skip the fallback reason on `DataWritingCommandExec` when its child
really was converted; a write with no native child now says why it fell back.
- Drop the node's duplicate `files_written`/`bytes_written`/`rows_written`.
`BasicWriteJobStatsTracker` is authoritative here, and the native
`bytes_written` reads 0 on HDFS.
Tests and docs:
- Rust `url_path_rewritten_characters` pins the `url` crate's path encode set,
which the JVM guard mirrors; a crate upgrade can no longer reopen the hole
with a green build.
- New JVM coverage: mixed-case `compression` (honored, and declined when
unsupported), the apache#3426 nested-name INSERT, an empty non-zero partition
writing no file, the third-party `WriteTaskStatsTracker` warning, and the
basename/committer-path guards.
- The abort test no longer claims to demonstrate task retry or speculation.
- `installation.md` says which Spark versions its EXPLAIN output applies to.
peterxcli
force-pushed
the
feat/spark-file-commit
branch
from
September 14, 2026 16:21
915b174 to
278f28e
Compare
peterxcli
changed the title
peterxcli
marked this pull request as ready for review
September 15, 2026 07:19
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Addresses the remaining Spark 3.x commit-protocol gaps in #2827 and #3015. This does not claim to close their remaining version and feature gaps.
Rationale for this change
Spark 3.4/3.5 native writes replace the whole
DataWritingCommandExec. The retained writer hardcodes the commit protocol, discards the filename returned bynewTaskTempFile, and does not consistently complete or abort the job from both execution entry points. Committers that track individual filenames can therefore commit a different file from the one native code writes.Dependency: #5763 is still open. The writer implementation is based on its current head,
5f5397f5049b1bfc6fe27e6c8e926ea2d0bee959, and the PR remains draft. The branch also includes the existing merge frommainat481aefea9. Functional follow-up changes are limited to the Spark 3.x writer, with focused tests and their CI registrations. Compare the writer implementation against the dependency. Once #5763 merges, this follow-up can be rebased onto main.What changes are included in this PR?
fileFormat.prepareWrite, and instantiate Spark's configuredFileCommitProtocol. Task contexts inherit the prepared configuration, including changes fromsetupJob.newTaskTempFilefilename through the existingoutput_pathfield withwork_dirunset. The previoustask_output_pathproposal is removed; protobuf field 9 remainsoutput_schema. This follow-up changes no protobuf or native Rust code.runJob, invokeonTaskCommit, and remove the collection accumulator.CometWriteFilesExecpath.CometNativeWriteSuitebeside the Parquet writer suite in both Linux and macOS CI matrices. This fixes the Preflight missing-suite failure.How are these changes tested?
The runtime results below were collected on writer commit
278f28e0a, before the subsequent merge frommainand CI registration fix. Built the native library withmake corebefore JVM tests. Maven ran from the repository root, without-pl, using JDK 17 and a debug native build. Spark profiles were built clean when switching versions.CometNativeWriteSuiteCometParquetWriterSuiteThe new suite uses a configured committer with an unusual exact filename containing spaces and
%, checks job-option/file-format preparation and task-message delivery, and verifies native row metrics and output read-back for both entry points. It injects failures in native iterator construction, native batch execution, final native metrics cleanup,commitTask,onTaskCommit, andcommitJob, including throwing abort callbacks. Assertions check task/job abort, preservation of the original exception, suppressed cleanup errors,_SUCCESS, and staging cleanup. Existing success-marker, Spark filename, and task-abort checks now also run on Spark 3.x; the writer suite covers SaveMode, schema/field IDs, codecs, and local URI regressions.Spotless, Scalastyle, and
git diff --checkpassed.After the merge from
main, CI registration commite9f5e5e7fpassed the suite inventory check, CI configuration checks, all 15 Iceberg shard validation tests, benchmark runner checks, andactionlint --shellcheck=off. A clean Spark 3.5/Scala 2.12/JDK 17 package build and semantic Scalafix check also passed (-DskipTests; runtime tests were not repeated for the two workflow-only additions).Limitations: validation used local Spark execution, not a real HDFS cluster, automatic task retries, or speculative attempts. #5763's unresolved percent-bearing HDFS basename admission issue remains a dependency limitation. This does not extend the legacy writer's catalog refresh, partitioning, file rolling, or other experimental writer behavior.