From 8a1c0f9893951a3b1bb047ac3b90d602cfe7414b Mon Sep 17 00:00:00 2001
From: Andy Grove <agrove@apache.org>
Date: Thu, 8 Oct 2026 16:58:48 -0600
Subject: [PATCH] fix: don't let S3A committer, delete and read defaults block
 native Iceberg writes

#6441 makes a native Iceberg write to S3 fall back when the Hadoop
configuration has an fs.s3a.* setting the native writer doesn't support.
Spark 4.1 sets fs.s3a.committer.magic.enabled and fs.s3a.committer.name
for every application when spark-hadoop-cloud is on the classpath
(SPARK-47618), and many deployments set fs.s3a.experimental.input.fadvise,
fs.s3a.bulk.delete.page.size or fs.s3a.committer.threads cluster-wide. Any
of them makes every native S3 write fall back.

None of these settings affects an Iceberg data-file write: the native
writer doesn't go through S3A, fadvise is a read hint, and Iceberg commits
through table metadata rather than a Hadoop output committer. Ignore them
like the Spark-seeded read settings already in IgnoredHadoopS3Keys.
---
 .../operator/CometIcebergNativeWrite.scala      | 11 ++++++++++-
 .../comet/CometIcebergWriteDetectionSuite.scala | 17 +++++++++++++++++
 2 files changed, 27 insertions(+), 1 deletion(-)

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 34c07ff8dd5..9c708e2d0d1 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
@@ -158,10 +158,19 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
   // 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.
+  //
+  // Some Spark distributions also set S3A delete, read and committer defaults cluster-wide. The
+  // native writer does not go through S3A, `fadvise` is a read hint, and Iceberg commits through
+  // table metadata rather than a Hadoop output committer, so these are harmless here as well.
   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")
+    "fs.s3a.vectored.read.min.seek.size",
+    "fs.s3a.bulk.delete.page.size",
+    "fs.s3a.experimental.input.fadvise",
+    "fs.s3a.committer.magic.enabled",
+    "fs.s3a.committer.name",
+    "fs.s3a.committer.threads")
 
   // 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
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
index 922b5d6073f..7caffd0e7ee 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
@@ -772,6 +772,23 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
     }
   }
 
+  test("cluster-wide S3A delete, read and committer defaults do not block a native write") {
+    val hadoopConf = new Configuration(false)
+    Seq(
+      "fs.s3a.bulk.delete.page.size" -> "1000",
+      "fs.s3a.experimental.input.fadvise" -> "random",
+      "fs.s3a.committer.magic.enabled" -> "true",
+      "fs.s3a.committer.name" -> "magic",
+      "fs.s3a.committer.threads" -> "8").foreach { case (k, v) => hadoopConf.set(k, v) }
+    assert(
+      CometIcebergNativeWrite.unsupportedHadoopS3Settings(hadoopConf, Some("target")).isEmpty)
+
+    hadoopConf.set("fs.s3a.encryption.algorithm", "SSE-KMS")
+    assert(
+      CometIcebergNativeWrite.unsupportedHadoopS3Settings(hadoopConf, Some("target")) == Seq(
+        "fs.s3a.encryption.algorithm"))
+  }
+
   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>