@@ -28,12 +28,14 @@ normalizes both operands before native execution, including noncanonical NaN lit
 applies wherever a comparison appears: projections, filters, aggregate arguments and `FILTER`
 clauses, join conditions, sort keys, and generator arguments.
 
-The data filters that a native Parquet scan uses to skip row groups compare values without
-normalizing them, so that statistics pruning still applies, and the filter above the scan
-evaluates them again with Spark's semantics. With
+A native Parquet scan skips row groups with its data filters. So that statistics pruning still
+applies, a `FLOAT` or `DOUBLE` column compared with a constant in a data filter is compared
+without normalizing the column, and the filter above the scan evaluates the comparison again with
+Spark's semantics. Every other comparison in a data filter is normalized. With
 `spark.comet.parquet.rowFilterPushdown.enabled=true` the scan also filters rows with these
-comparisons, so a noncanonical NaN, such as one with the sign bit set, can be filtered differently
-from Spark. Spark's Parquet writer only writes canonical NaNs.
+comparisons, so a noncanonical NaN stored in the file, such as one with the sign bit set, can be
+filtered differently from Spark when compared with a constant. Spark's Parquet writer only writes
+canonical NaNs.
 
 Top-level `IN`, `InSet`, and `NOT IN` membership also normalize dynamic candidates and lists
 containing NaN. When every candidate is a non-NaN literal, Comet keeps DataFusion's static filter
 
@@ -1019,9 +1019,9 @@ impl PhysicalPlanner {
 
     /// Create a data filter that a scan pushes into the Parquet reader. The filter prunes row
     /// groups and pages, and rows too when row-level pushdown is enabled. Pruning only recognizes
-    /// a column compared with a literal, so the filter's comparisons leave float operands as they
-    /// are rather than normalizing them. Spark's Filter above the scan evaluates the filter again
-    /// with Spark's semantics.
+    /// a column compared with a literal, so that shape leaves a float column as it is rather than
+    /// normalizing it, and every other comparison is normalized; see [`FloatOperands::Raw`].
+    /// Spark's Filter above the scan evaluates the filter again with Spark's semantics.
     fn create_data_filter(
         &self,
         spark_expr: &Expr,
 
@@ -443,3 +443,7 @@ harness = false
 [[bench]]
 name = "nested_comparison"
 harness = false
+
+[[bench]]
+name = "float_comparison"
+harness = false
@@ -0,0 +1,114 @@
+// 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.
+
+//! Compare Spark's float comparisons, which normalize each float operand first, with DataFusion's
+//! on the same data, and time `normalize_floats` on its own. Both comparisons return the same
+//! answer except on a NaN with the sign bit set, which is checked before timing; special-value
+//! semantics belong in tests.
+
+use arrow::array::{Array, ArrayRef, Float64Array};
+use arrow::datatypes::{DataType, Field, Schema};
+use arrow::record_batch::RecordBatch;
+use criterion::{criterion_group, criterion_main, BenchmarkId, Criterion};
+use datafusion::common::ScalarValue;
+use datafusion::logical_expr::Operator;
+use datafusion::physical_expr::expressions::{BinaryExpr, Column, Literal};
+use datafusion::physical_expr::PhysicalExpr;
+use datafusion_comet_spark_expr::{normalize_floats, spark_comparison, FloatOperands};
+use std::hint::black_box;
+use std::sync::Arc;
+use std::time::Duration;
+
+const ROWS: usize = 8192;
+
+/// A NaN with the sign bit set, which arithmetic produces on x86-64.
+const NEGATIVE_NAN: f64 = f64::from_bits(0xfff8_0000_0000_0000);
+
+/// Two columns of doubles, with `special` in every tenth row of each if given.
+fn batch(special: Option<f64>) -> RecordBatch {
+    let column = |seed: usize| -> ArrayRef {
+        Arc::new(Float64Array::from_iter_values((0..ROWS).map(
+            |i| match special {
+                Some(value) if i % 10 == seed => value,
+                _ => ((i * 7 + seed) % 1000) as f64 * 1.5,
+            },
+        )))
+    };
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("a", DataType::Float64, true),
+        Field::new("b", DataType::Float64, true),
+    ]));
+    RecordBatch::try_new(schema, vec![column(1), column(3)]).unwrap()
+}
+
+fn criterion_benchmark(c: &mut Criterion) {
+    let mut group = c.benchmark_group("float_comparison");
+    group.sample_size(20);
+    group.warm_up_time(Duration::from_millis(250));
+    group.measurement_time(Duration::from_secs(1));
+    for (data, special) in [
+        ("no_special", None),
+        ("neg_zero_10pct", Some(-0.0)),
+        ("nan_10pct", Some(f64::NAN)),
+        ("sign_bit_nan_10pct", Some(NEGATIVE_NAN)),
+    ] {
+        let batch = batch(special);
+        let schema = batch.schema();
+        let a: Arc<dyn PhysicalExpr> = Arc::new(Column::new("a", 0));
+        let b: Arc<dyn PhysicalExpr> = Arc::new(Column::new("b", 1));
+        let literal: Arc<dyn PhysicalExpr> =
+            Arc::new(Literal::new(ScalarValue::Float64(Some(500.0))));
+        for (shape, right) in [("column_column", b), ("column_literal", literal)] {
+            let datafusion: Arc<dyn PhysicalExpr> = Arc::new(BinaryExpr::new(
+                Arc::clone(&a),
+                Operator::Lt,
+                Arc::clone(&right),
+            ));
+            let spark = spark_comparison(
+                Arc::clone(&a),
+                Operator::Lt,
+                right,
+                &schema,
+                FloatOperands::Normalize,
+            )
+            .unwrap();
+            let evaluate = |expr: &Arc<dyn PhysicalExpr>| {
+                expr.evaluate(&batch)
+                    .unwrap()
+                    .into_array(ROWS)
+                    .unwrap()
+                    .to_data()
+            };
+            if special.map(f64::to_bits) != Some(NEGATIVE_NAN.to_bits()) {
+                assert_eq!(evaluate(&spark), evaluate(&datafusion));
+            }
+            for (engine, expr) in [("spark", &spark), ("datafusion", &datafusion)] {
+                group.bench_function(BenchmarkId::new(format!("{shape}_{engine}"), data), |b| {
+                    b.iter(|| black_box(expr.evaluate(black_box(&batch)).unwrap()))
+                });
+            }
+        }
+        let column = Arc::clone(batch.column(0));
+        group.bench_function(BenchmarkId::new("normalize_floats", data), |b| {
+            b.iter(|| black_box(normalize_floats(black_box(&column))))
+        });
+    }
+    group.finish();
+}
+
+criterion_group!(benches, criterion_benchmark);
+criterion_main!(benches);
@@ -19,8 +19,8 @@
 //! values without materializing normalized columns.
 
 use crate::float_semantics::{
-    has_float_leaf, normalize_comparison_operand, normalize_nested_floats, spark_equality,
-    NormalizeNestedFloats,
+    is_nested_with_float_leaf, normalize_comparison_operand, normalize_nested_floats,
+    spark_equality, NormalizeNestedFloats,
 };
 use arrow::array::{Array, ArrayRef, AsArray, BooleanArray};
 use arrow::buffer::{BooleanBuffer, NullBuffer};
@@ -29,17 +29,13 @@ use arrow::datatypes::{DataType, Schema};
 use arrow::record_batch::RecordBatch;
 use datafusion::common::{internal_err, DFSchema, Result, ScalarValue};
 use datafusion::logical_expr::{ColumnarValue, Operator};
-use datafusion::physical_expr::expressions::{in_list, BinaryExpr, InListExpr};
+use datafusion::physical_expr::expressions::{in_list, BinaryExpr, Column, InListExpr, Literal};
 use datafusion::physical_expr::PhysicalExpr;
 use datafusion::physical_expr_common::physical_expr::is_volatile;
 use std::fmt::{Display, Formatter};
 use std::hash::{Hash, Hasher};
 use std::sync::Arc;
 
-fn needs_spark_equality(dt: &DataType) -> bool {
-    dt.is_nested() && has_float_leaf(dt)
-}
-
 struct Operand {
     array: ArrayRef,
     scalar: bool,
@@ -226,9 +222,13 @@ impl PhysicalExpr for NestedPredicate {
 pub enum FloatOperands {
     /// Normalize them, so that the comparison follows Spark's SQL ordering.
     Normalize,
-    /// Leave them as they are. Only a scan's pushed-down data filters use this: Parquet pruning
-    /// recognizes a column compared with a literal but not a normalized column, and Spark's
-    /// Filter above the scan evaluates each of those filters again.
+    /// Leave a Float32 or Float64 column compared with a literal as it is, and normalize every
+    /// other operand. Only a scan's pushed-down data filters use this: Parquet pruning recognizes
+    /// a column compared with a literal but not a normalized column. With row-level pushdown the
+    /// reader also evaluates the filters on each row, and a row it drops never reaches Spark's
+    /// Filter above the scan, so any other shape, which pruning cannot use anyway, is normalized.
+    /// A computed operand such as `-d` can hold a NaN with the sign bit set, which a raw
+    /// comparison sorts below every other value.
     Raw,
 }
 
@@ -260,7 +260,7 @@ pub fn spark_comparison(
     let (Ok(left_type), Ok(_)) = (left.data_type(schema), right.data_type(schema)) else {
         return Ok(Arc::new(BinaryExpr::new(left, op, right)));
     };
-    if matches!(op, Eq | NotEq) && needs_spark_equality(&left_type) {
+    if matches!(op, Eq | NotEq) && is_nested_with_float_leaf(&left_type) {
         validate_types(&left, std::slice::from_ref(&right), schema)?;
         return Ok(Arc::new(NestedPredicate {
             value: left,
@@ -269,16 +269,33 @@ pub fn spark_comparison(
             membership: false,
         }));
     }
-    let (left, right) = match float_operands {
-        FloatOperands::Normalize => (
+    let raw = float_operands == FloatOperands::Raw;
+    let (left, right) = if raw && is_float_column(&left, schema) && is_literal(&right) {
+        (left, normalize_comparison_operand(right, schema)?)
+    } else if raw && is_literal(&left) && is_float_column(&right, schema) {
+        (normalize_comparison_operand(left, schema)?, right)
+    } else {
+        (
             normalize_comparison_operand(left, schema)?,
             normalize_comparison_operand(right, schema)?,
-        ),
-        FloatOperands::Raw => (left, right),
+        )
     };
     Ok(Arc::new(BinaryExpr::new(left, op, right)))
 }
 
+/// Whether `expr` is a Float32 or Float64 column, the operand that Parquet pruning reads.
+fn is_float_column(expr: &Arc<dyn PhysicalExpr>, schema: &Schema) -> bool {
+    expr.downcast_ref::<Column>().is_some()
+        && matches!(
+            expr.data_type(schema),
+            Ok(DataType::Float32 | DataType::Float64)
+        )
+}
+
+fn is_literal(expr: &Arc<dyn PhysicalExpr>) -> bool {
+    expr.downcast_ref::<Literal>().is_some()
+}
+
 fn validate_types(
     value: &Arc<dyn PhysicalExpr>,
     candidates: &[Arc<dyn PhysicalExpr>],
@@ -301,7 +318,7 @@ pub fn spark_in_list(
     negated: bool,
     schema: &Schema,
 ) -> Result<Arc<dyn PhysicalExpr>> {
-    if !needs_spark_equality(&value.data_type(schema)?) || candidates.is_empty() {
+    if !is_nested_with_float_leaf(&value.data_type(schema)?) || candidates.is_empty() {
         return in_list(value, candidates, &negated, schema);
     }
     validate_types(&value, &candidates, schema)?;
@@ -948,7 +965,8 @@ mod tests {
     }
 
     /// A literal operand is normalized while planning, so the comparison stays `column op
-    /// literal`. With `FloatOperands::Raw`, as for a scan's data filters, nothing is wrapped.
+    /// literal`. With `FloatOperands::Raw`, as for a scan's data filters, only the column of that
+    /// shape is left unwrapped.
     #[test]
     fn float_operand_shapes() -> Result<()> {
         use crate::float_semantics::NormalizeNaNAndZero;
@@ -974,6 +992,8 @@ mod tests {
         let folded = binary.right().downcast_ref::<Literal>().unwrap();
         assert!(matches!(folded.value(), ScalarValue::Float64(Some(v)) if v.to_bits() == 0));
 
+        // With `FloatOperands::Raw`, a float column compared with a literal keeps the column, on
+        // either side, and the literal is still normalized.
         let expr = spark_comparison(
             Arc::new(Column::new("a", 0)),
             Operator::Lt,
@@ -983,10 +1003,39 @@ mod tests {
         )?;
         let binary = expr.downcast_ref::<BinaryExpr>().unwrap();
         assert!(binary.left().downcast_ref::<Column>().is_some());
-        assert!(Arc::ptr_eq(binary.right(), &negative_zero));
+        let folded = binary.right().downcast_ref::<Literal>().unwrap();
+        assert!(matches!(folded.value(), ScalarValue::Float64(Some(v)) if v.to_bits() == 0));
+        let expr = spark_comparison(
+            Arc::clone(&negative_zero),
+            Operator::Lt,
+            Arc::new(Column::new("a", 0)),
+            &schema,
+            FloatOperands::Raw,
+        )?;
+        let binary = expr.downcast_ref::<BinaryExpr>().unwrap();
+        assert!(binary.right().downcast_ref::<Column>().is_some());
 
-        // Other operators are not comparisons and keep their operands.
+        // Any other shape is normalized even with `FloatOperands::Raw`, because a reader with
+        // row-level pushdown drops the rows it rejects.
         let (a, b) = columns();
+        let expr = spark_comparison(
+            Arc::clone(&a),
+            Operator::Lt,
+            Arc::clone(&b),
+            &schema,
+            FloatOperands::Raw,
+        )?;
+        let binary = expr.downcast_ref::<BinaryExpr>().unwrap();
+        assert!(binary
+            .left()
+            .downcast_ref::<NormalizeNaNAndZero>()
+            .is_some());
+        assert!(binary
+            .right()
+            .downcast_ref::<NormalizeNaNAndZero>()
+            .is_some());
+
+        // Other operators are not comparisons and keep their operands.
         let expr = spark_comparison(
             Arc::clone(&a),
             Operator::Plus,
 
@@ -45,6 +45,7 @@ mod compare;
 mod normalize;
 
 pub use compare::{spark_comparator, spark_equality};
+pub(crate) use normalize::is_nested_with_float_leaf;
 pub use normalize::{
     has_float_leaf, normalize_comparison_operand, normalize_floats, normalize_nested_floats,
     NormalizeNaNAndZero, NormalizeNestedFloats,
 
@@ -273,14 +273,9 @@ pub fn has_float_leaf(dt: &DataType) -> bool {
     }
 }
 
-fn is_nested_with_float_leaf(dt: &DataType) -> bool {
-    matches!(
-        dt,
-        DataType::List(_)
-            | DataType::LargeList(_)
-            | DataType::FixedSizeList(_, _)
-            | DataType::Struct(_)
-    ) && has_float_leaf(dt)
+/// Whether `dt` is a list or struct with a Float32 or Float64 field at any depth.
+pub(crate) fn is_nested_with_float_leaf(dt: &DataType) -> bool {
+    dt.is_nested() && has_float_leaf(dt)
 }
 
 /// Recursively rebuilds nested arrays with `-0.0` normalized to `0.0` and NaN canonicalized
 
@@ -22,6 +22,10 @@
 -- `-d` flips the sign bit, so `-d` of the NaN row is a NaN with the sign bit set on every
 -- platform. That is the NaN that arithmetic produces on x86-64, and Arrow's total order sorts
 -- it below -Infinity.
+--
+-- A scan pushes its filters into the Parquet reader, which with row-level pushdown drops rows
+-- itself, before Spark's Filter above the scan sees them.
+-- ConfigMatrix: spark.comet.parquet.rowFilterPushdown.enabled=false,true
 
 statement
 CREATE TABLE float_cmp(id INT, d DOUBLE, f FLOAT) USING parquet
@@ -57,6 +61,20 @@ SELECT id FROM float_cmp WHERE -d > 0.0D
 query
 SELECT id FROM float_cmp WHERE -f >= f
 
+-- A stored column compared with a constant, the filter shape that Parquet pruning reads. The
+-- scan still returns the -0.0 and NaN rows that Spark's ordering keeps.
+query
+SELECT id FROM float_cmp WHERE d >= 0.0D
+
+query
+SELECT id FROM float_cmp WHERE f <= -0.0F
+
+query
+SELECT id FROM float_cmp WHERE d = double('NaN')
+
+query
+SELECT id FROM float_cmp WHERE d > double('Infinity')
+
 -- Aggregate argument
 query
 SELECT sum(if(-d > 0.0D, 1, 0)), sum(if(-d = d, 1, 0)), sum(if(-f < f, 1, 0)) FROM float_cmp
Original file line number	Diff line number	Diff line change
`@@ -273,14 +273,9 @@ pub fn has_float_leaf(dt: &DataType) -> bool {`
`273`	`273`	`}`
`274`	`274`	`}`
`275`	`275`
`276`		`-fn is_nested_with_float_leaf(dt: &DataType) -> bool {`
`277`		`- matches!(`
`278`		`- dt,`
`279`		`- DataType::List(_)`
`280`		`- \| DataType::LargeList(_)`
`281`		`- \| DataType::FixedSizeList(_, _)`
`282`		`- \| DataType::Struct(_)`
`283`		`- ) && has_float_leaf(dt)`
	`276`	+/// Whether `dt` is a list or struct with a Float32 or Float64 field at any depth.
	`277`	`+pub(crate) fn is_nested_with_float_leaf(dt: &DataType) -> bool {`
	`278`	`+ dt.is_nested() && has_float_leaf(dt)`
`284`	`279`	`}`
`285`	`280`
`286`	`281`	/// Recursively rebuilds nested arrays with `-0.0` normalized to `0.0` and NaN canonicalized