This is an automated email from the ASF dual-hosted git repository.
alamb 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 a4e6b075b8 feat: Convert CumeDist to UDWF (#13051)
a4e6b075b8 is described below
commit a4e6b075b8fe448e0f1fde948626d7c43c1d9c4a
Author: Jonathan Chen <[email protected]>
AuthorDate: Wed Oct 23 07:00:20 2024 -0400
feat: Convert CumeDist to UDWF (#13051)
* Transferred cumedist
* fixes
* remove expr tests
* small fix
* small fix
* check
* clippy fix
* roundtrip fix
---
datafusion/expr/src/built_in_window_function.rs | 6 -
datafusion/expr/src/expr.rs | 23 +--
datafusion/expr/src/window_function.rs | 5 -
datafusion/functions-window/src/cume_dist.rs | 170 +++++++++++++++++++++
datafusion/functions-window/src/lib.rs | 3 +
datafusion/physical-expr/src/expressions/mod.rs | 1 -
datafusion/physical-expr/src/window/cume_dist.rs | 145 ------------------
datafusion/physical-expr/src/window/mod.rs | 1 -
datafusion/physical-plan/src/windows/mod.rs | 3 +-
datafusion/proto/proto/datafusion.proto | 2 +-
datafusion/proto/src/generated/pbjson.rs | 3 -
datafusion/proto/src/generated/prost.rs | 4 +-
datafusion/proto/src/logical_plan/from_proto.rs | 1 -
datafusion/proto/src/logical_plan/to_proto.rs | 1 -
datafusion/proto/src/physical_plan/to_proto.rs | 10 +-
.../proto/tests/cases/roundtrip_logical_plan.rs | 3 +-
datafusion/sqllogictest/test_files/window.slt | 4 +-
docs/source/user-guide/sql/window_functions_new.md | 9 ++
18 files changed, 195 insertions(+), 199 deletions(-)
diff --git a/datafusion/expr/src/built_in_window_function.rs
b/datafusion/expr/src/built_in_window_function.rs
index 2c70a07a4e..36916a6b59 100644
--- a/datafusion/expr/src/built_in_window_function.rs
+++ b/datafusion/expr/src/built_in_window_function.rs
@@ -40,8 +40,6 @@ impl fmt::Display for BuiltInWindowFunction {
/// [Window Function]: https://en.wikipedia.org/wiki/Window_function_(SQL)
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Hash, EnumIter)]
pub enum BuiltInWindowFunction {
- /// Relative rank of the current row: (number of rows preceding or peer
with current row) / (total rows)
- CumeDist,
/// Integer ranging from 1 to the argument value, dividing the partition
as equally as possible
Ntile,
/// returns value evaluated at the row that is the first row of the window
frame
@@ -56,7 +54,6 @@ impl BuiltInWindowFunction {
pub fn name(&self) -> &str {
use BuiltInWindowFunction::*;
match self {
- CumeDist => "CUME_DIST",
Ntile => "NTILE",
FirstValue => "first_value",
LastValue => "last_value",
@@ -69,7 +66,6 @@ impl FromStr for BuiltInWindowFunction {
type Err = DataFusionError;
fn from_str(name: &str) -> Result<BuiltInWindowFunction> {
Ok(match name.to_uppercase().as_str() {
- "CUME_DIST" => BuiltInWindowFunction::CumeDist,
"NTILE" => BuiltInWindowFunction::Ntile,
"FIRST_VALUE" => BuiltInWindowFunction::FirstValue,
"LAST_VALUE" => BuiltInWindowFunction::LastValue,
@@ -102,7 +98,6 @@ impl BuiltInWindowFunction {
match self {
BuiltInWindowFunction::Ntile => Ok(DataType::UInt64),
- BuiltInWindowFunction::CumeDist => Ok(DataType::Float64),
BuiltInWindowFunction::FirstValue
| BuiltInWindowFunction::LastValue
| BuiltInWindowFunction::NthValue =>
Ok(input_expr_types[0].clone()),
@@ -113,7 +108,6 @@ impl BuiltInWindowFunction {
pub fn signature(&self) -> Signature {
// Note: The physical expression must accept the type returned by this
function or the execution panics.
match self {
- BuiltInWindowFunction::CumeDist => Signature::any(0,
Volatility::Immutable),
BuiltInWindowFunction::FirstValue |
BuiltInWindowFunction::LastValue => {
Signature::any(1, Volatility::Immutable)
}
diff --git a/datafusion/expr/src/expr.rs b/datafusion/expr/src/expr.rs
index 691b65d344..7fadf6391b 100644
--- a/datafusion/expr/src/expr.rs
+++ b/datafusion/expr/src/expr.rs
@@ -2567,15 +2567,6 @@ mod test {
Ok(())
}
- #[test]
- fn test_cume_dist_return_type() -> Result<()> {
- let fun = find_df_window_func("cume_dist").unwrap();
- let observed = fun.return_type(&[], &[], "")?;
- assert_eq!(DataType::Float64, observed);
-
- Ok(())
- }
-
#[test]
fn test_ntile_return_type() -> Result<()> {
let fun = find_df_window_func("ntile").unwrap();
@@ -2587,13 +2578,7 @@ mod test {
#[test]
fn test_window_function_case_insensitive() -> Result<()> {
- let names = vec![
- "cume_dist",
- "ntile",
- "first_value",
- "last_value",
- "nth_value",
- ];
+ let names = vec!["ntile", "first_value", "last_value", "nth_value"];
for name in names {
let fun = find_df_window_func(name).unwrap();
let fun2 =
find_df_window_func(name.to_uppercase().as_str()).unwrap();
@@ -2609,12 +2594,6 @@ mod test {
#[test]
fn test_find_df_window_function() {
- assert_eq!(
- find_df_window_func("cume_dist"),
- Some(WindowFunctionDefinition::BuiltInWindowFunction(
- built_in_window_function::BuiltInWindowFunction::CumeDist
- ))
- );
assert_eq!(
find_df_window_func("first_value"),
Some(WindowFunctionDefinition::BuiltInWindowFunction(
diff --git a/datafusion/expr/src/window_function.rs
b/datafusion/expr/src/window_function.rs
index 3e1870c59c..c13a028e4a 100644
--- a/datafusion/expr/src/window_function.rs
+++ b/datafusion/expr/src/window_function.rs
@@ -17,11 +17,6 @@
use crate::{expr::WindowFunction, BuiltInWindowFunction, Expr, Literal};
-/// Create an expression to represent the `cume_dist` window function
-pub fn cume_dist() -> Expr {
- Expr::WindowFunction(WindowFunction::new(BuiltInWindowFunction::CumeDist,
vec![]))
-}
-
/// Create an expression to represent the `ntile` window function
pub fn ntile(arg: Expr) -> Expr {
Expr::WindowFunction(WindowFunction::new(BuiltInWindowFunction::Ntile,
vec![arg]))
diff --git a/datafusion/functions-window/src/cume_dist.rs
b/datafusion/functions-window/src/cume_dist.rs
new file mode 100644
index 0000000000..9e30c672fe
--- /dev/null
+++ b/datafusion/functions-window/src/cume_dist.rs
@@ -0,0 +1,170 @@
+// 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.
+
+//! `cume_dist` window function implementation
+
+use datafusion_common::arrow::array::{ArrayRef, Float64Array};
+use datafusion_common::arrow::datatypes::DataType;
+use datafusion_common::arrow::datatypes::Field;
+use datafusion_common::Result;
+use datafusion_expr::window_doc_sections::DOC_SECTION_RANKING;
+use datafusion_expr::{
+ Documentation, PartitionEvaluator, Signature, Volatility, WindowUDFImpl,
+};
+use datafusion_functions_window_common::field;
+use datafusion_functions_window_common::partition::PartitionEvaluatorArgs;
+use field::WindowUDFFieldArgs;
+use std::any::Any;
+use std::fmt::Debug;
+use std::iter;
+use std::ops::Range;
+use std::sync::{Arc, OnceLock};
+
+define_udwf_and_expr!(
+ CumeDist,
+ cume_dist,
+ "Calculates the cumulative distribution of a value in a group of values."
+);
+
+/// CumeDist calculates the cume_dist in the window function with order by
+#[derive(Debug)]
+pub struct CumeDist {
+ signature: Signature,
+}
+
+impl CumeDist {
+ pub fn new() -> Self {
+ Self {
+ signature: Signature::any(0, Volatility::Immutable),
+ }
+ }
+}
+
+impl Default for CumeDist {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl WindowUDFImpl for CumeDist {
+ /// Return a reference to Any that can be used for downcasting
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn name(&self) -> &str {
+ "cume_dist"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn partition_evaluator(
+ &self,
+ _partition_evaluator_args: PartitionEvaluatorArgs,
+ ) -> Result<Box<dyn PartitionEvaluator>> {
+ Ok(Box::<CumeDistEvaluator>::default())
+ }
+
+ fn field(&self, field_args: WindowUDFFieldArgs) -> Result<Field> {
+ Ok(Field::new(field_args.name(), DataType::Float64, false))
+ }
+
+ fn documentation(&self) -> Option<&Documentation> {
+ Some(get_cume_dist_doc())
+ }
+}
+
+static DOCUMENTATION: OnceLock<Documentation> = OnceLock::new();
+
+fn get_cume_dist_doc() -> &'static Documentation {
+ DOCUMENTATION.get_or_init(|| {
+ Documentation::builder()
+ .with_doc_section(DOC_SECTION_RANKING)
+ .with_description(
+ "Relative rank of the current row: (number of rows preceding
or peer with current row) / (total rows).",
+ )
+ .with_syntax_example("cume_dist()")
+ .build()
+ .unwrap()
+ })
+}
+
+#[derive(Debug, Default)]
+pub(crate) struct CumeDistEvaluator;
+
+impl PartitionEvaluator for CumeDistEvaluator {
+ /// Computes the cumulative distribution for all rows in the partition
+ fn evaluate_all_with_rank(
+ &self,
+ num_rows: usize,
+ ranks_in_partition: &[Range<usize>],
+ ) -> Result<ArrayRef> {
+ let scalar = num_rows as f64;
+ let result = Float64Array::from_iter_values(
+ ranks_in_partition
+ .iter()
+ .scan(0_u64, |acc, range| {
+ let len = range.end - range.start;
+ *acc += len as u64;
+ let value: f64 = (*acc as f64) / scalar;
+ let result = iter::repeat(value).take(len);
+ Some(result)
+ })
+ .flatten(),
+ );
+ Ok(Arc::new(result))
+ }
+
+ fn include_rank(&self) -> bool {
+ true
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use datafusion_common::cast::as_float64_array;
+
+ fn test_f64_result(
+ num_rows: usize,
+ ranks: Vec<Range<usize>>,
+ expected: Vec<f64>,
+ ) -> Result<()> {
+ let evaluator = CumeDistEvaluator;
+ let result = evaluator.evaluate_all_with_rank(num_rows, &ranks)?;
+ let result = as_float64_array(&result)?;
+ let result = result.values().to_vec();
+ assert_eq!(expected, result);
+ Ok(())
+ }
+
+ #[test]
+ #[allow(clippy::single_range_in_vec_init)]
+ fn test_cume_dist() -> Result<()> {
+ test_f64_result(0, vec![], vec![])?;
+
+ test_f64_result(1, vec![0..1], vec![1.0])?;
+
+ test_f64_result(2, vec![0..2], vec![1.0, 1.0])?;
+
+ test_f64_result(4, vec![0..2, 2..4], vec![0.5, 0.5, 1.0, 1.0])?;
+
+ Ok(())
+ }
+}
diff --git a/datafusion/functions-window/src/lib.rs
b/datafusion/functions-window/src/lib.rs
index 5a2aafa289..13a77977d5 100644
--- a/datafusion/functions-window/src/lib.rs
+++ b/datafusion/functions-window/src/lib.rs
@@ -32,6 +32,7 @@ use datafusion_expr::WindowUDF;
#[macro_use]
pub mod macros;
+pub mod cume_dist;
pub mod lead_lag;
pub mod rank;
@@ -40,6 +41,7 @@ mod utils;
/// Fluent-style API for creating `Expr`s
pub mod expr_fn {
+ pub use super::cume_dist::cume_dist;
pub use super::lead_lag::lag;
pub use super::lead_lag::lead;
pub use super::rank::{dense_rank, percent_rank, rank};
@@ -49,6 +51,7 @@ pub mod expr_fn {
/// Returns all default window functions
pub fn all_default_window_functions() -> Vec<Arc<WindowUDF>> {
vec![
+ cume_dist::cume_dist_udwf(),
row_number::row_number_udwf(),
lead_lag::lead_udwf(),
lead_lag::lag_udwf(),
diff --git a/datafusion/physical-expr/src/expressions/mod.rs
b/datafusion/physical-expr/src/expressions/mod.rs
index 54b8aafdb4..63047f6929 100644
--- a/datafusion/physical-expr/src/expressions/mod.rs
+++ b/datafusion/physical-expr/src/expressions/mod.rs
@@ -35,7 +35,6 @@ mod unknown_column;
/// Module with some convenient methods used in expression building
pub use crate::aggregate::stats::StatsType;
-pub use crate::window::cume_dist::{cume_dist, CumeDist};
pub use crate::window::nth_value::NthValue;
pub use crate::window::ntile::Ntile;
pub use crate::PhysicalSortExpr;
diff --git a/datafusion/physical-expr/src/window/cume_dist.rs
b/datafusion/physical-expr/src/window/cume_dist.rs
deleted file mode 100644
index 9720187ea8..0000000000
--- a/datafusion/physical-expr/src/window/cume_dist.rs
+++ /dev/null
@@ -1,145 +0,0 @@
-// 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.
-
-//! Defines physical expression for `cume_dist` that can evaluated
-//! at runtime during query execution
-
-use crate::window::BuiltInWindowFunctionExpr;
-use crate::PhysicalExpr;
-use arrow::array::ArrayRef;
-use arrow::array::Float64Array;
-use arrow::datatypes::{DataType, Field};
-use datafusion_common::Result;
-use datafusion_expr::PartitionEvaluator;
-use std::any::Any;
-use std::iter;
-use std::ops::Range;
-use std::sync::Arc;
-
-/// CumeDist calculates the cume_dist in the window function with order by
-#[derive(Debug)]
-pub struct CumeDist {
- name: String,
- /// Output data type
- data_type: DataType,
-}
-
-/// Create a cume_dist window function
-pub fn cume_dist(name: String, data_type: &DataType) -> CumeDist {
- CumeDist {
- name,
- data_type: data_type.clone(),
- }
-}
-
-impl BuiltInWindowFunctionExpr for CumeDist {
- /// Return a reference to Any that can be used for downcasting
- fn as_any(&self) -> &dyn Any {
- self
- }
-
- fn field(&self) -> Result<Field> {
- let nullable = false;
- Ok(Field::new(self.name(), self.data_type.clone(), nullable))
- }
-
- fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
- vec![]
- }
-
- fn name(&self) -> &str {
- &self.name
- }
-
- fn create_evaluator(&self) -> Result<Box<dyn PartitionEvaluator>> {
- Ok(Box::new(CumeDistEvaluator {}))
- }
-}
-
-#[derive(Debug)]
-pub(crate) struct CumeDistEvaluator;
-
-impl PartitionEvaluator for CumeDistEvaluator {
- fn evaluate_all_with_rank(
- &self,
- num_rows: usize,
- ranks_in_partition: &[Range<usize>],
- ) -> Result<ArrayRef> {
- let scalar = num_rows as f64;
- let result = Float64Array::from_iter_values(
- ranks_in_partition
- .iter()
- .scan(0_u64, |acc, range| {
- let len = range.end - range.start;
- *acc += len as u64;
- let value: f64 = (*acc as f64) / scalar;
- let result = iter::repeat(value).take(len);
- Some(result)
- })
- .flatten(),
- );
- Ok(Arc::new(result))
- }
-
- fn include_rank(&self) -> bool {
- true
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use datafusion_common::cast::as_float64_array;
-
- fn test_i32_result(
- expr: &CumeDist,
- num_rows: usize,
- ranks: Vec<Range<usize>>,
- expected: Vec<f64>,
- ) -> Result<()> {
- let result = expr
- .create_evaluator()?
- .evaluate_all_with_rank(num_rows, &ranks)?;
- let result = as_float64_array(&result)?;
- let result = result.values();
- assert_eq!(expected, *result);
- Ok(())
- }
-
- #[test]
- #[allow(clippy::single_range_in_vec_init)]
- fn test_cume_dist() -> Result<()> {
- let r = cume_dist("arr".into(), &DataType::Float64);
-
- let expected = vec![0.0; 0];
- test_i32_result(&r, 0, vec![], expected)?;
-
- let expected = vec![1.0; 1];
- test_i32_result(&r, 1, vec![0..1], expected)?;
-
- let expected = vec![1.0; 2];
- test_i32_result(&r, 2, vec![0..2], expected)?;
-
- let expected = vec![0.5, 0.5, 1.0, 1.0];
- test_i32_result(&r, 4, vec![0..2, 2..4], expected)?;
-
- let expected = vec![0.25, 0.5, 0.75, 1.0];
- test_i32_result(&r, 4, vec![0..1, 1..2, 2..3, 3..4], expected)?;
-
- Ok(())
- }
-}
diff --git a/datafusion/physical-expr/src/window/mod.rs
b/datafusion/physical-expr/src/window/mod.rs
index c0fe3c2933..7bab4dbc5a 100644
--- a/datafusion/physical-expr/src/window/mod.rs
+++ b/datafusion/physical-expr/src/window/mod.rs
@@ -18,7 +18,6 @@
mod aggregate;
mod built_in;
mod built_in_window_function_expr;
-pub(crate) mod cume_dist;
pub(crate) mod nth_value;
pub(crate) mod ntile;
mod sliding_aggregate;
diff --git a/datafusion/physical-plan/src/windows/mod.rs
b/datafusion/physical-plan/src/windows/mod.rs
index f6902fcbe2..39ff71496e 100644
--- a/datafusion/physical-plan/src/windows/mod.rs
+++ b/datafusion/physical-plan/src/windows/mod.rs
@@ -21,7 +21,7 @@ use std::borrow::Borrow;
use std::sync::Arc;
use crate::{
- expressions::{cume_dist, Literal, NthValue, Ntile, PhysicalSortExpr},
+ expressions::{Literal, NthValue, Ntile, PhysicalSortExpr},
ExecutionPlan, ExecutionPlanProperties, InputOrderMode, PhysicalExpr,
};
@@ -219,7 +219,6 @@ fn create_built_in_window_expr(
let out_data_type: &DataType =
input_schema.field_with_name(&name)?.data_type();
Ok(match fun {
- BuiltInWindowFunction::CumeDist => Arc::new(cume_dist(name,
out_data_type)),
BuiltInWindowFunction::Ntile => {
let n = get_scalar_value_from_args(args, 0)?.ok_or_else(|| {
DataFusionError::Execution(
diff --git a/datafusion/proto/proto/datafusion.proto
b/datafusion/proto/proto/datafusion.proto
index a15fa2c5f9..c92328278e 100644
--- a/datafusion/proto/proto/datafusion.proto
+++ b/datafusion/proto/proto/datafusion.proto
@@ -511,7 +511,7 @@ enum BuiltInWindowFunction {
// RANK = 1;
// DENSE_RANK = 2;
// PERCENT_RANK = 3;
- CUME_DIST = 4;
+ // CUME_DIST = 4;
NTILE = 5;
// LAG = 6;
// LEAD = 7;
diff --git a/datafusion/proto/src/generated/pbjson.rs
b/datafusion/proto/src/generated/pbjson.rs
index d223e3646b..ca331cdaa5 100644
--- a/datafusion/proto/src/generated/pbjson.rs
+++ b/datafusion/proto/src/generated/pbjson.rs
@@ -1662,7 +1662,6 @@ impl serde::Serialize for BuiltInWindowFunction {
{
let variant = match self {
Self::Unspecified => "UNSPECIFIED",
- Self::CumeDist => "CUME_DIST",
Self::Ntile => "NTILE",
Self::FirstValue => "FIRST_VALUE",
Self::LastValue => "LAST_VALUE",
@@ -1679,7 +1678,6 @@ impl<'de> serde::Deserialize<'de> for
BuiltInWindowFunction {
{
const FIELDS: &[&str] = &[
"UNSPECIFIED",
- "CUME_DIST",
"NTILE",
"FIRST_VALUE",
"LAST_VALUE",
@@ -1725,7 +1723,6 @@ impl<'de> serde::Deserialize<'de> for
BuiltInWindowFunction {
{
match value {
"UNSPECIFIED" => Ok(BuiltInWindowFunction::Unspecified),
- "CUME_DIST" => Ok(BuiltInWindowFunction::CumeDist),
"NTILE" => Ok(BuiltInWindowFunction::Ntile),
"FIRST_VALUE" => Ok(BuiltInWindowFunction::FirstValue),
"LAST_VALUE" => Ok(BuiltInWindowFunction::LastValue),
diff --git a/datafusion/proto/src/generated/prost.rs
b/datafusion/proto/src/generated/prost.rs
index 6b234be57a..fb0b3bcb2c 100644
--- a/datafusion/proto/src/generated/prost.rs
+++ b/datafusion/proto/src/generated/prost.rs
@@ -1818,7 +1818,7 @@ pub enum BuiltInWindowFunction {
/// RANK = 1;
/// DENSE_RANK = 2;
/// PERCENT_RANK = 3;
- CumeDist = 4,
+ /// CUME_DIST = 4;
Ntile = 5,
/// LAG = 6;
/// LEAD = 7;
@@ -1834,7 +1834,6 @@ impl BuiltInWindowFunction {
pub fn as_str_name(&self) -> &'static str {
match self {
Self::Unspecified => "UNSPECIFIED",
- Self::CumeDist => "CUME_DIST",
Self::Ntile => "NTILE",
Self::FirstValue => "FIRST_VALUE",
Self::LastValue => "LAST_VALUE",
@@ -1845,7 +1844,6 @@ impl BuiltInWindowFunction {
pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
match value {
"UNSPECIFIED" => Some(Self::Unspecified),
- "CUME_DIST" => Some(Self::CumeDist),
"NTILE" => Some(Self::Ntile),
"FIRST_VALUE" => Some(Self::FirstValue),
"LAST_VALUE" => Some(Self::LastValue),
diff --git a/datafusion/proto/src/logical_plan/from_proto.rs
b/datafusion/proto/src/logical_plan/from_proto.rs
index 99b11939e9..4587c090c9 100644
--- a/datafusion/proto/src/logical_plan/from_proto.rs
+++ b/datafusion/proto/src/logical_plan/from_proto.rs
@@ -152,7 +152,6 @@ impl From<protobuf::BuiltInWindowFunction> for
BuiltInWindowFunction {
match built_in_function {
protobuf::BuiltInWindowFunction::Unspecified => todo!(),
protobuf::BuiltInWindowFunction::FirstValue => Self::FirstValue,
- protobuf::BuiltInWindowFunction::CumeDist => Self::CumeDist,
protobuf::BuiltInWindowFunction::Ntile => Self::Ntile,
protobuf::BuiltInWindowFunction::NthValue => Self::NthValue,
protobuf::BuiltInWindowFunction::LastValue => Self::LastValue,
diff --git a/datafusion/proto/src/logical_plan/to_proto.rs
b/datafusion/proto/src/logical_plan/to_proto.rs
index a34a220e49..dce0cd741f 100644
--- a/datafusion/proto/src/logical_plan/to_proto.rs
+++ b/datafusion/proto/src/logical_plan/to_proto.rs
@@ -128,7 +128,6 @@ impl From<&BuiltInWindowFunction> for
protobuf::BuiltInWindowFunction {
BuiltInWindowFunction::LastValue => Self::LastValue,
BuiltInWindowFunction::NthValue => Self::NthValue,
BuiltInWindowFunction::Ntile => Self::Ntile,
- BuiltInWindowFunction::CumeDist => Self::CumeDist,
}
}
}
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs
b/datafusion/proto/src/physical_plan/to_proto.rs
index 33eca07231..37ea6a2b47 100644
--- a/datafusion/proto/src/physical_plan/to_proto.rs
+++ b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -23,8 +23,8 @@ use datafusion::datasource::file_format::parquet::ParquetSink;
use datafusion::physical_expr::window::{NthValueKind,
SlidingAggregateWindowExpr};
use datafusion::physical_expr::{PhysicalSortExpr, ScalarFunctionExpr};
use datafusion::physical_plan::expressions::{
- BinaryExpr, CaseExpr, CastExpr, Column, CumeDist, InListExpr,
IsNotNullExpr,
- IsNullExpr, Literal, NegativeExpr, NotExpr, NthValue, Ntile, TryCastExpr,
+ BinaryExpr, CaseExpr, CastExpr, Column, InListExpr, IsNotNullExpr,
IsNullExpr,
+ Literal, NegativeExpr, NotExpr, NthValue, Ntile, TryCastExpr,
};
use datafusion::physical_plan::udaf::AggregateFunctionExpr;
use datafusion::physical_plan::windows::{BuiltInWindowExpr,
PlainAggregateWindowExpr};
@@ -108,9 +108,9 @@ pub fn serialize_physical_window_expr(
let expr = built_in_window_expr.get_built_in_func_expr();
let built_in_fn_expr = expr.as_any();
- let builtin_fn = if
built_in_fn_expr.downcast_ref::<CumeDist>().is_some() {
- protobuf::BuiltInWindowFunction::CumeDist
- } else if let Some(ntile_expr) =
built_in_fn_expr.downcast_ref::<Ntile>() {
+ let builtin_fn = if let Some(ntile_expr) =
+ built_in_fn_expr.downcast_ref::<Ntile>()
+ {
args.insert(
0,
Arc::new(Literal::new(datafusion_common::ScalarValue::Int64(Some(
diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
index c017395d97..a8c82ff80f 100644
--- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
@@ -48,7 +48,7 @@ use datafusion::functions_aggregate::expr_fn::{
use datafusion::functions_aggregate::min_max::max_udaf;
use datafusion::functions_nested::map::map;
use datafusion::functions_window::expr_fn::{
- dense_rank, lag, lead, percent_rank, rank, row_number,
+ cume_dist, dense_rank, lag, lead, percent_rank, rank, row_number,
};
use datafusion::functions_window::rank::rank_udwf;
use datafusion::prelude::*;
@@ -940,6 +940,7 @@ async fn roundtrip_expr_api() -> Result<()> {
vec![lit(1), lit(2), lit(3)],
vec![lit(10), lit(20), lit(30)],
),
+ cume_dist(),
row_number(),
rank(),
dense_rank(),
diff --git a/datafusion/sqllogictest/test_files/window.slt
b/datafusion/sqllogictest/test_files/window.slt
index b3f2786d3d..51e8592755 100644
--- a/datafusion/sqllogictest/test_files/window.slt
+++ b/datafusion/sqllogictest/test_files/window.slt
@@ -1135,8 +1135,8 @@ SELECT
query IRR
SELECT
c8,
- CUME_DIST() OVER(ORDER BY c9) as cd1,
- CUME_DIST() OVER(ORDER BY c9 ROWS BETWEEN 10 PRECEDING and 1 FOLLOWING) as
cd2
+ cume_dist() OVER(ORDER BY c9) as cd1,
+ cume_dist() OVER(ORDER BY c9 ROWS BETWEEN 10 PRECEDING and 1 FOLLOWING) as
cd2
FROM aggregate_test_100
ORDER BY c8
LIMIT 5
diff --git a/docs/source/user-guide/sql/window_functions_new.md
b/docs/source/user-guide/sql/window_functions_new.md
index 462fc900d1..89ce2284a7 100644
--- a/docs/source/user-guide/sql/window_functions_new.md
+++ b/docs/source/user-guide/sql/window_functions_new.md
@@ -157,11 +157,20 @@ All [aggregate functions](aggregate_functions.md) can be
used as window function
## Ranking Functions
+- [cume_dist](#cume_dist)
- [dense_rank](#dense_rank)
- [percent_rank](#percent_rank)
- [rank](#rank)
- [row_number](#row_number)
+### `cume_dist`
+
+Relative rank of the current row: (number of rows preceding or peer with
current row) / (total rows).
+
+```
+cume_dist()
+```
+
### `dense_rank`
Returns the rank of the current row without gaps. This function ranks rows in
a dense manner, meaning consecutive ranks are assigned even for identical
values.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
