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 ba41a95728..b8b52af549 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 @@ -560,6 +568,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