Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
Loading