diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md
index fd333043013..8d44312f577 100644
--- a/docs/source/user-guide/latest/iceberg.md
+++ b/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
@@ -803,6 +813,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)).
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index 3863637fb37..4b89664cb45 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -2639,21 +2639,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
             };
@@ -3020,6 +3006,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
@@ -3142,7 +3146,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;
@@ -3163,6 +3167,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);
diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index bd30f12496e..798cfeb5f82 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/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.
    *
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
index 4070d785367..a1e4fb3de04 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
@@ -76,7 +76,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
       val OR = "Or"
       val NOT = "Not"
       // The only UnboundTerm shape this serde converts. The others, UnboundTransform and (on
-      // Iceberg versions that have it) UnboundExtract, are declined; see icebergExprToProto.
+      // Iceberg versions that have it) UnboundExtract, are declined; see referencedColumn.
       val NAMED_REFERENCE = "NamedReference"
     }
   }
@@ -679,34 +679,52 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
    *   - An exception means reflection over Iceberg's expression API did something unexpected, so
    *     no such decision was reached. It propagates instead of folding into `None`, which would
    *     make the two indistinguishable; see [[serializeResidual]].
+   *
+   * @param promotedColumns
+   *   columns whose older data files may hold a narrower type; see
+   *   [[IcebergReflection.promotedColumns]]
    */
   def icebergExprToProto(
       icebergExpr: Any,
       output: Seq[Attribute],
-      pageIndexUnsupportedColumns: Set[String]): Option[OperatorOuterClass.IcebergPredicate] = {
+      pageIndexUnsupportedColumns: Set[String],
+      promotedColumns: Set[String]): Option[OperatorOuterClass.IcebergPredicate] =
+    icebergExprToProto(
+      icebergExpr,
+      output.map(attr => attr.name -> attr).toMap,
+      pageIndexUnsupportedColumns,
+      promotedColumns,
+      underNot = false)
+
+  /**
+   * `underNot` is true beneath a NOT. The native side pushes that NOT down to the leaves, so a
+   * leaf whose negation the native reader evaluates differently from Spark is declined.
+   */
+  private def icebergExprToProto(
+      icebergExpr: Any,
+      attributes: Map[String, Attribute],
+      pageIndexUnsupportedColumns: Set[String],
+      promotedColumns: Set[String],
+      underNot: Boolean): Option[OperatorOuterClass.IcebergPredicate] = {
     val exprClass = icebergExpr.getClass
-    val attributeMap = output.map(attr => attr.name -> attr).toMap
+    def call(method: String): Any =
+      IcebergReflection.getMethod(exprClass, method).invoke(icebergExpr)
+    def convertChild(child: Any, underNot: Boolean, childAttributes: Map[String, Attribute]) =
+      icebergExprToProto(
+        child,
+        childAttributes,
+        pageIndexUnsupportedColumns,
+        promotedColumns,
+        underNot)
 
     if (exprClass.getName.endsWith(Constants.ExpressionTypes.UNBOUND_PREDICATE)) {
-      val operation = IcebergReflection.getMethod(exprClass, "op").invoke(icebergExpr).toString
-      val term = IcebergReflection.getMethod(exprClass, "term").invoke(icebergExpr)
-
-      // Only a bare column reference converts. UnboundTransform (and UnboundExtract, on Iceberg
-      // versions that have it) answers ref() with its *source* column, so reading the name off
-      // it would push `bucket(4, id) = 2` as `id = 2` and drop every matching row whose id is
-      // not 2. CometScanRule declines a non-identity transform at planning time, but only when
-      // the residual is a bare predicate rather than one under AND/OR/NOT, so the term shape is
-      // checked here too rather than assumed.
-      if (!term.getClass.getName.endsWith(Constants.ExpressionTypes.NAMED_REFERENCE)) {
-        None
-      } else {
-        val columnName =
-          IcebergReflection.getMethod(term.getClass, "name").invoke(term).asInstanceOf[String]
+      val operation = call("op").toString
 
+      referencedColumn(call("term")).flatMap { columnName =>
         // Iceberg names a nested reference by its dotted path ("struct.field"), which never matches
         // a top-level scan output attribute, so a residual on a nested field drops here. That miss
         // is also why the top-level-only pageIndexUnsupportedColumns gate below stays sound.
-        attributeMap.get(columnName).flatMap { attribute =>
+        attributes.get(columnName).flatMap { attribute =>
           import Constants.Operations._
           import OperatorOuterClass.IcebergPredicateOperator
           // iceberg-rust binds accessors only for primitive fields. Containers cannot be partition
@@ -720,6 +738,24 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
               case IS_NULL => Some(unaryPredicate(columnName, IcebergPredicateOperator.IsNull))
               case IS_NOT_NULL | NOT_NULL =>
                 Some(unaryPredicate(columnName, IcebergPredicateOperator.NotNull))
+              // Files written before the promotion still hold the older type, which iceberg-rust
+              // compares with a literal incorrectly (see IcebergReflection.promotedColumns),
+              // except for equality outside NOT: pruning keeps every row group and page for it,
+              // and a literal the older type cannot hold matches no row there anyway.
+              case op if promotedColumns.contains(columnName) && (op != EQ || underNot) =>
+                None
+              // Iceberg converts Spark's null-safe `id <=> 1` to `id = 1`, which Spark evaluates
+              // to false for a NULL id and the native reader to NULL. Both reject the row, but
+              // for `NOT (id <=> 1)` Spark keeps it while the native reader still drops it. A
+              // NOT_NULL conjunct of an enclosing AND makes the column non-nullable here.
+              case EQ if underNot && attribute.nullable => None
+              // Spark sorts every NaN above all numbers. The native row filter sorts negative NaNs
+              // below them, and iceberg-rust prunes row groups on min/max statistics that can
+              // leave NaN out, or hold it as both when parquet-mr 1.9 or earlier starts a row
+              // group with NaN (apache/iceberg-rust#3494).
+              case LT | LT_EQ | GT | GT_EQ
+                  if attribute.dataType == FloatType || attribute.dataType == DoubleType =>
+                None
               case op if binaryOps.contains(op) =>
                 binaryPredicate(exprClass, icebergExpr, columnName, attribute, binaryOps(op))
               case IN => setPredicate(exprClass, icebergExpr, columnName, attribute)
@@ -730,14 +766,15 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
         }
       }
     } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.AND)) {
-      val left = icebergExprToProto(
-        IcebergReflection.getMethod(exprClass, "left").invoke(icebergExpr),
-        output,
-        pageIndexUnsupportedColumns)
-      val right = icebergExprToProto(
-        IcebergReflection.getMethod(exprClass, "right").invoke(icebergExpr),
-        output,
-        pageIndexUnsupportedColumns)
+      // A row whose column is NULL fails a conjunct `NOT_NULL(column)`, so the whole AND is
+      // false in Spark and the native reader alike. The other conjuncts can then treat the column
+      // as non-nullable. Spark infers that conjunct for the columns a filter compares, so it keeps
+      // `c = false`, which Spark rewrites to `NOT c`, pushable on a nullable boolean.
+      val nonNullAttributes = attributes ++ notNullColumns(icebergExpr).flatMap { name =>
+        attributes.get(name).map(attribute => name -> attribute.withNullability(false))
+      }
+      val left = convertChild(call("left"), underNot, nonNullAttributes)
+      val right = convertChild(call("right"), underNot, nonNullAttributes)
       (left, right) match {
         // Push the residual only if it converts whole. Dropping a conjunct is safe in positive
         // position but strengthens the predicate under a NOT (De Morgan), which would wrongly
@@ -747,22 +784,15 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
         case _ => None
       }
     } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.OR)) {
-      val left = icebergExprToProto(
-        IcebergReflection.getMethod(exprClass, "left").invoke(icebergExpr),
-        output,
-        pageIndexUnsupportedColumns)
-      val right = icebergExprToProto(
-        IcebergReflection.getMethod(exprClass, "right").invoke(icebergExpr),
-        output,
-        pageIndexUnsupportedColumns)
+      val left = convertChild(call("left"), underNot, attributes)
+      val right = convertChild(call("right"), underNot, attributes)
       // Dropping a disjunct would strengthen the predicate and wrongly prune, so require both.
       (left, right) match {
         case (Some(l), Some(r)) => Some(logicalPredicate(isAnd = false, l, r))
         case _ => None
       }
     } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.NOT)) {
-      val child = IcebergReflection.getMethod(exprClass, "child").invoke(icebergExpr)
-      icebergExprToProto(child, output, pageIndexUnsupportedColumns).map(notPredicate)
+      convertChild(call("child"), underNot = true, attributes).map(notPredicate)
     } else {
       // Anything else, such as Expressions.alwaysTrue or alwaysFalse, or a node type a future
       // Iceberg introduces, carries no pushdown this serde can express.
@@ -770,6 +800,39 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
     }
   }
 
+  /**
+   * The column a predicate term names, if the term is a bare column reference. UnboundTransform
+   * (and UnboundExtract, on Iceberg versions that have it) answers ref() with its *source*
+   * column, so reading the name off it would push `bucket(4, id) = 2` as `id = 2` and drop every
+   * matching row whose id is not 2. CometScanRule declines a non-identity transform at planning
+   * time, but only when the residual is a bare predicate rather than one under AND/OR/NOT, so the
+   * term shape is checked here too rather than assumed.
+   */
+  private def referencedColumn(term: Any): Option[String] =
+    if (term.getClass.getName.endsWith(Constants.ExpressionTypes.NAMED_REFERENCE)) {
+      Some(IcebergReflection.getMethod(term.getClass, "name").invoke(term).asInstanceOf[String])
+    } else {
+      None
+    }
+
+  /** Columns that a `NOT_NULL` conjunct of this AND chain checks by bare reference. */
+  private def notNullColumns(icebergExpr: Any): Set[String] = {
+    import Constants.Operations._
+    val exprClass = icebergExpr.getClass
+    def call(method: String): Any =
+      IcebergReflection.getMethod(exprClass, method).invoke(icebergExpr)
+    if (exprClass.getName.endsWith(Constants.ExpressionTypes.AND)) {
+      notNullColumns(call("left")) ++ notNullColumns(call("right"))
+    } else if (exprClass.getName.endsWith(Constants.ExpressionTypes.UNBOUND_PREDICATE)) {
+      call("op").toString match {
+        case IS_NOT_NULL | NOT_NULL => referencedColumn(call("term")).toSet
+        case _ => Set.empty
+      }
+    } else {
+      Set.empty
+    }
+  }
+
   /**
    * Converts a FileScanTask's residual with [[icebergExprToProto]], failing the query if that
    * throws. The converter returns None only for a residual it deliberately leaves to the
@@ -780,9 +843,10 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
   private[operator] def serializeResidual(
       residual: Any,
       output: Seq[Attribute],
-      pageIndexUnsupportedColumns: Set[String]): Option[OperatorOuterClass.IcebergPredicate] =
+      pageIndexUnsupportedColumns: Set[String],
+      promotedColumns: Set[String]): Option[OperatorOuterClass.IcebergPredicate] =
     try {
-      icebergExprToProto(residual, output, pageIndexUnsupportedColumns)
+      icebergExprToProto(residual, output, pageIndexUnsupportedColumns, promotedColumns)
     } catch {
       case e: Exception =>
         val msg = "Iceberg reflection failure: Failed to convert residual expression " +
@@ -876,8 +940,8 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
 
   /**
    * Builds a predicate literal from its Iceberg value and the column's Spark type. Returns None
-   * for null (IS NULL is a separate op) and for Spark types this serde does not map to an
-   * IcebergLiteral, so the predicate degrades to no pushdown rather than an incorrect one.
+   * for null (IS NULL is a separate op), for Spark types that cannot be mapped safely to an
+   * IcebergLiteral, and for float literals the native reader compares differently from Spark.
    * Columns backed by Parquet FIXED_LEN_BYTE_ARRAY (decimal/uuid/fixed) or BYTE_ARRAY (binary)
    * are dropped earlier by the column gate in icebergExprToProto and never reach here.
    */
@@ -895,9 +959,22 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
       case _: ByteType | _: ShortType | _: IntegerType =>
         builder.setIntVal(value.asInstanceOf[java.lang.Number].intValue())
       case _: LongType => builder.setLongVal(value.asInstanceOf[java.lang.Number].longValue())
-      case _: FloatType => builder.setFloatVal(value.asInstanceOf[java.lang.Number].floatValue())
+      // The native reader compares floats by bit pattern, so -0.0 does not equal 0.0 as it does
+      // in Spark. Any other literal except NaN compares the same way in both. A FLOAT literal must
+      // also narrow exactly, or the native reader would compare against a different value.
+      case _: FloatType =>
+        val number = value.asInstanceOf[java.lang.Number]
+        val narrowed = number.floatValue()
+        if (narrowed.isNaN || narrowed == 0.0f || narrowed.toDouble != number.doubleValue()) {
+          return None
+        }
+        builder.setFloatVal(narrowed)
       case _: DoubleType =>
-        builder.setDoubleVal(value.asInstanceOf[java.lang.Number].doubleValue())
+        val double = value.asInstanceOf[java.lang.Number].doubleValue()
+        if (double.isNaN || double == 0.0) {
+          return None
+        }
+        builder.setDoubleVal(double)
       case _: DateType => builder.setDateVal(value.asInstanceOf[java.lang.Number].intValue())
       case _: StringType => builder.setStringVal(value.toString)
       case _: TimestampType =>
@@ -1009,6 +1086,12 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
     // schema so a filter column projected out of the scan output is still recognized.
     val pageIndexUnsupportedColumns =
       IcebergReflection.pageIndexUnsupportedColumns(metadata.tableSchema)
+    // Columns whose type changed over the table's history; residual predicates comparing them
+    // with a literal are dropped, except equality. The scan schema also covers a column that
+    // VERSION AS OF reads after it was dropped from the table.
+    val promotedColumns = Seq(metadata.tableSchema, metadata.scanSchema)
+      .flatMap(IcebergReflection.promotedColumns(metadata.table, _))
+      .toSet
 
     // iceberg-rust reads every leaf of a projected column that the task schema holds. Reading with
     // the pruned task schema skips the nested fields Spark's schema pruning removed, which the
@@ -1178,7 +1261,8 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
                   serializeResidual(
                     residualMethod.invoke(task),
                     output,
-                    pageIndexUnsupportedColumns)
+                    pageIndexUnsupportedColumns,
+                    promotedColumns)
 
                 residualExprOpt.foreach { residual =>
                   val residualIdx = residualToPoolIndex.getOrElseUpdate(
diff --git a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala
index ad9bd99b599..abcddd72575 100644
--- a/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometFuzzIcebergSuite.scala
@@ -215,10 +215,11 @@ class CometFuzzIcebergSuite extends CometFuzzIcebergBase {
         .collect()
         .headOption
         .map(_.get(0))
-        // Skip NaN, whose equality semantics differ, so the sampled literal is well-behaved.
+        // Skip NaN and zero, whose equality semantics differ. The serde leaves floating-point
+        // comparisons with them to the filter above the scan.
         .filterNot {
-          case d: Double => d.isNaN
-          case f: Float => f.isNaN
+          case d: Double => d.isNaN || d == 0.0
+          case f: Float => f.isNaN || f == 0.0f
           case _ => false
         }
 
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index 5dfbd19fcd0..ed47fbe9752 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -239,22 +239,22 @@ class CometIcebergNativeSuite
         spark.sql("""
           CREATE TABLE filter_cat.db.comparison_test (
             id INT,
-            value DOUBLE
+            value BIGINT
           ) USING iceberg
         """)
 
         spark.sql("""
           INSERT INTO filter_cat.db.comparison_test VALUES
-          (1, 10.5), (2, 20.3), (3, 30.7), (4, 15.2), (5, 25.8)
+          (1, 10), (2, 20), (3, 30), (4, 15), (5, 25)
         """)
 
-        checkIcebergNativeScan("SELECT * FROM filter_cat.db.comparison_test WHERE value > 20.0")
+        checkIcebergNativeScan("SELECT * FROM filter_cat.db.comparison_test WHERE value > 20")
 
-        checkIcebergNativeScan("SELECT * FROM filter_cat.db.comparison_test WHERE value >= 20.3")
+        checkIcebergNativeScan("SELECT * FROM filter_cat.db.comparison_test WHERE value >= 20")
 
-        checkIcebergNativeScan("SELECT * FROM filter_cat.db.comparison_test WHERE value < 20.0")
+        checkIcebergNativeScan("SELECT * FROM filter_cat.db.comparison_test WHERE value < 20")
 
-        checkIcebergNativeScan("SELECT * FROM filter_cat.db.comparison_test WHERE value <= 20.3")
+        checkIcebergNativeScan("SELECT * FROM filter_cat.db.comparison_test WHERE value <= 20")
 
         checkIcebergNativeScan("SELECT * FROM filter_cat.db.comparison_test WHERE id != 3")
 
@@ -279,24 +279,24 @@ class CometIcebergNativeSuite
           CREATE TABLE filter_cat.db.logical_test (
             id INT,
             category STRING,
-            value DOUBLE
+            value BIGINT
           ) USING iceberg
         """)
 
         spark.sql("""
           INSERT INTO filter_cat.db.logical_test VALUES
-          (1, 'A', 10.5), (2, 'B', 20.3), (3, 'A', 30.7),
-          (4, 'B', 15.2), (5, 'A', 25.8), (6, 'C', 35.0)
+          (1, 'A', 10), (2, 'B', 20), (3, 'A', 30),
+          (4, 'B', 15), (5, 'A', 25), (6, 'C', 35)
         """)
 
         checkIcebergNativeScan(
-          "SELECT * FROM filter_cat.db.logical_test WHERE category = 'A' AND value > 20.0")
+          "SELECT * FROM filter_cat.db.logical_test WHERE category = 'A' AND value > 20")
 
         checkIcebergNativeScan(
-          "SELECT * FROM filter_cat.db.logical_test WHERE category = 'B' OR value > 30.0")
+          "SELECT * FROM filter_cat.db.logical_test WHERE category = 'B' OR value > 30")
 
         checkIcebergNativeScan("""SELECT * FROM filter_cat.db.logical_test
-             WHERE (category = 'A' AND value > 20.0) OR category = 'C'""")
+             WHERE (category = 'A' AND value > 20) OR category = 'C'""")
 
         spark.sql("DROP TABLE filter_cat.db.logical_test")
       }
@@ -4306,6 +4306,229 @@ class CometIcebergNativeSuite
     }
   }
 
+  /**
+   * Compares `SELECT id FROM table WHERE predicate` with Spark for each predicate, alone and
+   * after `l IS NOT NULL AND`. That array null check is a conjunct the serde never pushes, as in
+   * the queries that found these bugs, so each predicate is also checked in a residual left
+   * unpushed.
+   */
+  private def checkResidualPredicates(table: String, predicates: Seq[String]): Unit =
+    for (predicate <- predicates; prefix <- Seq("", "l IS NOT NULL AND ")) {
+      checkIcebergNativeScan(s"SELECT id FROM $table WHERE $prefix($predicate)")
+    }
+
+  test("floating-point residuals preserve signed zeros in native scans") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+
+    withHadoopCatalog("test_cat") {
+      withSQLConf(
+        CometConf.COMET_ENABLED.key -> "true",
+        CometConf.COMET_EXEC_ENABLED.key -> "true",
+        CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+        val table = "test_cat.db.floating_residual_test"
+        spark.sql(s"CREATE TABLE $table (id INT, f FLOAT, d DOUBLE, l ARRAY<INT>) USING iceberg")
+        try {
+          withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+            // One file containing both zeros keeps file pruning from deciding the comparison.
+            // Distinct ids make a missing signed-zero row observable without comparing floats.
+            spark
+              .sql("""
+                SELECT id, CAST(v AS FLOAT) AS f, CAST(v AS DOUBLE) AS d, l
+                FROM VALUES
+                  (1, '-0.0', array(1)),
+                  (2, '0.0', array(1)),
+                  (3, '1.0', CAST(NULL AS ARRAY<INT>)),
+                  (4, '-1.0', array(1)),
+                  (5, NULL, array(1)),
+                  (6, 'NaN', array(1))
+                AS input(id, v, l)
+              """)
+              .coalesce(1)
+              .write
+              .format("iceberg")
+              .mode("append")
+              .saveAsTable(table)
+            assert(spark.sql(s"SELECT file_path FROM $table.files").collect().length == 1)
+            val rows = spark.sql(s"SELECT id, f, d FROM $table").collect().sortBy(_.getInt(0))
+            assert(java.lang.Float.floatToRawIntBits(rows(0).getFloat(1)) == Int.MinValue)
+            assert(java.lang.Double.doubleToRawLongBits(rows(0).getDouble(2)) == Long.MinValue)
+            assert(java.lang.Float.floatToRawIntBits(rows(1).getFloat(1)) == 0)
+            assert(java.lang.Double.doubleToRawLongBits(rows(1).getDouble(2)) == 0L)
+          }
+
+          for ((column, dataType) <- Seq("f" -> "FLOAT", "d" -> "DOUBLE")) {
+            val zero = s"CAST(0 AS $dataType)"
+            val negativeZero = s"CAST('-0.0' AS $dataType)"
+            val one = s"CAST(1 AS $dataType)"
+            val minusOne = s"CAST(-1 AS $dataType)"
+            val predicates = Seq(
+              s"$column = $zero",
+              s"$column = $negativeZero",
+              s"$column >= $zero",
+              s"$column <= $negativeZero",
+              s"$column IN ($zero, $one)",
+              s"NOT ($column < $zero)",
+              // Nonzero literals push equality and IN; range comparisons stay out.
+              s"$column = $one",
+              s"$column IN ($one, $minusOne)",
+              s"NOT ($column IN ($one, $minusOne) AND id > 1)",
+              s"$column < $one",
+              s"$column > $minusOne",
+              s"NOT ($column = $one)",
+              s"$column = $zero OR id = 3",
+              s"NOT ($column = $zero AND id > 1)",
+              s"$column IS NULL",
+              s"$column IS NOT NULL")
+            checkResidualPredicates(table, predicates)
+          }
+        } finally {
+          spark.sql(s"DROP TABLE $table")
+        }
+      }
+    }
+  }
+
+  test("negated null-safe equality keeps NULL rows in native scans") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+
+    withHadoopCatalog("test_cat") {
+      withSQLConf(
+        CometConf.COMET_ENABLED.key -> "true",
+        CometConf.COMET_EXEC_ENABLED.key -> "true",
+        CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+        val table = "test_cat.db.null_safe_residual_test"
+        spark.sql(s"CREATE TABLE $table (id INT, b BOOLEAN, l ARRAY<INT>) USING iceberg")
+        try {
+          withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+            // One file keeps file pruning from deciding the comparison.
+            spark
+              .sql("""
+                SELECT * FROM VALUES
+                  (NULL, NULL, array(1)), (1, true, array(1)), (2, false, array(1))
+                AS t(id, b, l)
+              """)
+              .coalesce(1)
+              .write
+              .format("iceberg")
+              .mode("append")
+              .saveAsTable(table)
+            assert(spark.sql(s"SELECT file_path FROM $table.files").collect().length == 1)
+          }
+          // Iceberg converts `id <=> 1` to an equality, which is false for a NULL id in Spark but
+          // NULL in the native reader, so only its negation can drop the NULL row. Spark also
+          // writes that negation as IS DISTINCT FROM and, for booleans, IS NOT TRUE. A null
+          // check ANDed with it rejects the NULL row either way.
+          val predicates = Seq(
+            "NOT (id <=> 1)",
+            "NOT (id <=> 1 AND id < 5)",
+            "NOT (id <=> 1) OR id = 2",
+            "id IS DISTINCT FROM 1",
+            "b IS NOT TRUE",
+            "NOT b",
+            "b = false",
+            "id IS NOT NULL AND NOT (id <=> 1)",
+            "NOT (id = 1)",
+            "id != 1")
+          checkResidualPredicates(table, predicates)
+          // `id != 1` reaches Iceberg as a not-equal, and `b = false` as `b IS NOT NULL AND NOT b`,
+          // whose null check covers the negated equality, so both are still pushed. A negated
+          // `<=>` has no such check and stays out.
+          // Planning commonData leaks manifest streams on Iceberg versions before 1.8.0.
+          if (icebergVersionAtLeast(1, 8)) {
+            for ((predicate, pushed) <- Seq(
+                "b = false" -> true,
+                "id != 1" -> true,
+                "id IS NOT NULL AND NOT (id <=> 1)" -> true,
+                "NOT (id <=> 1)" -> false,
+                "b IS NOT TRUE" -> false)) {
+              val (_, cometPlan) = checkSparkAnswer(s"SELECT id FROM $table WHERE $predicate")
+              val scans = collectIcebergNativeScans(cometPlan)
+              assert(scans.length == 1, s"$cometPlan")
+              val common = OperatorOuterClass.IcebergScanCommon.parseFrom(scans.head.commonData)
+              assert(common.getResidualPoolCount == (if (pushed) 1 else 0), s"$predicate")
+            }
+          }
+        } finally {
+          spark.sql(s"DROP TABLE $table")
+        }
+      }
+    }
+  }
+
+  test("literal residuals on promoted columns keep rows from older files") {
+    assume(icebergAvailable, "Iceberg not available in classpath")
+
+    withHadoopCatalog("test_cat") {
+      withSQLConf(
+        CometConf.COMET_ENABLED.key -> "true",
+        CometConf.COMET_EXEC_ENABLED.key -> "true",
+        CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") {
+        val table = "test_cat.db.promoted_residual_test"
+        spark.sql(s"CREATE TABLE $table (id INT, f FLOAT, l ARRAY<INT>) USING iceberg")
+        try {
+          withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+            // One file, so that its FLOAT bounds keep it for f = 0.1D below.
+            spark.sql(s"""INSERT INTO $table SELECT /*+ COALESCE(1) */ * FROM VALUES
+              (1, 0.5, array(1)), (2, 2.5, array(1)), (3, 0.1, array(1)), (6, 0.05, array(1))
+              AS v(id, f, l)""")
+            spark.sql(s"ALTER TABLE $table ALTER COLUMN id TYPE BIGINT")
+            spark.sql(s"ALTER TABLE $table ALTER COLUMN f TYPE DOUBLE")
+            spark.sql(s"INSERT INTO $table VALUES (4, 0.5D, array(1)), (5, 4.5D, array(1))")
+          }
+          // The first file still stores INT and FLOAT. iceberg-rust skips its pages for range and
+          // IN predicates on the promoted types, and casts 3000000000 to a NULL INT.
+          val predicates = Seq(
+            "id > 0",
+            "id < 3",
+            "id IN (1, 2)",
+            "id < 3000000000",
+            "id = 2",
+            "id = 2 OR id = 3000000000",
+            "f IN (0.5D, 4.5D)",
+            "f = 0.5D",
+            "f = 0.1D")
+          checkResidualPredicates(table, predicates)
+
+          // The native reader narrows 0.1D to the FLOAT 0.1, which Spark does not consider equal
+          // to it. The first file's bounds, 0.05 to 2.5, keep it in the scan for that.
+          val narrowed = s"SELECT id FROM $table WHERE f = 0.1D"
+          val df = spark.sql(narrowed)
+          df.collect()
+          val scanned =
+            collectIcebergNativeScans(df.queryExecution.executedPlan).head
+              .metrics("resultDataFiles")
+              .value
+          assert(scanned == 1, s"Expected the first file to be scanned, scanned $scanned files")
+
+          // Expiring the first snapshot with cleanExpiredMetadata (Iceberg 1.9+) drops the original
+          // schema, while the file that snapshot wrote stays in the table.
+          if (icebergVersionAtLeast(1, 9)) {
+            val icebergTable =
+              loadIcebergTable(spark, "test_cat", "db", "promoted_residual_test")
+                .asInstanceOf[org.apache.iceberg.Table]
+            val expire = icebergTable
+              .expireSnapshots()
+              .expireOlderThan(System.currentTimeMillis())
+              .retainLast(1)
+            classOf[org.apache.iceberg.ExpireSnapshots]
+              .getMethod("cleanExpiredMetadata", classOf[Boolean])
+              .invoke(expire, java.lang.Boolean.TRUE)
+            expire.commit()
+            spark.sql(s"REFRESH TABLE $table")
+            val schemas = loadIcebergTable(spark, "test_cat", "db", "promoted_residual_test")
+              .asInstanceOf[org.apache.iceberg.Table]
+              .schemas()
+            assert(schemas.size == 1 && !schemas.containsKey(0))
+            assert(spark.sql(s"SELECT * FROM $table").count() == 6)
+            checkResidualPredicates(table, predicates)
+          }
+        } finally {
+          spark.sql(s"DROP TABLE $table")
+        }
+      }
+    }
+  }
+
   // A residual can arrive as NOT over an AND that mixes a supported conjunct with an unsupported
   // (FIXED_LEN_BYTE_ARRAY) one. Dropping only the unsupported conjunct is safe in positive
   // position (it weakens the pruning predicate), but under a NOT it strengthens it: NOT(id < 200
@@ -7314,10 +7537,8 @@ class CometIcebergNativeSuite
     }
   }
 
-  // Mirrors TestSelect.simpleTypesInFilter. A predicate literal on a FLOAT column is carried as a
-  // double, and iceberg-rust's Datum::to() has no double-to-float arm, so binding the residual to
-  // the schema fails. The residual only drives row-group pruning, so the native scan skips the
-  // failed pushdown and the post-scan CometFilter still enforces correctness.
+  // Mirrors TestSelect.simpleTypesInFilter. Float comparisons stay in the post-scan filter,
+  // including when the predicate literal is a double rather than the column's float type.
   test("float column compared to a double predicate literal") {
     assume(icebergAvailable, "Iceberg not available in classpath")
     withTempIcebergDir { warehouseDir =>
diff --git a/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala b/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala
index 6ff5daf27fe..a297cb640ff 100644
--- a/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/serde/operator/CometIcebergNativeScanSuite.scala
@@ -21,13 +21,16 @@ package org.apache.comet.serde.operator
 
 import java.lang.reflect.InvocationTargetException
 
+import scala.jdk.CollectionConverters._
+
 import org.scalatest.funsuite.AnyFunSuite
 import org.scalatest.matchers.should.Matchers
 
-import org.apache.iceberg.{DeleteFile, FileContent}
-import org.apache.iceberg.expressions.Expressions
+import org.apache.iceberg.{DeleteFile, FileContent, Schema}
+import org.apache.iceberg.expressions.{Expression, Expressions}
+import org.apache.iceberg.types.Types
 import org.apache.spark.sql.catalyst.expressions.AttributeReference
-import org.apache.spark.sql.types.{ArrayType, IntegerType, MapType, StringType, StructType}
+import org.apache.spark.sql.types.{ArrayType, DoubleType, FloatType, IntegerType, MapType, StringType, StructType}
 
 import org.apache.comet.iceberg.IcebergReflection
 
@@ -43,7 +46,11 @@ class CometIcebergNativeScanSuite extends AnyFunSuite with Matchers {
       predicate <- Seq(Expressions.isNull("value"), Expressions.notNull("value"))) {
       withClue(s"$dataType: $predicate") {
         CometIcebergNativeScan
-          .icebergExprToProto(predicate, Seq(AttributeReference("value", dataType)()), Set.empty)
+          .icebergExprToProto(
+            predicate,
+            Seq(AttributeReference("value", dataType)()),
+            Set.empty,
+            Set.empty)
           .isEmpty shouldBe true
       }
     }
@@ -51,6 +58,7 @@ class CometIcebergNativeScanSuite extends AnyFunSuite with Matchers {
       .icebergExprToProto(
         Expressions.notNull("value"),
         Seq(AttributeReference("value", IntegerType)()),
+        Set.empty,
         Set.empty)
       .nonEmpty shouldBe true
   }
@@ -59,7 +67,7 @@ class CometIcebergNativeScanSuite extends AnyFunSuite with Matchers {
   private val intColumn = Seq(AttributeReference("value", IntegerType)())
 
   private def serialize(residual: Any) =
-    CometIcebergNativeScan.serializeResidual(residual, intColumn, Set.empty)
+    CometIcebergNativeScan.serializeResidual(residual, intColumn, Set.empty, Set.empty)
 
   private def causes(t: Throwable): Iterator[Throwable] =
     Iterator.iterate(t)(_.getCause).takeWhile(_ != null)
@@ -113,12 +121,230 @@ class CometIcebergNativeScanSuite extends AnyFunSuite with Matchers {
     }
     // iceberg-rust cannot use these columns in the page index; the post-scan filter has them.
     CometIcebergNativeScan
-      .serializeResidual(Expressions.equal("value", Integer.valueOf(1)), intColumn, Set("value"))
+      .serializeResidual(
+        Expressions.equal("value", Integer.valueOf(1)),
+        intColumn,
+        Set("value"),
+        Set.empty)
       .isEmpty shouldBe true
     // The same predicate converts when nothing declines it, so the cases above are not vacuous.
     serialize(Expressions.equal("value", Integer.valueOf(1))).nonEmpty shouldBe true
   }
 
+  test("floating-point ranges and zero literals are not pushed") {
+    val supported = Expressions.greaterThan("value", Integer.valueOf(5))
+    for (dataType <- Seq(FloatType, DoubleType)) {
+      val output = intColumn :+ AttributeReference("floating", dataType)()
+      def convert(expr: Any) =
+        CometIcebergNativeScan.serializeResidual(expr, output, Set.empty, Set.empty)
+      def declined(predicate: Expression): Unit =
+        withClue(s"$dataType: $predicate: ") {
+          convert(predicate).isEmpty shouldBe true
+          convert(Expressions.and(supported, predicate)).isEmpty shouldBe true
+          convert(Expressions.and(predicate, supported)).isEmpty shouldBe true
+          convert(Expressions.or(supported, predicate)).isEmpty shouldBe true
+          convert(Expressions.not(predicate)).isEmpty shouldBe true
+          convert(Expressions.not(Expressions.and(supported, predicate))).isEmpty shouldBe true
+        }
+
+      // The native reader orders floats differently from Spark: it separates the signed zeros
+      // and sorts negative NaNs below every number. Range comparisons stay out for every literal.
+      for (value <- Seq(-0.0, 0.0, 1.0, Double.NegativeInfinity, Double.PositiveInfinity);
+        predicate <- Seq(
+          Expressions.lessThan("floating", Double.box(value)),
+          Expressions.lessThanOrEqual("floating", Double.box(value)),
+          Expressions.greaterThan("floating", Double.box(value)),
+          Expressions.greaterThanOrEqual("floating", Double.box(value)))) {
+        declined(predicate)
+      }
+      // A zero literal is declined, including as any element of IN.
+      for (value <- Seq(-0.0, 0.0);
+        predicate <- Seq(
+          Expressions.equal("floating", Double.box(value)),
+          Expressions.notEqual("floating", Double.box(value)),
+          Expressions.in("floating", Double.box(value), Double.box(2.0)),
+          Expressions.in("floating", Double.box(2.0), Double.box(value)))) {
+        declined(predicate)
+      }
+      // Other literals compare the same way in both, so != and IN are pushed in any position.
+      for (value <- Seq(1.0, -1.0, Double.NegativeInfinity, Double.PositiveInfinity);
+        predicate <- Seq(
+          Expressions.notEqual("floating", Double.box(value)),
+          Expressions.in("floating", Double.box(value), Double.box(2.0)))) {
+        withClue(s"$dataType: $predicate: ") {
+          convert(predicate).nonEmpty shouldBe true
+          convert(Expressions.or(supported, predicate)).nonEmpty shouldBe true
+          convert(Expressions.not(predicate)).nonEmpty shouldBe true
+        }
+      }
+      // Equality under NOT still follows the nullable-column rule tested below.
+      val eq = Expressions.equal("floating", Double.box(1.0))
+      convert(eq).nonEmpty shouldBe true
+      convert(Expressions.not(eq)).isEmpty shouldBe true
+      for (predicate <- Seq(Expressions.isNull("floating"), Expressions.notNull("floating"))) {
+        convert(predicate).nonEmpty shouldBe true
+        convert(Expressions.not(predicate)).nonEmpty shouldBe true
+      }
+    }
+  }
+
+  test("NaN checks are not pushed") {
+    // Spark's isnan(f), f = NaN and f != NaN reach the serde as IS_NAN and NOT_NAN.
+    val supported = Expressions.greaterThan("value", Integer.valueOf(5))
+    for (dataType <- Seq(FloatType, DoubleType);
+      predicate <- Seq(Expressions.isNaN("floating"), Expressions.notNaN("floating"))) {
+      val output = intColumn :+ AttributeReference("floating", dataType)()
+      def convert(expr: Any) =
+        CometIcebergNativeScan.serializeResidual(expr, output, Set.empty, Set.empty)
+      withClue(s"$dataType: $predicate: ") {
+        convert(predicate).isEmpty shouldBe true
+        convert(Expressions.not(predicate)).isEmpty shouldBe true
+        convert(Expressions.and(supported, predicate)).isEmpty shouldBe true
+      }
+    }
+  }
+
+  test("a FLOAT literal is pushed only when it narrows exactly to a nonzero float") {
+    val output = Seq(AttributeReference("floating", FloatType)())
+    def convert(value: AnyRef) =
+      CometIcebergNativeScan.serializeResidual(
+        Expressions.equal("floating", value),
+        output,
+        Set.empty,
+        Set.empty)
+    convert(Float.box(0.1f)).nonEmpty shouldBe true
+    convert(Double.box(0.5)).nonEmpty shouldBe true
+    // 0.1 is not a float, and 1e-50 narrows to 0.0f.
+    convert(Double.box(0.1)).isEmpty shouldBe true
+    convert(Double.box(1e-50)).isEmpty shouldBe true
+  }
+
+  test("promoted columns push only null checks and equality outside NOT") {
+    // Required, so that NOT keeps equality for any reason other than the promotion.
+    val required = Seq(AttributeReference("value", IntegerType, nullable = false)())
+    def unpromoted(expr: Any) =
+      CometIcebergNativeScan.serializeResidual(expr, required, Set.empty, Set.empty)
+    def promoted(expr: Any) =
+      CometIcebergNativeScan.serializeResidual(expr, required, Set.empty, Set("value"))
+    val equal = Expressions.equal("value", Integer.valueOf(1))
+    for (predicate <- Seq(equal, Expressions.isNull("value"), Expressions.notNull("value"))) {
+      withClue(s"$predicate: ") {
+        promoted(predicate).nonEmpty shouldBe true
+        promoted(predicate) shouldBe unpromoted(predicate)
+      }
+    }
+    for (predicate <- Seq(
+        Expressions.not(equal),
+        Expressions.notEqual("value", Integer.valueOf(1)),
+        Expressions.lessThan("value", Integer.valueOf(1)),
+        Expressions.greaterThanOrEqual("value", Integer.valueOf(1)),
+        Expressions.in("value", Integer.valueOf(1), Integer.valueOf(2)))) {
+      withClue(s"$predicate: ") {
+        unpromoted(predicate).nonEmpty shouldBe true
+        promoted(predicate).isEmpty shouldBe true
+      }
+    }
+  }
+
+  /** Stands in for an Iceberg `Table`, of which promotedColumns reads only `schemas()`. */
+  class SchemaHistory(history: (Int, Schema)*) {
+    def schemas(): java.util.Map[Integer, Schema] =
+      history.map { case (id, schema) => Integer.valueOf(id) -> schema }.toMap.asJava
+  }
+
+  test("promoted columns are found by field id across the schema history") {
+    val original = new Schema(
+      Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+      Types.NestedField.optional(2, "f", Types.FloatType.get()),
+      Types.NestedField.optional(3, "s", Types.StringType.get()),
+      Types.NestedField.optional(4, "dropped", Types.IntegerType.get()))
+    val current = new Schema(
+      Types.NestedField.optional(1, "id", Types.LongType.get()),
+      Types.NestedField.optional(2, "f", Types.DoubleType.get()),
+      Types.NestedField.optional(3, "renamed", Types.StringType.get()),
+      Types.NestedField.optional(5, "added", Types.IntegerType.get()))
+    val table = new SchemaHistory(0 -> original, 1 -> current)
+    IcebergReflection.promotedColumns(table, current) shouldBe Set("id", "f")
+    IcebergReflection.promotedColumns(table, original) shouldBe Set("id", "f")
+    IcebergReflection.promotedColumns(new SchemaHistory(0 -> current), current) shouldBe empty
+  }
+
+  test("an incomplete schema history treats every promotable column as promoted") {
+    // Without schema 0, as after expiring snapshots with cleanExpiredMetadata, a long, double or
+    // decimal column may still hold files written as a narrower type.
+    val schema = new Schema(
+      Types.NestedField.optional(1, "id", Types.LongType.get()),
+      Types.NestedField.optional(2, "f", Types.DoubleType.get()),
+      Types.NestedField.optional(3, "amount", Types.DecimalType.of(10, 2)),
+      Types.NestedField.optional(4, "i", Types.IntegerType.get()),
+      Types.NestedField.optional(5, "s", Types.StringType.get()))
+    IcebergReflection.promotedColumns(new SchemaHistory(1 -> schema), schema) shouldBe
+      Set("id", "f", "amount")
+    IcebergReflection.promotedColumns(
+      new SchemaHistory(0 -> schema, 2 -> schema),
+      schema) shouldBe
+      Set("id", "f", "amount")
+    IcebergReflection.promotedColumns(
+      new SchemaHistory(0 -> schema, 1 -> schema),
+      schema) shouldBe
+      empty
+  }
+
+  test("NOT declines equality on a nullable column, which may be a null-safe comparison") {
+    // Iceberg converts Spark's `value <=> 1` to `value = 1`. Negated, Spark keeps NULL rows that
+    // the native reader's not-equal drops.
+    val eq = Expressions.equal("value", Integer.valueOf(1))
+    val lt = Expressions.lessThan("value", Integer.valueOf(5))
+    serialize(eq).nonEmpty shouldBe true
+    for (residual <- Seq(
+        Expressions.not(eq),
+        Expressions.not(Expressions.and(eq, lt)),
+        Expressions.not(Expressions.or(lt, eq)))) {
+      withClue(s"$residual: ") {
+        serialize(residual).isEmpty shouldBe true
+      }
+    }
+    serialize(Expressions.and(lt, Expressions.not(eq))).isEmpty shouldBe true
+    // Spark's other comparisons are NULL for a NULL input, as in the native reader.
+    serialize(Expressions.not(lt)).nonEmpty shouldBe true
+    serialize(Expressions.notEqual("value", Integer.valueOf(1))).nonEmpty shouldBe true
+    // A required column holds no NULLs.
+    val required = Seq(AttributeReference("value", IntegerType, nullable = false)())
+    CometIcebergNativeScan
+      .serializeResidual(Expressions.not(eq), required, Set.empty, Set.empty)
+      .nonEmpty shouldBe true
+  }
+
+  test("a NOT_NULL conjunct lets equality on the same column convert under NOT") {
+    // A NULL value fails the AND in Spark and in the native reader alike, so the negated
+    // equality only sees non-NULL values. Spark writes `b = false` as `b IS NOT NULL AND NOT b`.
+    val eq = Expressions.equal("value", Integer.valueOf(1))
+    val lt = Expressions.lessThan("value", Integer.valueOf(5))
+    val notNull = Expressions.notNull("value")
+    for (residual <- Seq(
+        Expressions.and(notNull, Expressions.not(eq)),
+        Expressions.and(Expressions.and(lt, notNull), Expressions.not(eq)),
+        Expressions.and(notNull, Expressions.or(lt, Expressions.not(eq))),
+        Expressions.not(Expressions.and(notNull, eq)))) {
+      withClue(s"$residual: ") {
+        serialize(residual).nonEmpty shouldBe true
+      }
+    }
+    // The check must be a conjunct of an AND that encloses the equality.
+    val twoColumns = intColumn :+ AttributeReference("other", IntegerType)()
+    for (residual <- Seq(
+        Expressions.or(notNull, Expressions.not(eq)),
+        Expressions.and(Expressions.or(notNull, lt), Expressions.not(eq)),
+        Expressions.and(Expressions.not(notNull), Expressions.not(eq)),
+        Expressions.and(Expressions.notNull("other"), Expressions.not(eq)))) {
+      withClue(s"$residual: ") {
+        CometIcebergNativeScan
+          .serializeResidual(residual, twoColumns, Set.empty, Set.empty)
+          .isEmpty shouldBe true
+      }
+    }
+  }
+
   test("a transform residual is declined rather than read as its source column") {
     // UnboundTransform answers ref() with the source column, so converting the term like a bare
     // reference would push bucket(4, value) = 1 as value = 1 and drop matching rows.