You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/06 12:46:20 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-6441-2739d376387d7e555f554f97442d5b3ec94c02ac created (now 8d8d1a26b1)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6441-2739d376387d7e555f554f97442d5b3ec94c02ac
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at 8d8d1a26b1 fix: gate unsupported Iceberg S3 write settings (#6441)
This branch includes the following new commits:
new 8d8d1a26b1 fix: gate unsupported Iceberg S3 write settings (#6441)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: fix: gate unsupported Iceberg S3 write settings (#6441)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6441-2739d376387d7e555f554f97442d5b3ec94c02ac
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 8d8d1a26b1563d14871e2881e01aebdc65df70e7
Author: Yu Hsun Hung <75...@users.noreply.github.com>
AuthorDate: Tue Oct 6 12:46:01 2026 +0000
fix: gate unsupported Iceberg S3 write settings (#6441)
* fix: gate unsupported Iceberg S3 write settings
* fix: reject dsse-kms and derive the S3A allow-list from NativeConfig
* docs: format the Iceberg write eligibility table
* fix: fall back when Iceberg AWS classes cannot be linked
Planning threw NoClassDefFoundError if the AWS SDK was absent, and a longer dotted bucket name was treated as a setting of the shorter target bucket.
* fix: preserve effective Iceberg S3 Hadoop configuration
* fix: attribute dotted S3 bucket keys using Hadoop property suffixes
* fix: preserve initialized FileIO configuration for native Iceberg writes
* fix: avoid forwarding Hadoop settings to non-Hadoop Iceberg FileIO
* fix: use resolved FileIO delegate for Iceberg write Hadoop settings
* docs: clarify Iceberg dictionary sampling memory limits
---------
Co-authored-by: Andy Grove <ag...@apache.org>
---
docs/source/contributor-guide/iceberg-writes.md | 37 +-
docs/source/user-guide/latest/iceberg-writes.md | 61 ++-
.../apache/comet/iceberg/IcebergReflection.scala | 38 +-
.../serde/operator/CometIcebergNativeWrite.scala | 337 ++++++++++++-
.../comet/CometIcebergWriteDetectionSuite.scala | 532 ++++++++++++++++++++-
5 files changed, 977 insertions(+), 28 deletions(-)
diff --git a/docs/source/contributor-guide/iceberg-writes.md b/docs/source/contributor-guide/iceberg-writes.md
index d52f943c46..073b7c836d 100644
--- a/docs/source/contributor-guide/iceberg-writes.md
+++ b/docs/source/contributor-guide/iceberg-writes.md
@@ -184,8 +184,15 @@ on the driver. Almost all of it is the per-write `IcebergWriteCommon`:
`IcebergWriteProtoTranslation`
- the sort order id, which the native side ignores and the JVM stamps onto the files afterwards
- `catalog_properties` for the native `FileIO`: the table's `FileIO.properties()` merged over the
- `fs.s3a.*` settings translated from the Hadoop configuration, the same translation the native
- scan uses
+ `fs.s3a.*` settings translated from the effective Hadoop configuration carried by the table's
+ `FileIO`, including SparkCatalog's catalog-specific `hadoop.*` overrides at initialization.
+ Later session or catalog changes do not replace values in that FileIO snapshot. Hadoop's
+ built-in default-source metadata is recovered separately for the S3 eligibility gate.
+ FileIO implementations without a Hadoop configuration, such as `S3FileIO`, use only their
+ initialized properties; session and catalog Hadoop options are neither checked nor forwarded.
+ For `ResolvingFileIO`, this decision uses the instantiated delegate for the data location,
+ including its `HadoopFileIO` fallback, rather than the wrapper's Hadoop configuration.
+ Failure to resolve that delegate causes a plan-time fallback.
The per-task `partition_id` and `task_attempt_id` are stamped onto a copy of the proto inside the
task closure in `CometIcebergWriteExec.doExecuteColumnar`. The native side refuses to run without
@@ -270,6 +277,32 @@ the storage backend from the data location's scheme and wires in Comet's S3 cred
one is configured. For writes the bridge fails closed: if a configured provider cannot initialize,
the task fails rather than writing with the default credential chain.
+The JVM eligibility gate fails closed before building that `FileIO` for an `s3` or `s3a` data
+location. It checks effective `fs.s3a.*` Hadoop settings against the six keys translated by
+`NativeConfig`, and separately checks `table.io().properties()` for `s3.*` / `client.*` keys
+against the properties consumed by the pinned iceberg-rust S3 parser and Comet credential bridge.
+That bridge set includes the explicit provider class and the built-in web-identity tuning
+properties documented in the S3 credential-provider guide. The `s3.sse.type` value is also
+checked: iceberg-rust supports `none`, `s3`, `kms`, and `custom`, but not Iceberg's `dsse-kms`.
+When a Comet credential provider is configured, vendor-owned `s3.*` / `client.*` keys that are not
+part of Iceberg's own S3 property vocabulary are preserved because the provider receives and may
+consume the unfiltered FileIO bag. Iceberg-defined properties remain subject to the storage
+allow-list, so configuring a provider does not make unsupported ACL, tag, storage-class, or other
+write settings eligible. That vocabulary is read from `S3FileIOProperties`,
+`AwsClientProperties`, and `AwsProperties`. Those classes link the AWS SDK, which HadoopFileIO does
+not always provide.
+If loading them fails with a linkage or class-not-found error, every non-allow-listed `s3.*` /
+`client.*` key falls back and planning continues; an empty vocabulary would admit `s3.acl`.
+Per-bucket `fs.s3a.bucket.<bucket>.*` keys are split into a complete bucket name and property
+suffix. `fs.s3a.bucket.target.other.endpoint` belongs to bucket `target.other`, so it does not
+make a write to `target` ineligible.
+The Hadoop check excludes keys whose only source is Hadoop's built-in `core-default.xml`, while
+retaining programmatic, site-XML, and custom `*-default.xml` settings. It also ignores Spark's
+session-wide S3A vectored-read and `downgrade.syncable.exceptions` settings, which cannot affect a
+data-file write request. Keep those allowlists aligned when adding storage support. Reaching
+`FileIOBuilder.with_prop` is not proof of support: the builder accepts unknown properties and the
+backend ignores them. A fallback reason must contain property names only, never values.
+
### Native to JVM: the task payload
Each task emits exactly one batch with one row and two `BINARY` columns (`build_output_schema`):
diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md
index 42d9a80223..a89c397d49 100644
--- a/docs/source/user-guide/latest/iceberg-writes.md
+++ b/docs/source/user-guide/latest/iceberg-writes.md
@@ -55,10 +55,10 @@ The JVM-side planner marshals everything iceberg-rust needs — the write schema
spec as JSON, the data location, the resolved parquet writer settings, the writer mode
(unpartitioned / fanout / clustered, mirroring `SparkWrite`'s own choice), object-store
configuration (the table's `FileIO` properties, e.g. REST-vended credentials, merged over
-`fs.s3a.*` settings translated from the session Hadoop configuration — the same translation
-the native scan uses, since `HadoopFileIO` carries its S3 configuration in the Hadoop
-Configuration rather than in `FileIO` properties), and per-task IDs — into the serialized
-native plan. On each task, iceberg-rust writes the Parquet files and
+`fs.s3a.*` settings translated from the effective Hadoop configuration carried by the table's
+`FileIO`, including catalog-specific `hadoop.*` overrides, since `HadoopFileIO` carries its S3
+configuration in the Hadoop Configuration rather than in `FileIO` properties), and per-task IDs —
+into the serialized native plan. On each task, iceberg-rust writes the Parquet files and
returns its `DataFile` metadata packed as a single in-memory Iceberg V2 data manifest; the JVM
decodes those bytes with Iceberg's own `ManifestFiles.read`, re-derives each file's manifest
metrics from the written Parquet footer with Iceberg's `MetricsConfig` logic (so metrics modes,
@@ -67,11 +67,18 @@ the same `TaskCommit` message the JVM writer would have produced. Everything ice
post-write — snapshot assignment, manifest-list aggregation, commit validation and retries —
is untouched: `IcebergCommit` performs the normal `BatchWrite.commit`.
-Before a task opens a partition's first file, it holds that partition's first
-`write.parquet.page-row-limit` rows in memory, so that it can choose which columns to
-dictionary-encode the way iceberg-java would (see the accepted divergences below). The rows a
-task holds back this way, across all of its partitions, stay within about
-`write.parquet.row-group-size-bytes`.
+For `ResolvingFileIO`, Hadoop settings are taken from the delegate opening the data location.
+An S3 location handled by `S3FileIO` uses its initialized FileIO properties, so Hadoop options
+on the wrapper do not change the native endpoint or encryption settings. If that delegate cannot
+be resolved, the write falls back to iceberg-java.
+
+Before a task opens a partition's first file, it holds the partition's initial rows in memory
+to choose which columns to dictionary-encode using parquet-mr's size accounting (see the
+accepted divergences below). It normally waits for at least `write.parquet.page-row-limit`
+rows (at least 100 when the configured limit is lower), or until the partition ends. The
+buffering threshold is `write.parquet.row-group-size-bytes`, shared across all partitions in
+a fanout write. Reaching it makes each partition still holding rows choose from the rows it
+already has, even if they do not fill its first page.
## Configuration
@@ -188,6 +195,8 @@ A write is eligible only when ALL of the following hold:
| `write.target-file-size-bytes` | any value (the two writers can choose different roll points; see accepted divergences) [...]
| data location URI scheme | `file`, `memory`, `s3`, `s3a`, `gs`, matched case-sensitively (`S3://` falls back). `s3`, `s3a` and `gs` need a bucket in the authority (`s3://bucket/...`), so a hostless form such as `s3:/bucket/key` falls back. `gs` only when the `FileIO` opening the data location is a `GCSFileIO`; see below [...]
| resolved `table.locationProvider()` | Iceberg's built-in `DefaultLocationProvider` [...]
+| Hadoop S3A settings for an `s3` / `s3a` data location | only `fs.s3a.access.key`, `secret.key`, `session.token`, `endpoint`, `endpoint.region`, and `path.style.access`, including their `fs.s3a.bucket.<data-bucket>.*` forms; any other effective `fs.s3a.*` setting falls back [...]
+| Iceberg `FileIO` S3 settings for an `s3` / `s3a` data location | the S3 endpoint, region, static/session credentials, path-style, SSE (`none`, `s3`, `kms`, or `custom`; not `dsse-kms`), assume-role, anonymous/config-chain settings parsed by the pinned iceberg-rust version, plus Comet's credential-provider class and built-in web-identity properties. When a custom provider is configured, its vendor-owned `s3.*` [...]
| partition spec | any, except an identity partition on a `float` or `double` column, or a `void` field whose source column was dropped beside a live field (see below) [...]
| column types | any except `uuid` (Spark plans it as a string; no Arrow cast reaches `fixed(16)`) and the v3 types `variant`, `unknown`, `timestamp_ns`, `geometry` and `geography` [...]
@@ -217,6 +226,28 @@ could resolve a different storage identity or endpoint than the JVM writer would
combination falls back; a `GCSFileIO` carries its `gcs.*` settings in `FileIO.properties()`,
which are forwarded.
+For an `s3` or `s3a` data location, the gate also inspects both the table FileIO's effective Hadoop
+configuration and `table.io().properties()`. These are separate allowlists because Hadoop S3A
+keys are translated before they reach iceberg-rust, while Iceberg `FileIO` keys are forwarded
+directly. When the FileIO exposes a Hadoop configuration, its initialized values govern both the
+gate and native translation, including after session or catalog options change. FileIO
+implementations without a Hadoop configuration, such as `S3FileIO`, use only their initialized
+properties; session and catalog Hadoop options are neither checked nor forwarded. Hadoop's built-in
+`core-default.xml` values are not treated as explicit settings, but
+programmatic settings and values from site or custom `*-default.xml` resources are. Spark's
+session-wide S3A vectored-read and `downgrade.syncable.exceptions` compatibility settings are also
+ignored because they cannot alter an Iceberg data-file write request. Unknown explicit
+`fs.s3a.*`, `s3.*`, or `client.*` settings therefore fall back at planning time instead of being
+silently ignored by the native storage backend. The exception is a vendor-owned `s3.*` /
+`client.*` property when
+`s3.comet.credential.provider.class` is configured: the provider receives the unfiltered FileIO
+bag and can consume that property. Iceberg-defined settings that the native storage path cannot
+honour still fall back even with a provider. A per-bucket Hadoop setting counts only for the exact
+data-bucket name, so configuration for a longer dotted bucket does not by itself disable the
+native write. If Iceberg's AWS property classes cannot be loaded, vendor `s3.*` / `client.*` keys
+fall back too and planning still completes. The fall-back reason reports only sorted property
+names, never their values, so credentials and tokens do not enter EXPLAIN or plan logs.
+
An identity partition on a `float` or `double` column falls back. iceberg-rust compares float
partition values with an equality that treats `-0.0` and `0.0` as one value, so the native writer
would put rows with either value in the same partition, where iceberg-java writes two. A read
@@ -319,13 +350,17 @@ a data file but not what any reader computes from it:
- Dictionary-encoded pages are labeled `RLE_DICTIONARY` (parquet-mr v1 files: `PLAIN_DICTIONARY`).
- Fixed-length binary columns (`uuid`, `fixed`, decimals with precision > 18) are not
dictionary-encoded (parquet-mr dictionary-encodes them).
-- Which columns are dictionary-encoded is decided as parquet-mr decides it: a column whose first
- data page shows the dictionary saving no space is written plain, with no dictionary page,
+- The native writer uses parquet-mr's size accounting to choose dictionary encoding: a column
+ whose sampled rows show the dictionary saving no space is written plain, with no dictionary page,
instead of carrying a dictionary page that every selective read of it would have to fetch
([#6114](https://github.com/apache/datafusion-comet/issues/6114)). The native writer decides
- once per partition, from that partition's first page of rows in the task, and keeps the
+ once per partition, normally from that partition's first page of rows in the task, and keeps the
decision for every file and row group it writes for the partition; parquet-mr decides again
- for every row group. Where the page size rather than `write.parquet.page-row-limit` ends a
+ for every row group. If buffered rows reach `write.parquet.row-group-size-bytes`, the choice
+ uses the rows collected so far. In a fanout write, all partitions share this threshold, so
+ each partition still buffering can make its choice before it has a full first page. Later
+ rows could have changed parquet-mr's decision, so this difference is not limited to columns
+ close to the size cut-off. Where the page size rather than `write.parquet.page-row-limit` ends a
column's first page, the native page ends at the first row past parquet-mr's size threshold,
while parquet-mr only ends it at its next periodic size check, so a column close to the
cut-off can be decided the other way. Close to the cut-off both encodings take about the same
diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index 6e6479ca80..9a0dfee85f 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -24,6 +24,7 @@ import java.util.concurrent.ConcurrentHashMap
import scala.util.control.NonFatal
+import org.apache.hadoop.conf.Configuration
import org.apache.spark.internal.Logging
import org.apache.spark.sql.SparkSession
@@ -554,12 +555,15 @@ object IcebergReflection extends Logging {
* reflection failure; callers must fail closed.
*/
def resolveFileIOClass(fileIO: Any, location: String): Option[Class[_]] =
+ resolveFileIO(fileIO, location).map(_.getClass)
+
+ private def resolveFileIO(fileIO: Any, location: String): Option[Any] =
if (!classNameInHierarchy(fileIO.getClass, Set(ClassNames.RESOLVING_FILE_IO))) {
- Some(fileIO.getClass)
+ Some(fileIO)
} else {
try {
findMethodInHierarchy(fileIO.getClass, "io", classOf[String]) match {
- case Some(ioMethod) => Option(ioMethod.invoke(fileIO, location)).map(_.getClass)
+ case Some(ioMethod) => Option(ioMethod.invoke(fileIO, location))
case None =>
logError(
s"Iceberg reflection failure: ${fileIO.getClass.getName} has no io(String) method")
@@ -624,6 +628,36 @@ object IcebergReflection extends Logging {
}
}
+ /**
+ * Gets the effective Hadoop configuration of the FileIO opening the table's data location.
+ *
+ * SparkCatalog overlays `spark.sql.catalog.<catalog>.hadoop.*` settings onto the configuration
+ * installed in HadoopConfigurable FileIO implementations. Reading the Spark session's Hadoop
+ * configuration directly misses those catalog-specific overrides. FileIO is test-scoped on the
+ * main classpath, so invoke `getConf` reflectively instead of linking HadoopConfigurable.
+ * ResolvingFileIO's own configuration does not imply that its delegate consumes Hadoop
+ * settings. Read the instantiated delegate instead, including its HadoopFileIO fallback. Throw
+ * when the delegate cannot be resolved so write detection fails closed; None means the resolved
+ * FileIO does not expose a Hadoop configuration.
+ */
+ def getFileIOHadoopConf(table: Any): Option[Configuration] =
+ getFileIO(table).flatMap { fileIO =>
+ val resolved =
+ if (classNameInHierarchy(fileIO.getClass, Set(ClassNames.RESOLVING_FILE_IO))) {
+ val location = getDataLocation(table)
+ .getOrElse(
+ throw new IllegalStateException("could not resolve the table data location"))
+ resolveFileIO(fileIO, location)
+ .getOrElse(
+ throw new IllegalStateException(s"could not resolve the FileIO for $location"))
+ } else {
+ fileIO
+ }
+ findMethodInHierarchy(resolved.getClass, "getConf").flatMap { confMethod =>
+ Option(confMethod.invoke(resolved)).collect { case conf: Configuration => conf }
+ }
+ }
+
/**
* Gets the schema from an Iceberg table.
*/
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
index 077c860cf8..c6741f3e40 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala
@@ -19,6 +19,7 @@
package org.apache.comet.serde.operator
+import java.lang.reflect.Modifier
import java.util.Locale
import scala.jdk.CollectionConverters._
@@ -99,6 +100,159 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
private val MaxDataFieldId = Int.MaxValue - 200
private val ParquetWritePropertyPrefix = "write.parquet."
private val ParquetMrPropertyPrefix = "parquet."
+ private val CometS3CredentialProviderClassProperty =
+ "s3.comet.credential.provider.class"
+
+ // Hadoop S3A settings are not forwarded wholesale. Keep this allow-list in lockstep with
+ // NativeConfig.s3aSuffixToIcebergGlobalKey: every admitted setting must be translated into the
+ // catalog properties consumed by iceberg-rust. Derive the set from the translation itself so a
+ // new mapping cannot be forwarded by the write path while this gate still rejects it.
+ // Per-bucket spellings for the data bucket are admitted through the same suffix list; settings
+ // for other buckets do not affect this write. The bucket name is the whole name left after the
+ // property suffix, so `fs.s3a.bucket.target.other.endpoint` belongs to `target.other`, not
+ // `target`.
+ private val SupportedHadoopS3Suffixes: Set[String] =
+ NativeConfig.s3aSuffixToIcebergGlobalKey.keySet
+
+ private val SupportedHadoopS3Keys: Set[String] =
+ SupportedHadoopS3Suffixes.map("fs.s3a." + _)
+
+ private val FsS3aPrefix = "fs.s3a."
+ private val FsS3aBucketPrefix = "fs.s3a.bucket."
+
+ private case class HadoopS3PropertyNames(exact: Set[String], prefixes: Seq[String]) {
+ def matchingSuffix(keyWithoutBucketPrefix: String): Option[String] = {
+ val exactMatches = exact.iterator.filter { suffix =>
+ keyWithoutBucketPrefix == suffix || keyWithoutBucketPrefix.endsWith("." + suffix)
+ }
+ val prefixMatches = prefixes.iterator.flatMap { prefix =>
+ val boundary = keyWithoutBucketPrefix.lastIndexOf("." + prefix)
+ if (boundary < 0) None else Some(keyWithoutBucketPrefix.substring(boundary + 1))
+ }
+ (exactMatches ++ prefixMatches).toSeq.sortBy(-_.length).headOption
+ }
+ }
+
+ // Hadoop's per-bucket key syntax has no separator between a dotted bucket name and its property
+ // suffix. Resolve the suffix against the property names declared by the runtime Hadoop version,
+ // so `bucket.target.other.encryption.algorithm` belongs to bucket `target.other`, even though
+ // `encryption.algorithm` is unsupported by the native writer. If hadoop-aws is unavailable,
+ // retain the supported suffixes and let unknown spellings take the conservative path below.
+ private lazy val HadoopS3Properties: HadoopS3PropertyNames =
+ try {
+ val suffixes = allStaticStringConstants(
+ IcebergReflection.loadClass("org.apache.hadoop.fs.s3a.Constants"))
+ .filter(key => key.startsWith(FsS3aPrefix) && !key.startsWith(FsS3aBucketPrefix))
+ .map(_.stripPrefix(FsS3aPrefix))
+ .filter(_.nonEmpty)
+ .toSet ++ SupportedHadoopS3Suffixes
+ val (prefixes, exact) = suffixes.partition(_.endsWith("."))
+ HadoopS3PropertyNames(exact, prefixes.toSeq.sorted)
+ } catch {
+ case _: ClassNotFoundException =>
+ HadoopS3PropertyNames(SupportedHadoopS3Suffixes, Seq.empty)
+ case _: LinkageError => HadoopS3PropertyNames(SupportedHadoopS3Suffixes, Seq.empty)
+ case NonFatal(_) => HadoopS3PropertyNames(SupportedHadoopS3Suffixes, Seq.empty)
+ }
+
+ // Spark seeds these Hadoop S3A compatibility/read settings into every session as if they came
+ // from spark.hadoop.*. They do not alter an Iceberg data-file write request, so they must not
+ // make every otherwise-clean S3 write ineligible. This also permits explicit overrides, which
+ // are harmless on the write path for the same reason.
+ private val IgnoredHadoopS3Keys: Set[String] = Set(
+ "fs.s3a.downgrade.syncable.exceptions",
+ "fs.s3a.vectored.read.max.merged.size",
+ "fs.s3a.vectored.read.min.seek.size")
+
+ // Audited against the pinned iceberg-rust S3 parser
+ // (`iceberg/src/io/storage/config/s3.rs` and `storage/opendal/src/s3.rs`). Do not broaden this
+ // to every s3.* / client.* property: FileIOBuilder accepts unknown keys, but the storage backend
+ // silently ignores them. The provider class, token expiry, and web-identity settings are
+ // consumed by Comet's credential paths rather than the storage parser. The expiry timestamp is
+ // needed by the documented REST-vended credential provider. The web-identity settings tune the
+ // built-in IRSA path when no explicit provider or credentials take precedence.
+ private val SupportedS3FileIOProperties: Set[String] = Set(
+ "s3.endpoint",
+ "s3.access-key-id",
+ "s3.secret-access-key",
+ "s3.session-token",
+ "s3.region",
+ "client.region",
+ "s3.path-style-access",
+ "s3.sse.type",
+ "s3.sse.key",
+ "s3.sse.md5",
+ "client.assume-role.arn",
+ "client.assume-role.external-id",
+ "client.assume-role.session-name",
+ "s3.allow-anonymous",
+ "s3.disable-ec2-metadata",
+ "s3.disable-config-load",
+ CometS3CredentialProviderClassProperty,
+ "s3.comet.credential.webIdentity.enabled",
+ "s3.comet.credential.webIdentity.maxAttempts",
+ "s3.comet.credential.webIdentity.minTtlSeconds",
+ "s3.comet.credential.webIdentity.refreshJitterSeconds",
+ "s3.session-token-expires-at-ms")
+
+ // iceberg-java also defines "dsse-kms", but the pinned iceberg-rust S3 backend cannot map
+ // that mode into an OpenDAL server-side-encryption configuration. Check the value as well as
+ // the property name so it falls back during planning instead of failing in the native task.
+ private val SupportedS3SseTypes: Set[String] = Set("none", "s3", "kms", "custom")
+
+ private[comet] case class IcebergAwsPropertyNames(exact: Set[String], prefixes: Seq[String]) {
+ def contains(key: String): Boolean =
+ exact.contains(key) || prefixes.exists(key.startsWith)
+ }
+
+ // Used when S3FileIOProperties / AwsClientProperties cannot be linked. Every non-allow-listed
+ // s3.* / client.* key then counts as Iceberg-owned. An empty vocabulary would do the opposite
+ // and admit s3.acl once a custom credential provider is configured.
+ private val UnclassifiedIcebergAwsProperties =
+ IcebergAwsPropertyNames(Set.empty, Seq("client.", "s3."))
+
+ private val IcebergAwsPropertyClasses = Seq(
+ "org.apache.iceberg.aws.s3.S3FileIOProperties",
+ "org.apache.iceberg.aws.AwsClientProperties",
+ "org.apache.iceberg.aws.AwsProperties")
+
+ // A configured Comet credential provider receives the complete, unfiltered FileIO property
+ // bag. It may therefore consume vendor-owned s3.* / client.* keys that neither iceberg-java nor
+ // iceberg-rust knows about. Keep rejecting the standard Iceberg properties that the native
+ // storage path cannot honour, however. Reading the constants from the runtime Iceberg version
+ // keeps this classification aligned with every supported profile and makes newly-added Iceberg
+ // properties fail closed without mistaking them for provider-owned configuration.
+ //
+ // Those classes reference the AWS SDK. HadoopFileIO and Comet's provider SPI do not require it,
+ // and linking the classes then throws NoClassDefFoundError. getSupportLevel only catches
+ // NonFatal, so the lookup itself must turn that linkage failure into a support decision.
+ private lazy val IcebergAwsProperties: IcebergAwsPropertyNames =
+ icebergAwsPropertyNames(IcebergReflection.loadClass)
+
+ private[comet] def icebergAwsPropertyNames(
+ loadClass: String => Class[_]): IcebergAwsPropertyNames =
+ try {
+ val propertyNames = IcebergAwsPropertyClasses.flatMap { className =>
+ staticStringConstants(loadClass(className))
+ }
+ val (prefixes, exact) = propertyNames.distinct.partition(_.endsWith("."))
+ IcebergAwsPropertyNames(exact.toSet, prefixes.sorted)
+ } catch {
+ case _: ClassNotFoundException => UnclassifiedIcebergAwsProperties
+ case _: LinkageError => UnclassifiedIcebergAwsProperties
+ }
+
+ private def staticStringConstants(cls: Class[_]): Iterator[String] =
+ allStaticStringConstants(cls)
+ .filter(key => key.startsWith("s3.") || key.startsWith("client."))
+
+ private def allStaticStringConstants(cls: Class[_]): Iterator[String] =
+ cls.getDeclaredFields.iterator
+ .filter(field => Modifier.isStatic(field.getModifiers) && field.getType == classOf[String])
+ .flatMap { field =>
+ field.setAccessible(true)
+ Option(field.get(null).asInstanceOf[String])
+ }
// Hadoop-side `parquet.*` keys that iceberg-java's writer never consumes, so seeing them
// in the session Hadoop configuration does not indicate the native writer would diverge.
@@ -162,7 +316,8 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
table,
tableProperties ++ writeProperties,
sparkWrite,
- op.session.sessionState.newHadoopConf())
+ op.session.sessionState.newHadoopConf(),
+ effectiveHadoopConf(op, table))
triggers.iterator.map(rule => rule(context)).collectFirst { case Some(reason) => reason }
}
@@ -170,10 +325,62 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
table: Any,
properties: Map[String, String],
sparkWrite: Any,
- hadoopConf: Configuration)
+ hadoopConf: Configuration,
+ s3HadoopConf: Configuration)
private type TriggerRule = TriggerContext => Option[String]
+ private def effectiveHadoopConf(op: IcebergWriteExec, table: Any): Configuration = {
+ val sessionConf = op.session.sessionState.newHadoopConf()
+ val catalogOverrides = IcebergReflection
+ .deriveCatalogName(table)
+ .map { catalogName =>
+ val prefix = s"spark.sql.catalog.$catalogName.hadoop."
+ op.session.sessionState.conf.getAllConfs.collect {
+ case (key, value) if key.startsWith(prefix) => key.substring(prefix.length) -> value
+ }
+ }
+ .getOrElse(Map.empty)
+
+ IcebergReflection.getFileIOHadoopConf(table) match {
+ case None =>
+ // S3FileIO and other non-Hadoop FileIO implementations use their initialized
+ // properties, not Spark's Hadoop options. Forwarding session or catalog settings here
+ // can redirect native writes away from the FileIO used for footer reads and cleanup.
+ new Configuration(false)
+ case Some(fileIOConf) =>
+ // HadoopFileIO stores its configuration in Iceberg's SerializableConfiguration. That
+ // class rebuilds a Configuration(false) by calling set() for every entry, so Hadoop's
+ // property-source metadata is lost and core-default.xml values look programmatic.
+ // The initialized FileIO remains authoritative for values: SparkCatalog does not
+ // reinitialize it when session or catalog options change. Recover only default-source
+ // metadata, without adding or replacing any FileIO configuration. Load core-default.xml
+ // separately so later session settings cannot change the default values used here.
+ val coreDefaults = new Configuration(false)
+ coreDefaults.addResource("core-default.xml")
+ val effectiveConf = new Configuration(fileIOConf)
+ fileIOConf.iterator().asScala.foreach { entry =>
+ val key = entry.getKey
+ val fileIOValue = fileIOConf.getRaw(key)
+ val explicitlyConfiguredValue = catalogOverrides.get(key).contains(fileIOValue) ||
+ (hasExplicitSource(sessionConf, key) && fileIOValue == sessionConf.getRaw(key))
+ if (!explicitlyConfiguredValue &&
+ Option(fileIOConf.getPropertySources(key))
+ .exists(_.toSeq == Seq("programmatically")) &&
+ fileIOValue == coreDefaults.getRaw(key)) {
+ effectiveConf.set(key, fileIOValue, "core-default.xml")
+ }
+ }
+ effectiveConf
+ }
+ }
+
+ private def hasExplicitSource(conf: Configuration, key: String): Boolean =
+ Option(conf.getPropertySources(key)) match {
+ case Some(sources) if sources.nonEmpty => !sources.forall(isCoreDefaultResource)
+ case _ => conf.getRaw(key) != null
+ }
+
private lazy val triggers: Seq[TriggerRule] = Seq(
requireFormatParquet,
requirePropertyAbsentOrNotTrue(
@@ -206,6 +413,8 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
requirePositiveIntParquetSizes,
requireNoParquetHadoopConfOverrides,
requireSupportedStorageScheme,
+ requireSupportedHadoopS3Settings,
+ requireSupportedS3FileIOProperties,
requireGcsFileIOForGcsDataLocation,
requireExecutorReflectionResolvable)
@@ -456,6 +665,123 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
}
}
+ private def s3DataLocation(ctx: TriggerContext): Option[String] =
+ IcebergReflection
+ .getDataLocation(ctx.table)
+ .filter(location => Set("s3", "s3a").contains(storageScheme(location)))
+
+ private def unsupportedSettingsReason(namespace: String, keys: Seq[String]): Option[String] =
+ keys match {
+ case Seq() => None
+ case Seq(key) => Some(s"unsupported $namespace setting: $key")
+ case _ => Some(s"unsupported $namespace settings: ${keys.mkString(", ")}")
+ }
+
+ /**
+ * Return effective Hadoop S3A keys that the native write path cannot reproduce. Global keys and
+ * keys scoped to the data bucket affect this write; per-bucket settings for other buckets do
+ * not. The data bucket must match in full: `fs.s3a.bucket.target.other.endpoint` is bucket
+ * `target.other` and suffix `endpoint`, so it does not affect a write to `target`. Values are
+ * deliberately never returned because this result is used in EXPLAIN fallback reasons and may
+ * include credentials.
+ */
+ private[comet] def unsupportedHadoopS3Settings(
+ hadoopConf: Configuration,
+ targetBucket: Option[String]): Seq[String] = {
+ val keys = hadoopConf
+ .iterator()
+ .asScala
+ .map(_.getKey)
+ .filter(_.startsWith("fs.s3a."))
+ .filterNot(IgnoredHadoopS3Keys.contains)
+ // Hadoop's iterator includes the many fs.s3a.* defaults loaded from core-default.xml.
+ // Those are library implementation defaults, not settings selected by the user, and
+ // treating them as explicit would reject every ordinary S3 write. Preserve settings from
+ // site XML and programmatic/Spark sources; exclude a key only when every recorded source is
+ // Hadoop's built-in core-default.xml resource. A user resource merely named
+ // `tenant-default.xml` is still explicit configuration.
+ .filter { key =>
+ Option(hadoopConf.getPropertySources(key))
+ .forall(sources => sources.isEmpty || !sources.forall(isCoreDefaultResource))
+ }
+ .toSeq
+
+ keys.filter(key => isUnsupportedHadoopS3Key(key, targetBucket)).sorted
+ }
+
+ private def isCoreDefaultResource(source: String): Boolean =
+ source == "core-default.xml" || source.endsWith("/core-default.xml")
+
+ /**
+ * Bucket of `fs.s3a.bucket.<bucket>.<suffix>` when `<suffix>` is exactly one supported S3A
+ * suffix. The longest suffix wins, so `endpoint.region` stays one property and a dotted bucket
+ * name is what remains.
+ */
+ private def perBucketProperty(key: String): Option[(String, String)] = {
+ if (!key.startsWith(FsS3aBucketPrefix)) {
+ None
+ } else {
+ val rest = key.substring(FsS3aBucketPrefix.length)
+ HadoopS3Properties.matchingSuffix(rest).flatMap { matched =>
+ val bucket = rest.substring(0, rest.length - matched.length).stripSuffix(".")
+ if (bucket.isEmpty) None else Some(bucket -> matched)
+ }
+ }
+ }
+
+ private def isUnsupportedHadoopS3Key(key: String, targetBucket: Option[String]): Boolean =
+ if (key.startsWith(FsS3aBucketPrefix)) {
+ perBucketProperty(key) match {
+ case Some((bucket, suffix)) =>
+ targetBucket.contains(bucket) && !SupportedHadoopS3Suffixes.contains(suffix)
+ case None =>
+ // Preserve fail-closed behavior for a property unknown to the runtime Hadoop version.
+ targetBucket.exists(bucket => key.startsWith(s"$FsS3aBucketPrefix$bucket."))
+ }
+ } else {
+ !SupportedHadoopS3Keys.contains(key)
+ }
+
+ /** Return unsupported FileIO S3/client property names in deterministic order. */
+ private[comet] def unsupportedS3FileIOProperties(properties: Map[String, String]): Seq[String] =
+ unsupportedS3FileIOProperties(properties, IcebergAwsProperties)
+
+ private[comet] def unsupportedS3FileIOProperties(
+ properties: Map[String, String],
+ icebergAwsProperties: IcebergAwsPropertyNames): Seq[String] = {
+ val customCredentialProviderConfigured = properties
+ .get(CometS3CredentialProviderClassProperty)
+ .exists(_.trim.nonEmpty)
+
+ properties.iterator
+ .filter { case (key, _) => key.startsWith("s3.") || key.startsWith("client.") }
+ .filter { case (key, value) =>
+ val unsupportedName = !SupportedS3FileIOProperties.contains(key)
+ val unsupportedValue =
+ key == "s3.sse.type" && !Option(value).exists(value =>
+ SupportedS3SseTypes.contains(value.toLowerCase(Locale.ROOT)))
+ unsupportedValue || (unsupportedName &&
+ (!customCredentialProviderConfigured || icebergAwsProperties.contains(key)))
+ }
+ .map(_._1)
+ .toSeq
+ .sorted
+ }
+
+ private val requireSupportedHadoopS3Settings: TriggerRule = ctx =>
+ s3DataLocation(ctx).flatMap { location =>
+ val dataBucket = NativeConfig.bucketForUri(new java.net.URI(location), Set.empty)
+ unsupportedSettingsReason(
+ "Hadoop S3A",
+ unsupportedHadoopS3Settings(ctx.s3HadoopConf, dataBucket))
+ }
+
+ private val requireSupportedS3FileIOProperties: TriggerRule = ctx =>
+ s3DataLocation(ctx).flatMap { _ =>
+ val properties = IcebergReflection.getFileIOProperties(ctx.table).getOrElse(Map.empty)
+ unsupportedSettingsReason("S3 FileIO", unsupportedS3FileIOProperties(properties))
+ }
+
// HadoopFileIO takes its GCS configuration from `fs.gs.*`, which is not forwarded to the
// native writer (only `fs.s3a.*` is bridged). Admit a gs:// data location only when the FileIO
// Iceberg resolves for it is a GCSFileIO, whose `gcs.*` settings are forwarded.
@@ -800,9 +1126,10 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
// Configuration instead (fs.s3a.* credentials, custom endpoint, path-style access), which
// the JVM writer would honour but iceberg-rust would never see. Mirror the scan side
// (`CometScanRule`): extract the object-store options for the data location from the
- // session Hadoop configuration, translate them to the s3.* keys iceberg-rust consumes, and
- // let FileIO/vended properties win on conflict.
- val writeHadoopConf = op.session.sessionState.newHadoopConf()
+ // effective FileIO Hadoop configuration, translate them to the s3.* keys iceberg-rust
+ // consumes, and let FileIO/vended properties win on conflict. The FileIO configuration
+ // includes SparkCatalog's catalog-specific `hadoop.*` overrides.
+ val writeHadoopConf = effectiveHadoopConf(op, table)
val dataUri = new java.net.URI(dataLocation)
// Promote the data bucket's per-bucket `fs.s3a.bucket.<b>.*` settings to global, mirroring the
// scan path: iceberg-rust's pinned S3 parser reads only global `s3.*`. Only an S3-family data
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
index dc40c1e1ff..da60c14ff0 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
@@ -26,6 +26,7 @@ import org.scalatest.Tag
import org.apache.hadoop.conf.Configuration
import org.apache.iceberg.{PartitionSpec, StructLike, TableMetadata, TableOperations}
+import org.apache.iceberg.aws.s3.S3FileIO
import org.apache.iceberg.catalog.TableIdentifier
import org.apache.iceberg.encryption.EncryptionManager
import org.apache.iceberg.hadoop.{HadoopCatalog, HadoopConfigurable, HadoopFileIO}
@@ -612,6 +613,505 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
}
}
+ test("S3 setting allow-lists reject unknown keys deterministically without values") {
+ val hadoopConf = new Configuration(false)
+ val secret = "SECRET_VALUE_MUST_NOT_APPEAR"
+ Seq(
+ "fs.s3a.access.key",
+ "fs.s3a.secret.key",
+ "fs.s3a.session.token",
+ "fs.s3a.endpoint",
+ "fs.s3a.endpoint.region",
+ "fs.s3a.path.style.access",
+ "fs.s3a.bucket.target.access.key",
+ "fs.s3a.bucket.target.path.style.access",
+ // A per-bucket property for another bucket is not effective for this write.
+ "fs.s3a.bucket.other.encryption.algorithm").foreach(hadoopConf.set(_, "supported"))
+ hadoopConf.set("fs.s3a.encryption.key", secret)
+ hadoopConf.set("fs.s3a.aws.credentials.provider", secret)
+ hadoopConf.set("fs.s3a.bucket.target.encryption.algorithm", secret)
+
+ val unsupportedHadoop =
+ CometIcebergNativeWrite.unsupportedHadoopS3Settings(hadoopConf, Some("target"))
+ assert(
+ unsupportedHadoop == Seq(
+ "fs.s3a.aws.credentials.provider",
+ "fs.s3a.bucket.target.encryption.algorithm",
+ "fs.s3a.encryption.key"),
+ unsupportedHadoop)
+ assert(!unsupportedHadoop.exists(_.contains(secret)), unsupportedHadoop)
+
+ val supportedFileIO = Map(
+ "s3.endpoint" -> "endpoint",
+ "s3.access-key-id" -> "access",
+ "s3.secret-access-key" -> secret,
+ "s3.session-token" -> secret,
+ "s3.region" -> "us-east-1",
+ "client.region" -> "us-east-1",
+ "s3.path-style-access" -> "true",
+ "s3.sse.type" -> "kms",
+ "s3.sse.key" -> "key-id",
+ "s3.sse.md5" -> "md5",
+ "client.assume-role.arn" -> "arn",
+ "client.assume-role.external-id" -> "external",
+ "client.assume-role.session-name" -> "session",
+ "s3.allow-anonymous" -> "false",
+ "s3.disable-ec2-metadata" -> "false",
+ "s3.disable-config-load" -> "false",
+ "s3.comet.credential.webIdentity.enabled" -> "true",
+ "s3.comet.credential.webIdentity.maxAttempts" -> "5",
+ "s3.comet.credential.webIdentity.minTtlSeconds" -> "300",
+ "s3.comet.credential.webIdentity.refreshJitterSeconds" -> "60",
+ "s3.session-token-expires-at-ms" -> "0")
+ val unsupportedFileIO = CometIcebergNativeWrite.unsupportedS3FileIOProperties(
+ supportedFileIO ++ Map(
+ "s3.acl" -> secret,
+ "s3.sse.type" -> "dsse-kms",
+ "s3.write.tags.foo" -> secret,
+ "s3.write.storage-class" -> secret,
+ "s3.access-points.bucket" -> secret,
+ "s3.remote-signing-enabled" -> secret,
+ "client.factory" -> secret,
+ "client.credentials-provider" -> secret,
+ "unrelated.property" -> secret))
+ assert(
+ unsupportedFileIO == Seq(
+ "client.credentials-provider",
+ "client.factory",
+ "s3.access-points.bucket",
+ "s3.acl",
+ "s3.remote-signing-enabled",
+ "s3.sse.type",
+ "s3.write.storage-class",
+ "s3.write.tags.foo"),
+ unsupportedFileIO)
+ assert(!unsupportedFileIO.exists(_.contains(secret)), unsupportedFileIO)
+ assert(
+ CometIcebergNativeWrite.unsupportedS3FileIOProperties(Map("s3.sse.type" -> " kms ")) == Seq(
+ "s3.sse.type"))
+
+ val withCustomProvider = supportedFileIO ++ Map(
+ "s3.comet.credential.provider.class" -> "provider",
+ "s3.vendor.credential-scope" -> secret,
+ "client.vendor.tenant-id" -> secret,
+ // Standard Iceberg settings remain unsupported even when a provider is configured.
+ "client.factory" -> secret,
+ "client.assume-role.tags.department" -> secret,
+ "s3.acl" -> secret,
+ "s3.write.tags.foo" -> secret)
+ val unsupportedWithCustomProvider =
+ CometIcebergNativeWrite.unsupportedS3FileIOProperties(withCustomProvider)
+ assert(
+ unsupportedWithCustomProvider == Seq(
+ "client.assume-role.tags.department",
+ "client.factory",
+ "s3.acl",
+ "s3.write.tags.foo"),
+ unsupportedWithCustomProvider)
+ assert(
+ !unsupportedWithCustomProvider.exists(_.contains(secret)),
+ unsupportedWithCustomProvider)
+ }
+
+ test("a longer dotted bucket is not the target bucket") {
+ val hadoopConf = new Configuration(false)
+ hadoopConf.set("fs.s3a.bucket.target.access.key", "access")
+ hadoopConf.set("fs.s3a.bucket.target.endpoint.region", "us-east-1")
+ hadoopConf.set("fs.s3a.bucket.target.path.style.access", "true")
+ // Bucket `target.other` plus suffix `endpoint`, not bucket `target` plus `other.endpoint`.
+ hadoopConf.set("fs.s3a.bucket.target.other.endpoint", "https://other.example")
+ hadoopConf.set("fs.s3a.bucket.target.other.encryption.algorithm", "SSE-KMS")
+ hadoopConf.set("fs.s3a.bucket.target.encryption.algorithm", "SSE-KMS")
+
+ assert(
+ CometIcebergNativeWrite.unsupportedHadoopS3Settings(hadoopConf, Some("target")) == Seq(
+ "fs.s3a.bucket.target.encryption.algorithm"))
+
+ val longerBucket = new Configuration(false)
+ longerBucket.set("fs.s3a.bucket.target.other.endpoint", "https://other.example")
+ longerBucket.set("fs.s3a.bucket.target.other.encryption.algorithm", "SSE-KMS")
+ assert(
+ CometIcebergNativeWrite.unsupportedHadoopS3Settings(longerBucket, Some("target.other")) ==
+ Seq("fs.s3a.bucket.target.other.encryption.algorithm"))
+ }
+
+ test("missing AWS SDK falls back instead of aborting S3 FileIO classification") {
+ val secret = "SECRET_VALUE_MUST_NOT_APPEAR"
+ val properties = Map(
+ "s3.endpoint" -> "https://example",
+ "s3.comet.credential.provider.class" -> "provider",
+ "s3.vendor.credential-scope" -> secret,
+ "client.vendor.tenant-id" -> secret,
+ "s3.acl" -> secret,
+ "s3.write.tags.foo" -> secret)
+ val expected =
+ Seq("client.vendor.tenant-id", "s3.acl", "s3.vendor.credential-scope", "s3.write.tags.foo")
+ val loadFailures = Seq[Throwable](
+ new NoClassDefFoundError("software/amazon/awssdk/auth/credentials/AwsCredentialsProvider"),
+ new ClassNotFoundException("org.apache.iceberg.aws.s3.S3FileIOProperties"))
+ loadFailures.foreach { failure =>
+ val names = CometIcebergNativeWrite.icebergAwsPropertyNames(_ => throw failure)
+ val unsupported =
+ CometIcebergNativeWrite.unsupportedS3FileIOProperties(properties, names)
+ assert(unsupported == expected, unsupported)
+ assert(!unsupported.exists(_.contains(secret)), unsupported)
+ }
+ }
+
+ test("Hadoop S3A built-in defaults are ignored but custom resources are effective") {
+ def xml(key: String, value: String): java.io.ByteArrayInputStream =
+ new java.io.ByteArrayInputStream(s"""<configuration>
+ | <property><name>$key</name><value>$value</value></property>
+ |</configuration>""".stripMargin.getBytes(java.nio.charset.StandardCharsets.UTF_8))
+
+ val key = "fs.s3a.encryption.algorithm"
+
+ val coreDefaultOnly = new Configuration(false)
+ coreDefaultOnly.addResource(xml(key, "SSE-KMS"), "core-default.xml")
+ assert(coreDefaultOnly.get(key) == "SSE-KMS")
+ assert(
+ CometIcebergNativeWrite
+ .unsupportedHadoopS3Settings(coreDefaultOnly, Some("target"))
+ .isEmpty)
+
+ val customDefault = new Configuration(false)
+ customDefault.addResource(xml(key, "SSE-KMS"), "tenant-default.xml")
+ assert(customDefault.get(key) == "SSE-KMS")
+ assert(
+ CometIcebergNativeWrite.unsupportedHadoopS3Settings(customDefault, Some("target")) == Seq(
+ key))
+
+ val site = new Configuration(false)
+ site.addResource(xml(key, "SSE-KMS"), "probe-site.xml")
+ assert(site.get(key) == "SSE-KMS")
+ assert(CometIcebergNativeWrite.unsupportedHadoopS3Settings(site, Some("target")) == Seq(key))
+
+ coreDefaultOnly.set(key, "SSE-S3")
+ assert(
+ CometIcebergNativeWrite.unsupportedHadoopS3Settings(coreDefaultOnly, Some("target")) == Seq(
+ key))
+ }
+
+ test("catalog Hadoop S3A overrides use the FileIO configuration") {
+ withTempIcebergDir { warehouseDir =>
+ val catalogWithOverride = "s3_hadoop_override_cat"
+ val hadoopKey = "fs.s3a.encryption.algorithm"
+ withSQLConf(
+ s"spark.sql.catalog.$catalogWithOverride" -> "org.apache.iceberg.spark.SparkCatalog",
+ s"spark.sql.catalog.$catalogWithOverride.type" -> "hadoop",
+ s"spark.sql.catalog.$catalogWithOverride.warehouse" -> warehouseDir.getAbsolutePath,
+ s"spark.sql.catalog.$catalogWithOverride.hadoop.$hadoopKey" -> "SSE-KMS") {
+ spark.sql(s"""
+ CREATE TABLE $catalogWithOverride.$ns.catalog_hadoop_override (
+ id INT,
+ region STRING,
+ amount DOUBLE
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.data.path'='s3a://probe-bucket/iceberg/db/catalog_hadoop_override'
+ )
+ """)
+
+ // SparkCatalog retains the initialized FileIO even after its options change.
+ withSQLConf(s"spark.sql.catalog.$catalogWithOverride.hadoop.$hadoopKey" -> "") {
+ val writeExec =
+ planInsertWriteExec(s"$catalogWithOverride.$ns.catalog_hadoop_override")
+ val sparkWrite = IcebergReflection
+ .getOuterSparkWrite(writeExec.batchWrite)
+ .getOrElse(fail("could not unwrap SparkWrite"))
+ val table = IcebergReflection
+ .getTableFromSparkWrite(sparkWrite)
+ .getOrElse(fail("could not extract Iceberg table"))
+ val fileIOConf = IcebergReflection
+ .getFileIOHadoopConf(table)
+ .getOrElse(fail("FileIO did not expose its Hadoop configuration"))
+ assert(fileIOConf.get(hadoopKey) == "SSE-KMS")
+ assertUnsupportedContains(
+ writeExec,
+ "catalog_hadoop_override",
+ s"unsupported Hadoop S3A setting: $hadoopKey")
+ }
+ }
+ }
+ }
+
+ test("supported catalog Hadoop S3A overrides are forwarded to the native write") {
+ withTempIcebergDir { warehouseDir =>
+ val catalogWithOverride = "s3_hadoop_forwarding_cat"
+ val endpoint = "https://s3.example.test"
+ withSQLConf(
+ "fs.s3a.endpoint" -> "https://session.example.test",
+ s"spark.sql.catalog.$catalogWithOverride" -> "org.apache.iceberg.spark.SparkCatalog",
+ s"spark.sql.catalog.$catalogWithOverride.type" -> "hadoop",
+ s"spark.sql.catalog.$catalogWithOverride.warehouse" -> warehouseDir.getAbsolutePath,
+ s"spark.sql.catalog.$catalogWithOverride.hadoop.fs.s3a.endpoint" -> endpoint,
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
+ spark.sql(s"""
+ CREATE TABLE $catalogWithOverride.$ns.catalog_hadoop_forwarding (
+ id INT,
+ region STRING,
+ amount DOUBLE
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.data.path'='s3a://probe-bucket/iceberg/db/catalog_hadoop_forwarding'
+ )
+ """)
+
+ val plan = captureWritePlan("catalog_hadoop_forwarding", allowWriteFailure = true) {
+ spark.sql(
+ s"INSERT INTO $catalogWithOverride.$ns.catalog_hadoop_forwarding " +
+ "VALUES (1, 'us', 1.0)")
+ }
+ val cometWrite = findCometWriteExec(plan)
+ .getOrElse(fail(s"expected CometIcebergWriteExec in:\n$plan"))
+ val properties = cometWrite.nativeOp.getIcebergWrite.getCommon.getCatalogPropertiesMap
+ assert(properties.get("s3.endpoint") == endpoint, properties)
+ }
+ }
+ }
+
+ test("native write preserves FileIO values after catalog initialization") {
+ withTempIcebergDir { warehouseDir =>
+ val initializedCatalog = "s3_initialized_cat"
+ val endpoint = "https://original.example.test"
+ withSQLConf(
+ "fs.s3a.endpoint" -> endpoint,
+ s"spark.sql.catalog.$initializedCatalog" -> "org.apache.iceberg.spark.SparkCatalog",
+ s"spark.sql.catalog.$initializedCatalog.type" -> "hadoop",
+ s"spark.sql.catalog.$initializedCatalog.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
+ spark.sql(s"""
+ CREATE TABLE $initializedCatalog.$ns.initialized_s3 (id INT, region STRING, amount DOUBLE)
+ USING iceberg
+ TBLPROPERTIES ('write.data.path'='s3a://probe-bucket/iceberg/db/initialized_s3')
+ """)
+
+ withSQLConf(
+ "fs.s3a.endpoint" -> "https://session-changed.example.test",
+ "fs.s3a.encryption.algorithm" -> "SSE-KMS",
+ s"spark.sql.catalog.$initializedCatalog.hadoop.fs.s3a.endpoint" ->
+ "https://changed.example.test",
+ s"spark.sql.catalog.$initializedCatalog.hadoop.fs.s3a.encryption.algorithm" ->
+ "SSE-KMS") {
+ // Planning only: inspect the native proto without issuing any S3 requests.
+ val insert = spark.sessionState.sqlParser.parsePlan(
+ s"INSERT INTO $initializedCatalog.$ns.initialized_s3 VALUES (1, 'us', 1.0)")
+ val plan =
+ spark.sessionState.executePlan(insert, CommandExecutionMode.SKIP).executedPlan
+ val cometWrite = findCometWriteExec(plan)
+ .getOrElse(fail(s"expected CometIcebergWriteExec in:\n$plan"))
+ val properties = cometWrite.nativeOp.getIcebergWrite.getCommon.getCatalogPropertiesMap
+ assert(properties.get("s3.endpoint") == endpoint, properties)
+ assert(!properties.containsKey("s3.sse.type"), properties)
+ }
+ }
+ }
+ }
+
+ test("S3FileIO ignores Hadoop options after catalog initialization") {
+ withTempIcebergDir { warehouseDir =>
+ val s3Catalog = "s3_non_hadoop_cat"
+ withSQLConf(
+ s"spark.sql.catalog.$s3Catalog" -> "org.apache.iceberg.spark.SparkCatalog",
+ s"spark.sql.catalog.$s3Catalog.catalog-impl" ->
+ classOf[DetectionS3FileIOHadoopCatalog].getName,
+ s"spark.sql.catalog.$s3Catalog.warehouse" -> warehouseDir.getAbsolutePath,
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
+ spark.sql(s"""
+ CREATE TABLE $s3Catalog.$ns.initialized_s3 (id INT, region STRING, amount DOUBLE)
+ USING iceberg
+ TBLPROPERTIES ('write.data.path'='s3://probe-bucket/iceberg/db/initialized_s3')
+ """)
+
+ withSQLConf(
+ "fs.s3a.endpoint" -> "https://session-changed.example.test",
+ "fs.s3a.encryption.algorithm" -> "SSE-KMS",
+ s"spark.sql.catalog.$s3Catalog.hadoop.fs.s3a.endpoint" ->
+ "https://changed.example.test",
+ s"spark.sql.catalog.$s3Catalog.hadoop.fs.s3a.encryption.algorithm" -> "SSE-KMS") {
+ // Plan only: the real S3FileIO retains its default endpoint and no S3 requests run.
+ val insert = spark.sessionState.sqlParser.parsePlan(
+ s"INSERT INTO $s3Catalog.$ns.initialized_s3 VALUES (1, 'us', 1.0)")
+ val plan =
+ spark.sessionState.executePlan(insert, CommandExecutionMode.SKIP).executedPlan
+ val cometWrite = findCometWriteExec(plan)
+ .getOrElse(fail(s"expected CometIcebergWriteExec in:\n$plan"))
+ assert(IcebergReflection.getFileIO(cometWrite.table).exists(_.isInstanceOf[S3FileIO]))
+ assert(IcebergReflection.getFileIOHadoopConf(cometWrite.table).isEmpty)
+ val properties = cometWrite.nativeOp.getIcebergWrite.getCommon.getCatalogPropertiesMap
+ assert(properties.get("client.region") == "us-east-1", properties)
+ assert(!properties.containsKey("s3.endpoint"), properties)
+ assert(!properties.containsKey("s3.sse.type"), properties)
+ }
+ }
+ }
+ }
+
+ test("ResolvingFileIO S3 delegate ignores wrapper Hadoop options") {
+ withTempIcebergDir { warehouseDir =>
+ val resolvingCatalog = "s3_resolving_hadoop_cat"
+ withSQLConf(
+ s"spark.sql.catalog.$resolvingCatalog" -> "org.apache.iceberg.spark.SparkCatalog",
+ s"spark.sql.catalog.$resolvingCatalog.type" -> "hadoop",
+ s"spark.sql.catalog.$resolvingCatalog.warehouse" -> warehouseDir.getAbsolutePath,
+ s"spark.sql.catalog.$resolvingCatalog.io-impl" -> classOf[ResolvingFileIO].getName,
+ s"spark.sql.catalog.$resolvingCatalog.client.region" -> "us-east-1",
+ s"spark.sql.catalog.$resolvingCatalog.hadoop.fs.s3a.endpoint" ->
+ "https://catalog.example.test",
+ s"spark.sql.catalog.$resolvingCatalog.hadoop.fs.s3a.encryption.algorithm" -> "SSE-KMS",
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
+ Seq("s3", "s3a").foreach { scheme =>
+ val tableName = s"resolving_$scheme"
+ val location = s"$scheme://probe-bucket/iceberg/db/$tableName"
+ spark.sql(s"""
+ CREATE TABLE $resolvingCatalog.$ns.$tableName (id INT, region STRING, amount DOUBLE)
+ USING iceberg
+ TBLPROPERTIES ('write.data.path'='$location')
+ """)
+
+ // Planning only: resolve the real S3FileIO without issuing S3 requests.
+ val insert = spark.sessionState.sqlParser.parsePlan(
+ s"INSERT INTO $resolvingCatalog.$ns.$tableName VALUES (1, 'us', 1.0)")
+ val plan =
+ spark.sessionState.executePlan(insert, CommandExecutionMode.SKIP).executedPlan
+ val cometWrite = findCometWriteExec(plan)
+ .getOrElse(fail(s"expected CometIcebergWriteExec in:\n$plan"))
+ val fileIO = IcebergReflection
+ .getFileIO(cometWrite.table)
+ .getOrElse(fail("could not resolve table FileIO"))
+ assert(fileIO.isInstanceOf[ResolvingFileIO])
+ assert(
+ fileIO.asInstanceOf[ResolvingFileIO].getConf.get("fs.s3a.endpoint") ==
+ "https://catalog.example.test")
+ assert(
+ IcebergReflection.resolveFileIOClass(fileIO, location).contains(classOf[S3FileIO]))
+ assert(IcebergReflection.getFileIOHadoopConf(cometWrite.table).isEmpty)
+ val properties = cometWrite.nativeOp.getIcebergWrite.getCommon.getCatalogPropertiesMap
+ assert(properties.get("client.region") == "us-east-1", properties)
+ assert(!properties.containsKey("s3.endpoint"), properties)
+ assert(!properties.containsKey("s3.sse.type"), properties)
+ }
+ }
+ }
+ }
+
+ test("fall-back: unsupported Hadoop S3A setting on an S3 data location") {
+ val secret = "SECRET_VALUE_MUST_NOT_APPEAR"
+ withSQLConf("fs.s3a.encryption.algorithm" -> secret) {
+ withDetectionCatalog { dir =>
+ createTable(
+ dir,
+ "s3a_hadoop_unsupported",
+ partitionSpec = "",
+ properties =
+ Some("'write.data.path'='s3a://probe-bucket/iceberg/db/s3a_hadoop_unsupported'"))
+ val writeExec = planInsertWriteExec(s"$catalog.$ns.s3a_hadoop_unsupported")
+ val support = CometIcebergNativeWrite.getSupportLevel(writeExec)
+ support match {
+ case Unsupported(Some(reason)) =>
+ assert(reason == "unsupported Hadoop S3A setting: fs.s3a.encryption.algorithm")
+ assert(!reason.contains(secret), reason)
+ case other => fail(s"expected Unsupported, got $other")
+ }
+ val planReasons =
+ writeExec.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty)
+ assert(planReasons.exists(_.contains("fs.s3a.encryption.algorithm")), planReasons)
+ assert(!planReasons.exists(_.contains(secret)), planReasons)
+ assert(!writeExec.toString.contains(secret), writeExec.toString)
+ }
+ }
+ }
+
+ test("S3-only setting gates do not affect a local data location") {
+ withDetectionCatalog { dir =>
+ createTable(dir, "local_with_s3_conf", partitionSpec = "")
+ withSQLConf("fs.s3a.encryption.algorithm" -> "SSE-KMS") {
+ assertSupportLevelIs[Compatible]("local_with_s3_conf")
+ }
+ }
+ }
+
+ test("fall-back: unsupported S3 FileIO properties") {
+ withTempIcebergDir { warehouseDir =>
+ val fileIOCat = "s3_file_io_probe_cat"
+ val secret = "SECRET_VALUE_MUST_NOT_APPEAR"
+ withSQLConf(
+ s"spark.sql.catalog.$fileIOCat" -> "org.apache.iceberg.spark.SparkCatalog",
+ s"spark.sql.catalog.$fileIOCat.type" -> "hadoop",
+ s"spark.sql.catalog.$fileIOCat.warehouse" -> warehouseDir.getAbsolutePath,
+ s"spark.sql.catalog.$fileIOCat.io-impl" -> classOf[ResolvingFileIO].getName,
+ s"spark.sql.catalog.$fileIOCat.s3.acl" -> secret,
+ s"spark.sql.catalog.$fileIOCat.s3.sse.type" -> "dsse-kms",
+ s"spark.sql.catalog.$fileIOCat.s3.write.tags.foo" -> secret) {
+ spark.sql(s"""
+ CREATE TABLE $fileIOCat.$ns.unsupported_file_io (
+ id INT,
+ region STRING,
+ amount DOUBLE
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.data.path'='s3a://probe-bucket/iceberg/db/unsupported_file_io'
+ )
+ """)
+ val writeExec = planInsertWriteExec(s"$fileIOCat.$ns.unsupported_file_io")
+ val support = CometIcebergNativeWrite.getSupportLevel(writeExec)
+ support match {
+ case Unsupported(Some(reason)) =>
+ assert(
+ reason ==
+ "unsupported S3 FileIO settings: s3.acl, s3.sse.type, s3.write.tags.foo",
+ reason)
+ assert(!reason.contains(secret), reason)
+ case other => fail(s"expected Unsupported, got $other")
+ }
+ val planReasons =
+ writeExec.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty)
+ assert(planReasons.exists(_.contains("s3.acl")), planReasons)
+ assert(planReasons.exists(_.contains("s3.write.tags.foo")), planReasons)
+ assert(!planReasons.exists(_.contains(secret)), planReasons)
+ }
+ }
+ }
+
+ test("custom credential-provider S3/client properties keep the native writer engaged") {
+ withTempIcebergDir { warehouseDir =>
+ val fileIOCat = "s3_provider_props_cat"
+ val secret = "SECRET_PROVIDER_VALUE_MUST_NOT_APPEAR"
+ val providerClass =
+ classOf[org.apache.comet.cloud.s3.MinioCometS3CredentialProvider].getName
+ withSQLConf(
+ s"spark.sql.catalog.$fileIOCat" -> "org.apache.iceberg.spark.SparkCatalog",
+ s"spark.sql.catalog.$fileIOCat.type" -> "hadoop",
+ s"spark.sql.catalog.$fileIOCat.warehouse" -> warehouseDir.getAbsolutePath,
+ s"spark.sql.catalog.$fileIOCat.io-impl" -> classOf[ResolvingFileIO].getName,
+ s"spark.sql.catalog.$fileIOCat.s3.comet.credential.provider.class" -> providerClass,
+ s"spark.sql.catalog.$fileIOCat.s3.vendor.credential-scope" -> secret,
+ s"spark.sql.catalog.$fileIOCat.client.vendor.tenant-id" -> "tenant-A",
+ CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key -> "true") {
+ spark.sql(s"""
+ CREATE TABLE $fileIOCat.$ns.provider_properties (
+ id INT,
+ region STRING,
+ amount DOUBLE
+ ) USING iceberg
+ TBLPROPERTIES (
+ 'write.data.path'='s3a://probe-bucket/iceberg/db/provider_properties'
+ )
+ """)
+
+ val plan = captureWritePlan("provider_properties", allowWriteFailure = true) {
+ spark.sql(s"INSERT INTO $fileIOCat.$ns.provider_properties VALUES (1, 'us', 1.0)")
+ }
+ val cometWrite = findCometWriteExec(plan)
+ .getOrElse(fail(s"expected CometIcebergWriteExec in:\n$plan"))
+ val properties = cometWrite.nativeOp.getIcebergWrite.getCommon.getCatalogPropertiesMap
+ assert(properties.get("s3.vendor.credential-scope") == secret, properties)
+ assert(properties.get("client.vendor.tenant-id") == "tenant-A", properties)
+ assert(!cometWrite.simpleString(Int.MaxValue).contains(secret))
+ }
+ }
+ }
+
test("Compatible when the data location scheme is memory") {
withDetectionCatalog { dir =>
createTable(
@@ -813,17 +1313,17 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
// proto assembly see. LocalTableScan conversion is enabled the same way the write action
// suite does, so the VALUES insert converts and the built proto is inspectable.
withDetectionCatalog { dir =>
- createTable(
- dir,
- "s3a_props",
- partitionSpec = "",
- properties = Some("'write.data.path'='s3a://probe-bucket/iceberg/db/s3a_props'"))
val conf = spark.sessionState.conf
conf.setConfString(CometConf.COMET_EXEC_LOCAL_TABLE_SCAN_ENABLED.key, "true")
conf.setConfString("fs.s3a.endpoint", "http://localhost:9000")
conf.setConfString("fs.s3a.access.key", "probe-access-key")
conf.setConfString("fs.s3a.path.style.access", "true")
try {
+ createTable(
+ dir,
+ "s3a_props",
+ partitionSpec = "",
+ properties = Some("'write.data.path'='s3a://probe-bucket/iceberg/db/s3a_props'"))
val plan = captureWritePlan("s3a_props", allowWriteFailure = true) {
spark.sql(s"INSERT INTO $catalog.$ns.s3a_props VALUES (1, 'us', 1.0)")
}
@@ -1141,10 +1641,13 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
}
}
- private val catalog = "cat"
+ private var catalog = "cat"
private val ns = "db"
private def withDetectionCatalog(f: File => Unit): Unit = withTempIcebergDir { warehouseDir =>
+ // Spark caches catalog instances, including their initialized Hadoop configurations.
+ // Each fixture needs a fresh catalog to observe the settings selected by this test.
+ catalog = "cat_" + java.util.UUID.randomUUID().toString.replace("-", "")
withSQLConf(
s"spark.sql.catalog.$catalog" -> "org.apache.iceberg.spark.SparkCatalog",
s"spark.sql.catalog.$catalog.type" -> "hadoop",
@@ -1463,3 +1966,20 @@ class DetectionDelegatingLocationProvider(delegate: LocationProvider) extends Lo
filename: String): String =
delegate.newDataLocation(spec, partitionData, filename)
}
+
+/** A real S3FileIO for data, with local metadata managed by HadoopCatalog. Planning only. */
+class DetectionS3FileIOHadoopCatalog extends HadoopCatalog {
+ override protected def newTableOps(identifier: TableIdentifier): TableOperations =
+ new DetectionS3FileIOTableOperations(super.newTableOps(identifier))
+}
+
+class DetectionS3FileIOTableOperations(delegate: TableOperations)
+ extends DetectionLocationProviderTableOperations(delegate) {
+ private val s3FileIO = new S3FileIO()
+ s3FileIO.initialize(java.util.Collections.singletonMap("client.region", "us-east-1"))
+
+ override def io(): FileIO = s3FileIO
+ override def locationProvider(): LocationProvider = delegate.locationProvider()
+ override def temp(uncommittedMetadata: TableMetadata): TableOperations =
+ new DetectionS3FileIOTableOperations(delegate.temp(uncommittedMetadata))
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org