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 @@ -89,6 +89,16 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] with
private val ParquetWritePropertyPrefix = "write.parquet."
private val ParquetMrPropertyPrefix = "parquet."

// 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.
// `parquet.hadoop.vectored.io.enabled` is a reader-side vectored-IO knob declared by
// parquet-hadoop as `ParquetInputFormat.HADOOP_VECTORED_IO_ENABLED` (default `true` in
// parquet-hadoop 1.16+) and only consulted by parquet-mr's Hadoop reader path. Keep it
// out of the writer-compatibility gate so that environments which seed it into the
// session Hadoop configuration do not silently disable native Iceberg writes.
private val IgnoredHadoopParquetConfKeys: Set[String] = Set(
"parquet.hadoop.vectored.io.enabled")

private lazy val vettedParquetWriteKeys: Set[String] = Set(
PropertyKeys.ParquetCompressionCodec,
PropertyKeys.ParquetCompressionLevel,
Expand Down Expand Up @@ -271,7 +281,8 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] with
private val requireNoParquetHadoopConfOverrides: TriggerRule = ctx =>
ctx.hadoopConf.asScala
.map(_.getKey)
.find(_.startsWith(ParquetMrPropertyPrefix))
.filter(_.startsWith(ParquetMrPropertyPrefix))
.find(k => !IgnoredHadoopParquetConfKeys.contains(k))
.map(k => s"Hadoop configuration sets $k (reaches iceberg-java's writer but not native)")

private val requireSupportedStorageScheme: TriggerRule = ctx =>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -357,6 +357,20 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
}
}

// parquet.hadoop.vectored.io.enabled is a reader-side vectored-IO knob declared by
// parquet-hadoop (ParquetInputFormat.HADOOP_VECTORED_IO_ENABLED, default true in
// parquet-hadoop 1.16+). iceberg-java's writer never consumes it, so it must not
// disable native Iceberg writes when it happens to be present in the session
// Hadoop configuration.
test("Compatible when only parquet.hadoop.vectored.io.enabled is set in Hadoop configuration") {
withDetectionCatalog { dir =>
createTable(dir, "vectored_io_only", partitionSpec = "")
withSQLConf("parquet.hadoop.vectored.io.enabled" -> "true") {
assertSupportLevelIs[Compatible]("vectored_io_only")
}
}
}

test("fall-back: io-impl set") {
withDetectionCatalog { dir =>
createTable(
Expand Down
Loading