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/09/18 22:47:01 UTC
(datafusion-comet) 01/01: fix: fall back to Spark for native Iceberg writes to gs:// through HadoopFileIO (#5935)
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-5935-ba0a7eec7227b14b38e4bb1b4919837ac31e75ff
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 5fdc96199685061b7c67ad28651b4c0a3dcd6541
Author: Feng Zhang <fe...@apache.org>
AuthorDate: Fri Sep 18 22:21:59 2026 +0000
fix: fall back to Spark for native Iceberg writes to gs:// through HadoopFileIO (#5935)
* fix: Fall back to Spark for native Iceberg writes to gs:// through HadoopFileIO
HadoopFileIO takes its GCS configuration from fs.gs.*, which is not
forwarded to the native writer, so a native write could run with a
different storage identity or endpoint than the JVM writer. Decline a
gs:// data location unless the FileIO Iceberg resolves for it is a
GCSFileIO.
Closes #5637
* test: resolve the ResolvingFileIO delegate through the reflection helper
ResolvingFileIO.ioClass throws a NoClassDefFoundError when the GCS
client libraries are absent, which aborted the suite on the Spark 3.4
and 3.5 CI jobs. Resolve the delegate the way the gate does and expect a
decline when it cannot be resolved.
* fix: judge a ResolvingFileIO by the delegate it instantiates
ResolvingFileIO.ioClass only maps the scheme to a class; io(location)
instantiates it and falls back to HadoopFileIO when that fails, so the
gate must inspect the instantiated delegate.
* test: make the GCSFileIO initialization-failure fixture version-aware
Iceberg 1.10+ defers gcs.* parsing to client construction, so an
unparseable chunk size no longer fails GCSFileIO.initialize there.
---
docs/source/user-guide/latest/iceberg-writes.md | 11 +-
.../apache/comet/iceberg/IcebergReflection.scala | 52 ++++++-
.../serde/operator/CometIcebergNativeWrite.scala | 43 +++++-
.../comet/CometIcebergWriteDetectionSuite.scala | 172 +++++++++++++++++++--
4 files changed, 256 insertions(+), 22 deletions(-)
diff --git a/docs/source/user-guide/latest/iceberg-writes.md b/docs/source/user-guide/latest/iceberg-writes.md
index e179606a9b..2ffcd08d05 100644
--- a/docs/source/user-guide/latest/iceberg-writes.md
+++ b/docs/source/user-guide/latest/iceberg-writes.md
@@ -158,7 +158,7 @@ A write is eligible only when ALL of the following hold:
| `write.metadata.metrics.*` | any value (manifest metrics are re-derived on the JVM with Iceberg's own logic) [...]
| `write.spark.fanout.enabled` | any value (the native writer implements both clustered and fanout modes) [...]
| `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` [...]
+| data location URI scheme | `file`, `memory`, `s3`, `s3a`, `gs` (`gs` only when the `FileIO` opening the data location is a `GCSFileIO`; see below) [...]
| partition spec | any [...]
| column types | any except `uuid` (Spark plans it as a string; no Arrow cast reaches `fixed(16)`) [...]
@@ -177,6 +177,15 @@ scan uses, minus the `EncryptingFileIO` family — the native writer produces pl
so an encrypting `FileIO` is rejected on the write side), and `table.encryption()` must be
Iceberg's `PlaintextEncryptionManager`. Anything else falls back.
+A `gs` data location additionally requires that the `FileIO` actually opening it is a
+`GCSFileIO` (for a `ResolvingFileIO`, the delegate it instantiates for that location,
+which is a `HadoopFileIO` when the GCS `FileIO` cannot be loaded or initialized). A
+`HadoopFileIO` takes its GCS credentials, endpoint and project from `fs.gs.*` in the Hadoop
+Configuration, and only `fs.s3a.*` is translated into the native `FileIO`, so the native writer
+could resolve a different storage identity or endpoint than the JVM writer would. That
+combination falls back; a `GCSFileIO` carries its `gcs.*` settings in `FileIO.properties()`,
+which are forwarded.
+
Other `write.*` properties are intentionally not gated because they cannot make the native
writer produce different data files: distribution and ordering settings shape the Spark plan
identically on both paths, WAP / branch / snapshot properties act on the JVM committer,
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 cd05c98178..c1cb43b153 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -57,6 +57,8 @@ object IcebergReflection extends Logging {
val SPARK_STAGED_SCAN = "org.apache.iceberg.spark.source.SparkStagedScan"
val SPARK_SCHEMA_UTIL = "org.apache.iceberg.spark.SparkSchemaUtil"
val TABLE = "org.apache.iceberg.Table"
+ val RESOLVING_FILE_IO = "org.apache.iceberg.io.ResolvingFileIO"
+ val GCS_FILE_IO = "org.apache.iceberg.gcp.gcs.GCSFileIO"
val PARTITIONING = "org.apache.iceberg.Partitioning"
val SPARK_WRITE = "org.apache.iceberg.spark.source.SparkWrite"
val TABLE_PROPERTIES = "org.apache.iceberg.TableProperties"
@@ -310,8 +312,11 @@ object IcebergReflection extends Logging {
method
}
- private def declaredMethod(clazz: Class[_], methodName: String): Option[Method] =
- try Some(makeAccessible(clazz.getDeclaredMethod(methodName)))
+ private def declaredMethod(
+ clazz: Class[_],
+ methodName: String,
+ paramTypes: Class[_]*): Option[Method] =
+ try Some(makeAccessible(clazz.getDeclaredMethod(methodName, paramTypes: _*)))
catch { case _: NoSuchMethodException => None }
/**
@@ -343,12 +348,15 @@ object IcebergReflection extends Logging {
/**
* Searches through class hierarchy to find a method (including protected methods).
*/
- def findMethodInHierarchy(clazz: Class[_], methodName: String): Option[Method] =
- cachedLookup(clazz, "hierarchy:" + methodName) {
+ def findMethodInHierarchy(
+ clazz: Class[_],
+ methodName: String,
+ paramTypes: Class[_]*): Option[Method] =
+ cachedLookup(clazz, "hierarchy:" + lookupKey(methodName, paramTypes)) {
var current: Class[_] = clazz
var found: Option[Method] = None
while (found.isEmpty && current != null) {
- found = declaredMethod(current, methodName)
+ found = declaredMethod(current, methodName, paramTypes: _*)
if (found.isEmpty) current = current.getSuperclass
}
found
@@ -535,6 +543,40 @@ object IcebergReflection extends Logging {
}
}
+ /**
+ * The FileIO class that actually opens `location`: for a `ResolvingFileIO`, the delegate it
+ * instantiates for the location (`io(location)`, which falls back to HadoopFileIO when the
+ * scheme's FileIO cannot be loaded or initialized -- `ioClass(location)` only maps the scheme
+ * to a class and misses that fallback), or the FileIO's own class otherwise. The delegate is
+ * cached by the ResolvingFileIO, so this is the instance the JVM writer would use. `None` on
+ * reflection failure; callers must fail closed.
+ */
+ def resolveFileIOClass(fileIO: Any, location: String): Option[Class[_]] =
+ if (!classNameInHierarchy(fileIO.getClass, Set(ClassNames.RESOLVING_FILE_IO))) {
+ Some(fileIO.getClass)
+ } else {
+ try {
+ findMethodInHierarchy(fileIO.getClass, "io", classOf[String]) match {
+ case Some(ioMethod) => Option(ioMethod.invoke(fileIO, location)).map(_.getClass)
+ case None =>
+ logError(
+ s"Iceberg reflection failure: ${fileIO.getClass.getName} has no io(String) method")
+ None
+ }
+ } catch {
+ case e: Exception =>
+ // Method.invoke wraps whatever io(location) throws; report that, not the wrapper.
+ val cause = e match {
+ case ite: java.lang.reflect.InvocationTargetException => ite.getCause
+ case other => other
+ }
+ logError(
+ "Iceberg reflection failure: Failed to resolve the FileIO delegate for " +
+ s"$location: $cause")
+ None
+ }
+ }
+
/**
* The table's `EncryptionManager` (`table.encryption()`). Unlike the `encryption.*` property
* prefix, this reflects what the table's `TableOperations` actually installed, so it also
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 4aee4123ff..94f6b13927 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
@@ -84,6 +84,7 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
// `oss` is deliberately absent: iceberg-rust has an OSS backend, but Comet does not forward
// `oss.*` catalog properties to it and no functional test covers the path, so an OSS write
// could silently drop endpoint/credential configuration. Fail closed until it is covered.
+ // `gs` is additionally gated on the resolved FileIO (`requireGcsFileIOForGcsDataLocation`).
private val SupportedStorageSchemes: Set[String] =
Set("file", "memory", "s3", "s3a", "gs")
private val MinUnsupportedFormatVersion = 3
@@ -192,6 +193,7 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
requirePositiveIntParquetSizes,
requireNoParquetHadoopConfOverrides,
requireSupportedStorageScheme,
+ requireGcsFileIOForGcsDataLocation,
requireExecutorReflectionResolvable)
private val requireFormatParquet: TriggerRule = ctx =>
@@ -305,19 +307,50 @@ object CometIcebergNativeWrite extends CometOperatorSerde[IcebergWriteExec] {
.find(k => !IgnoredHadoopParquetConfKeys.contains(k))
.map(k => s"Hadoop configuration sets $k (reaches iceberg-java's writer but not native)")
+ private def storageScheme(location: String): String =
+ if (location.contains("://")) {
+ location.substring(0, location.indexOf("://")).toLowerCase(Locale.ROOT)
+ } else {
+ "file"
+ }
+
private val requireSupportedStorageScheme: TriggerRule = ctx =>
IcebergReflection.getDataLocation(ctx.table) match {
case None => Some("could not resolve the table data location")
case Some(location) =>
- val scheme = if (location.contains("://")) {
- location.substring(0, location.indexOf("://")).toLowerCase(Locale.ROOT)
- } else {
- "file"
- }
+ val scheme = storageScheme(location)
if (SupportedStorageSchemes.contains(scheme)) None
else Some(s"unsupported storage scheme: $scheme")
}
+ // 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.
+ private val requireGcsFileIOForGcsDataLocation: TriggerRule = ctx =>
+ IcebergReflection.getDataLocation(ctx.table).filter(storageScheme(_) == "gs").flatMap {
+ location =>
+ val resolved = IcebergReflection
+ .getFileIO(ctx.table)
+ .flatMap(io => IcebergReflection.resolveFileIOClass(io, location))
+ gcsDataLocationRejection(location, resolved)
+ }
+
+ private[comet] def gcsDataLocationRejection(
+ location: String,
+ resolvedFileIO: Option[Class[_]]): Option[String] =
+ resolvedFileIO match {
+ case Some(cls)
+ if IcebergReflection
+ .classNameInHierarchy(cls, Set(IcebergReflection.ClassNames.GCS_FILE_IO)) =>
+ None
+ case Some(cls) =>
+ Some(
+ s"gs:// data location $location is written through ${cls.getName}, whose fs.gs.* " +
+ "Hadoop configuration is not forwarded to the native writer")
+ case None =>
+ Some(s"could not resolve the FileIO for the gs:// data location $location")
+ }
+
// The commit-message assembly that runs on executors after iceberg-rust has already written
// the task's data files is pure reflection over iceberg-java internals. Resolving the whole
// surface up front turns an Iceberg release that moves any of it into a plan-time fallback
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
index 7c5631640c..37ddb1872e 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteDetectionSuite.scala
@@ -26,7 +26,7 @@ import org.scalatest.Tag
import org.apache.hadoop.conf.Configuration
import org.apache.iceberg.hadoop.{HadoopConfigurable, HadoopFileIO}
-import org.apache.iceberg.io.{FileIO, InputFile, OutputFile}
+import org.apache.iceberg.io.{FileIO, InputFile, OutputFile, ResolvingFileIO}
import org.apache.iceberg.util.SerializableSupplier
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
@@ -34,7 +34,7 @@ import org.apache.spark.sql.CometTestBase
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference}
import org.apache.spark.sql.comet.{CometIcebergWriteExec, CometSparkToColumnarExec, IcebergWriteExec}
-import org.apache.spark.sql.execution.{ApplyColumnarRulesAndInsertTransitions, ColumnarToRowExec, LeafExecNode, SparkPlan}
+import org.apache.spark.sql.execution.{ApplyColumnarRulesAndInsertTransitions, ColumnarToRowExec, CommandExecutionMode, LeafExecNode, SparkPlan}
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.vectorized.ColumnarBatch
@@ -455,16 +455,154 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
}
}
- test("Compatible for the remaining supported data location schemes") {
+ test("Compatible when the data location scheme is memory") {
withDetectionCatalog { dir =>
- Seq("gs", "memory").foreach { scheme =>
- val table = s"${scheme}_scheme"
- createTable(
- dir,
- table,
- partitionSpec = "",
- properties = Some(s"'write.data.path'='$scheme://nonexistent/iceberg/db/$table'"))
- assertSupportLevelIs[Compatible](table, allowWriteFailure = true)
+ createTable(
+ dir,
+ "memory_scheme",
+ partitionSpec = "",
+ properties = Some("'write.data.path'='memory://nonexistent/iceberg/db/memory_scheme'"))
+ assertSupportLevelIs[Compatible]("memory_scheme", allowWriteFailure = true)
+ }
+ }
+
+ test("fall-back: gs data location under HadoopFileIO (fs.gs.* is not forwarded)") {
+ // The hadoop catalog's table.io() is a HadoopFileIO. Planned only, never executed: running
+ // the write would have the Hadoop GCS connector look for credentials over the network.
+ withDetectionCatalog { dir =>
+ createTable(
+ dir,
+ "gs_hadoop_io",
+ partitionSpec = "",
+ properties = Some("'write.data.path'='gs://nonexistent/iceberg/db/gs_hadoop_io'"))
+ assertUnsupportedContains(
+ planInsertWriteExec(s"$catalog.$ns.gs_hadoop_io"),
+ "gs_hadoop_io",
+ "gs://",
+ classOf[HadoopFileIO].getName)
+ }
+ }
+
+ test("gs data location gate decides on the resolved FileIO class") {
+ // Deterministic coverage of every branch; the ResolvingFileIO test below depends on which
+ // delegate this classpath yields. GCSFileIO is loaded without initialization so the
+ // optional GCS client libraries are never touched.
+ val location = "gs://bucket/iceberg/db/t"
+ val gcsFileIO =
+ Class.forName(IcebergReflection.ClassNames.GCS_FILE_IO, false, getClass.getClassLoader)
+ assert(CometIcebergNativeWrite.gcsDataLocationRejection(location, Some(gcsFileIO)).isEmpty)
+ val hadoop =
+ CometIcebergNativeWrite.gcsDataLocationRejection(location, Some(classOf[HadoopFileIO]))
+ assert(
+ hadoop.exists(r => r.contains("gs://") && r.contains(classOf[HadoopFileIO].getName)),
+ hadoop)
+ val unresolved = CometIcebergNativeWrite.gcsDataLocationRejection(location, None)
+ assert(unresolved.exists(_.contains("gs://")), unresolved)
+ }
+
+ test("gs data location under ResolvingFileIO with a GCSFileIO that fails to initialize") {
+ // ResolvingFileIO.ioClass maps gs:// to GCSFileIO, but the delegate it instantiates is a
+ // HadoopFileIO whenever loading or initializing GCSFileIO throws an IllegalArgumentException,
+ // so the gate must judge the instantiated delegate. Iceberg before 1.10 parses gcs.* in
+ // GCSFileIO.initialize, so an unparseable chunk size takes that fallback wherever GCSFileIO
+ // can be constructed (where the class does not load, the same fallback runs; where its
+ // construction fails, Iceberg does not fall back and the delegate is unresolvable). 1.10+
+ // defers the parsing to client construction, so there the property leaves the delegate
+ // unchanged and the gate must follow whatever Iceberg instantiates.
+ withTempIcebergDir { warehouseDir =>
+ val location = "gs://nonexistent/iceberg/db/gs_resolving_bad"
+ val badProperty = "gcs.channel.read.chunk-size-bytes" -> "invalid"
+ def resolveWith(props: java.util.Map[String, String]): Option[Class[_]] = {
+ val resolving = new ResolvingFileIO()
+ resolving.setConf(new Configuration())
+ resolving.initialize(props)
+ try IcebergReflection.resolveFileIOClass(resolving, location)
+ finally resolving.close()
+ }
+ val eagerInit = !icebergVersionAtLeast(1, 10)
+ val delegate =
+ resolveWith(java.util.Collections.singletonMap(badProperty._1, badProperty._2))
+ logInfo(
+ s"ResolvingFileIO delegate with $badProperty on this classpath: $delegate " +
+ s"(eager initialization: $eagerInit)")
+ if (eagerInit) {
+ assert(delegate.forall(_ == classOf[HadoopFileIO]), delegate)
+ } else {
+ val unaffected = resolveWith(java.util.Collections.emptyMap[String, String]())
+ assert(delegate == unaffected, s"$delegate differs from $unaffected without the property")
+ }
+ val badCat = "resolving_bad_io_cat"
+ withSQLConf(
+ s"spark.sql.catalog.$badCat" -> "org.apache.iceberg.spark.SparkCatalog",
+ s"spark.sql.catalog.$badCat.type" -> "hadoop",
+ s"spark.sql.catalog.$badCat.warehouse" -> warehouseDir.getAbsolutePath,
+ s"spark.sql.catalog.$badCat.io-impl" -> classOf[ResolvingFileIO].getName,
+ s"spark.sql.catalog.$badCat.${badProperty._1}" -> badProperty._2) {
+ spark.sql(s"""
+ CREATE TABLE $badCat.$ns.gs_resolving_bad (
+ id INT,
+ region STRING,
+ amount DOUBLE
+ ) USING iceberg
+ TBLPROPERTIES ('write.data.path'='$location')
+ """)
+ val support = CometIcebergNativeWrite.getSupportLevel(
+ planInsertWriteExec(s"$badCat.$ns.gs_resolving_bad"))
+ if (delegate.exists(_.getName == IcebergReflection.ClassNames.GCS_FILE_IO)) {
+ assert(!eagerInit, "an unparseable chunk size must fail eager initialization")
+ assert(
+ support.isInstanceOf[Compatible],
+ s"expected Compatible via GCSFileIO, got $support")
+ } else {
+ support match {
+ case Unsupported(Some(reason)) =>
+ assert(reason.contains("gs://") && !reason.contains("GCSFileIO"), reason)
+ case other => fail(s"expected Unsupported with a reason, got $other")
+ }
+ }
+ }
+ }
+ }
+
+ test("gs data location under ResolvingFileIO is judged by the resolved delegate") {
+ // ResolvingFileIO (the REST catalog default) instantiates GCSFileIO for gs:// when the GCS
+ // client libraries are present and HadoopFileIO otherwise; the expectation follows whichever
+ // this classpath yields.
+ withTempIcebergDir { warehouseDir =>
+ val location = "gs://nonexistent/iceberg/db/gs_resolving"
+ val resolving = new ResolvingFileIO()
+ resolving.setConf(new Configuration())
+ resolving.initialize(java.util.Collections.emptyMap[String, String]())
+ val delegate =
+ try IcebergReflection.resolveFileIOClass(resolving, location)
+ finally resolving.close()
+ logInfo(s"ResolvingFileIO delegate for $location on this classpath: $delegate")
+ val resolvingCat = "resolving_io_cat"
+ withSQLConf(
+ s"spark.sql.catalog.$resolvingCat" -> "org.apache.iceberg.spark.SparkCatalog",
+ s"spark.sql.catalog.$resolvingCat.type" -> "hadoop",
+ s"spark.sql.catalog.$resolvingCat.warehouse" -> warehouseDir.getAbsolutePath,
+ s"spark.sql.catalog.$resolvingCat.io-impl" -> classOf[ResolvingFileIO].getName) {
+ spark.sql(s"""
+ CREATE TABLE $resolvingCat.$ns.gs_resolving (
+ id INT,
+ region STRING,
+ amount DOUBLE
+ ) USING iceberg
+ TBLPROPERTIES ('write.data.path'='$location')
+ """)
+ val writeExec = planInsertWriteExec(s"$resolvingCat.$ns.gs_resolving")
+ delegate match {
+ case Some(cls) if cls.getName == IcebergReflection.ClassNames.GCS_FILE_IO =>
+ val support = CometIcebergNativeWrite.getSupportLevel(writeExec)
+ assert(
+ support.isInstanceOf[Compatible],
+ s"expected Compatible via GCSFileIO, got $support")
+ case Some(cls) =>
+ assertUnsupportedContains(writeExec, "gs_resolving", "gs://", cls.getName)
+ case None =>
+ assertUnsupportedContains(writeExec, "gs_resolving", "gs://", "could not resolve")
+ }
}
}
}
@@ -744,6 +882,18 @@ class CometIcebergWriteDetectionSuite extends CometTestBase with CometIcebergTes
spark.sql(s"INSERT INTO $catalog.$ns.$tableName VALUES (1, 'us', 1.0)")
}
+ /**
+ * Plans an INSERT into `qualifiedTable` without executing it and returns its IcebergWriteExec.
+ * `CommandExecutionMode.SKIP` keeps `QueryExecution` from eagerly running the write command, so
+ * a data location no filesystem on this classpath can reach never triggers a write.
+ */
+ private def planInsertWriteExec(qualifiedTable: String): IcebergWriteExec = {
+ val plan =
+ spark.sessionState.sqlParser.parsePlan(s"INSERT INTO $qualifiedTable VALUES (1, 'us', 1.0)")
+ findWriteExecOrFail(
+ spark.sessionState.executePlan(plan, CommandExecutionMode.SKIP).executedPlan)
+ }
+
private def dfWriteExec(tableName: String, options: (String, String)*): IcebergWriteExec =
captureWriteExec(tableName, allowWriteFailure = false) {
val df = spark
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org