refactor: remove dead nonIdentityTransform detection from CometScanRule

Author: Shekharrajak

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

@@ -531,32 +531,12 @@ case class CometScanRule(session: SparkSession)
             false
           }
 
-        // Check for non-identity transform functions in residual expressions.
-        // Non-identity transforms (truncate, bucket, year, month, day, hour) are handled
-        // differently based on whether delete files are present:
-        //   - Read-only (no delete files): native scan proceeds; iceberg-rust skips row-group
-        //     filtering for these predicates and CometFilter applies them post-scan.
-        //   - With delete files: must fall back to Spark. convertIcebergExpression() cannot
-        //     convert BoundTerm with transforms, causing residuals to be silently dropped,
-        //     which breaks delete correctness.
-        // Future: convert transforms to Spark expressions (bucket->pmod, year->Year, etc.)
-        // to enable native execution with delete files too.
-        val transformFunctionsSupported = taskValidation.nonIdentityTransform 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 residual contains transform '$transformType' - " +
-                  "post-scan filtering will apply.")
-              true
-            }
-          case None =>
-            true
-        }
+        // 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
 
         // Check for unsupported struct types in delete files
         val deleteFileTypesSupported = {
@@ -639,7 +619,7 @@ case class CometScanRule(session: SparkSession)
 
         if (schemaSupported && fileIOCompatible && formatVersionSupported &&
           taskValidation.allParquet && allSupportedFilesystems && partitionTypesSupported &&
-          complexTypePredicatesSupported && transformFunctionsSupported &&
+          complexTypePredicatesSupported &&
           deleteFileTypesSupported && dppSubqueriesSupported) {
           CometBatchScanExec(
             scanExec.clone().asInstanceOf[BatchScanExec],
@@ -786,23 +766,19 @@ object CometScanRule extends Logging {
     val contentScanTaskClass = Class.forName(IcebergReflection.ClassNames.CONTENT_SCAN_TASK)
     val contentFileClass = Class.forName(IcebergReflection.ClassNames.CONTENT_FILE)
     val fileScanTaskClass = Class.forName(IcebergReflection.ClassNames.FILE_SCAN_TASK)
-    val unboundPredicateClass = Class.forName(IcebergReflection.ClassNames.UNBOUND_PREDICATE)
     // scalastyle:on classforname
 
     // Cache all method lookups outside the loop
     val fileMethod = contentScanTaskClass.getMethod("file")
     val formatMethod = contentFileClass.getMethod("format")
     val pathMethod = contentFileClass.getMethod("path")
-    val residualMethod = contentScanTaskClass.getMethod("residual")
     val deletesMethod = fileScanTaskClass.getMethod("deletes")
-    val termMethod = unboundPredicateClass.getMethod("term")
 
     val supportedSchemes =
       Set("file", "s3", "s3a", "gs", "gcs", "oss", "abfss", "abfs", "wasbs", "wasb")
 
     var allParquet = true
     val unsupportedSchemes = mutable.Set[String]()
-    var nonIdentityTransform: Option[String] = None
     val deleteFiles = new java.util.ArrayList[Any]()
 
     tasks.asScala.foreach { task =>
@@ -826,28 +802,11 @@ object CometScanRule extends Logging {
         case _: java.net.URISyntaxException => // ignore
       }
 
-      // Residual transform check (short-circuit if already found unsupported)
-      if (nonIdentityTransform.isEmpty && fileScanTaskClass.isInstance(task)) {
-        try {
-          val residual = residualMethod.invoke(task)
-          if (unboundPredicateClass.isInstance(residual)) {
-            val term = termMethod.invoke(residual)
-            try {
-              val transformMethod = term.getClass.getMethod("transform")
-              transformMethod.setAccessible(true)
-              val transform = transformMethod.invoke(term)
-              val transformStr = transform.toString
-              if (transformStr != IcebergReflection.Transforms.IDENTITY) {
-                nonIdentityTransform = Some(transformStr)
-              }
-            } catch {
-              case _: NoSuchMethodException => // No transform = simple reference, OK
-            }
-          }
-        } catch {
-          case _: Exception => // Skip tasks where we can't get residual
-        }
-      }
+      // TODO: Non-identity transform detection via residual inspection was removed
+      // because Iceberg residuals use NamedReference terms (not BoundTransform),
+      // so the transform() method is never available. To properly detect non-identity
+      // transforms, inspect the table's PartitionSpec instead of residual expressions.
+      // See: https://github.com/apache/datafusion-comet/issues/XXXX
 
       // Collect delete files and check their schemes
       if (fileScanTaskClass.isInstance(task)) {
@@ -875,11 +834,7 @@ object CometScanRule extends Logging {
       }
     }
 
-    IcebergTaskValidationResult(
-      allParquet,
-      unsupportedSchemes.toSet,
-      nonIdentityTransform,
-      deleteFiles)
+    IcebergTaskValidationResult(allParquet, unsupportedSchemes.toSet, deleteFiles)
   }
 }
 
@@ -889,5 +844,4 @@ object CometScanRule extends Logging {
 case class IcebergTaskValidationResult(
     allParquet: Boolean,
     unsupportedSchemes: Set[String],
-    nonIdentityTransform: Option[String],
     deleteFiles: java.util.List[_])