feat: implement PartitionSpec-based non-identity transform detection for delete file safety

Author: Shekharrajak

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

@@ -528,6 +528,44 @@ object IcebergReflection extends Logging {
    *   List of unsupported partition types (empty if all supported). Each entry is (fieldName,
    *   typeStr, reason)
    */
+  /**
+   * Checks whether a partition spec contains any non-identity transforms.
+   *
+   * Non-identity transforms include bucket, truncate, year, month, day, hour. These transforms
+   * mean that Iceberg's partition pruning cannot fully resolve equality filters on the source
+   * column, producing residual expressions that require post-scan filtering.
+   *
+   * @param partitionSpec
+   *   The Iceberg PartitionSpec object
+   * @return
+   *   Some(transformStr) for the first non-identity transform found, or None if all are identity
+   */
+  def findNonIdentityTransform(partitionSpec: Any): Option[String] = {
+    import scala.jdk.CollectionConverters._
+
+    try {
+      val fieldsMethod = partitionSpec.getClass.getMethod("fields")
+      val fields = fieldsMethod.invoke(partitionSpec).asInstanceOf[java.util.List[_]]
+
+      val partitionFieldClass = loadClass(ClassNames.PARTITION_FIELD)
+      val transformMethod = partitionFieldClass.getMethod("transform")
+
+      fields.asScala.foreach { field =>
+        val transform = transformMethod.invoke(field)
+        val transformStr = transform.toString
+        if (transformStr != Transforms.IDENTITY) {
+          return Some(transformStr)
+        }
+      }
+      None
+    } catch {
+      case e: Exception =>
+        logError(
+          s"Iceberg reflection failure: Failed to inspect partition transforms: ${e.getMessage}")
+        None
+    }
+  }
+
   def validatePartitionTypes(partitionSpec: Any, schema: Any): List[(String, String, String)] = {
     import scala.jdk.CollectionConverters._
 

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

@@ -531,12 +531,33 @@ case class CometScanRule(session: SparkSession)
             false
           }
 
-        // TODO: A safety guard for non-identity transforms (truncate, bucket, year, etc.)
-        // with delete files should be re-implemented using PartitionSpec inspection.
-        // The previous residual-based detection was ineffective because Iceberg residuals
-        // use NamedReference terms without transform metadata. Correctness is currently
-        // maintained by Iceberg/Spark's MOR delete processing and CometFilter post-scan
-        // filtering. See: https://github.com/apache/datafusion-comet/issues/XXXX
+        // Safety guard: non-identity transforms with delete files must fall back.
+        // Non-identity transforms (truncate, bucket, year, etc.) produce residual
+        // expressions that require post-scan filtering. When delete files are also
+        // present, the native scan cannot correctly apply both the residual filter
+        // and delete file processing together, so we fall back to Spark.
+        // Detection uses PartitionSpec (table-level) rather than residual expressions,
+        // since Iceberg residuals use NamedReference terms without transform metadata.
+        val transformFunctionsSupported =
+          IcebergReflection.getPartitionSpec(metadata.table) match {
+            case Some(partitionSpec) =>
+              IcebergReflection.findNonIdentityTransform(partitionSpec) match {
+                case Some(transformType) =>
+                  if (!taskValidation.deleteFiles.isEmpty) {
+                    fallbackReasons +=
+                      s"Iceberg transform '$transformType' with delete files present. " +
+                        "Falling back to ensure correct delete operation."
+                    false
+                  } else {
+                    logInfo(
+                      s"Iceberg partition uses transform '$transformType' - " +
+                        "post-scan filtering will apply via CometFilter.")
+                    true
+                  }
+                case None => true
+              }
+            case None => true // Cannot inspect spec, allow through
+          }
 
         // Check for unsupported struct types in delete files
         val deleteFileTypesSupported = {
@@ -619,7 +640,7 @@ case class CometScanRule(session: SparkSession)
 
         if (schemaSupported && fileIOCompatible && formatVersionSupported &&
           taskValidation.allParquet && allSupportedFilesystems && partitionTypesSupported &&
-          complexTypePredicatesSupported &&
+          complexTypePredicatesSupported && transformFunctionsSupported &&
           deleteFileTypesSupported && dppSubqueriesSupported) {
           CometBatchScanExec(
             scanExec.clone().asInstanceOf[BatchScanExec],

spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala

@@ -2860,12 +2860,13 @@ class CometIcebergNativeSuite extends CometTestBase with RESTCatalogHelper {
   // Additional integration tests for non-identity transform residuals
   // =========================================================================
 
-  // Test A: Non-identity transform with delete files - verify correctness.
-  // When delete files and non-identity transforms coexist, query results must
-  // be correct regardless of whether the native scan or Spark handles it.
+  // Test A: Non-identity transform with delete files must fall back to Spark.
+  // When the table has a non-identity partition transform (truncate) AND MOR delete
+  // files are present, the native Iceberg scan must fall back to Spark to ensure
+  // correct delete processing. Detection uses PartitionSpec inspection.
   // Uses truncate(3, name) so the query and deleted row share the same partition
   // (truncate(3, 'alpha') = truncate(3, 'alpine') = 'alp').
-  test("non-identity transform residual - correct results with delete files present") {
+  test("non-identity transform residual - falls back with delete files present") {
     assume(icebergAvailable, "Iceberg not available in classpath")
 
     withTempIcebergDir { warehouseDir =>
@@ -2878,7 +2879,7 @@ class CometIcebergNativeSuite extends CometTestBase with RESTCatalogHelper {
         CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
 
         spark.sql("""
-          CREATE TABLE test_cat.db.truncate_delete_test (
+          CREATE TABLE test_cat.db.truncate_delete_fallback (
             id INT,
             name STRING,
             value DOUBLE
@@ -2892,38 +2893,43 @@ class CometIcebergNativeSuite extends CometTestBase with RESTCatalogHelper {
         """)
 
         spark.sql("""
-          INSERT INTO test_cat.db.truncate_delete_test VALUES
+          INSERT INTO test_cat.db.truncate_delete_fallback VALUES
           (1, 'alpha', 10.0), (2, 'alpine', 20.0), (3, 'bravo', 30.0),
           (4, 'bridge', 40.0), (5, 'charlie', 50.0), (6, 'cherry', 60.0)
         """)
 
         // Delete 'alpine' which shares truncate(3)='alp' partition with 'alpha'.
-        spark.sql("DELETE FROM test_cat.db.truncate_delete_test WHERE name = 'alpine'")
+        spark.sql("DELETE FROM test_cat.db.truncate_delete_fallback WHERE name = 'alpine'")
 
-        // Query for 'alpha' creates a residual on the truncate transform.
-        // The deleted row 'alpine' must not appear, and 'alpha' must be returned.
+        // Query with filter. Because the table has truncate transform AND delete files,
+        // native scan must fall back to Spark.
         val query =
-          "SELECT * FROM test_cat.db.truncate_delete_test WHERE name = 'alpha' ORDER BY id"
+          "SELECT * FROM test_cat.db.truncate_delete_fallback WHERE name = 'alpha' ORDER BY id"
         val (_, cometPlan) = checkSparkAnswer(query)
 
+        // Assert fallback: no CometIcebergNativeScanExec in plan
+        val icebergScans = collectIcebergNativeScans(cometPlan)
+        assert(
+          icebergScans.isEmpty,
+          "Expected fallback to Spark (no CometIcebergNativeScanExec) when " +
+            s"non-identity transform has delete files. Plan:\n$cometPlan")
+
         // Verify correct results: only 'alpha' returned, 'alpine' is deleted
         val result = spark.sql(query).collect()
         assert(result.length == 1, s"Expected 1 row, got ${result.length}")
         assert(result(0).getInt(0) == 1, s"Expected id=1, got ${result(0).getInt(0)}")
         assert(result(0).getString(1) == "alpha")
 
         // Verify 'alpine' is truly gone from broader query
-        val allAlpResult = spark
-          .sql("SELECT * FROM test_cat.db.truncate_delete_test ORDER BY id")
+        val allResult = spark
+          .sql("SELECT * FROM test_cat.db.truncate_delete_fallback ORDER BY id")
           .collect()
+        assert(allResult.length == 5, s"Expected 5 rows after delete, got ${allResult.length}")
         assert(
-          allAlpResult.length == 5,
-          s"Expected 5 rows after delete, got ${allAlpResult.length}")
-        assert(
-          !allAlpResult.exists(_.getString(1) == "alpine"),
+          !allResult.exists(_.getString(1) == "alpine"),
           "Deleted row 'alpine' should not appear in results")
 
-        spark.sql("DROP TABLE test_cat.db.truncate_delete_test")
+        spark.sql("DROP TABLE test_cat.db.truncate_delete_fallback")
       }
     }
   }