fix: fall back to Spark for native Iceberg writes to gs:// through HadoopFileIO

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,

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

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