@@ -408,6 +408,7 @@ For full documentation on the test file format, including directives like `Confi
 
 - **Cover both column references and literals.** Comet often uses different code paths for each. The Comet SQL Tests suite automatically disables constant folding, so all-literal queries are evaluated natively.
 - **Include edge cases** such as `NULL`, empty strings, boundary values, `NaN`, and multibyte UTF-8 characters.
+- **For `FLOAT` and `DOUBLE` input, include `-0.0` and a NaN with the sign bit set.** [Floating-Point Semantics](floating_point.md) explains which Spark rule an expression follows and how to test it.
 - **Keep one file per expression** to make failures easy to locate.
 - **Pin the mechanism when it depends on the input.** If your serde routes some input types natively and others through the JVM codegen dispatcher, say so with `expect_native(...)` and `expect_dispatch(...)`. A plain `query` cannot tell the two apart, so without these a later change that swaps one for the other passes silently. See [Comet SQL Tests](sql-file-tests.md).
 
 
@@ -47,6 +47,7 @@ Development Guide <development>
 Comet Plugin Overview <plugin_overview>
 Arrow FFI <ffi>
 Timezone Handling <timezones>
+Floating-Point Semantics <floating_point>
 JVM Shuffle <jvm_shuffle>
 Native Shuffle <native_shuffle>
 Memory Management <memory_management>
 
@@ -23,20 +23,43 @@ Spark normalizes NaN and zero for floating point numbers for several cases. See
 However, one exception is comparison. Spark does not normalize NaN and zero when comparing values
 because they are handled well in Spark (e.g., `SQLOrderingUtil.compareFloats`). But the comparison
 functions of arrow-rs used by DataFusion do not normalize NaN and zero (e.g., [arrow::compute::kernels::cmp::eq](https://docs.rs/arrow/latest/arrow/compute/kernels/cmp/fn.eq.html#)).
-For top-level `FLOAT` and `DOUBLE` comparisons, Comet normalizes both operands before native
-execution, including noncanonical NaN literals. 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 and pruning path, enumerating both signed-zero
-forms when a list contains zero.
+For `FLOAT` and `DOUBLE` comparisons (`=`, `<>`, `<=>`, `<`, `<=`, `>` and `>=`), Comet
+normalizes both operands before native execution, including noncanonical NaN literals. This
+applies wherever a comparison appears: projections, filters, aggregate arguments and `FILTER`
+clauses, join conditions, sort keys, and generator arguments.
+
+A native Parquet scan prunes row groups and pages with its data filters. So that this pruning
+still applies, a `FLOAT` or `DOUBLE` column compared with a constant other than NaN 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.
+Bloom filters store `-0.0` and `0.0` separately, so `=` against either zero checks them for both.
+With `spark.comet.parquet.rowFilterPushdown.enabled=true` the scan also drops the rows its data
+filters reject, so every comparison in a data filter is normalized, and a `FLOAT` or `DOUBLE`
+comparison in a data filter does not prune row groups or pages
+([#6702](https://github.com/apache/datafusion-comet/issues/6702)).
+
+Spark's own Parquet reader keeps `-0.0` and `0.0` apart when it prunes row groups with dictionaries
+and bloom filters, so it can skip a row group holding rows that its filter matches. For example,
+`WHERE d = -0.0D` can skip a row group that holds `0.0` but not `-0.0`, and so can `<=>` or `IN`
+against `-0.0D`. `WHERE d >= 0.0D` can skip a dictionary-encoded row group whose largest value is
+`-0.0`, and `WHERE d <= -0.0D` one whose smallest value is `0.0`. Comet reads those row groups, so
+it can return rows that Spark does not. Comet's result is the one that Spark's comparison semantics
+call for, and Spark returns the same rows with `spark.sql.parquet.filterPushdown=false`.
+
+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
+and pruning path, enumerating both signed-zero forms when a list contains zero.
 
 This scalar membership handling does not yet recurse into floating-point leaves nested in arrays
 or structs; see [#6019](https://github.com/apache/datafusion-comet/issues/6019).
 
-## Nested equality and membership
+## Nested comparisons and membership
 
 For arrays and structs containing `FLOAT` or `DOUBLE`, native `=`, `<>`, `IN`, and `NOT IN`
 compare signed zeros as equal and all NaN representations as equal, matching Spark. This also
-covers single-candidate membership that Spark rewrites into equality.
+covers single-candidate membership that Spark rewrites into equality. `<=>`, `<`, `<=`, `>`, and
+`>=` normalize the floating-point values inside both operands first, so they follow Spark's order
+too, in which NaN sorts above every other value.
 
 Equality and dynamic membership compare nested elements directly and stop at the first mismatch.
 Constant membership sets use normalized comparison values for static lookup. These operations
 
@@ -28,7 +28,8 @@ decode step and lazily materializes projected columns for surviving rows. This c
 CPU and memory when the filter is highly selective on a small subset of columns. It is disabled by default
 because it can hurt when the filter is not selective or when most columns must be read anyway. Row-group,
 page-index, and bloom-filter pruning happen regardless of this flag whenever Spark's
-`spark.sql.parquet.filterPushdown` is on.
+`spark.sql.parquet.filterPushdown` is on, except for filters that compare `FLOAT` or `DOUBLE` values,
+which do not prune while this flag is on (see [Floating-point Number Comparison](../compatibility/floating-point.md)).
 
 ### Parquet Native Scans
 
 
@@ -144,10 +144,10 @@ use datafusion_comet_proto::{
 use datafusion_comet_spark_expr::{
     create_case_when, create_if_expr, jvm_udf::JvmScalarUdfExpr, spark_in_list, ApproxPercentile,
     ArrayInsert, Avg, AvgDecimal, Cast, CheckOverflow, Correlation, Covariance, CreateNamedStruct,
-    DecimalRescaleCheckOverflow, GetArrayStructFields, GetStructField, HllPlusPlus, HllSketchAgg,
-    HllUnionAgg, IfExpr, ListExtract, MaxMinBy, Mode, NormalizeNaNAndZero, NormalizeNestedFloats,
-    Regr, RegrType, SparkCastOptions, SparkMinMax, Stddev, SumDecimal, ToJson, UnboundColumn,
-    Variance, WideDecimalBinaryExpr, WideDecimalOp,
+    DecimalRescaleCheckOverflow, FloatOperands, GetArrayStructFields, GetStructField, HllPlusPlus,
+    HllSketchAgg, HllUnionAgg, IfExpr, ListExtract, MaxMinBy, Mode, NormalizeNaNAndZero,
+    NormalizeNestedFloats, Regr, RegrType, SparkCastOptions, SparkMinMax, Stddev, SumDecimal,
+    ToJson, UnboundColumn, Variance, WideDecimalBinaryExpr, WideDecimalOp,
 };
 use itertools::Itertools;
 use jni::objects::{Global, JObject};
@@ -325,6 +325,7 @@ pub struct BinaryExprOptions {
 pub const TEST_EXEC_CONTEXT_ID: i64 = -1;
 
 /// The query planner for converting Spark query plans to DataFusion query plans.
+#[derive(Clone)]
 pub struct PhysicalPlanner {
     // The execution context id of this planner.
     exec_context_id: i64,
@@ -346,6 +347,9 @@ pub struct PhysicalPlanner {
     /// Task-owned destination for remote shuffle blocks, registered on the driving Spark task
     /// thread before native planning. Only explicit RSS destinations may use it.
     shuffle_partition_pusher: Option<Arc<dyn ShufflePartitionPusher>>,
+    /// How comparisons treat floating-point operands. `Raw` only while planning a scan's data
+    /// filters; see [`Self::create_data_filter`].
+    float_operands: FloatOperands,
 }
 
 impl Default for PhysicalPlanner {
@@ -365,6 +369,7 @@ impl PhysicalPlanner {
             task_context: None,
             class_loader: None,
             shuffle_partition_pusher: None,
+            float_operands: FloatOperands::Normalize,
         }
     }
 
@@ -477,6 +482,11 @@ impl PhysicalPlanner {
         self
     }
 
+    /// How comparisons treat floating-point operands.
+    pub fn float_operands(&self) -> FloatOperands {
+        self.float_operands
+    }
+
     /// Return session context of this planner.
     pub fn session_ctx(&self) -> &Arc<SessionContext> {
         &self.session_ctx
@@ -1129,6 +1139,45 @@ impl PhysicalPlanner {
         }
     }
 
+    /// Create a data filter that a scan pushes into the Parquet reader, with float operands
+    /// treated as [`Self::data_filter_float_operands`] decides.
+    fn create_data_filter(
+        &self,
+        spark_expr: &Expr,
+        input_schema: SchemaRef,
+        float_operands: FloatOperands,
+    ) -> Result<Arc<dyn PhysicalExpr>, ExecutionError> {
+        let planner = Self {
+            float_operands,
+            ..self.clone()
+        };
+        planner.create_expr(spark_expr, input_schema)
+    }
+
+    /// How a scan's data filters treat float operands. The Parquet reader prunes row groups and
+    /// pages with them, and with row-level pushdown (`pushdown_filters`) it also drops the rows
+    /// they reject, which Spark's Filter above the scan then never sees.
+    ///
+    /// Without row-level pushdown the filters only prune, and pruning only recognizes a column
+    /// compared with a literal, so that shape keeps the raw column ([`FloatOperands::Raw`]). With
+    /// it, every operand is normalized and float comparisons give up pruning: a raw column would
+    /// drop a stored NaN that Spark matches, such as one with the sign bit set, which Arrow orders
+    /// below every other value.
+    fn data_filter_float_operands(&self) -> FloatOperands {
+        if self
+            .session_ctx
+            .copied_config()
+            .options()
+            .execution
+            .parquet
+            .pushdown_filters
+        {
+            FloatOperands::Normalize
+        } else {
+            FloatOperands::Raw
+        }
+    }
+
     /// Normalize floating-point comparison keys without changing output values: a `FLOAT` or
     /// `DOUBLE` key, and an array or struct key with a float at any depth, whose order Arrow
     /// otherwise takes from the raw bits. Sort, Window, and WindowGroupLimit must use identical
@@ -1910,10 +1959,17 @@ impl PhysicalPlanner {
                                 .cloned()
                                 .collect::<Vec<FieldRef>>(),
                         ));
+                        let float_operands = self.data_filter_float_operands();
                         common
                             .data_filters
                             .iter()
-                            .map(|expr| self.create_expr(expr, Arc::clone(&filter_schema)))
+                            .map(|expr| {
+                                self.create_data_filter(
+                                    expr,
+                                    Arc::clone(&filter_schema),
+                                    float_operands,
+                                )
+                            })
                             .collect()
                     };
 
@@ -5826,6 +5882,121 @@ mod tests {
         assert_eq!(0, filter_exec.additional_native_plans.len());
     }
 
+    /// Comparisons normalize float operands, except in the data filters that a scan pushes into
+    /// the Parquet reader, where pruning has to see the column itself.
+    #[test]
+    fn scan_data_filters_compare_float_columns_directly() {
+        use datafusion::physical_expr::expressions::BinaryExpr;
+        use datafusion_comet_spark_expr::NormalizeNaNAndZero;
+        let double = spark_expression::DataType {
+            type_id: 6,
+            type_info: None,
+        };
+        let operand = |expr_struct| Expr {
+            expr_struct: Some(expr_struct),
+            query_context: None,
+            expr_id: None,
+        };
+        let expr = operand(Gt(Box::new(spark_expression::BinaryExpr {
+            left: Some(Box::new(operand(Bound(spark_expression::BoundReference {
+                index: 0,
+                datatype: Some(double.clone()),
+            })))),
+            right: Some(Box::new(operand(Literal(spark_expression::Literal {
+                value: Some(literal::Value::DoubleVal(500.0)),
+                datatype: Some(double),
+                is_null: false,
+            })))),
+        })));
+        let schema = Arc::new(Schema::new(vec![Field::new("d", DataType::Float64, true)]));
+        let left_operand = |expr: Arc<dyn PhysicalExpr>| {
+            let comparison = expr.downcast_ref::<BinaryExpr>().expect("a comparison");
+            Arc::clone(comparison.left())
+        };
+        let planner = PhysicalPlanner::default();
+        let comparison = planner.create_expr(&expr, Arc::clone(&schema)).unwrap();
+        assert!(left_operand(comparison)
+            .downcast_ref::<NormalizeNaNAndZero>()
+            .is_some());
+        let data_filter = planner
+            .create_data_filter(&expr, schema, planner.data_filter_float_operands())
+            .unwrap();
+        assert!(left_operand(data_filter).downcast_ref::<Column>().is_some());
+    }
+
+    /// With row-level pushdown the Parquet reader drops the rows a data filter rejects, so the
+    /// filter normalizes the column as well as the literal. A raw column would drop a stored NaN
+    /// whose bits differ from the normalized literal, and a stored NaN with the sign bit set
+    /// under any ordering comparison, both of which Spark matches.
+    #[test]
+    fn scan_data_filters_normalize_float_columns_with_row_level_pushdown() {
+        use arrow::array::{AsArray, BooleanArray};
+        use datafusion::physical_expr::expressions::BinaryExpr;
+        use datafusion_comet_spark_expr::{FloatOperands, NormalizeNaNAndZero};
+        let double = spark_expression::DataType {
+            type_id: 6,
+            type_info: None,
+        };
+        let operand = |expr_struct| Expr {
+            expr_struct: Some(expr_struct),
+            query_context: None,
+            expr_id: None,
+        };
+        let column_and = |value: f64| {
+            Box::new(spark_expression::BinaryExpr {
+                left: Some(Box::new(operand(Bound(spark_expression::BoundReference {
+                    index: 0,
+                    datatype: Some(double.clone()),
+                })))),
+                right: Some(Box::new(operand(Literal(spark_expression::Literal {
+                    value: Some(literal::Value::DoubleVal(value)),
+                    datatype: Some(double.clone()),
+                    is_null: false,
+                })))),
+            })
+        };
+        // Spark folds `-double('NaN')` into a literal with the sign bit set.
+        let negative_nan = f64::from_bits(0xfff8_0000_0000_0000);
+        let schema = Arc::new(Schema::new(vec![Field::new("d", DataType::Float64, true)]));
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![Arc::new(Float64Array::from(vec![
+                negative_nan,
+                f64::NAN,
+                1.0,
+            ]))],
+        )
+        .unwrap();
+        let config =
+            SessionConfig::new().set_bool("datafusion.execution.parquet.pushdown_filters", true);
+        let planner = PhysicalPlanner::new(Arc::new(SessionContext::new_with_config(config)), 0);
+        assert_eq!(
+            planner.data_filter_float_operands(),
+            FloatOperands::Normalize
+        );
+        for (expr, expected) in [
+            (operand(Eq(column_and(negative_nan))), [true, true, false]),
+            (operand(Gt(column_and(0.0))), [true, true, true]),
+        ] {
+            let data_filter = planner
+                .create_data_filter(
+                    &expr,
+                    Arc::clone(&schema),
+                    planner.data_filter_float_operands(),
+                )
+                .unwrap();
+            let comparison = data_filter
+                .downcast_ref::<BinaryExpr>()
+                .expect("a comparison");
+            assert!(comparison
+                .left()
+                .downcast_ref::<NormalizeNaNAndZero>()
+                .is_some());
+            let matched = data_filter.evaluate(&batch).unwrap().into_array(3).unwrap();
+            assert_eq!(matched.as_boolean(), &BooleanArray::from(expected.to_vec()));
+        }
+    }
+
     #[test]
     fn spark_plan_metrics_hash_join() {
         let op_scan = create_scan();
 
@@ -98,6 +98,7 @@ macro_rules! binary_expr_builder {
                     $operator,
                     right,
                     input_schema.as_ref(),
+                    planner.float_operands(),
                 )
                 .map_err(Into::into)
             }
Original file line number	Diff line number	Diff line change
`@@ -98,6 +98,7 @@ macro_rules! binary_expr_builder {`
`98`	`98`	`$operator,`
`99`	`99`	`right,`
`100`	`100`	`input_schema.as_ref(),`
	`101`	`+ planner.float_operands(),`
`101`	`102`	`)`
`102`	`103`	`.map_err(Into::into)`
`103`	`104`	`}`