You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by ad...@apache.org on 2025/08/19 13:28:43 UTC
(datafusion) branch main updated: Support `distinct` and `ignore_nulls` in window expressions (#17235)
This is an automated email from the ASF dual-hosted git repository.
adriangb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 944d8c0f50 Support `distinct` and `ignore_nulls` in window expressions (#17235)
944d8c0f50 is described below
commit 944d8c0f50401d59aedccde8e02ef5dd2cca930f
Author: Peter L <ce...@hotmail.com>
AuthorDate: Tue Aug 19 22:58:40 2025 +0930
Support `distinct` and `ignore_nulls` in window expressions (#17235)
---
datafusion/proto/proto/datafusion.proto | 2 +
datafusion/proto/src/generated/pbjson.rs | 35 +++++++
datafusion/proto/src/generated/prost.rs | 4 +
datafusion/proto/src/physical_plan/from_proto.rs | 4 +-
datafusion/proto/src/physical_plan/to_proto.rs | 88 +++++++++--------
.../proto/tests/cases/roundtrip_physical_plan.rs | 108 +++++++++++++++++++++
6 files changed, 199 insertions(+), 42 deletions(-)
diff --git a/datafusion/proto/proto/datafusion.proto b/datafusion/proto/proto/datafusion.proto
index dab7975243..81d5016810 100644
--- a/datafusion/proto/proto/datafusion.proto
+++ b/datafusion/proto/proto/datafusion.proto
@@ -888,6 +888,8 @@ message PhysicalWindowExprNode {
WindowFrame window_frame = 7;
string name = 8;
optional bytes fun_definition = 9;
+ bool ignore_nulls = 11;
+ bool distinct = 12;
}
message PhysicalIsNull {
diff --git a/datafusion/proto/src/generated/pbjson.rs b/datafusion/proto/src/generated/pbjson.rs
index deb3ab842f..d274d5c518 100644
--- a/datafusion/proto/src/generated/pbjson.rs
+++ b/datafusion/proto/src/generated/pbjson.rs
@@ -17851,6 +17851,12 @@ impl serde::Serialize for PhysicalWindowExprNode {
if self.fun_definition.is_some() {
len += 1;
}
+ if self.ignore_nulls {
+ len += 1;
+ }
+ if self.distinct {
+ len += 1;
+ }
if self.window_function.is_some() {
len += 1;
}
@@ -17875,6 +17881,12 @@ impl serde::Serialize for PhysicalWindowExprNode {
#[allow(clippy::needless_borrows_for_generic_args)]
struct_ser.serialize_field("funDefinition", pbjson::private::base64::encode(&v).as_str())?;
}
+ if self.ignore_nulls {
+ struct_ser.serialize_field("ignoreNulls", &self.ignore_nulls)?;
+ }
+ if self.distinct {
+ struct_ser.serialize_field("distinct", &self.distinct)?;
+ }
if let Some(v) = self.window_function.as_ref() {
match v {
physical_window_expr_node::WindowFunction::UserDefinedAggrFunction(v) => {
@@ -17905,6 +17917,9 @@ impl<'de> serde::Deserialize<'de> for PhysicalWindowExprNode {
"name",
"fun_definition",
"funDefinition",
+ "ignore_nulls",
+ "ignoreNulls",
+ "distinct",
"user_defined_aggr_function",
"userDefinedAggrFunction",
"user_defined_window_function",
@@ -17919,6 +17934,8 @@ impl<'de> serde::Deserialize<'de> for PhysicalWindowExprNode {
WindowFrame,
Name,
FunDefinition,
+ IgnoreNulls,
+ Distinct,
UserDefinedAggrFunction,
UserDefinedWindowFunction,
}
@@ -17948,6 +17965,8 @@ impl<'de> serde::Deserialize<'de> for PhysicalWindowExprNode {
"windowFrame" | "window_frame" => Ok(GeneratedField::WindowFrame),
"name" => Ok(GeneratedField::Name),
"funDefinition" | "fun_definition" => Ok(GeneratedField::FunDefinition),
+ "ignoreNulls" | "ignore_nulls" => Ok(GeneratedField::IgnoreNulls),
+ "distinct" => Ok(GeneratedField::Distinct),
"userDefinedAggrFunction" | "user_defined_aggr_function" => Ok(GeneratedField::UserDefinedAggrFunction),
"userDefinedWindowFunction" | "user_defined_window_function" => Ok(GeneratedField::UserDefinedWindowFunction),
_ => Err(serde::de::Error::unknown_field(value, FIELDS)),
@@ -17975,6 +17994,8 @@ impl<'de> serde::Deserialize<'de> for PhysicalWindowExprNode {
let mut window_frame__ = None;
let mut name__ = None;
let mut fun_definition__ = None;
+ let mut ignore_nulls__ = None;
+ let mut distinct__ = None;
let mut window_function__ = None;
while let Some(k) = map_.next_key()? {
match k {
@@ -18016,6 +18037,18 @@ impl<'de> serde::Deserialize<'de> for PhysicalWindowExprNode {
map_.next_value::<::std::option::Option<::pbjson::private::BytesDeserialize<_>>>()?.map(|x| x.0)
;
}
+ GeneratedField::IgnoreNulls => {
+ if ignore_nulls__.is_some() {
+ return Err(serde::de::Error::duplicate_field("ignoreNulls"));
+ }
+ ignore_nulls__ = Some(map_.next_value()?);
+ }
+ GeneratedField::Distinct => {
+ if distinct__.is_some() {
+ return Err(serde::de::Error::duplicate_field("distinct"));
+ }
+ distinct__ = Some(map_.next_value()?);
+ }
GeneratedField::UserDefinedAggrFunction => {
if window_function__.is_some() {
return Err(serde::de::Error::duplicate_field("userDefinedAggrFunction"));
@@ -18037,6 +18070,8 @@ impl<'de> serde::Deserialize<'de> for PhysicalWindowExprNode {
window_frame: window_frame__,
name: name__.unwrap_or_default(),
fun_definition: fun_definition__,
+ ignore_nulls: ignore_nulls__.unwrap_or_default(),
+ distinct: distinct__.unwrap_or_default(),
window_function: window_function__,
})
}
diff --git a/datafusion/proto/src/generated/prost.rs b/datafusion/proto/src/generated/prost.rs
index ef5a3b4bb9..8118adf323 100644
--- a/datafusion/proto/src/generated/prost.rs
+++ b/datafusion/proto/src/generated/prost.rs
@@ -1356,6 +1356,10 @@ pub struct PhysicalWindowExprNode {
pub name: ::prost::alloc::string::String,
#[prost(bytes = "vec", optional, tag = "9")]
pub fun_definition: ::core::option::Option<::prost::alloc::vec::Vec<u8>>,
+ #[prost(bool, tag = "11")]
+ pub ignore_nulls: bool,
+ #[prost(bool, tag = "12")]
+ pub distinct: bool,
#[prost(oneof = "physical_window_expr_node::WindowFunction", tags = "3, 10")]
pub window_function: ::core::option::Option<
physical_window_expr_node::WindowFunction,
diff --git a/datafusion/proto/src/physical_plan/from_proto.rs b/datafusion/proto/src/physical_plan/from_proto.rs
index 53f4b2bd84..39ad52d46a 100644
--- a/datafusion/proto/src/physical_plan/from_proto.rs
+++ b/datafusion/proto/src/physical_plan/from_proto.rs
@@ -176,8 +176,8 @@ pub fn parse_physical_window_expr(
&order_by,
Arc::new(window_frame),
&extended_schema,
- false,
- false,
+ proto.ignore_nulls,
+ proto.distinct,
)
}
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs b/datafusion/proto/src/physical_plan/to_proto.rs
index 85ced4933a..64960e39f7 100644
--- a/datafusion/proto/src/physical_plan/to_proto.rs
+++ b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -81,12 +81,7 @@ fn serialize_physical_window_aggr_expr(
_window_frame: &WindowFrame,
codec: &dyn PhysicalExtensionCodec,
) -> Result<(physical_window_expr_node::WindowFunction, Option<Vec<u8>>)> {
- if aggr_expr.is_distinct() || aggr_expr.ignore_nulls() {
- // TODO
- return not_impl_err!(
- "Distinct aggregate functions not supported in window expressions"
- );
- }
+ // Distinct and ignore_nulls are now supported in window expressions
let mut buf = Vec::new();
codec.try_encode_udaf(aggr_expr.fun(), &mut buf)?;
@@ -106,44 +101,55 @@ pub fn serialize_physical_window_expr(
let args = window_expr.expressions().to_vec();
let window_frame = window_expr.get_window_frame();
- let (window_function, fun_definition) = if let Some(plain_aggr_window_expr) =
- expr.downcast_ref::<PlainAggregateWindowExpr>()
- {
- serialize_physical_window_aggr_expr(
- plain_aggr_window_expr.get_aggregate_expr(),
- window_frame,
- codec,
- )?
- } else if let Some(sliding_aggr_window_expr) =
- expr.downcast_ref::<SlidingAggregateWindowExpr>()
- {
- serialize_physical_window_aggr_expr(
- sliding_aggr_window_expr.get_aggregate_expr(),
- window_frame,
- codec,
- )?
- } else if let Some(udf_window_expr) = expr.downcast_ref::<StandardWindowExpr>() {
- if let Some(expr) = udf_window_expr
- .get_standard_func_expr()
- .as_any()
- .downcast_ref::<WindowUDFExpr>()
+ let (window_function, fun_definition, ignore_nulls, distinct) =
+ if let Some(plain_aggr_window_expr) =
+ expr.downcast_ref::<PlainAggregateWindowExpr>()
{
- let mut buf = Vec::new();
- codec.try_encode_udwf(expr.fun(), &mut buf)?;
+ let aggr_expr = plain_aggr_window_expr.get_aggregate_expr();
+ let (window_function, fun_definition) =
+ serialize_physical_window_aggr_expr(aggr_expr, window_frame, codec)?;
(
- physical_window_expr_node::WindowFunction::UserDefinedWindowFunction(
- expr.fun().name().to_string(),
- ),
- (!buf.is_empty()).then_some(buf),
+ window_function,
+ fun_definition,
+ aggr_expr.ignore_nulls(),
+ aggr_expr.is_distinct(),
+ )
+ } else if let Some(sliding_aggr_window_expr) =
+ expr.downcast_ref::<SlidingAggregateWindowExpr>()
+ {
+ let aggr_expr = sliding_aggr_window_expr.get_aggregate_expr();
+ let (window_function, fun_definition) =
+ serialize_physical_window_aggr_expr(aggr_expr, window_frame, codec)?;
+ (
+ window_function,
+ fun_definition,
+ aggr_expr.ignore_nulls(),
+ aggr_expr.is_distinct(),
)
+ } else if let Some(udf_window_expr) = expr.downcast_ref::<StandardWindowExpr>() {
+ if let Some(expr) = udf_window_expr
+ .get_standard_func_expr()
+ .as_any()
+ .downcast_ref::<WindowUDFExpr>()
+ {
+ let mut buf = Vec::new();
+ codec.try_encode_udwf(expr.fun(), &mut buf)?;
+ (
+ physical_window_expr_node::WindowFunction::UserDefinedWindowFunction(
+ expr.fun().name().to_string(),
+ ),
+ (!buf.is_empty()).then_some(buf),
+ false, // WindowUDFExpr doesn't have ignore_nulls/distinct
+ false,
+ )
+ } else {
+ return not_impl_err!(
+ "User-defined window function not supported: {window_expr:?}"
+ );
+ }
} else {
- return not_impl_err!(
- "User-defined window function not supported: {window_expr:?}"
- );
- }
- } else {
- return not_impl_err!("WindowExpr not supported: {window_expr:?}");
- };
+ return not_impl_err!("WindowExpr not supported: {window_expr:?}");
+ };
let args = serialize_physical_exprs(&args, codec)?;
let partition_by = serialize_physical_exprs(window_expr.partition_by(), codec)?;
@@ -161,6 +167,8 @@ pub fn serialize_physical_window_expr(
window_function: Some(window_function),
name: window_expr.name().to_string(),
fun_definition,
+ ignore_nulls,
+ distinct,
})
}
diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
index 1b48dd2a4b..a554489d31 100644
--- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs
@@ -54,6 +54,7 @@ use datafusion::datasource::physical_plan::{
use datafusion::datasource::sink::DataSinkExec;
use datafusion::datasource::source::DataSourceExec;
use datafusion::execution::FunctionRegistry;
+use datafusion::functions_aggregate::count::count_udaf;
use datafusion::functions_aggregate::sum::sum_udaf;
use datafusion::functions_window::nth_value::nth_value_udwf;
use datafusion::functions_window::row_number::row_number_udwf;
@@ -418,6 +419,113 @@ fn roundtrip_window() -> Result<()> {
)?))
}
+#[test]
+fn roundtrip_window_distinct() -> Result<()> {
+ let field_a = Field::new("a", DataType::Int64, false);
+ let field_b = Field::new("b", DataType::Int64, false);
+ let schema = Arc::new(Schema::new(vec![field_a, field_b]));
+
+ // Create a distinct count window expression with unbounded frame (becomes PlainAggregateWindowExpr)
+ let distinct_count_expr = Arc::new(PlainAggregateWindowExpr::new(
+ AggregateExprBuilder::new(count_udaf(), vec![col("a", &schema)?])
+ .schema(Arc::clone(&schema))
+ .alias("count(DISTINCT a)")
+ .distinct() // Enable distinct
+ .build()
+ .map(Arc::new)?,
+ &[col("b", &schema)?], // partition by b
+ &[], // no order by
+ Arc::new(WindowFrame::new(None)), // unbounded frame
+ ));
+
+ // Create a distinct sum window expression with bounded frame (becomes SlidingAggregateWindowExpr)
+ let bounded_frame = WindowFrame::new_bounds(
+ datafusion_expr::WindowFrameUnits::Rows,
+ WindowFrameBound::Preceding(ScalarValue::UInt64(Some(1))),
+ WindowFrameBound::CurrentRow,
+ );
+
+ let distinct_sum_expr = Arc::new(SlidingAggregateWindowExpr::new(
+ AggregateExprBuilder::new(
+ sum_udaf(),
+ vec![cast(col("a", &schema)?, &schema, DataType::Float64)?],
+ )
+ .schema(Arc::clone(&schema))
+ .alias("sum(DISTINCT a)")
+ .distinct() // Enable distinct
+ .with_ignore_nulls(true) // Enable ignore nulls
+ .build()
+ .map(Arc::new)?,
+ &[], // no partition by
+ &[], // no order by
+ Arc::new(bounded_frame), // bounded frame
+ ));
+
+ let input = Arc::new(EmptyExec::new(schema.clone()));
+
+ roundtrip_test(Arc::new(WindowAggExec::try_new(
+ vec![distinct_count_expr, distinct_sum_expr],
+ input,
+ false,
+ )?))
+}
+
+#[test]
+fn test_distinct_window_serialization_end_to_end() -> Result<()> {
+ // Create a more comprehensive test that verifies distinct window functions
+ // work properly through the entire serialization/deserialization pipeline
+ let field_a = Field::new("a", DataType::Int64, false);
+ let field_b = Field::new("b", DataType::Int64, false);
+ let schema = Arc::new(Schema::new(vec![field_a, field_b]));
+
+ // Test 1: DISTINCT COUNT with IGNORE NULLS
+ let distinct_count_ignore_nulls = Arc::new(PlainAggregateWindowExpr::new(
+ AggregateExprBuilder::new(count_udaf(), vec![col("a", &schema)?])
+ .schema(Arc::clone(&schema))
+ .alias("count_distinct_ignore_nulls")
+ .distinct()
+ .with_ignore_nulls(true)
+ .build()
+ .map(Arc::new)?,
+ &[col("b", &schema)?],
+ &[],
+ Arc::new(WindowFrame::new(None)),
+ ));
+
+ // Test 2: DISTINCT SUM (without ignore nulls)
+ let bounded_frame = WindowFrame::new_bounds(
+ datafusion_expr::WindowFrameUnits::Rows,
+ WindowFrameBound::Preceding(ScalarValue::UInt64(Some(2))),
+ WindowFrameBound::CurrentRow,
+ );
+
+ let distinct_sum = Arc::new(SlidingAggregateWindowExpr::new(
+ AggregateExprBuilder::new(
+ sum_udaf(),
+ vec![cast(col("a", &schema)?, &schema, DataType::Float64)?],
+ )
+ .schema(Arc::clone(&schema))
+ .alias("sum_distinct")
+ .distinct()
+ .build()
+ .map(Arc::new)?,
+ &[],
+ &[],
+ Arc::new(bounded_frame),
+ ));
+
+ let input = Arc::new(EmptyExec::new(schema.clone()));
+
+ let window_exec = Arc::new(WindowAggExec::try_new(
+ vec![distinct_count_ignore_nulls, distinct_sum],
+ input,
+ false,
+ )?);
+
+ // Perform the roundtrip test
+ roundtrip_test(window_exec)
+}
+
#[test]
fn rountrip_aggregate() -> Result<()> {
let field_a = Field::new("a", DataType::Int64, false);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org