Repository navigation
Conversation
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.
github-actions
Bot
added
enhancement
…r'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.
|
@parthchandra @comphead I'm still iterating om this, but let me know what you think of the direction |
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.
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.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Reviewed the full four-commit, 24-file diff from 474b5ece841a7fc6dff37ae04bd6787e654d7be7 to 4756615e289db86dda118aec72d12b87e638a7ee. The PR is not a draft. Read the supplied discussions and linked issues. No introduced P1/P2 issues found within this review, and no substantiated existing P1/P2 blockers remain.
Routed skills: review-comet-pr, review-comet-expression-pr for serde/planning, review-comet-ffi-pr, review-comet-memory-pr, and review-comet-shuffle-pr for the shared callback-registration refactor.
Summary
- Prior state and problem: Native Parquet writes admitted local and HDFS destinations but fell back for S3. Writing magic-committer paths directly would bypass S3A’s pending-upload tracking and could leave a successful job without committed data.
- Design approach: Hadoop creates the task file and owns its upload lifecycle. Native code supplies encoded Parquet bytes through a registered JNI sink. Planning admits recognized magic-committer configurations, while execution selects the transport from the actual task filename.
- Correctness: Checked Spark’s commit protocols, file-writer cleanup and statistics ordering, Hadoop’s committer selection and stream lifecycle, and parquet-rs’s footer flushing. The stream closes before statistics and task commit. Failed native writes abort the Hadoop stream, and stream exceptions retain their Java throwable through the native boundary. No introduced P1/P2 correctness issue was identified.
- Compatibility analysis: Compared relevant sources for Spark
3.4.3,3.5.9,4.0.4,4.1.3and4.2.0, plus Hadoop3.3.4and3.4.1. The factory and committer-name precedence matches Hadoop. Both Spark writer entry points use the shared transport, without introducing Spark-version logic into Rust. Native writes remain experimental and disabled by default. - Key design decisions: S3A retains responsibility for magic paths, pending commits, encryption and upload settings. The per-task
jvm_output_streamflag is checked against the registered sink, preventing a missing or unexpected stream from silently selecting another destination. - Implementation sketch:
HadoopFileOutputhandles create/write/close/abort.JavaOutputStreamSinkreuses a Java byte array.OutputStreamWriterbatches writes and stops sending after a sink failure.CometWriteFilesExec.writeFileshares execution and cleanup across Spark 3.x and 4.x. Local byte metrics now use the writer’s byte count. - Performance: The new route adds two copies of encoded bytes, approximately one JNI upcall per MiB, a 1 MiB native buffer included in writer accounting, and a reusable 1 MiB Java array. The author’s measurements in #6944 show little local routing overhead, but do not establish real-S3 upload performance. No evidence-backed regression was identified on the existing local/HDFS paths.
- Design: Keeping S3A responsible for its own commit protocol is straightforward and avoids duplicating Hadoop-internal metadata formats. Choosing transport from the committer’s returned path also makes execution resilient to planning-time assumptions.
- Abstraction & complexity: The sink interface provides a focused boundary between encoding and transport. Shared file execution removes duplication between Spark versions. The extracted callback-registration helper preserves remote shuffle’s existing once-only, before-execution checks, with its regression test passing at this head.
- Behavioral changes worth calling out: Compared with
branch-1.1at8ac6fccde53d96fec22184be287747eeae481cac, eligible magic-committer writes can now use native encoding when native writes are enabled. Other S3 configurations still fall back. Earlier Spark 3.x commit-lifecycle changes already present in the base were distinguished from this PR’s changes. No unintended release-relative regression was identified. - Suggested improvements: No additional implementation change met the P1/P2 reporting bar. Broader Spark SQL and object-store validation remain the limitations described below.
Exact-head CI: run 38475818148 has 26 successful and 18 skipped checks, with no failed checks. Logs confirm 2,429 Rust tests passed, including the new stream and planner tests. The Spark 4.1 scans job passed the new Hadoop-routing, abort/exception and committer-selection tests. Spark SQL jobs across supported versions, Iceberg integration jobs, macOS and benchmarks were skipped.
Local validation: compiled the exact HadoopFileOutput and RecordingFileSystem sources with JDK 17, cached Hadoop 3.4.2 and parquet-java 1.16.0. A disposable probe passed byte preservation, escaped/Unicode paths, explicit close, no-overwrite behavior, original IOException propagation and abort cleanup. Suite registration and git diff --check also passed. Project code remains unchanged.
Validation limits: local Rust tests stopped before compilation because the configured registry could not resolve locked DataFusion 55.2.0, both offline and online. Full local Comet integration suites were therefore not rerun. I did not independently run MinIO, real S3, or the multi-version magic-committer suite. The author’s reported S3Proxy runs across all five Spark profiles are supplementary evidence, not independently reproduced results.
viirya
left a comment
viirya
left a comment
There was a problem hiding this comment.
Thanks for this. Having S3A create the file and letting the native writer supply only the bytes keeps the magic committer's protocol entirely inside Hadoop. Choosing the route per task from the committer's actual path is a nice property. I traced the upcalls and they run on the Spark task thread through block_on, since the writer's plan always has a JVM-fed scan, so S3A blocking on its upload queue never ties up a tokio worker. The close-before-closeFile ordering and the abort path also look right to me. I left a few inline comments on the S3 gate and the comments around it. Two more things don't anchor to a line in the diff.
Validation. The end-to-end evidence for this PR comes from ParquetWriteWithMagicCommitterSuite. As committed, that suite runs against MinIO through CometS3TestBase, but the runs so far used S3Proxy through a local patch, so the code that would merge hasn't run yet. Could you run it once as committed on 3.5 and 4.1? If Docker isn't available, making the backend selectable in CometS3TestBase, so the S3Proxy setup is reproducible from the PR itself, would work too. This PR also changes how every native write's task file is routed, so could you trigger the manual Spark SQL Writer Tests workflow for 3.5 and 4.1?
Memory docs. The ParquetWriterExec bullet under "Native writers" in memory_management.md should mention the 1 MiB output-stream buffer that the reservation now includes. It should also say that S3A's own upload buffers (fs.s3a.fast.upload.buffer) sit outside Comet's pool, as they do for Spark's writer.
| * `None` when it can. Local and HDFS files are created natively. A write that S3A's magic | ||
| * committer commits is supported too, since each of its files is created through S3A (see | ||
| * `CometWriteFilesExec.writeFile`), which is the only way to give the committer the pending | ||
| * upload it completes. Other S3 writes stay on Spark until the native writer reproduces what |
There was a problem hiding this comment.
The per-task routing makes every S3A-served write correct. Staging committers hand out local file: paths, which the native writer creates itself, and the file committer's _temporary paths are created through Hadoop. So this gate now only decides scope, and "until the native writer reproduces what S3A does at create time" no longer describes why other S3 writes fall back. The same reason appears in the last test of ParquetWriteWithMagicCommitterSuite.
Since #6355 hasn't landed, the alternative for a non-magic S3A write today is Spark's writer, not object_store. Would admitting every S3A destination be the simpler choice? It would drop usesS3AMagicCommitter and its test, which mirror Hadoop's committer resolution and could drift as Hadoop changes. If you'd rather keep the scope, could the comments here and in the suite say the real reason (benchmarks against #6355)?
| factory match { | ||
| case Some(MagicCommitterFactoryClass) => true | ||
| case Some(S3ACommitterFactoryClass) => | ||
| // S3ACommitterFactory takes the name from the job's configuration when it has one, and |
There was a problem hiding this comment.
core-default.xml sets fs.s3a.committer.name=file in both 3.3.4 and 3.4.1, so with a defaults-loaded configuration this lookup always finds a job-level name, and the bucket fallback never applies. That matches S3ACommitterFactory, which also reads the task conf first, so it isn't wrong. But Spark never reaches this branch. Could it be dropped?
| !usesMagic(pathOutput, s3aSchemeFactory, "fs.s3a.committer.name" -> name)(), | ||
| s"committer name '$name'") | ||
| } | ||
| // A bucket's committer name applies when the job sets none of its own, and a global name in |
There was a problem hiding this comment.
These cases build new Configuration(false), so "a bucket's committer name applies when the job sets none" describes a setup Spark never has. Spark's Hadoop conf carries fs.s3a.committer.name=file from core-default.xml and, once JobConf has loaded, the s3a scheme factory from mapred-default.xml. Could you add a case with new Configuration() that pins the behavior Spark actually sees, or drop these assertions together with the fallback?
| // Whether the JVM has already created `output_path` through Hadoop's FileSystem API and | ||
| // registered its output stream with the plan (Native.setOutputStream). The native writer then | ||
| // writes the file's bytes to that stream instead of opening the path itself, and the JVM closes | ||
| // the file afterwards. Set per task by CometWriteFilesExec and CometNativeWriteExec for every |
There was a problem hiding this comment.
Only CometWriteFilesExec.writeFile sets this flag. CometNativeWriteExec reaches it through that helper. More generally, the explanation of why the magic committer needs S3A to create the file now appears in about eight places: here, HadoopFileOutput, both exec scaladocs, writeFile, NativeWriteUtils, parquet_writer.rs and operators.md. Could it live once, in HadoopFileOutput or writeFile, with the other places referring to it? This sentence shows the copies are already drifting.
3 participants
Which issue does this PR close?
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). The magic committer needs S3A itself to create each file. S3A turns a file created under
__magic_job-<id>/into a multipart upload to the file's final key, leaves the upload pending, and records it for task commit, either in a.pendingfile or, withfs.s3a.committer.magic.track.commits.in.memory.enabled, in memory. A writer that creates the object on its own bypasses all of that, and the job then succeeds without its data.This takes the first approach from the issue: S3A creates the file, and the native writer only supplies its bytes. The magic committer's files never go through
object_store, so this does not depend on #6355.It covers Spark 3.4 and 3.5 as well. The issue limits this to Spark 4.0+ because the 3.x writer used to build its Hadoop contexts without the session's settings, but #4746 has since made it run the session's commit protocol with the write's Hadoop configuration, so it reaches S3A's committers the same way Spark's writer does.
What changes are included in this PR?
This changes both native writers:
CometWriteFilesExecon Spark 4.0+ andCometNativeWriteExecon Spark 3.4 and 3.5.Per task, in both writers (
CometWriteFilesExec.writeFile):__magic…/__base/part-…paths, is created by the JVM with the task's Hadoop configuration (HadoopFileOutput). It uses the sameFileSystem.createcall that parquet-mr'sHadoopOutputFile.createmakes for Spark's writer. The native writer streams the encoded file into that stream. The JVM then closes the stream, which is where S3A records the pending upload, before the write's statistics ask for the file's size.Native:
OutputStreamSinkandJavaOutputStreamSinkin the JNI bridge. The JVM registers the sink with the plan before execution throughNative.setOutputStream, as it does for the remote shuffle pusher. Each upcall copies up to 1 MiB into a Java array the sink allocates once.ParquetWriterExecgets anOutputStreamwriter: anArrowWriterover a bufferingWriteadapter. It never opensoutput_path. After a failed write it sends nothing more, not even what dropping the writer flushes, and the task fails with that write's error unchanged. Itsbytes_writtenmetric reports what it wrote, as the local writer's now does too, rather than asking the file system.ParquetWriter.jvm_output_streamin the proto. The planner rejects a writer planned to write to the stream when none is registered, and a registered stream for a writer that creates its own file.Planning, in
CometWriteFilesandCometDataWritingCommand(NativeWriteUtils.unsupportedFileSystem):NativeWriteUtils.usesS3AMagicCommitter). That meansPathOutputCommitProtocol, or Spark's default protocol withBindingParquetOutputCommitter, and a committer factory, taken frommapreduce.outputcommitter.factory.classor elsemapreduce.outputcommitter.factory.scheme.<scheme>, that is eitherMagicS3GuardCommitterFactoryorS3ACommitterFactorywith committer namemagic. The name is resolved the wayS3ACommitterFactoryresolves it: the job'sfs.s3a.committer.namefirst, then the bucket's.object_storeupload.Tests and docs:
PathOutputCommitProtocolandBindingParquetOutputCommitter. With it on the classpath, Spark 4.1 would enable the magic committer for every test session, so the scalatest system properties setspark.hadoop.fs.s3a.committer.magic.enabled=false, and the magic committer suite turns it back on.operators.mddocuments the supported destinations.Cost of the JVM route:
RawLocalFileSystemso that only the route differs, the route costs nothing beyond run-to-run noise: median 2172 ms against 2176 ms for snappy output (269 MiB), and 1926 ms against 1896 ms uncompressed (643 MiB, about 1% more CPU). GC time does not change. Spark's own writer takes about 5.8 s on the same data. The setup and full numbers are in Upload S3A magic committer files natively instead of through the JVM #6944.How are these changes tested?
bytes_writtenmatches the file.Writeadapter sends nothing more, not even what a later flush would send, and a failed send consumes none of the bytes it was given.CometParquetWriterSuite, which runs in CI:IOException.ParquetWriteWithMagicCommitterSuite, new and manual like the other MinIO suites, listed indev/ci/check-suites.py. It uses the magic committer as Spark itself configures it, by default on 4.1+ and through the per-bucket key on earlier versions:_SUCCESSmanifest names the magic committer, no__magicdirectory or pending upload is left, and the write's reported byte count matches the objects. On 4.0+ that count comes from the marker S3A writes when it closes a magic file, so the check also covers the close ordering. On 3.x it comes from the native writer's own count.fs.s3a.committer.abort.pending.uploads, whichcore-default.xmlturns on, because S3A's job abort would otherwise abort the upload itself and hide a writer that never aborts its file.fileanddirectorycommitters stay on Spark.Run locally on macOS (arm64):
cargo test -p datafusion-comet --lib, filtered to the writer and registration tests: 12 passed, and the 4 HDFS tests are ignored as before.CometParquetWriterSuiteplusParquetWriteWithMagicCommitterSuite:CometS3TestBasepatch that is not part of this PR.flushthat sends nothing, and a writer that forgets a failed write.cargo clippy --all-targets --workspace -- -D warningson stable and on nightly 1.99,cargo fmt, spotless, scalastyle, scalafix in check mode on 3.5, and the strict-warnings compile on 3.5 with the delta contrib.Not run: the magic committer suite against real MinIO.