fix: treat signed zeros as equal in window PARTITION BY

datafusion/core/src/physical_planner.rs

@@ -71,6 +71,7 @@ use datafusion_common::config::ConfigOptions;
 use datafusion_common::display::ToStringifiedPlan;
 use datafusion_common::format::ExplainAnalyzeCategories;
 use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
+use datafusion_common::utils::has_float_leaf;
 use datafusion_common::{
     DFSchema, DFSchemaRef, ScalarValue, exec_err, internal_datafusion_err, internal_err,
     not_impl_err, plan_err,
@@ -101,7 +102,7 @@ use datafusion_expr::{
 use datafusion_physical_expr::aggregate::{
     AggregateFunctionExpr, LoweredAggregate, LoweredAggregateBuilder,
 };
-use datafusion_physical_expr::expressions::Literal;
+use datafusion_physical_expr::expressions::{Literal, NormalizeFloatZeroExpr};
 use datafusion_physical_expr::{
     LexOrdering, PhysicalSortExpr, create_physical_sort_exprs,
 };
@@ -2901,7 +2902,19 @@ pub fn create_window_expr_with_name(
                 logical_schema,
                 execution_props,
                 planning_ctx,
-            )?;
+            )?
+            .into_iter()
+            .map(|expr| {
+                // Normalize before ordering/distribution requirements are derived so
+                // sorting and both window executors use the same partition keys.
+                if has_float_leaf(&expr.data_type(&physical_schema)?) {
+                    Ok(Arc::new(NormalizeFloatZeroExpr::new(expr))
+                        as Arc<dyn PhysicalExpr>)
+                } else {
+                    Ok(expr)
+                }
+            })
+            .collect::<Result<Vec<_>>>()?;
             let order_by = create_physical_sort_exprs(
                 order_by,
                 logical_schema,

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};

datafusion/physical-expr/src/expressions/normalize_float_zero.rs

@@ -0,0 +1,226 @@
+// 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.
+
+//! Canonical floating-point zeros in window partition keys.
+
+use std::hash::Hash;
+use std::sync::Arc;
+
+use arrow::datatypes::{DataType, FieldRef, Schema};
+use arrow::record_batch::RecordBatch;
+use datafusion_common::float_semantics::{normalize_float_scalar, normalize_floats};
+use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::ColumnarValue;
+use datafusion_expr::interval_arithmetic::Interval;
+use datafusion_expr::sort_properties::ExprProperties;
+
+use crate::PhysicalExpr;
+
+/// Replaces negative floating-point zeros with positive zeros, including nested values.
+///
+/// # Public Only for Internal Use
+///
+/// Used by window planning and physical-plan serialization. Normalizing the partition
+/// expression makes sorting, partition boundaries, and cross-batch keys agree without
+/// changing the original input columns.
+#[doc(hidden)]
+#[derive(Debug, Eq)]
+pub struct NormalizeFloatZeroExpr {
+    arg: Arc<dyn PhysicalExpr>,
+}
+
+// Manually implement PartialEq and Hash to work around rust-lang/rust#78808.
+impl PartialEq for NormalizeFloatZeroExpr {
+    fn eq(&self, other: &Self) -> bool {
+        self.arg.eq(&other.arg)
+    }
+}
+
+impl Hash for NormalizeFloatZeroExpr {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.arg.hash(state);
+    }
+}
+
+impl NormalizeFloatZeroExpr {
+    /// Create a normalized window partition key.
+    pub fn new(arg: Arc<dyn PhysicalExpr>) -> Self {
+        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<DataType> {
+        self.arg.data_type(input_schema)
+    }
+
+    fn nullable(&self, input_schema: &Schema) -> Result<bool> {
+        self.arg.nullable(input_schema)
+    }
+
+    fn return_field(&self, input_schema: &Schema) -> Result<FieldRef> {
+        self.arg.return_field(input_schema)
+    }
+
+    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
+        Ok(match self.arg.evaluate(batch)? {
+            ColumnarValue::Array(array) => ColumnarValue::Array(normalize_floats(&array)),
+            ColumnarValue::Scalar(scalar) if scalar.data_type().is_floating() => {
+                ColumnarValue::Scalar(normalize_float_scalar(scalar))
+            }
+            ColumnarValue::Scalar(scalar) => {
+                // Nested scalar values need the recursive array normalization too.
+                let array = normalize_floats(&scalar.to_array()?);
+                ColumnarValue::Scalar(ScalarValue::try_from_array(&array, 0)?)
+            }
+        })
+    }
+
+    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
+        vec![&self.arg]
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        children: Vec<Arc<dyn PhysicalExpr>>,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        Ok(Arc::new(Self::new(Arc::clone(&children[0]))))
+    }
+
+    fn get_properties(&self, children: &[ExprProperties]) -> Result<ExprProperties> {
+        let child = &children[0];
+        if child.range.data_type().is_floating() {
+            Ok(ExprProperties {
+                sort_properties: child.sort_properties,
+                range: Interval::try_new(
+                    normalize_float_scalar(child.range.lower().clone()),
+                    normalize_float_scalar(child.range.upper().clone()),
+                )?,
+                preserves_lex_ordering: true,
+                // Merging signed-zero groups can invalidate suffix sort keys.
+                strictly_order_preserving: false,
+            })
+        } else {
+            // Nested values are not monotone: [-0, 2] < [+0, 1], but [0, 2] > [0, 1].
+            Ok(ExprProperties::new_unknown())
+        }
+    }
+
+    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<Option<datafusion_proto_models::protobuf::PhysicalExprNode>> {
+        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 {
+    /// Reconstruct a normalized partition key from its protobuf representation.
+    pub fn try_from_proto(
+        node: &datafusion_proto_models::protobuf::PhysicalExprNode,
+        ctx: &datafusion_physical_expr_common::physical_expr::proto_decode::PhysicalExprDecodeCtx<'_>,
+    ) -> Result<Arc<dyn PhysicalExpr>> {
+        use datafusion_physical_expr_common::expect_expr_variant;
+        use datafusion_proto_models::protobuf;
+
+        let n = expect_expr_variant!(
+            node,
+            protobuf::physical_expr_node::ExprType::NormalizeFloatZero,
+            "NormalizeFloatZero",
+        );
+        let expr = ctx.decode_required_expression(
+            n.expr.as_deref(),
+            "NormalizeFloatZeroExpr",
+            "expr",
+        )?;
+        Ok(Arc::new(Self::new(expr)))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::expressions::Literal;
+    use arrow::array::{ArrayRef, Float64Array, ListArray};
+    use arrow::datatypes::Float64Type;
+
+    #[test]
+    fn normalize_scalar_and_nested_scalar() -> Result<()> {
+        let batch = RecordBatch::new_empty(Arc::new(Schema::empty()));
+        let nested: ArrayRef =
+            Arc::new(ListArray::from_iter_primitive::<Float64Type, _, _>(vec![
+                Some(vec![Some(-0.0), None, Some(1.0)]),
+            ]));
+        for scalar in [
+            ScalarValue::Float64(Some(-0.0)),
+            ScalarValue::try_from_array(&nested, 0)?,
+        ] {
+            let expected =
+                ScalarValue::try_from_array(&normalize_floats(&scalar.to_array()?), 0)?;
+            let expr = NormalizeFloatZeroExpr::new(Arc::new(Literal::new(scalar)));
+            let ColumnarValue::Scalar(actual) = expr.evaluate(&batch)? else {
+                panic!("scalar input must remain scalar");
+            };
+            assert_eq!(actual, expected);
+        }
+        // The array path must preserve nulls and the original input buffer.
+        let input: ArrayRef = Arc::new(Float64Array::from(vec![Some(-0.0), None]));
+        let schema = Arc::new(Schema::new(vec![arrow::datatypes::Field::new(
+            "a",
+            DataType::Float64,
+            true,
+        )]));
+        let batch = RecordBatch::try_new(schema, vec![Arc::clone(&input)])?;
+        let expr = NormalizeFloatZeroExpr::new(Arc::new(
+            crate::expressions::Column::new("a", 0),
+        ));
+        let actual = expr.evaluate(&batch)?.into_array(2)?;
+        assert_eq!(
+            ScalarValue::try_from_array(&actual, 0)?,
+            ScalarValue::Float64(Some(0.0))
+        );
+        assert!(actual.is_null(1));
+        assert_eq!(
+            ScalarValue::try_from_array(&input, 0)?,
+            ScalarValue::Float64(Some(-0.0))
+        );
+        Ok(())
+    }
+}

datafusion/physical-plan/src/windows/bounded_window_agg_exec.rs

@@ -1660,6 +1660,89 @@ mod tests {
     use itertools::Itertools;
     use tokio::time::timeout;
 
+    #[tokio::test]
+    async fn signed_zero_partition_across_batches() -> Result<()> {
+        use arrow::array::{Float64Array, Int64Array, UInt64Array};
+        use datafusion_physical_expr::expressions::NormalizeFloatZeroExpr;
+
+        let schema = Arc::new(Schema::new(vec![
+            Field::new("a", DataType::Float64, false),
+            Field::new("b", DataType::Int64, false),
+        ]));
+        let key: Arc<dyn PhysicalExpr> =
+            Arc::new(NormalizeFloatZeroExpr::new(col("a", &schema)?));
+        let b = col("b", &schema)?;
+        // Each signed zero starts a separate input batch. The row number must
+        // continue within a partition in all three search modes.
+        let batches = [(0.0, 1), (-0.0, 1), (0.0, 2), (-0.0, 2)]
+            .into_iter()
+            .map(|(a, b)| {
+                Ok(RecordBatch::try_new(
+                    Arc::clone(&schema),
+                    vec![
+                        Arc::new(Float64Array::from(vec![a])),
+                        Arc::new(Int64Array::from(vec![b])),
+                    ],
+                )?)
+            })
+            .collect::<Result<Vec<_>>>()?;
+        for mode in [
+            InputOrderMode::Sorted,
+            InputOrderMode::Linear,
+            InputOrderMode::PartiallySorted(vec![1]),
+        ] {
+            let sort_exprs = match &mode {
+                InputOrderMode::Sorted => vec![
+                    PhysicalSortExpr::new_default(Arc::clone(&key)),
+                    PhysicalSortExpr::new_default(Arc::clone(&b)),
+                ],
+                _ => vec![PhysicalSortExpr::new_default(Arc::clone(&b))],
+            };
+            let input = TestMemoryExec::try_new(
+                std::slice::from_ref(&batches),
+                Arc::clone(&schema),
+                None,
+            )?
+            .try_with_sort_information(vec![LexOrdering::new(sort_exprs).unwrap()])?;
+            let input: Arc<dyn ExecutionPlan> =
+                Arc::new(TestMemoryExec::update_cache(&Arc::new(input)));
+            let window_expr = create_window_expr(
+                &WindowFunctionDefinition::WindowUDF(row_number_udwf()),
+                "rn".to_string(),
+                &[],
+                &[Arc::clone(&key), Arc::clone(&b)],
+                &[],
+                Arc::new(WindowFrame::new(None)),
+                Arc::clone(&schema),
+                false,
+                false,
+                None,
+            )?;
+            let plan: Arc<dyn ExecutionPlan> = Arc::new(BoundedWindowAggExec::try_new(
+                vec![window_expr],
+                input,
+                mode.clone(),
+                false,
+            )?);
+            let output =
+                collect(execute_stream(plan, Arc::new(TaskContext::default()))?).await?;
+            let row_numbers = output
+                .iter()
+                .flat_map(|batch| {
+                    batch
+                        .column(2)
+                        .as_any()
+                        .downcast_ref::<UInt64Array>()
+                        .unwrap()
+                        .values()
+                        .to_vec()
+                })
+                .collect::<Vec<_>>();
+            assert_eq!(row_numbers, vec![1, 2, 1, 2], "{mode:?}");
+        }
+        Ok(())
+    }
+
     #[derive(Debug, Clone)]
     struct TestStreamPartition {
         schema: SchemaRef,

datafusion/proto-models/proto/datafusion.proto

@@ -1117,6 +1117,7 @@ message PhysicalExprNode {
     PhysicalRangeExprNode range_expr = 27;
     PhysicalSqlSimilarToPatternNode sql_similar_to_pattern = 28;
     PhysicalLiteralNode literal_with_metadata = 29;
+    PhysicalNormalizeFloatZeroNode normalize_float_zero = 31;
   }
 }
 
@@ -1125,6 +1126,10 @@ message PhysicalLiteralNode {
   map<string, string> metadata = 2;
 }
 
+message PhysicalNormalizeFloatZeroNode {
+  PhysicalExprNode expr = 1;
+}
+
 message PhysicalDynamicFilterNode {
   repeated PhysicalExprNode children = 1;
   repeated PhysicalExprNode remapped_children = 2;