You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "comphead (via GitHub)" <gi...@apache.org> on 2026/09/28 20:17:44 UTC
[PR] feat: native Parquet writes to S3 [datafusion-comet]
comphead opened a new pull request, #6355:
URL: https://github.com/apache/datafusion-comet/pull/6355
## Which issue does this PR close?
Closes #6007.
## Rationale for this change
With `spark.comet.parquet.write.enabled=true`, both native write serdes, `CometWriteFiles` on Spark 4.0+ and `CometDataWritingCommand` on Spark 3.x, decline every output path that is not `file:` or `hdfs:`. Native Parquet writes therefore never run against S3, where most production writes go.
Admitting `s3a:` alone would not be safe, for three reasons found on `main`:
- `objectstore::s3::create_store` hardcodes `AccessMode::Read`, and the process-wide `object_store` cache has no access mode in its key. A write would reuse a scan's store and present whatever a `CometS3CredentialProvider` grants for reads.
- The writer receives Hadoop's unescaped `Path.toString` but derives the object key the way the scan does, by percent-decoding the URL path. A name holding `%`, `?` or `#` would be written under a different key from the one Spark commits, and the job would succeed without its data.
- S3A does work at `create()` that an `object_store` upload bypasses: the magic committer's pending uploads, and the encryption, ACL, storage class, content encoding and headers it applies to every object it creates.
## What changes are included in this PR?
Native:
- `parquet_writer.rs`: `ParquetWriter::ObjectStore` wraps an `AsyncArrowWriter` over `object_store::buffered::BufWriter`. A file under 10 MiB goes up in one PUT, a larger one as a multipart upload that only becomes an object when `close` completes it. A write that fails before `close` aborts the upload. The `bytes_written` metric now reports the uploaded size. It read 0 for anything but a local file. URLs that normalize to `s3` (`s3`, `s3a` and configured aliases) route to the new writer, and other object store schemes still fail with `Unsupported storage scheme`.
- `parquet_support.rs`: `prepare_object_store_with_configs` is split into `resolve_object_store` (normalization, backend selection and the cache) and the `RuntimeEnv` registration. The new `object_store_for_write` resolves under `AccessMode::Write` and registers nothing. The cache key gains the access mode. Scans resolve and register exactly as before.
- `objectstore/s3.rs`: `create_store` takes the access mode it forwards to the `CometS3CredentialBridge`.
- `native/core/Cargo.toml` names the `parquet` crate's `async` and `object_store` features, which DataFusion already enables.
JVM:
- `NativeWriteUtils.unsupportedDestination` is now the filesystem gate of both serdes. Local and HDFS destinations are gated as before. An S3 destination is admitted only when:
- its scheme is `s3a` or `s3`, and Hadoop serves it with `org.apache.hadoop.fs.s3a.S3AFileSystem` itself. `s3://` on EMRFS, subclasses of S3A and the `fs.comet.s3Compliant.schemes` aliases stay on Spark.
- `fs.comet.libhdfs.schemes` does not route it through libhdfs.
- neither the path nor `mapreduce.output.basename` contains `%`, `?`, `#` or an ASCII control character. Spaces and non-ASCII names survive the native URL round trip and stay native.
- it is not under a `__magic*` directory, and `fs.s3a.committer.name` (global or per-bucket) is not `magic`.
- none of `fs.s3a.encryption.algorithm`, `fs.s3a.server-side-encryption-algorithm`, `fs.s3a.acl.default`, `fs.s3a.create.storage.class`, `fs.s3a.object.content.encoding` or `fs.s3a.create.header.*` is set for the bucket.
- `checkNativeWriteDestination`, which runs on the path the commit protocol hands each task on Spark 4.0+, repeats the path and magic-directory checks.
- Docs: `operators.md`, `s3-credential-providers.md` and `s3-credential-provider-design.md`, the `operator.proto` comment that counted the S3A magic committer among the committers that work, and the `CometS3AccessMode.WRITE` Javadoc.
Left out on purpose, possible follow-ups:
- Translating S3A's SSE settings into `object_store`'s encryption options instead of falling back.
- `gs://`, `abfss://` and the `fs.comet.s3Compliant.schemes` aliases.
- Aborting the upload when a task is killed rather than failed. The writer is then dropped without an abort, and the incomplete upload is left for the bucket's lifecycle rules.
- Accounting for the upload buffers (10 MiB parts, up to 8 in flight per writer) in Comet's memory pool.
## How are these changes tested?
- Rust unit tests, against an in-memory store seeded into the `object_store` cache so that no request leaves the process:
- a small file lands in one PUT at the key the scan reads, through a directory with a space and a non-ASCII name, and `bytes_written` matches its size,
- a row group of about 16 MiB is streamed as a multipart upload and completed,
- a write that fails after its multipart upload started aborts it and leaves no object,
- other object store schemes are rejected before any store is built,
- a writer and a scan of the same bucket and configuration get separate stores.
- `CometParquetWriterSuite`: `S3 destinations are admitted only where the native writer matches S3A` covers each rule above without an S3 endpoint, plus the task-time check and its agreement with planning.
- `ParquetWriteToS3Suite` is new and runs against MinIO. It is a manual suite like the other MinIO suites, so it is listed in `dev/ci/check-suites.py`. It checks that native writes are committed where Spark reads them (and `_SUCCESS` on 4.0+), that a file larger than the upload buffer is uploaded in parts (from its ETag), `Append` and `Overwrite`, that a directory with a space and a non-ASCII name stays native, that `%` and the magic committer fall back, and that a per-bucket `CometS3CredentialProvider` is asked for `WRITE` only.
Verified locally on the default Spark 4.1 profile: `cargo clippy --all-targets --workspace -- -D warnings`, `cargo fmt --check`, and `./mvnw test-compile` with scalastyle. The new tests have not been run yet, and neither have the other Spark profiles or scalafix. `ParquetWriteToS3Suite` uses `CometS3TestBase` and its `minio/minio:latest` image, like the other MinIO suites, and anonymous pulls of MinIO images have been reported failing (apache/datafusion#25215).
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native Parquet writes to S3 [datafusion-comet]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on code in PR #6355:
URL: https://github.com/apache/datafusion-comet/pull/6355#discussion_r4253280136
##########
spark/src/main/scala/org/apache/comet/serde/operator/NativeWriteUtils.scala:
##########
@@ -172,19 +177,215 @@ object NativeWriteUtils {
/**
* Fail the task if the native writer would not write to exactly `filePath`.
*
- * [[escapedHdfsDestination]] declines the shapes Comet can predict at planning time, but the
+ * [[unsupportedDestination]] declines the shapes Comet can predict at planning time, but the
* path a write actually uses comes from `FileCommitProtocol.newTaskTempFile`, and a custom
* commit protocol can return anything. This is the backstop: it runs before the writer opens
* anything, so the task fails with a clear message rather than committing successfully with the
* data left somewhere else.
*/
- def checkNativeWriteDestination(filePath: String): Unit =
+ def checkNativeWriteDestination(filePath: String): Unit = {
hdfsPathDivergence(filePath).foreach { shown =>
throw new UnsupportedOperationException(
s"Comet's native Parquet writer cannot write to '$filePath': the path it would create " +
s"on HDFS is not the one the commit protocol chose ($shown). Set " +
"spark.comet.write.parquet.enabled=false to write this table with Spark.")
}
+ if (s3Scheme(filePath).isDefined &&
+ (s3PathDivergence(filePath).isDefined || isS3AMagicPath(filePath))) {
+ throw new UnsupportedOperationException(
+ s"Comet's native Parquet writer cannot write to '$filePath': the object it would create " +
+ "on S3 is not the one the commit protocol expects. Set " +
+ "spark.comet.write.parquet.enabled=false to write this table with Spark.")
+ }
+ }
+
+ /**
+ * A fallback reason when the native writer cannot write `outputPath`'s files, or `None` when it
+ * can. Both native write serdes gate on this, so they admit exactly the same destinations.
+ *
+ * @param fileNamePrefix
+ * the basename every file name starts with, see [[escapedHdfsDestination]]
+ * @param hadoopConf
+ * the write's Hadoop configuration, the one its object store options are extracted from
+ */
+ def unsupportedDestination(
+ outputPath: String,
+ fileNamePrefix: String,
+ hadoopConf: Configuration): Option[String] = {
+ if (outputPath.startsWith("file:")) {
+ None
+ } else if (outputPath.startsWith("hdfs:")) {
+ escapedHdfsDestination(outputPath, fileNamePrefix)
+ } else {
+ s3Scheme(outputPath) match {
+ case Some(scheme) =>
+ unsupportedS3Destination(outputPath, scheme, fileNamePrefix, hadoopConf)
+ case None => Some("Supported output filesystems: local, HDFS, S3 (through S3A)")
+ }
+ }
+ }
+
+ /**
+ * The schemes the native writer uploads through `object_store`'s S3 client. Configured
+ * `fs.comet.s3Compliant.schemes` aliases are not among them even though the scan reads them:
+ * the native side reads their settings through vendor key translation, so a write could reach a
+ * different endpoint than the one Spark's committer then lists.
+ */
+ private val S3Schemes: Seq[String] = Seq("s3", "s3a")
+
+ /**
+ * The only Hadoop FileSystem whose writes the native S3 writer reproduces. Named rather than
+ * referenced, so that a deployment without hadoop-aws still plans writes and falls back.
+ */
+ private val S3AFileSystemClassName = "org.apache.hadoop.fs.s3a.S3AFileSystem"
+
+ /**
+ * S3A settings, without their `fs.s3a.` prefix, that change the objects S3A creates. The native
+ * writer does not apply them, so a write that sets one stays on Spark's writer rather than
+ * produce files without the encryption, ACL, storage class or content encoding it asks for. The
+ * list covers Hadoop 3.3 and 3.4. Each setting is its current name followed by any deprecated
+ * one, see [[effectiveS3AKey]].
+ */
+ private val UnsupportedS3AWriteOptions: Seq[Seq[String]] = Seq(
+ // Server-side and client-side encryption, under its current and its deprecated name
+ Seq("encryption.algorithm", "server-side-encryption-algorithm"),
+ Seq("acl.default"),
+ Seq("create.storage.class"),
+ Seq("object.content.encoding"))
+
+ /** Custom object headers S3A adds to every file it creates. */
+ private val S3ACreateHeaderPrefix = "create.header."
+
+ private def s3Scheme(path: String): Option[String] =
+ S3Schemes.find(scheme => path.startsWith(s"$scheme:"))
+
+ /**
+ * The first character of `path` that the native S3 writer would not carry into the object key,
+ * if any.
+ *
+ * The native writer derives the key the way the Parquet scan does: it parses the path as a URL
+ * and percent-decodes the URL's path. It is handed `Path.toString`, which Hadoop leaves
+ * unescaped, so that is only right where escaping and decoding cancel out. They do for the
+ * characters the URL parser escapes, among them spaces and every non-ASCII character, which
+ * come back unchanged. They do not for:
+ *
+ * - `%`, which can start an escape sequence that decoding then replaces. `50%25` is written
+ * as `50%`, and `%2F` even adds a directory level.
+ * - `?` and `#`, which the parser takes for the start of the query and the fragment, so the
+ * path is truncated there.
+ * - ASCII control characters. The parser drops tabs and line breaks outright, and
+ * `object_store` rejects the rest, which would fail the task.
+ *
+ * Returns the offending character as a fallback reason would show it.
+ */
+ private def s3PathDivergence(path: String): Option[String] =
+ path.find(c => c == '%' || c == '?' || c == '#' || c < ' ' || c == '\u007f').map { c =>
+ if (c == '%' || c == '?' || c == '#') s"'$c'" else f"control character U+${c.toInt}%04X"
+ }
+
+ /**
+ * Whether `path` is under an S3A magic committer directory: an element that starts with
+ * `__magic`, which is `__magic` on Hadoop 3.3 and `__magic_job-<id>` on 3.4. S3A intercepts a
+ * file created there, uploading it to its final destination as a multipart upload it leaves
+ * incomplete for the committer to complete at job commit. The native writer talks to S3
+ * directly, so it would create a real object in the magic directory instead, which the
+ * committer never lists and job cleanup deletes: the job would succeed without its data.
+ */
+ private def isS3AMagicPath(path: String): Boolean =
+ path.split('/').exists(_.startsWith("__magic"))
+
+ /**
+ * The key S3A takes the setting `names` from for `bucket`, when it holds a value. As in S3A's
+ * `buildEncryptionSecrets`, that is the first key set among `fs.s3a.bucket.<bucket>.<name>` for
+ * each name, then `fs.s3a.<name>` for each name. A bucket key under either name overrides the
+ * global keys under both, which matters because Hadoop aliases the two global keys once S3A is
+ * loaded but leaves the bucket keys apart. A blank value, which S3A treats as no setting, does
+ * not count.
+ */
+ private def effectiveS3AKey(
+ hadoopConf: Configuration,
+ bucket: String,
+ names: Seq[String]): Option[String] =
+ (names.map(name => s"fs.s3a.bucket.$bucket.$name") ++ names.map(name => s"fs.s3a.$name"))
+ .find(key => hadoopConf.getTrimmed(key) != null)
+ .filter(key => hadoopConf.getTrimmed(key).nonEmpty)
+
+ /** The FileSystem class Hadoop would serve `scheme` with, or `None` when it has none. */
+ private def fileSystemClassName(scheme: String, hadoopConf: Configuration): Option[String] =
+ try {
+ Some(FileSystem.getFileSystemClass(scheme, hadoopConf).getName)
+ } catch {
+ case NonFatal(_) => None
+ }
+
+ /**
+ * Why the native writer cannot write an S3 destination, or `None` when it can. The native
+ * writer uploads with `object_store` rather than through S3A, so this admits only destinations
+ * where the two produce the same objects.
+ */
+ private def unsupportedS3Destination(
+ outputPath: String,
+ scheme: String,
+ fileNamePrefix: String,
+ hadoopConf: Configuration): Option[String] = {
+ // The same configuration `NativeConfig.extractObjectStoreOptions` forwards, so this matches
+ // the native routing, which would send the write to the HDFS writer instead.
+ if (NativeConfig.parseSchemeSet(hadoopConf.get(COMET_LIBHDFS_SCHEMES_KEY)).contains(scheme)) {
+ return Some(
+ s"$scheme:// output routed through libhdfs by $COMET_LIBHDFS_SCHEMES_KEY is " +
+ "not supported")
+ }
+ // Asking Hadoop rather than trusting the scheme is what keeps an `s3://` served by something
+ // other than S3A, such as EMRFS, on Spark's writer: its committers can depend on its own
+ // output streams, which the native writer would bypass.
+ val fileSystem = fileSystemClassName(scheme, hadoopConf)
+ if (!fileSystem.contains(S3AFileSystemClassName)) {
+ return Some(
+ s"S3 output is only supported through S3A, but $scheme:// is served by " +
+ fileSystem.getOrElse("no loadable FileSystem"))
+ }
+ s3PathDivergence(outputPath).foreach { shown =>
+ return Some(
+ s"S3 output paths containing $shown are not supported: the native writer would write " +
+ s"to a different key from the one Spark commits ($outputPath)")
+ }
+ // The basename goes into every file name, see `escapedHdfsDestination`.
+ s3PathDivergence(fileNamePrefix).foreach { shown =>
+ return Some(
+ s"S3 output file names containing $shown are not supported: " +
+ s"$BASE_OUTPUT_NAME=$fileNamePrefix would make the native writer create a different " +
+ "object from the one Spark commits")
+ }
+ if (isS3AMagicPath(outputPath)) {
Review Comment:
Is this needed if we are using Spark 4 or newer? Since Spark 4, the file writign and commit protocol are decoupled via `WriteFilesExec`.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native Parquet writes to S3 [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6355:
URL: https://github.com/apache/datafusion-comet/pull/6355#issuecomment-6267946225
@comphead should we just route the bytes through the JVM using Hadoop writer, like I am doing in the magic committer PR?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native Parquet writes to S3 [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on PR #6355:
URL: https://github.com/apache/datafusion-comet/pull/6355#issuecomment-6028286150
@parthchandra thanks for the review, its a draft PR to shape the direction and I'm planning to test it out this week
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native Parquet writes to S3 [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6355:
URL: https://github.com/apache/datafusion-comet/pull/6355#issuecomment-6134071631
Thanks @comphead. Some notes on the S3A magic committer handling. Falling back is the right call, since `object_store` bypasses the S3A magic-path handling the committer depends on. I opened #6929 to track supporting it.
1. **Spark 4.1 turns the magic committer on by default.** When spark-hadoop-cloud is on the classpath, `SparkContext.enableMagicCommitterIfNeeded` (SPARK-47618) sets `fs.s3a.committer.name=magic`, `PathOutputCommitProtocol` and `BindingParquetOutputCommitter` unless they are already set. Earlier versions only do this when a bucket sets `fs.s3a.bucket.<bucket>.committer.magic.enabled=true`. So on 4.1 deployments that ship spark-hadoop-cloud, the native S3 path here will fall back unless the user opts out of magic. Worth saying in `operators.md`.
2. **The magic check misses the committer factory keys.** When the S3A committers are wired up (`PathOutputCommitProtocol` or `BindingParquetOutputCommitter`), `PathOutputCommitterFactory` picks the factory from `mapreduce.outputcommitter.factory.class` and then `mapreduce.outputcommitter.factory.scheme.s3a`. Naming `org.apache.hadoop.fs.s3a.commit.magic.MagicS3GuardCommitterFactory` in either one selects the magic committer without `fs.s3a.committer.name`. `unsupportedS3Destination` admits that write, then `checkNativeWriteDestination` fails every task on the `__magic_job-*` path, so the job fails where Spark's writer would succeed. Declining when either key names `MagicS3GuardCommitterFactory` would close the gap.
3. **No test writes through the magic committer.** Comet's test classpath only has spark-hadoop-cloud's `tests` jar, so `PathOutputCommitProtocol` can't load and Spark 4.1's default never applies in our tests. The MinIO test sets `fs.s3a.committer.name=magic` while Spark's default protocol still uses `FileOutputCommitter`, so it shows the planning check fires but not what happens on a real magic write. An end-to-end test needs the main jar in test scope. That turns on the 4.1 magic defaults for every test session unless the shared test config sets `spark.hadoop.fs.s3a.committer.magic.enabled=false`.
4. **On Spark 3.x no S3A committer is ever used.** `CometNativeWriteExec` hard-codes `SQLHadoopMapReduceCommitProtocol` and builds its Hadoop contexts with `Job.getInstance()`, which has neither the session's Hadoop settings nor an output path. Replaying those calls in a Spark 3.5.8 shell gives a `FileOutputCommitter` whose output path and work path are both null. `newTaskTempFile` then returns a path directly in the destination, so nothing is staged and no `_SUCCESS` is written (#3015). Admitting S3 on 3.x would silently ignore a configured `directory` or `partitioned` committer, so I'd keep native S3 writes to Spark 4.0+ in this PR.
5. **A direction question, since this is a draft.** Supporting the magic committer probably means letting S3A create the file. If we go that way for magic, it may be worth sending all S3A writes through S3A, which would also cover the encryption, ACL, storage class, content encoding and header settings this PR falls back on. That's a question for #6929 rather than a blocker here, but it may affect how far to take the `object_store` path.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native Parquet writes to S3 [datafusion-comet]
Posted by "comphead (via GitHub) via github" <gi...@datafusion.apache.org>.
comphead closed pull request #6355: feat: native Parquet writes to S3
URL: https://github.com/apache/datafusion-comet/pull/6355
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: native Parquet writes to S3 [datafusion-comet]
Posted by "parthchandra (via GitHub)" <gi...@apache.org>.
parthchandra commented on code in PR #6355:
URL: https://github.com/apache/datafusion-comet/pull/6355#discussion_r4190659247
##########
native/core/src/execution/operators/parquet_writer.rs:
##########
@@ -402,6 +435,25 @@ impl ParquetWriterExec {
})?;
Ok(ParquetWriter::LocalFile(writer))
}
+ } else if Self::is_s3_destination(output_file_path, object_store_options)? {
+ // The store is resolved exactly as the Parquet scan resolves it, including the
+ // `fs.s3a.*` endpoint, region and credential settings, but under the write access mode.
+ let (store, path) = object_store_for_write(output_file_path, object_store_options)
+ .map_err(|e| {
+ DataFusionError::Execution(format!(
+ "Failed to prepare object store for '{}': {}",
+ output_file_path, e
+ ))
+ })?;
+ let writer =
+ AsyncArrowWriter::try_new(BufWriter::new(store, path), schema, Some(props))
Review Comment:
Can you confirm what the buffer size here will be? This is probably not tracked by the Comet allocator and could potentially be quite large.
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetWriteToS3Suite.scala:
##########
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.parquet
+
+import java.net.URI
+
+import org.apache.hadoop.fs.{FileStatus, Path}
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{DataFrame, Row, SaveMode}
+
+import org.apache.comet.{CometConf, CometS3TestBase}
+import org.apache.comet.CometSparkSessionExtensions.isSpark40Plus
+import org.apache.comet.cloud.s3.{CometS3AccessMode, MinioCometS3CredentialProvider}
+
+import software.amazon.awssdk.auth.credentials.{AwsBasicCredentials, StaticCredentialsProvider}
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.s3.S3Client
+import software.amazon.awssdk.services.s3.model.HeadObjectRequest
+
+/**
+ * Native Parquet writes to S3, against MinIO. Which S3 destinations the native writer accepts is
+ * covered without an S3 endpoint in `CometParquetWriterSuite`; this suite checks that the ones it
+ * accepts are written where Spark's commit protocol expects them.
+ *
+ * A manual suite, like the other MinIO suites: it needs Docker, so CI does not run it (see
+ * `dev/ci/check-suites.py`).
+ */
+class ParquetWriteToS3Suite extends CometParquetWriterTestBase with CometS3TestBase {
+
+ import testImplicits._
+
+ override protected val testBucketName = "native-write-bucket"
+
+ /** A bucket whose native access goes through [[MinioCometS3CredentialProvider]]. */
+ private val credentialBucket = "native-write-credential-bucket"
+
+ override protected def sparkConf: SparkConf = {
+ val conf = super.sparkConf
+ conf.set(
+ s"spark.hadoop.fs.s3a.bucket.$credentialBucket.comet.credential.provider.class",
+ classOf[MinioCometS3CredentialProvider].getName)
+ conf
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ MinioCometS3CredentialProvider.installCredentials(userName, password)
+ createBucketIfNotExists(credentialBucket)
+ }
+
+ private def s3Path(key: String, bucket: String = testBucketName): String =
+ s"s3a://$bucket/$key"
+
+ /**
+ * Persist `df` to S3 with Spark's writer and return a DataFrame that reads it back, so that the
+ * write under test has the Comet scan below it that native writes require.
+ */
+ private def cometSource(df: DataFrame, name: String): DataFrame = {
+ val path = s3Path(s"sources/$name")
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ df.write.mode(SaveMode.Overwrite).parquet(path)
+ }
+ spark.read.parquet(path)
+ }
+
+ private def someRows(name: String): DataFrame =
+ cometSource((1 to 1000).map(i => (i, s"name_$i")).toDF("id", "name").repartition(4), name)
Review Comment:
Can we add coverage with decimal, timestamp, and nested types (nested struct with maps, arrays).
##########
spark/src/test/scala/org/apache/comet/parquet/ParquetWriteToS3Suite.scala:
##########
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.parquet
Review Comment:
The PR says these haven't been run. Can we run the tests manually and post the results?
##########
native/core/src/execution/operators/parquet_writer.rs:
##########
@@ -550,47 +620,67 @@ impl ExecutionPlan for ParquetWriterExec {
let mut stream = input;
let mut total_rows = 0i64;
- while let Some(batch_result) = stream.try_next().await.transpose() {
- let batch = batch_result?;
-
- // Track row count
- total_rows += batch.num_rows() as i64;
-
- // Rename columns in the batch to match output schema
- let renamed_batch = if !column_names.is_empty() {
- // Collection field IDs exist on the target schema, not on arrays produced by
- // the placeholder Scan. Both schemas use the same Catalyst data types, and
- // disabling field-name matching still recursively validates nested nullability;
- // only nested field names and metadata are ignored.
- RecordBatch::try_new_with_options(
- Arc::clone(&schema_for_write),
- batch.columns().to_vec(),
- &RecordBatchOptions::new().with_match_field_names(false),
- )
- .map_err(|e| {
- DataFusionError::Execution(format!("Failed to rename batch columns: {}", e))
- })?
- } else {
- batch
- };
-
- writer.write(&renamed_batch).await.map_err(|e| {
- DataFusionError::Execution(format!("Failed to write batch: {}", e))
- })?;
+ // A failure here leaves the file unfinished, so keep it rather than return it at once:
+ // the writer has to discard what it may already have uploaded.
+ let written: Result<()> = async {
+ while let Some(batch_result) = stream.try_next().await.transpose() {
+ let batch = batch_result?;
+
+ // Track row count
+ total_rows += batch.num_rows() as i64;
+
+ // Rename columns in the batch to match output schema
+ let renamed_batch = if !column_names.is_empty() {
+ // Collection field IDs exist on the target schema, not on arrays produced
+ // by the placeholder Scan. Both schemas use the same Catalyst data types,
+ // and disabling field-name matching still recursively validates nested
+ // nullability; only nested field names and metadata are ignored.
+ RecordBatch::try_new_with_options(
+ Arc::clone(&schema_for_write),
+ batch.columns().to_vec(),
+ &RecordBatchOptions::new().with_match_field_names(false),
+ )
+ .map_err(|e| {
+ DataFusionError::Execution(format!(
+ "Failed to rename batch columns: {}",
+ e
+ ))
+ })?
+ } else {
+ batch
+ };
+
+ writer.write(&renamed_batch).await.map_err(|e| {
+ DataFusionError::Execution(format!("Failed to write batch: {}", e))
+ })?;
+ }
+ Ok(())
+ }
+ .await;
+ if let Err(e) = written {
+ if let Err(abort_error) = writer.abort().await {
+ log::warn!("Failed to abort the write of '{part_file}': {abort_error}");
+ }
+ return Err(e);
}
- writer.close().await.map_err(|e| {
+ let uploaded_size = writer.close().await.map_err(|e| {
Review Comment:
This ultimately calls `complete` which might fail too, but in that case we are not calling `abort`
##########
spark/src/main/scala/org/apache/comet/serde/operator/NativeWriteUtils.scala:
##########
@@ -172,19 +177,215 @@ object NativeWriteUtils {
/**
* Fail the task if the native writer would not write to exactly `filePath`.
*
- * [[escapedHdfsDestination]] declines the shapes Comet can predict at planning time, but the
+ * [[unsupportedDestination]] declines the shapes Comet can predict at planning time, but the
* path a write actually uses comes from `FileCommitProtocol.newTaskTempFile`, and a custom
* commit protocol can return anything. This is the backstop: it runs before the writer opens
* anything, so the task fails with a clear message rather than committing successfully with the
* data left somewhere else.
*/
- def checkNativeWriteDestination(filePath: String): Unit =
+ def checkNativeWriteDestination(filePath: String): Unit = {
hdfsPathDivergence(filePath).foreach { shown =>
throw new UnsupportedOperationException(
s"Comet's native Parquet writer cannot write to '$filePath': the path it would create " +
s"on HDFS is not the one the commit protocol chose ($shown). Set " +
"spark.comet.write.parquet.enabled=false to write this table with Spark.")
}
+ if (s3Scheme(filePath).isDefined &&
+ (s3PathDivergence(filePath).isDefined || isS3AMagicPath(filePath))) {
+ throw new UnsupportedOperationException(
+ s"Comet's native Parquet writer cannot write to '$filePath': the object it would create " +
+ "on S3 is not the one the commit protocol expects. Set " +
+ "spark.comet.parquet.write.enabled=false to write this table with Spark.")
Review Comment:
In the message on line 191, the config is `spark.comet.write.parquet.enabled` whereas here it is `spark.comet.parquet.write.enabled`.
Can you correct to the right config.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org