fix: Iceberg reflection for current() on TableOperations hierarchy (#3895)

Author: karuppayya

.github/workflows/pr_build_linux.yml

@@ -298,6 +298,7 @@ jobs:
               org.apache.spark.sql.comet.ParquetEncryptionITCase
               org.apache.comet.exec.CometNativeReaderSuite
               org.apache.comet.CometIcebergNativeSuite
+              org.apache.comet.iceberg.IcebergReflectionSuite
           - name: "csv"
             value: |
               org.apache.comet.csv.CometCsvNativeReadSuite

.github/workflows/pr_build_macos.yml

@@ -175,6 +175,7 @@ jobs:
               org.apache.spark.sql.comet.ParquetEncryptionITCase
               org.apache.comet.exec.CometNativeReaderSuite
               org.apache.comet.CometIcebergNativeSuite
+              org.apache.comet.iceberg.IcebergReflectionSuite
           - name: "csv"
             value: |
               org.apache.comet.csv.CometCsvNativeReadSuite

dev/diffs/iceberg/1.10.0.diff

@@ -1395,6 +1395,12 @@ index 16fa726032..64e367cf47 100644
              .getOrCreate();
 
      catalog =
+@@ -202,3 +222,4 @@ public class TestSparkReaderDeletes extends DeleteReadTests {
+   protected boolean countDeletes() {
+-    return true;
++    // TODO: Enable once iceberg-rust exposes delete count metrics to Comet
++    return false;
+   }
 diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java
 index baf7fa8f88..665946ad82 100644
 --- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java

dev/diffs/iceberg/1.8.1.diff

@@ -1497,6 +1497,12 @@ index dda49b4946..529992de6b 100644
              .getOrCreate();
 
      catalog =
+@@ -201,3 +221,4 @@ public class TestSparkReaderDeletes extends DeleteReadTests {
+   protected boolean countDeletes() {
+-    return true;
++    // TODO: Enable once iceberg-rust exposes delete count metrics to Comet
++    return false;
+   }
 diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java
 index e5831b76e4..5c45a111d9 100644
 --- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java

dev/diffs/iceberg/1.9.1.diff

@@ -1471,6 +1471,12 @@ index dda49b4946..529992de6b 100644
              .getOrCreate();
 
      catalog =
+@@ -201,3 +221,4 @@ public class TestSparkReaderDeletes extends DeleteReadTests {
+   protected boolean countDeletes() {
+-    return true;
++    // TODO: Enable once iceberg-rust exposes delete count metrics to Comet
++    return false;
+   }
 diff --git a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java b/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java
 index e5831b76e4..5c45a111d9 100644
 --- a/spark/v3.4/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkReaderWithBloomFilter.java

spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala

@@ -228,11 +228,18 @@ object IcebergReflection extends Logging {
           val opsMethod = table.getClass.getDeclaredMethod("operations")
           opsMethod.setAccessible(true)
           val ops = opsMethod.invoke(table)
-          val currentMethod = ops.getClass.getDeclaredMethod("current")
-          currentMethod.setAccessible(true)
-          val metadata = currentMethod.invoke(ops)
-          val formatVersionMethod = metadata.getClass.getMethod("formatVersion")
-          Some(formatVersionMethod.invoke(metadata).asInstanceOf[Int])
+          findMethodInHierarchy(ops.getClass, "current")
+            .flatMap { currentMethod =>
+              val metadata = currentMethod.invoke(ops)
+              val formatVersionMethod = metadata.getClass.getMethod("formatVersion")
+              Some(formatVersionMethod.invoke(metadata).asInstanceOf[Int])
+            }
+            .orElse {
+              logError(
+                "Iceberg reflection failure: Failed to get format version: " +
+                  "current() method not found in operations class hierarchy")
+              None
+            }
         } catch {
           case e: Exception =>
             logError(s"Iceberg reflection failure: Failed to get format version: ${e.getMessage}")
@@ -327,9 +334,12 @@ object IcebergReflection extends Logging {
       operationsMethod.setAccessible(true)
       val operations = operationsMethod.invoke(table)
 
-      val currentMethod = operations.getClass.getDeclaredMethod("current")
-      currentMethod.setAccessible(true)
-      Some(currentMethod.invoke(operations))
+      findMethodInHierarchy(operations.getClass, "current").map(_.invoke(operations)).orElse {
+        logError(
+          "Iceberg reflection failure: Failed to get table metadata: " +
+            "current() method not found in operations class hierarchy")
+        None
+      }
     } catch {
       case e: Exception =>
         logError(s"Iceberg reflection failure: Failed to get table metadata: ${e.getMessage}")

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala

@@ -399,8 +399,11 @@ case class CometScanRule(session: SparkSession)
         // Check if table uses a FileIO implementation compatible with iceberg-rust
 
         val fileIOCompatible = IcebergReflection.getFileIO(metadata.table) match {
+          case Some(fileIO)
+              if fileIO.getClass.getName == "org.apache.iceberg.inmemory.InMemoryFileIO" =>
+            fallbackReasons += "InMemoryFileIO is not supported by Comet's native reader"
+            false
           case Some(_) =>
-            // InMemoryFileIO is now supported with table location fallback for REST catalogs
             true
           case None =>
             fallbackReasons += "Could not check FileIO compatibility"

spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.comet.iceberg
+
+import java.util.Collections
+
+import org.scalatest.funsuite.AnyFunSuite
+
+import org.apache.iceberg.BaseMetastoreTableOperations
+import org.apache.iceberg.BaseTable
+import org.apache.iceberg.Schema
+import org.apache.iceberg.TableMetadata
+import org.apache.iceberg.io.FileIO
+import org.apache.iceberg.types.Types
+
+class IcebergReflectionSuite extends AnyFunSuite {
+
+  /** Mimics HiveTableOperations/GlueTableOperations which inherit current(). */
+  class StubTableOperations extends BaseMetastoreTableOperations {
+    override protected def tableName(): String = "test"
+    override def refresh(): TableMetadata = null
+    override def io(): FileIO = null
+  }
+
+  test("getTableMetadata succeeds when operations class inherits current()") {
+    val ops = new StubTableOperations()
+    val schema = new Schema(Types.NestedField.required(1, "id", Types.IntegerType.get()))
+    val expectedMetadata = TableMetadata.newTableMetadata(
+      schema,
+      org.apache.iceberg.PartitionSpec.unpartitioned(),
+      "file:///tmp/test-table",
+      Collections.emptyMap[String, String]())
+    val metadataField = classOf[BaseMetastoreTableOperations]
+      .getDeclaredField("currentMetadata")
+    metadataField.setAccessible(true)
+    metadataField.set(ops, expectedMetadata)
+    // current() checks shouldRefresh (default true) and calls refresh() instead of
+    // returning currentMetadata. Set to false so current() returns our stubbed metadata.
+    val refreshField = classOf[BaseMetastoreTableOperations]
+      .getDeclaredField("shouldRefresh")
+    refreshField.setAccessible(true)
+    refreshField.set(ops, false)
+
+    val table = new BaseTable(ops, "test-table")
+    val metadata = IcebergReflection.getTableMetadata(table)
+    assert(metadata.isDefined)
+    assert(metadata.get.isInstanceOf[TableMetadata])
+  }
+}