From edb18eb62ab9f58fe0b824ad64992a9903987fa3 Mon Sep 17 00:00:00 2001 From: Ma Zhengxuan <1319614897@qq.com> Date: Thu, 24 Sep 2026 01:12:12 +0800 Subject: [PATCH 1/4] [AURON #2517] Align native float ordering with Spark --- .../datafusion-ext-plans/src/agg/maxmin.rs | 239 ++++++++++++++---- .../org/apache/auron/AuronFunctionSuite.scala | 36 +++ .../spark/sql/auron/NativeConverters.scala | 12 +- 3 files changed, 239 insertions(+), 48 deletions(-) diff --git a/native-engine/datafusion-ext-plans/src/agg/maxmin.rs b/native-engine/datafusion-ext-plans/src/agg/maxmin.rs index c9cb564c4..eaf1947c0 100644 --- a/native-engine/datafusion-ext-plans/src/agg/maxmin.rs +++ b/native-engine/datafusion-ext-plans/src/agg/maxmin.rs @@ -40,6 +40,30 @@ use crate::{ pub type AggMax = AggMaxMin; pub type AggMin = AggMaxMin; +fn spark_float_cmp(left: T, right: T) -> Ordering { + // Spark places all NaNs last and treats signed zeros as equal. + if left == right || (left.is_nan() && right.is_nan()) { + Ordering::Equal + } else if left.is_nan() { + Ordering::Greater + } else if right.is_nan() { + Ordering::Less + } else if left < right { + Ordering::Less + } else { + Ordering::Greater + } +} + +macro_rules! compare_primitive { + (spark, $left:expr, $right:expr) => { + Some(spark_float_cmp($left, $right)) + }; + (partial, $left:expr, $right:expr) => { + $left.partial_cmp(&$right) + }; +} + pub struct AggMaxMin { child: PhysicalExprRef, data_type: DataType, @@ -108,15 +132,16 @@ impl Agg for AggMaxMin

{ accs.ensure_size(acc_idx); macro_rules! handle_primitive { - ($array:expr) => {{ + ($array:expr, $compare:ident) => {{ let partial_arg = $array; let accs = downcast_any!(accs, mut AccPrimColumn<_>)?; idx_for_zipped! { ((acc_idx, partial_arg_idx) in (acc_idx, partial_arg_idx)) => { if partial_arg.is_valid(partial_arg_idx) { - let partial_value = partial_arg.value(partial_arg_idx); - accs.update_value(acc_idx, partial_value, |v| { - if v.partial_cmp(&partial_value) == Some(P::ORD) { + let partial_value = partial_arg.value(partial_arg_idx); + accs.update_value(acc_idx, partial_value, |v| { + let ord = compare_primitive!($compare, v, partial_value); + if ord == Some(P::ORD) || ord == Some(Ordering::Equal) { v } else { partial_value @@ -167,30 +192,38 @@ impl Agg for AggMaxMin

{ }}; } - downcast_primitive_array! { - partial_arg => handle_primitive!(partial_arg), - DataType::Boolean => handle_boolean!(downcast_any!(partial_arg, BooleanArray)?), - DataType::Binary => handle_bytes!(downcast_any!(partial_arg, BinaryArray)?), - DataType::Utf8 => handle_bytes!(downcast_any!(partial_arg, StringArray)?), - DataType::Null => {} - _ => { - let accs = downcast_any!(accs, mut AccScalarValueColumn)?; - idx_for_zipped! { - ((acc_idx, partial_arg_idx) in (acc_idx, partial_arg_idx)) => { - if partial_args[0].is_valid(partial_arg_idx) { - let partial_arg_scalar = compacted_scalar_value_from_array( - &partial_args[0], - partial_arg_idx, - )?; - let acc_scalar = accs.value(acc_idx); - if !acc_scalar.is_null() && acc_scalar.partial_cmp(&partial_arg_scalar) == Some(P::ORD) { - continue; + match partial_arg.data_type() { + DataType::Float32 => { + handle_primitive!(downcast_any!(partial_arg, Float32Array)?, spark) + } + DataType::Float64 => { + handle_primitive!(downcast_any!(partial_arg, Float64Array)?, spark) + } + _ => downcast_primitive_array! { + partial_arg => handle_primitive!(partial_arg, partial), + DataType::Boolean => handle_boolean!(downcast_any!(partial_arg, BooleanArray)?), + DataType::Binary => handle_bytes!(downcast_any!(partial_arg, BinaryArray)?), + DataType::Utf8 => handle_bytes!(downcast_any!(partial_arg, StringArray)?), + DataType::Null => {} + _ => { + let accs = downcast_any!(accs, mut AccScalarValueColumn)?; + idx_for_zipped! { + ((acc_idx, partial_arg_idx) in (acc_idx, partial_arg_idx)) => { + if partial_args[0].is_valid(partial_arg_idx) { + let partial_arg_scalar = compacted_scalar_value_from_array( + &partial_args[0], + partial_arg_idx, + )?; + let acc_scalar = accs.value(acc_idx); + if !acc_scalar.is_null() && acc_scalar.partial_cmp(&partial_arg_scalar) == Some(P::ORD) { + continue; + } + accs.set_value(acc_idx, partial_arg_scalar); } - accs.set_value(acc_idx, partial_arg_scalar); } } } - } + }, } Ok(()) } @@ -205,7 +238,10 @@ impl Agg for AggMaxMin

{ accs.ensure_size(acc_idx); macro_rules! handle_primitive { - ($ty:ty) => {{ + ($ty:ty) => { + handle_primitive!($ty, partial) + }; + ($ty:ty, $compare:ident) => {{ type TNative = <$ty as ArrowPrimitiveType>::Native; let accs = downcast_any!(accs, mut AccPrimColumn)?; let merging_accs = downcast_any!(merging_accs, mut AccPrimColumn<_>)?; @@ -213,7 +249,8 @@ impl Agg for AggMaxMin

{ ((acc_idx, merging_acc_idx) in (acc_idx, merging_acc_idx)) => { if let Some(merging_value) = merging_accs.value(merging_acc_idx) { accs.update_value(acc_idx, merging_value, |v| { - if v.partial_cmp(&merging_value) == Some(P::ORD) { + let ord = compare_primitive!($compare, v, merging_value); + if ord == Some(P::ORD) || ord == Some(Ordering::Equal) { v } else { merging_value @@ -266,28 +303,32 @@ impl Agg for AggMaxMin

{ } }}; } - downcast_primitive! { - (&self.data_type) => (handle_primitive), - DataType::Boolean => handle_boolean!(), - DataType::Utf8 | DataType::Binary => handle_bytes!(), - DataType::Null => {}, - _ => { - let accs = downcast_any!(accs, mut AccScalarValueColumn)?; - let merging_accs = downcast_any!(merging_accs, mut AccScalarValueColumn)?; - idx_for_zipped! { - ((acc_idx, merging_acc_idx) in (acc_idx, merging_acc_idx)) => { - let merging_value = merging_accs.take_value(merging_acc_idx); - if merging_value.is_null() { - continue; - } - let w = accs.value(acc_idx); - if !w.is_null() && w.partial_cmp(&merging_value) == Some(P::ORD) { - continue; + match &self.data_type { + DataType::Float32 => handle_primitive!(Float32Type, spark), + DataType::Float64 => handle_primitive!(Float64Type, spark), + _ => downcast_primitive! { + (&self.data_type) => (handle_primitive), + DataType::Boolean => handle_boolean!(), + DataType::Utf8 | DataType::Binary => handle_bytes!(), + DataType::Null => {}, + _ => { + let accs = downcast_any!(accs, mut AccScalarValueColumn)?; + let merging_accs = downcast_any!(merging_accs, mut AccScalarValueColumn)?; + idx_for_zipped! { + ((acc_idx, merging_acc_idx) in (acc_idx, merging_acc_idx)) => { + let merging_value = merging_accs.take_value(merging_acc_idx); + if merging_value.is_null() { + continue; + } + let w = accs.value(acc_idx); + if !w.is_null() && w.partial_cmp(&merging_value) == Some(P::ORD) { + continue; + } + accs.set_value(acc_idx, merging_value); } - accs.set_value(acc_idx, merging_value); } } - } + }, } Ok(()) } @@ -314,3 +355,111 @@ impl AggMaxMinParams for AggMinParams { const NAME: &'static str = "min"; const ORD: Ordering = Ordering::Less; } + +#[cfg(test)] +mod tests { + use datafusion::{common::ScalarValue, physical_expr::expressions::Literal}; + + use super::*; + + fn result(agg: &dyn Agg, acc: &mut AccColumnRef) -> Result { + let array = agg.final_merge(acc, IdxSelection::Single(0))?; + Ok(downcast_any!(array, Float64Array)?.value(0)) + } + + #[test] + fn float_ordering() { + assert_eq!(spark_float_cmp(0.0_f64, -0.0), Ordering::Equal); + assert_eq!(spark_float_cmp(f64::NAN, f64::NAN), Ordering::Equal); + assert_eq!( + spark_float_cmp(f64::from_bits(0xfff8_0000_0000_0001), f64::INFINITY), + Ordering::Greater + ); + assert_eq!(spark_float_cmp(0.0_f32, -0.0), Ordering::Equal); + assert_eq!( + spark_float_cmp(f32::from_bits(0xffc0_0001), f32::INFINITY), + Ordering::Greater + ); + } + + #[test] + fn float_max_min_update_and_merge() -> Result<()> { + let child: PhysicalExprRef = Arc::new(Literal::new(ScalarValue::Float64(None))); + let max = AggMax::try_new(child.clone(), DataType::Float64)?; + let min = AggMin::try_new(child, DataType::Float64)?; + let negative_nan = f64::from_bits(0xfff8_0000_0000_0001); + + let mut max_acc = max.create_acc_column(1); + let max_values: ArrayRef = Arc::new(Float64Array::from(vec![ + f64::INFINITY, + negative_nan, + f64::MAX, + ])); + max.partial_update( + &mut max_acc, + IdxSelection::Indices(&[0, 0, 0]), + &[max_values], + IdxSelection::Range(0, 3), + )?; + let mut max_merged = max.create_acc_column(1); + max.partial_update( + &mut max_merged, + IdxSelection::Single(0), + &[Arc::new(Float64Array::from(vec![f64::INFINITY]))], + IdxSelection::Single(0), + )?; + max.partial_merge( + &mut max_merged, + IdxSelection::Single(0), + &mut max_acc, + IdxSelection::Single(0), + )?; + assert!(result(&max, &mut max_merged)?.is_nan()); + + let mut min_acc = min.create_acc_column(1); + let min_values: ArrayRef = Arc::new(Float64Array::from(vec![ + -1.0, + f64::NEG_INFINITY, + negative_nan, + ])); + min.partial_update( + &mut min_acc, + IdxSelection::Indices(&[0, 0, 0]), + &[min_values], + IdxSelection::Range(0, 3), + )?; + let mut min_merged = min.create_acc_column(1); + min.partial_update( + &mut min_merged, + IdxSelection::Single(0), + &[Arc::new(Float64Array::from(vec![f64::NAN]))], + IdxSelection::Single(0), + )?; + min.partial_merge( + &mut min_acc, + IdxSelection::Single(0), + &mut min_merged, + IdxSelection::Single(0), + )?; + assert_eq!(result(&min, &mut min_acc)?, f64::NEG_INFINITY); + + let mut max_zero = max.create_acc_column(1); + max.partial_update( + &mut max_zero, + IdxSelection::Indices(&[0, 0]), + &[Arc::new(Float64Array::from(vec![-0.0, 0.0]))], + IdxSelection::Range(0, 2), + )?; + assert_eq!(result(&max, &mut max_zero)?.to_bits(), (-0.0_f64).to_bits()); + + let mut min_zero = min.create_acc_column(1); + min.partial_update( + &mut min_zero, + IdxSelection::Indices(&[0, 0]), + &[Arc::new(Float64Array::from(vec![0.0, -0.0]))], + IdxSelection::Range(0, 2), + )?; + assert_eq!(result(&min, &mut min_zero)?.to_bits(), 0.0_f64.to_bits()); + Ok(()) + } +} diff --git a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala index 58fce9b34..5369b3432 100644 --- a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala +++ b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala @@ -26,6 +26,42 @@ import org.apache.auron.util.AuronTestUtils class AuronFunctionSuite extends AuronQueryTest with BaseAuronSQLSuite { + test("float comparisons follow Spark NaN and signed zero ordering") { + withTable("t_float_ordering") { + sql( + "create table t_float_ordering (d1 double, d2 double, f1 float, f2 float) using parquet") + sql("""insert into t_float_ordering values + | (0.0, -0.0, cast(0.0 as float), cast(-0.0 as float)), + | (cast('NaN' as double), cast('NaN' as double), + | cast('NaN' as float), cast('NaN' as float)), + | (cast('NaN' as double), cast('Infinity' as double), + | cast('NaN' as float), cast('Infinity' as float)), + | (null, 1.0, null, cast(1.0 as float)) + |""".stripMargin) + + checkSparkAnswerAndOperator("""select + | d1 = d2, not (d1 = d2), d1 < d2, d1 <= d2, d1 > d2, d1 >= d2, d1 <=> d2, + | f1 = f2, not (f1 = f2), f1 < f2, f1 <= f2, f1 > f2, f1 >= f2, f1 <=> f2 + |from t_float_ordering + |""".stripMargin) + } + } + + test("float min and max follow Spark NaN ordering") { + withTable("t_float_ordering") { + sql("create table t_float_ordering (d double, f float) using parquet") + sql("""insert into t_float_ordering values + | (cast('Infinity' as double), cast('Infinity' as float)), + | (cast('NaN' as double), cast('NaN' as float)), + | (-1.0, cast(-1.0 as float)), + | (null, null) + |""".stripMargin) + + checkSparkAnswerAndOperator( + "select isnan(max(d)), min(d), isnan(max(f)), min(f) from t_float_ordering") + } + } + test("sum function with float input") { if (AuronTestUtils.isSparkV31OrGreater) { withTable("t1") { diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala index 2180487cd..776f186ed 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala @@ -1415,15 +1415,21 @@ object NativeConverters extends Logging { right: Expression, op: String, isPruningExpr: Boolean, - fallback: Expression => pb.PhysicalExprNode): pb.PhysicalExprNode = + fallback: Expression => pb.PhysicalExprNode): pb.PhysicalExprNode = { + val isFloatComparison = (left.dataType == FloatType || left.dataType == DoubleType) && + (op == "Eq" || op == "NotEq" || op == "Lt" || op == "LtEq" || op == "Gt" || + op == "GtEq" || op == "IsNotDistinctFrom") + val lhs = if (isFloatComparison) NormalizeNaNAndZero(left) else left + val rhs = if (isFloatComparison) NormalizeNaNAndZero(right) else right buildExprNode { _.setBinaryExpr( pb.PhysicalBinaryExprNode .newBuilder() - .setL(convertExprWithFallback(left, isPruningExpr, fallback)) - .setR(convertExprWithFallback(right, isPruningExpr, fallback)) + .setL(convertExprWithFallback(lhs, isPruningExpr, fallback)) + .setR(convertExprWithFallback(rhs, isPruningExpr, fallback)) .setOp(op)) } + } def buildScalarFunctionNode( fn: pb.ScalarFunction, From 1b245fbe40ef0534ae10119ac7c208b55ee68c3f Mon Sep 17 00:00:00 2001 From: Ma Zhengxuan <1319614897@qq.com> Date: Fri, 25 Sep 2026 00:56:25 +0800 Subject: [PATCH 2/4] [AURON #2517] Fix Clippy warning in float comparator --- native-engine/datafusion-ext-plans/src/agg/maxmin.rs | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/native-engine/datafusion-ext-plans/src/agg/maxmin.rs b/native-engine/datafusion-ext-plans/src/agg/maxmin.rs index eaf1947c0..89f1706e0 100644 --- a/native-engine/datafusion-ext-plans/src/agg/maxmin.rs +++ b/native-engine/datafusion-ext-plans/src/agg/maxmin.rs @@ -46,9 +46,7 @@ fn spark_float_cmp(left: T, right: T) -> Ordering { Ordering::Equal } else if left.is_nan() { Ordering::Greater - } else if right.is_nan() { - Ordering::Less - } else if left < right { + } else if right.is_nan() || left < right { Ordering::Less } else { Ordering::Greater From be9af9106ff0a523b593112651467fda711f6708 Mon Sep 17 00:00:00 2001 From: Ma Zhengxuan <1319614897@qq.com> Date: Thu, 8 Oct 2026 00:32:30 +0800 Subject: [PATCH 3/4] [AURON #2517] Preserve float scan pruning --- ...uron-preserve-float-pruning-with-nan.patch | 98 +++++++++++++++++++ .../org/apache/auron/AuronFunctionSuite.scala | 14 +++ .../apache/auron/NativeConvertersSuite.scala | 20 +++- .../spark/sql/auron/NativeConverters.scala | 8 +- 4 files changed, 135 insertions(+), 5 deletions(-) create mode 100644 dev/vendors/patches/orc-rust/0002-auron-preserve-float-pruning-with-nan.patch diff --git a/dev/vendors/patches/orc-rust/0002-auron-preserve-float-pruning-with-nan.patch b/dev/vendors/patches/orc-rust/0002-auron-preserve-float-pruning-with-nan.patch new file mode 100644 index 000000000..f8f8e27e4 --- /dev/null +++ b/dev/vendors/patches/orc-rust/0002-auron-preserve-float-pruning-with-nan.patch @@ -0,0 +1,98 @@ +diff --git a/src/row_group_filter.rs b/src/row_group_filter.rs +index 6478a8a..f57b775 100644 +--- a/src/row_group_filter.rs ++++ b/src/row_group_filter.rs +@@ -200,7 +200,7 @@ fn evaluate_comparison_with_stats( + } + + // Float comparisons +- TypeStatistics::Double { min, max, .. } => { ++ TypeStatistics::Double { min, max, sum } => { + let v = match value { + PredicateValue::Float32(Some(v)) => *v as f64, + PredicateValue::Float64(Some(v)) => *v, +@@ -211,7 +211,17 @@ fn evaluate_comparison_with_stats( + .build()); + } + }; +- evaluate_float_comparison(*min, *max, op, v) ++ // NaN may be absent from min/max while still present in the row group. ++ // ORC writers include it in the sum; missing or NaN sums are not safe to prune. ++ if v.is_nan() ++ || min.is_nan() ++ || max.is_nan() ++ || sum.as_ref().map_or(true, |sum| sum.is_nan()) ++ { ++ true ++ } else { ++ evaluate_float_comparison(*min, *max, op, v) ++ } + } + + // String comparisons +@@ -364,9 +374,7 @@ fn evaluate_float_comparison(min: f64, max: f64, op: ComparisonOp, value: f64) - + } + ComparisonOp::NotEqual => { + // col != value: keep if value is not the only value +- // If min and max are very close to value, skip +- const EPSILON: f64 = 1e-9; +- !((min - value).abs() < EPSILON && (max - value).abs() < EPSILON) ++ min != value || max != value + } + ComparisonOp::LessThan => { + // col < value: keep if min < value +@@ -571,6 +579,54 @@ mod tests { + crate::schema::RootDataType::from_proto(&types).unwrap() + } + ++ #[test] ++ fn test_float_comparison_keeps_uncertain_nan_statistics() { ++ use crate::predicate::{ComparisonOp, PredicateValue}; ++ ++ for (sum, keep) in [(Some(f64::NAN), true), (None, true), (Some(0.0), false)] { ++ let stats = ColumnStatistics::try_from(&proto::ColumnStatistics { ++ number_of_values: Some(2), ++ double_statistics: Some(proto::DoubleStatistics { ++ minimum: Some(0.0), ++ maximum: Some(0.0), ++ sum, ++ }), ++ ..Default::default() ++ }) ++ .unwrap(); ++ let actual = super::evaluate_comparison_with_stats( ++ &stats, ++ ComparisonOp::GreaterThan, ++ &PredicateValue::Float64(Some(100.0)), ++ ) ++ .unwrap(); ++ assert_eq!(actual, keep); ++ } ++ } ++ ++ #[test] ++ fn test_float_not_equal_does_not_skip_nearby_values() { ++ use crate::predicate::{ComparisonOp, PredicateValue}; ++ ++ let value = 100.0000000001; ++ let stats = ColumnStatistics::try_from(&proto::ColumnStatistics { ++ number_of_values: Some(1), ++ double_statistics: Some(proto::DoubleStatistics { ++ minimum: Some(value), ++ maximum: Some(value), ++ sum: Some(value), ++ }), ++ ..Default::default() ++ }) ++ .unwrap(); ++ assert!(super::evaluate_comparison_with_stats( ++ &stats, ++ ComparisonOp::NotEqual, ++ &PredicateValue::Float64(Some(100.0)), ++ ) ++ .unwrap()); ++ } ++ + #[test] + fn test_evaluate_predicate_integer_greater_than() { + use crate::predicate::{Predicate, PredicateValue}; diff --git a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala index 5369b3432..e5c5361eb 100644 --- a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala +++ b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala @@ -47,6 +47,20 @@ class AuronFunctionSuite extends AuronQueryTest with BaseAuronSQLSuite { } } + test("float scan filters preserve NaN and signed zero results") { + Seq("parquet", "orc").foreach { format => + withTable("t_float_scan") { + sql(s"create table t_float_scan (d double) using $format") + sql("""insert into t_float_scan values + | (0.0), (-0.0), (cast('NaN' as double)) + |""".stripMargin) + + checkSparkAnswerAndOperator("select d from t_float_scan where d > 100.0") + checkSparkAnswerAndOperator("select d from t_float_scan where d = 0.0") + } + } + } + test("float min and max follow Spark NaN ordering") { withTable("t_float_ordering") { sql("create table t_float_ordering (d double, f float) using parquet") diff --git a/spark-extension-shims-spark/src/test/scala/org/apache/auron/NativeConvertersSuite.scala b/spark-extension-shims-spark/src/test/scala/org/apache/auron/NativeConvertersSuite.scala index 8bc931905..7aac711cd 100644 --- a/spark-extension-shims-spark/src/test/scala/org/apache/auron/NativeConvertersSuite.scala +++ b/spark-extension-shims-spark/src/test/scala/org/apache/auron/NativeConvertersSuite.scala @@ -18,8 +18,8 @@ package org.apache.auron import org.apache.spark.sql.AuronQueryTest import org.apache.spark.sql.auron.NativeConverters -import org.apache.spark.sql.catalyst.expressions.{Cast, Literal} -import org.apache.spark.sql.types.{BooleanType, DataType, IntegerType, StringType} +import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast, GreaterThan, Literal} +import org.apache.spark.sql.types.{BooleanType, DataType, DoubleType, FloatType, IntegerType, StringType} import org.apache.auron.protobuf.ScalarFunction @@ -83,4 +83,20 @@ class NativeConvertersSuite assert(!childExpr.hasScalarFunction) assert(childExpr.hasLiteral) } + + test("float scan pruning keeps column and literal comparisons") { + val comparisons = Seq( + GreaterThan(AttributeReference("d", DoubleType)(), Literal(100.0)), + GreaterThan(AttributeReference("f", FloatType)(), Literal(100.0f))) + + comparisons.foreach { expr => + val pruning = NativeConverters.convertScanPruningExpr(expr).getBinaryExpr + assert(pruning.getL.hasColumn) + assert(pruning.getR.hasLiteral) + + val rowFilter = NativeConverters.convertExpr(expr).getBinaryExpr + assert(rowFilter.getL.getScalarFunction.getName == "Spark_NormalizeNanAndZero") + assert(rowFilter.getR.getScalarFunction.getName == "Spark_NormalizeNanAndZero") + } + } } diff --git a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala index 776f186ed..235cde766 100644 --- a/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala +++ b/spark-extension/src/main/scala/org/apache/spark/sql/auron/NativeConverters.scala @@ -1416,11 +1416,13 @@ object NativeConverters extends Logging { op: String, isPruningExpr: Boolean, fallback: Expression => pb.PhysicalExprNode): pb.PhysicalExprNode = { - val isFloatComparison = (left.dataType == FloatType || left.dataType == DoubleType) && + // Pruning needs a direct column/literal comparison; row filters still normalize floats. + val normalizeFloatComparison = !isPruningExpr && + (left.dataType == FloatType || left.dataType == DoubleType) && (op == "Eq" || op == "NotEq" || op == "Lt" || op == "LtEq" || op == "Gt" || op == "GtEq" || op == "IsNotDistinctFrom") - val lhs = if (isFloatComparison) NormalizeNaNAndZero(left) else left - val rhs = if (isFloatComparison) NormalizeNaNAndZero(right) else right + val lhs = if (normalizeFloatComparison) NormalizeNaNAndZero(left) else left + val rhs = if (normalizeFloatComparison) NormalizeNaNAndZero(right) else right buildExprNode { _.setBinaryExpr( pb.PhysicalBinaryExprNode From 393a07f24ce338140e8fc241dcb850c160873f1a Mon Sep 17 00:00:00 2001 From: Ma Zhengxuan <1319614897@qq.com> Date: Fri, 9 Oct 2026 10:04:49 +0800 Subject: [PATCH 4/4] [AURON #2517] Fix float scan regression test on Spark 3.0/3.1 --- .../src/test/scala/org/apache/auron/AuronFunctionSuite.scala | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala index e5c5361eb..08c9cfda7 100644 --- a/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala +++ b/spark-extension-shims-spark/src/test/scala/org/apache/auron/AuronFunctionSuite.scala @@ -55,7 +55,10 @@ class AuronFunctionSuite extends AuronQueryTest with BaseAuronSQLSuite { | (0.0), (-0.0), (cast('NaN' as double)) |""".stripMargin) - checkSparkAnswerAndOperator("select d from t_float_scan where d > 100.0") + // Spark 3.0/3.1 ORC pushdown can drop NaN before evaluating this filter. + val greaterThan = sql("select d from t_float_scan where d > 100.0") + checkAnswer(greaterThan, Seq(Row(Double.NaN))) + assertPlanIsNative(greaterThan) checkSparkAnswerAndOperator("select d from t_float_scan where d = 0.0") } }