diff --git a/datafusion/physical-expr/src/expressions/mod.rs b/datafusion/physical-expr/src/expressions/mod.rs index 521b8b87e305c..00edff33f8777 100644 --- a/datafusion/physical-expr/src/expressions/mod.rs +++ b/datafusion/physical-expr/src/expressions/mod.rs @@ -32,6 +32,7 @@ mod like; mod literal; mod negative; mod no_op; +mod normalize_float_zero; mod not; mod similar_to_pattern; mod try_cast; @@ -58,6 +59,7 @@ pub use like::{LikeExpr, like}; pub use literal::{Literal, lit}; pub use negative::{NegativeExpr, negative}; pub use no_op::NoOp; +pub use normalize_float_zero::NormalizeFloatZeroExpr; pub use not::{NotExpr, not}; pub(crate) use similar_to_pattern::translate_scalar; pub use similar_to_pattern::{SqlSimilarToPattern, sql_similar_to_regex}; diff --git a/datafusion/physical-expr/src/expressions/normalize_float_zero.rs b/datafusion/physical-expr/src/expressions/normalize_float_zero.rs new file mode 100644 index 0000000000000..7e4dc3db150d1 --- /dev/null +++ b/datafusion/physical-expr/src/expressions/normalize_float_zero.rs @@ -0,0 +1,255 @@ +// 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. + +//! Floating-point signed-zero normalization expression. + +use std::hash::Hash; +use std::sync::Arc; + +use arrow::datatypes::{DataType, FieldRef, Schema}; +use arrow::record_batch::RecordBatch; +use datafusion_common::Result; +use datafusion_common::utils::{normalize_float_zero, normalize_float_zero_scalar}; +use datafusion_expr::ColumnarValue; +use datafusion_expr::interval_arithmetic::Interval; +use datafusion_expr::sort_properties::ExprProperties; + +use crate::PhysicalExpr; + +/// Replaces floating-point `-0.0` values with `+0.0`. +/// +/// Other values and data types are returned unchanged. This expression is +/// order-preserving but not strictly order-preserving because it collapses the +/// two signed-zero representations. +#[derive(Debug, Eq)] +pub struct NormalizeFloatZeroExpr { + arg: Arc, +} + +impl PartialEq for NormalizeFloatZeroExpr { + fn eq(&self, other: &Self) -> bool { + self.arg.eq(&other.arg) + } +} + +impl Hash for NormalizeFloatZeroExpr { + fn hash(&self, state: &mut H) { + self.arg.hash(state); + } +} + +impl NormalizeFloatZeroExpr { + /// Creates a signed-zero normalization expression. + pub fn new(arg: Arc) -> Self { + Self { arg } + } + + /// Returns the input expression. + pub fn arg(&self) -> &Arc { + &self.arg + } +} + +impl std::fmt::Display for NormalizeFloatZeroExpr { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "normalize_float_zero({})", self.arg) + } +} + +impl PhysicalExpr for NormalizeFloatZeroExpr { + fn data_type(&self, input_schema: &Schema) -> Result { + self.arg.data_type(input_schema) + } + + fn nullable(&self, input_schema: &Schema) -> Result { + self.arg.nullable(input_schema) + } + + fn evaluate(&self, batch: &RecordBatch) -> Result { + Ok(match self.arg.evaluate(batch)? { + ColumnarValue::Array(array) => { + ColumnarValue::Array(normalize_float_zero(&array)) + } + ColumnarValue::Scalar(scalar) => { + ColumnarValue::Scalar(normalize_float_zero_scalar(scalar)) + } + }) + } + + fn return_field(&self, input_schema: &Schema) -> Result { + self.arg.return_field(input_schema) + } + + fn children(&self) -> Vec<&Arc> { + vec![&self.arg] + } + + fn with_new_children( + self: Arc, + children: Vec>, + ) -> Result> { + Ok(Arc::new(Self::new(Arc::clone(&children[0])))) + } + + fn evaluate_bounds(&self, children: &[&Interval]) -> Result { + Interval::try_new( + normalize_float_zero_scalar(children[0].lower().clone()), + normalize_float_zero_scalar(children[0].upper().clone()), + ) + } + + fn get_properties(&self, children: &[ExprProperties]) -> Result { + let range = self.evaluate_bounds(&[&children[0].range])?; + Ok(children[0] + .clone() + .with_range(range) + .with_strictly_order_preserving(false)) + } + + fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "normalize_float_zero(")?; + self.arg.fmt_sql(f)?; + write!(f, ")") + } + + #[cfg(feature = "proto")] + fn try_to_proto( + &self, + ctx: &datafusion_physical_expr_common::physical_expr::proto_encode::PhysicalExprEncodeCtx<'_>, + ) -> Result> { + use datafusion_proto_models::protobuf; + + Ok(Some(protobuf::PhysicalExprNode { + expr_id: None, + expr_type: Some(protobuf::physical_expr_node::ExprType::NormalizeFloatZero( + Box::new(protobuf::PhysicalNormalizeFloatZeroNode { + expr: Some(Box::new(ctx.encode_child(&self.arg)?)), + }), + )), + })) + } +} + +#[cfg(feature = "proto")] +impl NormalizeFloatZeroExpr { + /// Reconstructs a [`NormalizeFloatZeroExpr`] from protobuf. + pub fn try_from_proto( + node: &datafusion_proto_models::protobuf::PhysicalExprNode, + ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>, + ) -> Result> { + use datafusion_physical_expr_common::expect_expr_variant; + use datafusion_proto_models::protobuf; + + let node = expect_expr_variant!( + node, + protobuf::physical_expr_node::ExprType::NormalizeFloatZero, + "NormalizeFloatZero", + ); + let arg = ctx.decode_required_expression( + node.expr.as_deref(), + "NormalizeFloatZeroExpr", + "expr", + )?; + Ok(Arc::new(Self::new(arg))) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use arrow::array::{ArrayRef, AsArray, Float64Array}; + use datafusion_common::ScalarValue; + use datafusion_expr::sort_properties::SortProperties; + use half::f16; + + use crate::expressions::{Column, Literal}; + + #[test] + fn normalizes_array_and_scalar_signed_zero() -> Result<()> { + let batch = RecordBatch::try_from_iter(vec![( + "a", + Arc::new(Float64Array::from(vec![-0.0, 0.0, 1.0])) as ArrayRef, + )])?; + let expr = NormalizeFloatZeroExpr::new(Arc::new(Column::new("a", 0))); + let ColumnarValue::Array(array) = expr.evaluate(&batch)? else { + panic!("column evaluation must return an array"); + }; + let array = array.as_primitive::(); + assert_eq!(array.value(0).to_bits(), 0.0_f64.to_bits()); + assert_eq!(array.value(1).to_bits(), 0.0_f64.to_bits()); + assert_eq!(array.value(2), 1.0); + + let expr = NormalizeFloatZeroExpr::new(Arc::new(Literal::new( + ScalarValue::Float64(Some(-0.0)), + ))); + let ColumnarValue::Scalar(ScalarValue::Float64(Some(value))) = + expr.evaluate(&RecordBatch::new_empty(Arc::new(Schema::empty())))? + else { + panic!("literal evaluation must return a Float64 scalar"); + }; + assert_eq!(value.to_bits(), 0.0_f64.to_bits()); + Ok(()) + } + + #[test] + fn normalizes_signed_zero_bounds_and_properties() -> Result<()> { + let batch = RecordBatch::new_empty(Arc::new(Schema::empty())); + let cases = [ + ( + ScalarValue::Float16(Some(f16::NEG_ZERO)), + ScalarValue::Float16(Some(f16::ZERO)), + ), + ( + ScalarValue::Float32(Some(-0.0)), + ScalarValue::Float32(Some(0.0)), + ), + ( + ScalarValue::Float64(Some(-0.0)), + ScalarValue::Float64(Some(0.0)), + ), + ]; + + for (negative_zero, positive_zero) in cases { + let child_range = + Interval::try_new(negative_zero.clone(), negative_zero.clone())?; + let expected_range = + Interval::try_new(positive_zero.clone(), positive_zero.clone())?; + let child_properties = ExprProperties::new_unknown() + .with_order(SortProperties::Singleton) + .with_range(child_range.clone()) + .with_preserves_lex_ordering(true) + .with_strictly_order_preserving(true); + let expr = NormalizeFloatZeroExpr::new(Arc::new(Literal::new(negative_zero))); + + let ColumnarValue::Scalar(value) = expr.evaluate(&batch)? else { + panic!("literal evaluation must return a scalar"); + }; + let bounds = expr.evaluate_bounds(&[&child_range])?; + assert_eq!(value, positive_zero); + assert_eq!(bounds, expected_range); + assert!(bounds.contains_value(&value)?); + + let properties = expr.get_properties(&[child_properties])?; + assert_eq!(properties.sort_properties, SortProperties::Singleton); + assert_eq!(properties.range, expected_range); + assert!(properties.preserves_lex_ordering); + assert!(!properties.strictly_order_preserving); + } + Ok(()) + } +} diff --git a/datafusion/physical-plan/src/joins/asof_join.rs b/datafusion/physical-plan/src/joins/asof_join.rs index 22625afca1f49..54540d89ebf68 100644 --- a/datafusion/physical-plan/src/joins/asof_join.rs +++ b/datafusion/physical-plan/src/joins/asof_join.rs @@ -88,7 +88,9 @@ use datafusion_execution::TaskContext; use datafusion_execution::memory_pool::{MemoryConsumer, MemoryReservation}; use datafusion_expr::Operator; use datafusion_physical_expr::PhysicalSortExpr; -use datafusion_physical_expr::expressions::Column as PhysicalColumn; +use datafusion_physical_expr::expressions::{ + Column as PhysicalColumn, NormalizeFloatZeroExpr, +}; use datafusion_physical_expr::projection::{ProjectionMapping, ProjectionRef}; use datafusion_physical_expr::utils::collect_columns; use datafusion_physical_expr_common::physical_expr::{ @@ -162,10 +164,8 @@ impl AsOfJoinExec { /// /// The match operator must be `<`, `<=`, `>`, or `>=`. Equality and match /// expressions must be deterministic, reference only their corresponding - /// input, and have matching input types. Equality types must support hashing; - /// floating-point equality keys are not supported because Arrow sorting - /// distinguishes signed zero while SQL equality does not. Projection indices - /// refer to the full left-then-right join schema. + /// input, and have matching input types. Equality types must support hashing. + /// Projection indices refer to the full left-then-right join schema. pub fn try_new( left: Arc, right: Arc, @@ -189,24 +189,33 @@ impl AsOfJoinExec { descending, nulls_first: true, }; - let mut left_sort_exprs = on - .iter() - .map(|(left, _)| PhysicalSortExpr { - expr: Arc::clone(left), + let mut left_sort_exprs = Vec::with_capacity(on.len() + 1); + let mut right_sort_exprs = Vec::with_capacity(on.len() + 1); + for (left, right) in &on { + let left_expr = if left.data_type(&left_schema)?.is_floating() { + Arc::new(NormalizeFloatZeroExpr::new(Arc::clone(left))) as PhysicalExprRef + } else { + Arc::clone(left) + }; + let right_expr = if right.data_type(&right_schema)?.is_floating() { + Arc::new(NormalizeFloatZeroExpr::new(Arc::clone(right))) + as PhysicalExprRef + } else { + Arc::clone(right) + }; + left_sort_exprs.push(PhysicalSortExpr { + expr: left_expr, options: equality_options, - }) - .collect::>(); + }); + right_sort_exprs.push(PhysicalSortExpr { + expr: right_expr, + options: equality_options, + }); + } left_sort_exprs.push(PhysicalSortExpr { expr: Arc::clone(&match_condition.left), options: match_options, }); - let mut right_sort_exprs = on - .iter() - .map(|(_, right)| PhysicalSortExpr { - expr: Arc::clone(right), - options: equality_options, - }) - .collect::>(); right_sort_exprs.push(PhysicalSortExpr { expr: Arc::clone(&match_condition.right), options: match_options, @@ -1196,11 +1205,6 @@ fn validate_asof_join( "AsOfJoinExec equality expressions have unsupported hash type {left_type}" ); } - if left_type.is_floating() { - return plan_err!( - "AsOfJoinExec equality expressions do not support floating-point type {left_type}" - ); - } } let left_match_type = match_condition.left.data_type(&left_schema)?; let right_match_type = match_condition.right.data_type(&right_schema)?; @@ -1248,8 +1252,9 @@ mod tests { use super::*; use crate::collect; + use crate::sorts::sort::SortExec; use crate::test::TestMemoryExec; - use arrow::array::{Int32Array, Int64Array, StringArray}; + use arrow::array::{Float64Array, Int32Array, Int64Array, StringArray}; use arrow::datatypes::{DataType, Field}; use datafusion_common::test_util::batches_to_sort_string; use datafusion_execution::config::SessionConfig; @@ -1712,34 +1717,88 @@ mod tests { Ok(()) } - #[test] - fn rejects_floating_equality_expressions() -> Result<()> { - let exec = test_exec()?; + #[tokio::test] + async fn floating_equality_keys_treat_signed_zero_as_equal() -> Result<()> { + let left_batch = RecordBatch::try_from_iter(vec![ + ("key", Arc::new(Float64Array::from(vec![0.0])) as ArrayRef), + ("ts", Arc::new(Int64Array::from(vec![5])) as ArrayRef), + ("id", Arc::new(Int32Array::from(vec![1])) as ArrayRef), + ])?; + let right_batch = RecordBatch::try_from_iter(vec![ + ( + "key", + Arc::new(Float64Array::from(vec![-0.0, 0.0])) as ArrayRef, + ), + ("ts", Arc::new(Int64Array::from(vec![10, 1])) as ArrayRef), + ( + "price", + Arc::new(Int32Array::from(vec![100, 10])) as ArrayRef, + ), + ])?; for data_type in [DataType::Float16, DataType::Float32, DataType::Float64] { - let left = Arc::new(CastExpr::new( - Arc::new(PhysicalColumn::new("ts", 1)), - data_type.clone(), + let left = TestMemoryExec::try_new_exec( + &[vec![left_batch.clone()]], + left_batch.schema(), None, - )); - let right = Arc::new(CastExpr::new( - Arc::new(PhysicalColumn::new("ts", 1)), - data_type.clone(), + )?; + let right = TestMemoryExec::try_new_exec( + &[vec![right_batch.clone()]], + right_batch.schema(), None, - )); - let error = AsOfJoinExec::try_new( - Arc::clone(&exec.left), - Arc::clone(&exec.right), - vec![(left, right)], - exec.match_condition.clone(), - Some(vec![0, 1, 2, 5]), - ) - .expect_err("floating equality expressions must be rejected"); - assert!( - error.to_string().contains(&format!( - "equality expressions do not support floating-point type {data_type}" + )?; + let on: JoinOn = vec![( + Arc::new(CastExpr::new( + Arc::new(PhysicalColumn::new("key", 0)), + data_type.clone(), + None, + )), + Arc::new(CastExpr::new( + Arc::new(PhysicalColumn::new("key", 0)), + data_type.clone(), + None, )), - "unexpected error: {error}" + )]; + let match_condition = AsOfMatchExpr::new( + Arc::new(PhysicalColumn::new("ts", 1)), + Operator::GtEq, + Arc::new(PhysicalColumn::new("ts", 1)), ); + let unsorted = AsOfJoinExec::try_new( + left, + right, + on.clone(), + match_condition.clone(), + Some(vec![2, 5]), + )?; + let left = Arc::new(SortExec::new( + unsorted.left_ordering.clone(), + Arc::clone(&unsorted.left), + )); + let right = Arc::new(SortExec::new( + unsorted.right_ordering.clone(), + Arc::clone(&unsorted.right), + )); + let exec = Arc::new(AsOfJoinExec::try_new( + left, + right, + on, + match_condition, + Some(vec![2, 5]), + )?); + + let batches = collect(exec, Arc::new(TaskContext::default())).await?; + let prices = batches + .iter() + .flat_map(|batch| { + batch + .column(1) + .as_any() + .downcast_ref::() + .unwrap() + .iter() + }) + .collect::>(); + assert_eq!(prices, vec![Some(10)], "data type: {data_type}"); } Ok(()) } diff --git a/datafusion/proto-models/proto/datafusion.proto b/datafusion/proto-models/proto/datafusion.proto index d98b67a66e0a9..efaf8e210d9cc 100644 --- a/datafusion/proto-models/proto/datafusion.proto +++ b/datafusion/proto-models/proto/datafusion.proto @@ -1052,6 +1052,7 @@ message PhysicalExprNode { PhysicalLambdaVariableExprNode lambda_variable = 26; PhysicalRangeExprNode range_expr = 27; PhysicalSqlSimilarToPatternNode sql_similar_to_pattern = 28; + PhysicalNormalizeFloatZeroNode normalize_float_zero = 29; } } @@ -1197,6 +1198,10 @@ message PhysicalNegativeNode { PhysicalExprNode expr = 1; } +message PhysicalNormalizeFloatZeroNode { + PhysicalExprNode expr = 1; +} + message PhysicalExtensionExprNode { bytes expr = 1; repeated PhysicalExprNode inputs = 2; diff --git a/datafusion/proto-models/src/generated/pbjson.rs b/datafusion/proto-models/src/generated/pbjson.rs index 21309bb2d0941..27ea8b663d8ed 100644 --- a/datafusion/proto-models/src/generated/pbjson.rs +++ b/datafusion/proto-models/src/generated/pbjson.rs @@ -18585,6 +18585,9 @@ impl serde::Serialize for PhysicalExprNode { physical_expr_node::ExprType::SqlSimilarToPattern(v) => { struct_ser.serialize_field("sqlSimilarToPattern", v)?; } + physical_expr_node::ExprType::NormalizeFloatZero(v) => { + struct_ser.serialize_field("normalizeFloatZero", v)?; + } } } struct_ser.end() @@ -18644,6 +18647,8 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode { "rangeExpr", "sql_similar_to_pattern", "sqlSimilarToPattern", + "normalize_float_zero", + "normalizeFloatZero", ]; #[allow(clippy::enum_variant_names)] @@ -18675,6 +18680,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode { LambdaVariable, RangeExpr, SqlSimilarToPattern, + NormalizeFloatZero, } impl<'de> serde::Deserialize<'de> for GeneratedField { fn deserialize(deserializer: D) -> std::result::Result @@ -18723,6 +18729,7 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode { "lambdaVariable" | "lambda_variable" => Ok(GeneratedField::LambdaVariable), "rangeExpr" | "range_expr" => Ok(GeneratedField::RangeExpr), "sqlSimilarToPattern" | "sql_similar_to_pattern" => Ok(GeneratedField::SqlSimilarToPattern), + "normalizeFloatZero" | "normalize_float_zero" => Ok(GeneratedField::NormalizeFloatZero), _ => Err(serde::de::Error::unknown_field(value, FIELDS)), } } @@ -18934,6 +18941,13 @@ impl<'de> serde::Deserialize<'de> for PhysicalExprNode { return Err(serde::de::Error::duplicate_field("sqlSimilarToPattern")); } expr_type__ = map_.next_value::<::std::option::Option<_>>()?.map(physical_expr_node::ExprType::SqlSimilarToPattern) +; + } + GeneratedField::NormalizeFloatZero => { + if expr_type__.is_some() { + return Err(serde::de::Error::duplicate_field("normalizeFloatZero")); + } + expr_type__ = map_.next_value::<::std::option::Option<_>>()?.map(physical_expr_node::ExprType::NormalizeFloatZero) ; } } @@ -20304,6 +20318,97 @@ impl<'de> serde::Deserialize<'de> for PhysicalNegativeNode { deserializer.deserialize_struct("datafusion.PhysicalNegativeNode", FIELDS, GeneratedVisitor) } } +impl serde::Serialize for PhysicalNormalizeFloatZeroNode { + #[allow(deprecated)] + fn serialize(&self, serializer: S) -> std::result::Result + where + S: serde::Serializer, + { + use serde::ser::SerializeStruct; + let mut len = 0; + if self.expr.is_some() { + len += 1; + } + let mut struct_ser = serializer.serialize_struct("datafusion.PhysicalNormalizeFloatZeroNode", len)?; + if let Some(v) = self.expr.as_ref() { + struct_ser.serialize_field("expr", v)?; + } + struct_ser.end() + } +} +impl<'de> serde::Deserialize<'de> for PhysicalNormalizeFloatZeroNode { + #[allow(deprecated)] + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + const FIELDS: &[&str] = &[ + "expr", + ]; + + #[allow(clippy::enum_variant_names)] + enum GeneratedField { + Expr, + } + impl<'de> serde::Deserialize<'de> for GeneratedField { + fn deserialize(deserializer: D) -> std::result::Result + where + D: serde::Deserializer<'de>, + { + struct GeneratedVisitor; + + impl serde::de::Visitor<'_> for GeneratedVisitor { + type Value = GeneratedField; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(formatter, "expected one of: {:?}", &FIELDS) + } + + #[allow(unused_variables)] + fn visit_str(self, value: &str) -> std::result::Result + where + E: serde::de::Error, + { + match value { + "expr" => Ok(GeneratedField::Expr), + _ => Err(serde::de::Error::unknown_field(value, FIELDS)), + } + } + } + deserializer.deserialize_identifier(GeneratedVisitor) + } + } + struct GeneratedVisitor; + impl<'de> serde::de::Visitor<'de> for GeneratedVisitor { + type Value = PhysicalNormalizeFloatZeroNode; + + fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("struct datafusion.PhysicalNormalizeFloatZeroNode") + } + + fn visit_map(self, mut map_: V) -> std::result::Result + where + V: serde::de::MapAccess<'de>, + { + let mut expr__ = None; + while let Some(k) = map_.next_key()? { + match k { + GeneratedField::Expr => { + if expr__.is_some() { + return Err(serde::de::Error::duplicate_field("expr")); + } + expr__ = map_.next_value()?; + } + } + } + Ok(PhysicalNormalizeFloatZeroNode { + expr: expr__, + }) + } + } + deserializer.deserialize_struct("datafusion.PhysicalNormalizeFloatZeroNode", FIELDS, GeneratedVisitor) + } +} impl serde::Serialize for PhysicalNot { #[allow(deprecated)] fn serialize(&self, serializer: S) -> std::result::Result diff --git a/datafusion/proto-models/src/generated/prost.rs b/datafusion/proto-models/src/generated/prost.rs index d830624322e14..ff28cd65bc277 100644 --- a/datafusion/proto-models/src/generated/prost.rs +++ b/datafusion/proto-models/src/generated/prost.rs @@ -1557,7 +1557,7 @@ pub struct PhysicalExprNode { pub expr_id: ::core::option::Option, #[prost( oneof = "physical_expr_node::ExprType", - tags = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 14, 15, 16, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28" + tags = "1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 14, 15, 16, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29" )] pub expr_type: ::core::option::Option, } @@ -1626,6 +1626,10 @@ pub mod physical_expr_node { SqlSimilarToPattern( ::prost::alloc::boxed::Box, ), + #[prost(message, tag = "29")] + NormalizeFloatZero( + ::prost::alloc::boxed::Box, + ), } } #[derive(Clone, PartialEq, ::prost::Message)] @@ -1856,6 +1860,11 @@ pub struct PhysicalNegativeNode { pub expr: ::core::option::Option<::prost::alloc::boxed::Box>, } #[derive(Clone, PartialEq, ::prost::Message)] +pub struct PhysicalNormalizeFloatZeroNode { + #[prost(message, optional, boxed, tag = "1")] + pub expr: ::core::option::Option<::prost::alloc::boxed::Box>, +} +#[derive(Clone, PartialEq, ::prost::Message)] pub struct PhysicalExtensionExprNode { #[prost(bytes = "vec", tag = "1")] pub expr: ::prost::alloc::vec::Vec, diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs index bc443149df413..a9a41a15d435f 100644 --- a/datafusion/proto/src/physical_plan/from_proto.rs +++ b/datafusion/proto/src/physical_plan/from_proto.rs @@ -36,8 +36,8 @@ use datafusion_physical_expr::{ }; use datafusion_physical_plan::expressions::{ BinaryExpr, CaseExpr, CastExpr, Column, InListExpr, IsNotNullExpr, IsNullExpr, - LikeExpr, Literal, NegativeExpr, NotExpr, SqlSimilarToPattern, TryCastExpr, - UnKnownColumn, + LikeExpr, Literal, NegativeExpr, NormalizeFloatZeroExpr, NotExpr, + SqlSimilarToPattern, TryCastExpr, UnKnownColumn, }; use datafusion_physical_plan::joins::HashExpr; use datafusion_physical_plan::proto::ExecutionPlanDecodeCtx; @@ -288,6 +288,9 @@ pub fn parse_physical_expr_with_converter( ExprType::IsNotNullExpr(_) => IsNotNullExpr::try_from_proto(proto, &decode_ctx)?, ExprType::NotExpr(_) => NotExpr::try_from_proto(proto, &decode_ctx)?, ExprType::Negative(_) => NegativeExpr::try_from_proto(proto, &decode_ctx)?, + ExprType::NormalizeFloatZero(_) => { + NormalizeFloatZeroExpr::try_from_proto(proto, &decode_ctx)? + } ExprType::InList(_) => InListExpr::try_from_proto(proto, &decode_ctx)?, ExprType::Case(_) => CaseExpr::try_from_proto(proto, &decode_ctx)?, ExprType::Cast(_) => CastExpr::try_from_proto(proto, &decode_ctx)?, diff --git a/datafusion/proto/tests/cases/plans/sorts.rs b/datafusion/proto/tests/cases/plans/sorts.rs index 1172775b1bad7..92a4d5173b944 100644 --- a/datafusion/proto/tests/cases/plans/sorts.rs +++ b/datafusion/proto/tests/cases/plans/sorts.rs @@ -22,7 +22,9 @@ use datafusion::arrow::compute::kernels::sort::SortOptions; use datafusion::arrow::datatypes::{DataType, Field, Schema}; use datafusion::physical_expr::LexOrdering; use datafusion::physical_plan::empty::EmptyExec; -use datafusion::physical_plan::expressions::{PhysicalSortExpr, col}; +use datafusion::physical_plan::expressions::{ + NormalizeFloatZeroExpr, PhysicalSortExpr, col, +}; use datafusion::physical_plan::sorts::sort::SortExec; use datafusion::physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec; use datafusion::prelude::SessionContext; @@ -61,6 +63,20 @@ fn roundtrip_sort() -> Result<()> { ))) } +#[test] +fn roundtrip_sort_with_normalized_float_zero() -> Result<()> { + let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); + let sort_exprs = [PhysicalSortExpr { + expr: Arc::new(NormalizeFloatZeroExpr::new(col("a", &schema)?)), + options: SortOptions::default(), + }] + .into(); + roundtrip_test(Arc::new(SortExec::new( + sort_exprs, + Arc::new(EmptyExec::new(schema)), + ))) +} + #[test] fn roundtrip_sort_preserve_partitioning() -> Result<()> { let field_a = Field::new("a", DataType::Boolean, false); diff --git a/docs/source/library-user-guide/upgrading/55.0.0.md b/docs/source/library-user-guide/upgrading/55.0.0.md index e88722bbf4a3e..9e9b5ea5af594 100644 --- a/docs/source/library-user-guide/upgrading/55.0.0.md +++ b/docs/source/library-user-guide/upgrading/55.0.0.md @@ -1332,6 +1332,37 @@ readers that don't recognize the new tag. See [PR #23188](https://github.com/apache/datafusion/pull/23188) for details. +### `ExprType` gained a `NormalizeFloatZero` variant + +Floating-point equality keys in ASOF joins now use a +`NormalizeFloatZeroExpr` physical expression to ensure that `-0.0` and `+0.0` +have the same ordering and equality semantics. The generated `ExprType` enum +on `PhysicalExprNode` gained a matching `NormalizeFloatZero` variant (tag 29). + +**Who is affected:** + +- Users matching exhaustively on `ExprType` (for example in a custom + physical-plan encoder/decoder). + +**Migration guide:** + +Add a `NormalizeFloatZero` arm, or fall back to a wildcard arm: + +```rust,ignore +match expr_type { + // ... + ExprType::NormalizeFloatZero(node) => { /* ... */ } + _ => { /* ... */ } +} +``` + +Plans encoded before this variant existed are unaffected: they never produced +this variant, so decoding is unchanged. Plans containing a +`NormalizeFloatZeroExpr` encoded after this change fail to decode on older +readers that don't recognize the new tag. + +See [PR #24375](https://github.com/apache/datafusion/pull/24375) for details. + ### `ParquetObjectReader` / `ParquetObjectWriter` deprecated upstream The [`parquet` crate] deprecated [`ParquetObjectReader`]