Re: [PR] fix: gate unsupported Iceberg S3 write settings [datafusion-comet]

["sunchao (via GitHub)" <gi...@apache.org>]

sunchao commented on code in PR #6441:
URL: https://github.com/apache/datafusion-comet/pull/6441#discussion_r4154010515


##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -90,6 +91,96 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
   private val MinUnsupportedFormatVersion = 3
   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.
+  private val SupportedHadoopS3Suffixes: Set[String] =
+    NativeConfig.s3aSuffixToIcebergGlobalKey.keySet
+
+  private val SupportedHadoopS3Keys: Set[String] =
+    SupportedHadoopS3Suffixes.map("fs.s3a." + _)
+
+  // 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 case class IcebergAwsPropertyNames(exact: Set[String], prefixes: Seq[String]) {
+    def contains(key: String): Boolean =
+      exact.contains(key) || prefixes.exists(key.startsWith)
+  }
+
+  // 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.
+  private lazy val IcebergAwsProperties: IcebergAwsPropertyNames = {
+    val propertyNames = Seq(
+      "org.apache.iceberg.aws.s3.S3FileIOProperties",
+      "org.apache.iceberg.aws.AwsClientProperties").flatMap { className =>
+      IcebergReflection
+        .loadClass(className)

Review Comment:
   [P2] Loading these AWS classes can abort query planning instead of returning a support decision. On a Spark 3.5/HadoopFileIO deployment without AWS SDK v2, configuring `s3.comet.credential.provider.class` plus a vendor property such as `s3.vendor.credential-scope` reaches this lookup and throws `NoClassDefFoundError`. HadoopFileIO and Comet’s provider SPI do not require SDK v2, and its S3 dependencies are test-scoped for this profile. `getSupportLevel` catches only `NonFatal`, which excludes this error, so previously eligible writes now fail before execution. Avoid loading optional AWS-dependent classes for classification, or explicitly handle linkage failures with a safe fallback.
   
   Evidence: A JDK 17 harness compiled the exact-head helpers and repository `ClassLoaders.java`. With Hadoop 3.3.4 and Iceberg runtime 1.8.1, HadoopFileIO initialized and returned the provider properties successfully. The new classifier then threw `NoClassDefFoundError: software/amazon/awssdk/auth/credentials/AwsCredentialsProvider`, with `NonFatal(error) == false`. Adding SDK v2 made the same classification succeed. The behavior also reproduced with Iceberg 1.11.0.



##########
spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeWrite.scala:
##########
@@ -347,6 +440,94 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
         else Some(s"unsupported storage scheme: $scheme")
     }
 
+  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. 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 targetBucketPrefix = targetBucket.map(bucket => s"fs.s3a.bucket.$bucket.")
+
+    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
+      // a Hadoop *-default.xml resource.
+      .filter { key =>
+        Option(hadoopConf.getPropertySources(key))
+          .forall(sources => sources.isEmpty || !sources.forall(_.endsWith("-default.xml")))
+      }
+      .filter { key =>
+        targetBucketPrefix match {
+          case Some(prefix) if key.startsWith(prefix) =>

Review Comment:
   [P2] The prefix match mistakes settings for a longer dotted bucket name for settings of the target bucket. For a write to `s3a://target/...`, setting `fs.s3a.bucket.target.other.endpoint` for the separate bucket `target.other` produces an unsupported-setting rejection because the stripped suffix becomes `other.endpoint`. That endpoint does not affect writes to `target`, so the expected behavior is to preserve native eligibility. This change instead disables native writes to `target` whenever the unrelated configuration is present. Distinguish the complete bucket name and property suffix, and add a regression case with overlapping dotted bucket names.
   
   Evidence: The exact-head helper returned `Seq("fs.s3a.bucket.target.other.endpoint")` for target bucket `target`. The unchanged `hadoopToIcebergS3Properties` helper dropped that endpoint and forwarded only the target bucket’s access key. Hadoop 3.3.4’s `propagateBucketOptions` likewise maps the overlapping entry to `fs.s3a.other.endpoint`, not the consumed `fs.s3a.endpoint`. Removing the unrelated entry restored an empty rejection list.



-- 
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