fix: keep Iceberg residual pushdown within Spark semantics

docs/source/user-guide/latest/iceberg.md

@@ -125,6 +125,16 @@ results, but are not pushed into iceberg-rust, which binds accessors only for pr
 These residuals provide no native row-group pruning, nor do conjunctions containing them;
 safe partial pruning is tracked in [#5883](https://github.com/apache/datafusion-comet/issues/5883).
 
+Some predicates are left to the filter above the scan because iceberg-rust would return
+different rows than Spark:
+
+- Range comparisons on `FLOAT` and `DOUBLE` columns, and comparisons with `0.0`, `-0.0` or NaN.
+- `NOT` over equality on a nullable column, as in `NOT (id <=> 1)`, unless it is ANDed with an
+  `IS NOT NULL` check on that column, written in the query or inferred by Spark as for `b = false`.
+  `!=` reaches iceberg-rust as a separate comparison and is not affected.
+- Comparisons other than `=` on a column whose type was promoted. This applies to every `BIGINT`
+  and `DOUBLE` column when expiring snapshots removed older schemas.
+
 **Partitioning:**
 
 - Standard partitioning with partition pruning
@@ -801,6 +811,13 @@ Which entries apply depends on which native paths are enabled:
 - **Data files with neither field ids nor a name mapping.** Spark reads every nested field of such
   a file as NULL, and the native scan returns the values stored in the file
   ([#6794](https://github.com/apache/datafusion-comet/issues/6794)).
+- **Data files that store a narrower type than their column.** Files written before a type
+  promotion, such as `INT` to `BIGINT`, keep the older type, and files imported by `add_files`,
+  `migrate` or `snapshot` can store any narrower type. iceberg-rust compares such a file with a
+  filter literal incorrectly, so comparisons other than `=` pushed to the native scan can drop
+  matching rows. Comet avoids this for promotions that the schema history records, but not for
+  imported files or promotions made before Iceberg 0.12
+  ([#7138](https://github.com/apache/datafusion-comet/issues/7138)).
 - **`VERSION AS OF` reads of a column renamed after the snapshot.** The renamed column reads as
   NULL, and two columns that swapped names return each other's values
   ([#6822](https://github.com/apache/datafusion-comet/issues/6822)).

native/core/src/execution/planner.rs

@@ -2572,21 +2572,7 @@ fn parse_file_scan_tasks_from_common(
                     .residual_pool
                     .get(idx as usize)
                     .and_then(iceberg_predicate_to_predicate)
-                    .and_then(|pred| {
-                        // The residual predicate only drives row-group pruning; the post-scan
-                        // filter still enforces correctness. iceberg-rust cannot bind a datum
-                        // whose type has no conversion to the column type, so on a bind failure
-                        // we skip pushdown rather than fail the scan.
-                        // rewrite_not first: iceberg-rust's scan evaluators require negation-normal
-                        // form and reject a bare NOT (e.g. from a `NOT (a = b)` residual).
-                        match pred.rewrite_not().bind(Arc::clone(&schema_ref), true) {
-                            Ok(bound) => Some(bound),
-                            Err(e) => {
-                                log::warn!("Skipping Iceberg predicate pushdown; bind failed: {e}");
-                                None
-                            }
-                        }
-                    })
+                    .and_then(|pred| bind_residual(pred, Arc::clone(&schema_ref)))
             } else {
                 None
             };
@@ -2953,6 +2939,24 @@ fn iceberg_predicate_to_predicate(
     }
 }
 
+/// Binds a residual predicate to the file's schema. The residual prunes reads and filters rows; the
+/// post-scan filter enforces any omitted predicates. iceberg-rust cannot bind a datum whose type
+/// has no conversion to the column type, so on a bind failure we skip pushdown rather than fail
+/// the scan. rewrite_not first: iceberg-rust's scan evaluators require negation-normal form and
+/// reject a bare NOT (e.g. from a `NOT (a = b)` residual).
+fn bind_residual(
+    pred: iceberg::expr::Predicate,
+    schema: Arc<iceberg::spec::Schema>,
+) -> Option<iceberg::expr::BoundPredicate> {
+    match pred.rewrite_not().bind(schema, true) {
+        Ok(bound) => Some(bound),
+        Err(e) => {
+            log::warn!("Skipping Iceberg predicate pushdown; bind failed: {e}");
+            None
+        }
+    }
+}
+
 /// Combines the two children of a logical residual node (And/Or), returning `None` unless both
 /// converted. A missing child is not expected: the Scala serde emits a logical node only when both
 /// children convert, and Rust decodes every node it emits, so both sides always convert for a
@@ -3071,7 +3075,7 @@ mod tests {
 
     use crate::execution::operators::ExecutionError;
     use crate::execution::planner::{
-        convert_spark_types_to_arrow_schema, literal_to_array_ref,
+        bind_residual, convert_spark_types_to_arrow_schema, literal_to_array_ref,
         parse_file_scan_tasks_from_common,
     };
     use crate::execution::shuffle::CometPartitioning;
@@ -3092,6 +3096,26 @@ mod tests {
     };
     use datafusion_comet_spark_expr::EvalMode;
 
+    #[test]
+    fn iceberg_residual_bind_failure_skips_pushdown() {
+        use iceberg::expr::Reference;
+        use iceberg::spec::{Datum, NestedField, PrimitiveType, Schema as IcebergSchema, Type};
+
+        let schema = Arc::new(
+            IcebergSchema::builder()
+                .with_fields(vec![Arc::new(NestedField::optional(
+                    1,
+                    "id",
+                    Type::Primitive(PrimitiveType::Int),
+                ))])
+                .build()
+                .unwrap(),
+        );
+        let id = Reference::new("id");
+        assert!(bind_residual(!id.clone().equal_to(Datum::int(1)), Arc::clone(&schema)).is_some());
+        assert!(bind_residual(id.equal_to(Datum::bool(true)), schema).is_none());
+    }
+
     #[test]
     fn scan_default_rejects_struct_expressions() {
         let planner = PhysicalPlanner::new(Arc::new(SessionContext::new()), 0);

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

@@ -1216,6 +1216,48 @@ object IcebergReflection extends Logging {
     }
   }
 
+  /**
+   * Top-level column names of `schema` whose type differs in some schema of the table's history
+   * (`table.schemas()`), such as an int column promoted to long. Data files written before the
+   * promotion keep the older type, and iceberg-rust compares literals against it incorrectly: the
+   * page index skips every page for range and IN predicates, and the row filter casts the literal
+   * to the older type, so an out-of-range value becomes NULL (apache/iceberg-rust#3122).
+   *
+   * Expiring snapshots with `cleanExpiredMetadata` removes schemas that no retained snapshot
+   * uses, even though files written under them can remain. Iceberg numbers schemas consecutively
+   * from 0, so a gap means the history is incomplete, and every column of a type that a promotion
+   * can produce (long, double, decimal) is returned. The history cannot reveal a file whose type
+   * differs without a recorded schema change, such as one imported by `add_files`, `migrate` or
+   * `snapshot`, or one promoted under metadata from before Iceberg 0.12, which kept only the
+   * current schema. Called at serialization time, so it throws on reflection failure.
+   */
+  def promotedColumns(table: Any, schema: Any): Set[String] = {
+    import scala.jdk.CollectionConverters._
+    val findField = getMethod(schema.getClass, "findField", classOf[Int])
+    def typeOf(field: Any): String = getMethod(field.getClass, "type").invoke(field).toString
+    val schemasById = getMethod(table.getClass, "schemas")
+      .invoke(table)
+      .asInstanceOf[java.util.Map[_, _]]
+      .asScala
+    val history = schemasById.values.toSeq
+    val ids = schemasById.keys.map(_.asInstanceOf[Integer].intValue).toSet
+    val historyComplete = ids.nonEmpty && ids.min == 0 && ids.max == ids.size - 1
+    getMethod(schema.getClass, "columns")
+      .invoke(schema)
+      .asInstanceOf[java.util.List[_]]
+      .asScala
+      .filter { column =>
+        val id = fieldIdOf(column).asInstanceOf[AnyRef]
+        val current = typeOf(column)
+        val promotable =
+          current == "long" || current == "double" || current.startsWith("decimal(")
+        (promotable && !historyComplete) ||
+        history.exists(older => Option(findField.invoke(older, id)).exists(typeOf(_) != current))
+      }
+      .map(fieldName)
+      .toSet
+  }
+
   /**
    * Validates partition column types for compatibility with iceberg-rust.
    *