andygrove opened a new pull request, #6940:
URL: https://github.com/apache/datafusion-comet/pull/6940
## 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 `.pending` file or, with
`fs.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.
## What changes are included in this PR?
This is Spark 4.0+ only (`CometWriteFiles` and `CometWriteFilesExec`). The
Spark 3.x writer is unchanged.
Per task, in `CometWriteFilesExec`:
- The native writer still creates local and HDFS files itself. Any other
file the commit protocol returns, such as the magic committer's
`__magic_job-…/__base/part-…` paths, is created by the JVM with the task's
Hadoop configuration (`HadoopFileOutput`). It uses the same `FileSystem.create`
call that parquet-mr's `HadoopOutputFile.create` makes 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 stats
trackers ask for the file's size.
- When the write fails, the stream is aborted if it supports that (S3A's
streams do), so a truncated file is never recorded as a pending upload. Other
streams are closed, as Spark's writer closes its file on failure. The task
rethrows the exception the stream threw, not just its text.
- The route is chosen from the path the commit protocol actually returned,
so a wrong guess at planning cannot misplace data: whichever committer runs, a
file that is neither local nor on HDFS is created through Hadoop.
Native:
- `OutputStreamSink` and `JavaOutputStreamSink` in the JNI bridge. The JVM
registers the sink with the plan before execution through
`Native.setOutputStream`, as it does for the remote shuffle pusher. Each upcall
copies up to 1 MiB into a Java array the sink allocates once.
- `ParquetWriterExec` gets an `OutputStream` writer: an `ArrowWriter` over a
buffering `Write` adapter. It never opens `output_path`. After a failed write
it sends nothing more, not even what dropping the writer flushes. Its
`bytes_written` metric reports what it wrote.
- `ParquetWriter.jvm_output_stream` in 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 `CometWriteFiles`:
- An S3 destination is admitted when the magic committer will commit it
(`NativeWriteUtils.usesS3AMagicCommitter`). That means
`PathOutputCommitProtocol`, or Spark's default protocol with
`BindingParquetOutputCommitter`, and a committer factory, taken from
`mapreduce.outputcommitter.factory.class` or else
`mapreduce.outputcommitter.factory.scheme.<scheme>`, that is either
`MagicS3GuardCommitterFactory` or `S3ACommitterFactory` with committer name
`magic`. The name is resolved the way `S3ACommitterFactory` resolves it: the
job's `fs.s3a.committer.name` first, then the bucket's.
- Other S3 writes still fall back to Spark. Routing every S3A write through
S3A would now be a one-line change to this gate, and would also cover the
encryption, ACL, storage class and header settings #6355 falls back on. That is
the open question in #6929, and it needs benchmarks against the `object_store`
upload.
Tests and docs:
- spark-hadoop-cloud's main jar is now in test scope, for
`PathOutputCommitProtocol` and `BindingParquetOutputCommitter`. With it on the
classpath, Spark 4.1 would enable the magic committer for every test session,
so the scalatest system properties set
`spark.hadoop.fs.s3a.committer.magic.enabled=false`, and the magic committer
suite turns it back on.
- `operators.md` documents the supported destinations.
## How are these changes tested?
- Rust unit tests:
- The stream writer's output is byte-identical to the local writer's, sent
in writes no larger than the sink's limit. It never opens its path, here a
magic committer path, and its `bytes_written` matches the file.
- A failed write to the sink fails the task, and nothing is sent after it.
- The output stream can be registered only once, and only before execution.
- The planner validates the stream against the proto flag.
- `CometParquetWriterSuite`, which runs in CI:
- A commit protocol hands out task files on a local-disk test FileSystem
under its own scheme, which the native writer cannot open. The write runs
natively, and every file is created through Hadoop and then closed.
- A failing stream gets its file aborted, and the task fails with the
stream's own `IOException`.
- The planning check resolves the committer correctly across bindings,
factory keys, committer names, and per-bucket settings.
- `ParquetWriteWithMagicCommitterSuite`, new and manual like the other MinIO
suites, listed in `dev/ci/check-suites.py`. It uses the magic committer as
Spark itself configures it:
- The `_SUCCESS` manifest names the magic committer, no `__magic`
directory or pending upload is left, and `numOutputBytes` matches the objects,
which S3A reports for a magic file only through the marker written when it is
closed.
- The same with commits tracked in memory.
- The same through `BindingParquetOutputCommitter` under Spark's default
protocol.
- A failed write leaves no pending upload.
- Writes under other committers stay on Spark.
Run locally on macOS (arm64), on the default Spark 4.1 profile:
- `cargo test -p datafusion-comet --lib`, filtered to the writer and
registration tests: 11 passed, and the 4 HDFS tests are ignored as before.
- `CometParquetWriterSuite`: 67 passed.
- `ParquetWriteWithMagicCommitterSuite`, against S3Proxy 2.9.0 in place of
MinIO (no Docker here, via a local-only `CometS3TestBase` patch that is not
part of this PR): 4 of 5 pass. The failed-write test currently fails because
the job's `__magic_job-<uuid>` directory is still there after the job aborts.
I'm checking whether Spark's own writer leaves it as well.
Not run yet: the Spark 4.0 and 4.2 profiles, compiling the 3.x profiles,
clippy, and scalafix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
