Upload S3A magic committer files natively instead of through the JVM #6944

What is the problem the feature request solves?

#6940 adds S3A magic committer support to native Parquet writes by letting S3A create each file. The JVM opens the file with Hadoop's FileSystem.create, and the native writer streams its encoded Parquet bytes into that stream over JNI. S3A then turns the file into the pending multipart upload that the committer completes at job commit, and applies its own write settings. The encoding is native. The last leg of the bytes, and the upload itself, are not.

The JNI leg costs little. I measured the native writer writing local files directly against the same writer writing them through the #6940 route. The route writes to local disk through Hadoop's RawLocalFileSystem under its own scheme, so only the route differs. Medians of 5 interleaved rounds:

The difference is within run-to-run noise, about 1% of CPU uncompressed, which is the route's worst case. GC time does not change. Per file, the route costs:

• two extra copies of the encoded bytes, into the writer's 1 MiB buffer and then into a Java array;
• one JNI upcall per MiB;
• 1 MiB of native memory, counted against the task's memory pool, and one 1 MiB Java array, allocated once per file.

Setup: macOS arm64, Spark 4.1.3, libcomet built --release without LTO and with 16 codegen units, local[2], 20M rows of five columns (long, int, double, string, long) read through a native scan, two write tasks, and a warm-up round.

What the route leaves in the JVM is the upload, and it is exactly the upload Spark's writer makes. By default S3A stages each file on local disk in 64 MiB blocks, up to four per file (fs.s3a.fast.upload.buffer=disk, fs.s3a.multipart.size, fs.s3a.fast.upload.active.blocks), and uploads them with the AWS Java SDK. A native upload through object_store would skip the local disk and the JVM entirely. It would hold parts in native memory instead: the uploader in #6355 keeps up to nine 10 MiB parts per file. Nobody has measured whether that is faster or cheaper on real S3.

Describe the potential solution

Options, from the least to the most of S3A's job taken over:

1. Keep the JVM route and trim it. Writing straight into the Java array would save one of the two extra copies. Given the numbers above, this alone is worth little.
2. Upload natively, and record the commit through S3A. The native writer uploads the parts and, on close, hands the upload ID and part ETags to the JVM. The JVM records the pending commit through the tracker S3A would have created for the path, so S3A still writes its own .pending file, or its in-memory entry under fs.s3a.committer.magic.track.commits.in.memory.enabled. This leaves S3A's commit formats with S3A, but it reaches into non-public S3A classes (MagicCommitIntegration, PutTracker). The upload settings would also have to be reproduced natively, since S3A applies them when it starts an upload. A variant has the JVM start the upload through S3A and pass the upload ID to native code, which then only uploads parts.
3. Reproduce the magic committer's file protocol natively. For a file under a magic path, the native writer would compute the final key as MagicCommitPaths.finalDestination does, start a multipart upload to it, and upload the parts. On close it would write what S3MagicCommitTracker.aboutToComplete writes: a zero-byte marker at the magic path with the file's length in the x-hadoop-s3a-magic-data-length header, and the .pending JSON (SinglePendingCommit: upload ID, bucket, destination key, part ETags, length). Task and job commit stay with S3A. object_store's MultipartStore exposes the upload ID and ETags this needs, while put_multipart hides them. The costs:
   • Comet would depend on S3A's internal, versioned .pending format and on path rules that differ across Hadoop lines (__magic on 3.3, __magic_job-<id> on 3.4).
   • It cannot serve the in-memory tracker, where task commit reads S3A's in-process state rather than .pending files, so that setting would keep the JVM route.
   • Every S3A setting that applies to an upload would have to be translated to object_store, or force the JVM route: encryption, canned ACL, storage class, content encoding, fs.s3a.create.header.*, checksums. object_store 0.13 supports SSE-KMS, DSSE-KMS, SSE-C, storage class, content encoding and user metadata, but not canned ACLs.
4. A Comet commit protocol. A FileCommitProtocol of Comet's own for S3 destinations: tasks upload natively and return their upload IDs and ETags in their commit messages, and job commit completes the uploads. The bytes are native end to end and nothing depends on S3A internals. But Comet would own a committer, including speculation, task retries, cleanup after a job abort, and _SUCCESS compatibility, and users would have to configure it in place of S3A's magic committer.

The suggested next step is a benchmark of S3A's upload against object_store's on real S3: throughput, CPU, local disk I/O and memory, with S3A's default disk buffering and with bytebuffer. #6929 asks for the same measurement before sending every S3A write through S3A, and it decides whether any option above is worth its coupling. If a native upload wins, option 3 restricted to the configurations it can reproduce, with the #6940 route as the fallback, looks like the smallest step. It would share #6355's upload code, moved from put_multipart to MultipartStore.

Additional context

Follows #6940 and #6929. Related to #6355 and #1625.