Repository navigation
feat: [branch-1.2] support the S3A magic committer in native Parquet writes (#6940) - #7281
Merged
Merged
Conversation
…e#6940) * feat: support the S3A magic committer in native Parquet writes On Spark 4.0+, a task file that is neither local nor on HDFS is created by the JVM through Hadoop's FileSystem API, as Spark's Parquet writer creates it, and the native writer streams the file's bytes into that stream over JNI. S3A therefore creates the magic committer's files itself and records their pending uploads. A failed write aborts the stream. Planning admits an S3 destination when the S3A magic committer commits it, through PathOutputCommitProtocol or BindingParquetOutputCommitter. Other S3 writes still fall back to Spark. * test: make the magic committer's failed-write check isolate the writer's abort S3A's job abort also aborts pending uploads under the destination, through core-default.xml's fs.s3a.committer.abort.pending.uploads=true, which hid a writer that never aborted its file. Turn that off in the test, use one task so job abort cannot race other tasks, and run each committer binding alone. Add a unit test that the output stream writer sends nothing after a failed write. * feat: support the S3A magic committer in Spark 3.4 and 3.5 native writes The Spark 3.x writer runs the session's commit protocol with the write's Hadoop configuration since apache#4746, so it reaches S3A's committers as Spark's writer does. Both writers now create task files through one helper, NativeWriteUtils.writeFile, which creates any file that is neither local nor on HDFS through Hadoop and closes or aborts it, and both serdes share one file system gate. The magic committer tests run on every supported Spark version; in-memory commit tracking is skipped on Hadoop 3.3.4, which lacks it. * refactor: simplify the native write path through the JVM The native writer now fails a task with the output stream's own error, which keeps the Java exception for the JNI boundary to rethrow as it was, so HadoopFileOutput no longer records failures and the JVM no longer swaps exceptions. Both writers write a task file through one helper, CometWriteFilesExec.writeFile, which sets the file and stream flag, runs the plan, and closes or aborts the file. Also: the pusher and output stream registrations share their rules, the writer takes an optional stream, local writes report their counted bytes, planning skips object store options and Hadoop configurations it does not need, the committer factory keys come from Hadoop's constants, and the tests share their setup and read S3A's success manifest through SuccessData. * address review: drop the unreachable bucket committer fallback and say once why S3A creates the file * address review: enrich write errors as Spark does, test the Hadoop route's failure paths Fail a native write's task with TASK_WRITE_FAILED as Spark's writer does, keeping fetch failures and fast-failing on an existing file. Admit subclasses of PathOutputCommitProtocol at planning. Build each task's plan once and create its iterator through CometExec, register the output stream through register_before_execution directly, and move the supported destinations to the Parquet page's Writing section. Test a file of several MiB, a failure partway through it, a failed close, a stream that cannot abort, a retried task and a path with special characters, and against MinIO, a failure after S3A uploaded a part and failed task and job commits. * test: compare what a failed job commit leaves pending with Spark's writer S3A's magic committer leaves the uploads of committed tasks pending when job commit fails, for Spark's own writer as well, so check that the native write leaves the same and nothing visible. Scope each pending upload check to its table, and abort what a test leaves behind. * address review: test the fast-fail and commit paths, nested and temporal columns, and the sink's guards Cover the branches of the Hadoop route that had no test: a task file that already exists, a file that exists at task commit (with and without fastFailOnFileFormatOutput), nested and temporal columns compared with Spark's writer through chunked writes and the abort path, an INSERT into a table of the V2 file source, and the output stream sink's oversize and null checks. * fix: handle the fallible execution context lookup in setOutputStream * test: write complex and temporal types through the Hadoop route * test: leave the non-ASCII character out of the output path when the JVM cannot encode it (cherry picked from commit f1121ca)
github-actions
Bot
added
enhancement
comphead
approved these changes
Oct 28, 2026
comphead
left a comment
comphead
left a comment
Contributor
There was a problem hiding this comment.
Thanks @andygrove lets check it out
This was referenced Oct 28, 2026
andygrove
added a commit
that referenced
this pull request
Oct 28, 2026
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
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?
Backport of #6940, which closes #6929.
Rationale for this change
Native Parquet writes cannot run where S3A's magic committer commits the write, and Spark 4.1 makes it the default whenever spark-hadoop-cloud is on the classpath (SPARK-47618). #6940 lets the native writer produce such a write by having the JVM create each file through S3A and the native writer supply only its bytes.
This is a feature rather than a bug fix.
branch-1.2has not shipped its.0release, so it is still taking whatmaingains after the cut (see Which Branches Take Backports), and the release manager decides whether it stays. It is not going tobranch-1.1orbranch-1.0, which take regression fixes only. There is no newer release branch.What changes are included in this PR?
The cherry-pick of f1121ca (
git cherry-pick -x), with one adaptation:native/core/src/execution/jni_api.rs:Native.setOutputStreamcallsget_execution_context(exec_context)without the?. Onmainthat function returns aCometResult(it rejects a null handle with anInternalerror), but onbranch-1.2it returns the&mut ExecutionContextdirectly.setOutputStreamalready rejects a zero handle with aNullPointererror before the call, so no check is lost. The change that made the function fallible is not backported.Everything else is identical to the source commit.
How are these changes tested?
Built on
branch-1.2with the cherry-pick, because a clean pick can still depend on code that landed after the cut:cargo check --all-targets --workspace,cargo clippy --all-targets --workspace -- -D warningsandcargo fmt --all --checkinnative/../mvnw test-compileon the default profile (Spark 4.1) and on-Pspark-3.5.datafusion-comet.CometParquetWriterSuite,CometParquetWriterRetrySuiteand the other suites the source PR changed run in CI, which runs every tier on a pull request against a release branch.ParquetWriteWithMagicCommitterSuiteneeds MinIO and is manual, as onmain; I have not run it on this branch.