This is an automated email from the ASF dual-hosted git repository.
jayzhan 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 bd2d4eeaf3 Convert built-in `row_number` to user-defined window
function (#12030)
bd2d4eeaf3 is described below
commit bd2d4eeaf39bc4fbc4694aa332c695019847e25b
Author: jcsherin <[email protected]>
AuthorDate: Sat Aug 17 15:19:25 2024 +0530
Convert built-in `row_number` to user-defined window function (#12030)
* Adds new crate for window functions
* Moves `row_number` to window functions crate
* Fixes build errors
* Regenerates protobuf
* Makes `row_number` no-op temporarily
* Minor: fixes formatting
* Implements `WindowUDF` for `row_number`
* Minor: fixes formatting
* Adds singleton instance of UDWF: `row_number`
* Adds partition evaluator
* Registers default window functions
* Implements `evaluate_all`
* Fixes: allow non-uppercase globals
* Minor: prefix underscore for unused variable
* Minor: fixes formatting
* Uses `row_number_udwf`
* Fixes: unparser test for `row_number`
* Uses row number to represent functional dependency
* Minor: fixes formatting
* Removes `row_number` from case-insensitive name test
* Deletes wrapper for `row_number` window expression
* Fixes: lowercase name in error statement
* Fixes: `row_number` fields are not nullable
* Fixes: lowercase name in explain output
* Updates Cargo.lock
* Fixes: lowercase name in explain output
* Adds support for result ordering
* Minor: add newline between methods
* Fixes: re-export crate name in doc comments
* Adds doc comment for `WindowUDFImpl::nullable`
* Minor: renames variable
* Minor: update doc comments
* Deletes code
* Minor: update doc comments
* Minor: adds period
* Adds doc comment for `row_number` window UDF
* Adds fluent API for creating `row_number` expression
* Minor: removes unnecessary path prefix
* Adds roundtrip logical plan test case
* Updates unit tests for `row_number`
* Deletes code
* Minor: copy edit doc comments
* Minor: deletes comment
* Minor: copy edits udwf doc comments
---
Cargo.toml | 2 +
datafusion-cli/Cargo.lock | 11 ++
datafusion/core/Cargo.toml | 1 +
datafusion/core/src/execution/session_state.rs | 1 +
.../core/src/execution/session_state_defaults.rs | 9 +-
datafusion/core/src/lib.rs | 5 +
datafusion/core/tests/fuzz_cases/window_fuzz.rs | 13 +-
datafusion/expr/src/built_in_window_function.rs | 10 +-
datafusion/expr/src/expr.rs | 1 -
datafusion/expr/src/logical_plan/plan.rs | 14 +-
datafusion/expr/src/udwf.rs | 34 ++++
datafusion/expr/src/window_function.rs | 8 -
datafusion/{sql => functions-window}/Cargo.toml | 29 +---
datafusion/functions-window/README.md | 26 +++
datafusion/functions-window/src/lib.rs | 58 +++++++
datafusion/functions-window/src/row_number.rs | 183 +++++++++++++++++++++
datafusion/physical-expr/src/expressions/mod.rs | 1 -
datafusion/physical-expr/src/window/mod.rs | 1 -
datafusion/physical-expr/src/window/row_number.rs | 166 -------------------
datafusion/physical-expr/src/window/window_expr.rs | 6 -
datafusion/physical-plan/src/windows/mod.rs | 21 ++-
datafusion/proto/proto/datafusion.proto | 3 +-
datafusion/proto/src/generated/pbjson.rs | 6 +-
datafusion/proto/src/generated/prost.rs | 8 +-
datafusion/proto/src/logical_plan/from_proto.rs | 2 +-
datafusion/proto/src/logical_plan/to_proto.rs | 1 -
datafusion/proto/src/physical_plan/to_proto.rs | 7 +-
.../proto/tests/cases/roundtrip_logical_plan.rs | 2 +
datafusion/sql/Cargo.toml | 1 +
datafusion/sql/src/unparser/expr.rs | 7 +-
datafusion/sqllogictest/test_files/functions.slt | 2 +-
datafusion/sqllogictest/test_files/group_by.slt | 8 +-
datafusion/sqllogictest/test_files/joins.slt | 56 +++----
datafusion/sqllogictest/test_files/window.slt | 124 +++++++-------
34 files changed, 480 insertions(+), 347 deletions(-)
diff --git a/Cargo.toml b/Cargo.toml
index 02b1f1ccd9..dfb70dd1eb 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -29,6 +29,7 @@ members = [
"datafusion/functions-aggregate",
"datafusion/functions-aggregate-common",
"datafusion/functions-nested",
+ "datafusion/functions-window",
"datafusion/optimizer",
"datafusion/physical-expr",
"datafusion/physical-expr-common",
@@ -102,6 +103,7 @@ datafusion-functions = { path = "datafusion/functions",
version = "41.0.0" }
datafusion-functions-aggregate = { path = "datafusion/functions-aggregate",
version = "41.0.0" }
datafusion-functions-aggregate-common = { path =
"datafusion/functions-aggregate-common", version = "41.0.0" }
datafusion-functions-nested = { path = "datafusion/functions-nested", version
= "41.0.0" }
+datafusion-functions-window = { path = "datafusion/functions-window", version
= "41.0.0" }
datafusion-optimizer = { path = "datafusion/optimizer", version = "41.0.0",
default-features = false }
datafusion-physical-expr = { path = "datafusion/physical-expr", version =
"41.0.0", default-features = false }
datafusion-physical-expr-common = { path = "datafusion/physical-expr-common",
version = "41.0.0", default-features = false }
diff --git a/datafusion-cli/Cargo.lock b/datafusion-cli/Cargo.lock
index 22f889de80..cda57ba9b4 100644
--- a/datafusion-cli/Cargo.lock
+++ b/datafusion-cli/Cargo.lock
@@ -1152,6 +1152,7 @@ dependencies = [
"datafusion-functions",
"datafusion-functions-aggregate",
"datafusion-functions-nested",
+ "datafusion-functions-window",
"datafusion-optimizer",
"datafusion-physical-expr",
"datafusion-physical-expr-common",
@@ -1376,6 +1377,16 @@ dependencies = [
"rand",
]
+[[package]]
+name = "datafusion-functions-window"
+version = "41.0.0"
+dependencies = [
+ "datafusion-common",
+ "datafusion-expr",
+ "datafusion-physical-expr-common",
+ "log",
+]
+
[[package]]
name = "datafusion-optimizer"
version = "41.0.0"
diff --git a/datafusion/core/Cargo.toml b/datafusion/core/Cargo.toml
index e678c93ede..adbba3eb31 100644
--- a/datafusion/core/Cargo.toml
+++ b/datafusion/core/Cargo.toml
@@ -106,6 +106,7 @@ datafusion-expr = { workspace = true }
datafusion-functions = { workspace = true }
datafusion-functions-aggregate = { workspace = true }
datafusion-functions-nested = { workspace = true, optional = true }
+datafusion-functions-window = { workspace = true }
datafusion-optimizer = { workspace = true }
datafusion-physical-expr = { workspace = true }
datafusion-physical-expr-common = { workspace = true }
diff --git a/datafusion/core/src/execution/session_state.rs
b/datafusion/core/src/execution/session_state.rs
index e9c8762918..88a90e1e1d 100644
--- a/datafusion/core/src/execution/session_state.rs
+++ b/datafusion/core/src/execution/session_state.rs
@@ -1044,6 +1044,7 @@ impl SessionStateBuilder {
self.scalar_functions =
Some(SessionStateDefaults::default_scalar_functions());
self.aggregate_functions =
Some(SessionStateDefaults::default_aggregate_functions());
+ self.window_functions =
Some(SessionStateDefaults::default_window_functions());
self
}
diff --git a/datafusion/core/src/execution/session_state_defaults.rs
b/datafusion/core/src/execution/session_state_defaults.rs
index 07420afe84..bc7e194cae 100644
--- a/datafusion/core/src/execution/session_state_defaults.rs
+++ b/datafusion/core/src/execution/session_state_defaults.rs
@@ -29,12 +29,12 @@ use crate::datasource::provider::DefaultTableFactory;
use crate::execution::context::SessionState;
#[cfg(feature = "nested_expressions")]
use crate::functions_nested;
-use crate::{functions, functions_aggregate};
+use crate::{functions, functions_aggregate, functions_window};
use datafusion_execution::config::SessionConfig;
use datafusion_execution::object_store::ObjectStoreUrl;
use datafusion_execution::runtime_env::RuntimeEnv;
use datafusion_expr::planner::ExprPlanner;
-use datafusion_expr::{AggregateUDF, ScalarUDF};
+use datafusion_expr::{AggregateUDF, ScalarUDF, WindowUDF};
use std::collections::HashMap;
use std::sync::Arc;
use url::Url;
@@ -112,6 +112,11 @@ impl SessionStateDefaults {
functions_aggregate::all_default_aggregate_functions()
}
+ /// returns the list of default [`WindowUDF']'s
+ pub fn default_window_functions() -> Vec<Arc<WindowUDF>> {
+ functions_window::all_default_window_functions()
+ }
+
/// returns the list of default [`FileFormatFactory']'s
pub fn default_file_formats() -> Vec<Arc<dyn FileFormatFactory>> {
let file_formats: Vec<Arc<dyn FileFormatFactory>> = vec![
diff --git a/datafusion/core/src/lib.rs b/datafusion/core/src/lib.rs
index 6b3773e4f6..daeb21db9d 100644
--- a/datafusion/core/src/lib.rs
+++ b/datafusion/core/src/lib.rs
@@ -603,6 +603,11 @@ pub mod functions_aggregate {
pub use datafusion_functions_aggregate::*;
}
+/// re-export of [`datafusion_functions_window`] crate
+pub mod functions_window {
+ pub use datafusion_functions_window::*;
+}
+
#[cfg(test)]
pub mod test;
pub mod test_util;
diff --git a/datafusion/core/tests/fuzz_cases/window_fuzz.rs
b/datafusion/core/tests/fuzz_cases/window_fuzz.rs
index d75d8e4337..a6c2cf700c 100644
--- a/datafusion/core/tests/fuzz_cases/window_fuzz.rs
+++ b/datafusion/core/tests/fuzz_cases/window_fuzz.rs
@@ -44,6 +44,7 @@ use datafusion_physical_expr::expressions::{cast, col, lit};
use datafusion_physical_expr::{PhysicalExpr, PhysicalSortExpr};
use test_utils::add_empty_batches;
+use datafusion::functions_window::row_number::row_number_udwf;
use hashbrown::HashMap;
use rand::distributions::Alphanumeric;
use rand::rngs::StdRng;
@@ -180,12 +181,10 @@ async fn bounded_window_causal_non_causal() -> Result<()>
{
// ROWS BETWEEN UNBOUNDED PRECEDING AND <end_bound>
PRECEDING/FOLLOWING
// )
(
- // Window function
- WindowFunctionDefinition::BuiltInWindowFunction(
- BuiltInWindowFunction::RowNumber,
- ),
+ // user-defined window function
+ WindowFunctionDefinition::WindowUDF(row_number_udwf()),
// its name
- "ROW_NUMBER",
+ "row_number",
// no argument
vec![],
// Expected causality, for None cases causality will be determined
from window frame boundaries
@@ -377,9 +376,7 @@ fn get_random_function(
window_fn_map.insert(
"row_number",
(
- WindowFunctionDefinition::BuiltInWindowFunction(
- BuiltInWindowFunction::RowNumber,
- ),
+ WindowFunctionDefinition::WindowUDF(row_number_udwf()),
vec![],
),
);
diff --git a/datafusion/expr/src/built_in_window_function.rs
b/datafusion/expr/src/built_in_window_function.rs
index 3885d70049..597e4e68a0 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, Hash, EnumIter)]
pub enum BuiltInWindowFunction {
- /// number of the current row within its partition, counting from 1
- RowNumber,
/// rank of the current row with gaps; same as row_number of its first peer
Rank,
/// rank of the current row without gaps; this function counts peer groups
@@ -74,7 +72,6 @@ impl BuiltInWindowFunction {
pub fn name(&self) -> &str {
use BuiltInWindowFunction::*;
match self {
- RowNumber => "ROW_NUMBER",
Rank => "RANK",
DenseRank => "DENSE_RANK",
PercentRank => "PERCENT_RANK",
@@ -93,7 +90,6 @@ impl FromStr for BuiltInWindowFunction {
type Err = DataFusionError;
fn from_str(name: &str) -> Result<BuiltInWindowFunction> {
Ok(match name.to_uppercase().as_str() {
- "ROW_NUMBER" => BuiltInWindowFunction::RowNumber,
"RANK" => BuiltInWindowFunction::Rank,
"DENSE_RANK" => BuiltInWindowFunction::DenseRank,
"PERCENT_RANK" => BuiltInWindowFunction::PercentRank,
@@ -131,8 +127,7 @@ impl BuiltInWindowFunction {
})?;
match self {
- BuiltInWindowFunction::RowNumber
- | BuiltInWindowFunction::Rank
+ BuiltInWindowFunction::Rank
| BuiltInWindowFunction::DenseRank
| BuiltInWindowFunction::Ntile => Ok(DataType::UInt64),
BuiltInWindowFunction::PercentRank |
BuiltInWindowFunction::CumeDist => {
@@ -150,8 +145,7 @@ 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::RowNumber
- | BuiltInWindowFunction::Rank
+ BuiltInWindowFunction::Rank
| BuiltInWindowFunction::DenseRank
| BuiltInWindowFunction::PercentRank
| BuiltInWindowFunction::CumeDist => Signature::any(0,
Volatility::Immutable),
diff --git a/datafusion/expr/src/expr.rs b/datafusion/expr/src/expr.rs
index b4d489cc7c..88939ccf41 100644
--- a/datafusion/expr/src/expr.rs
+++ b/datafusion/expr/src/expr.rs
@@ -2896,7 +2896,6 @@ mod test {
#[test]
fn test_window_function_case_insensitive() -> Result<()> {
let names = vec![
- "row_number",
"rank",
"dense_rank",
"percent_rank",
diff --git a/datafusion/expr/src/logical_plan/plan.rs
b/datafusion/expr/src/logical_plan/plan.rs
index 2bab6d516a..f9b3035167 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -36,9 +36,9 @@ use crate::utils::{
split_conjunction,
};
use crate::{
- build_join_schema, expr_vec_fmt, BinaryExpr, BuiltInWindowFunction,
- CreateMemoryTable, CreateView, Expr, ExprSchemable, LogicalPlanBuilder,
Operator,
- TableProviderFilterPushDown, TableSource, WindowFunctionDefinition,
+ build_join_schema, expr_vec_fmt, BinaryExpr, CreateMemoryTable,
CreateView, Expr,
+ ExprSchemable, LogicalPlanBuilder, Operator, TableProviderFilterPushDown,
+ TableSource, WindowFunctionDefinition,
};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
@@ -2214,18 +2214,14 @@ impl Window {
.enumerate()
.filter_map(|(idx, expr)| {
if let Expr::WindowFunction(WindowFunction {
- // Function is ROW_NUMBER
- fun:
- WindowFunctionDefinition::BuiltInWindowFunction(
- BuiltInWindowFunction::RowNumber,
- ),
+ fun: WindowFunctionDefinition::WindowUDF(udwf),
partition_by,
..
}) = expr
{
// When there is no PARTITION BY, row number will be unique
// across the entire table.
- if partition_by.is_empty() {
+ if udwf.name() == "row_number" && partition_by.is_empty() {
return Some(idx + input_len);
}
}
diff --git a/datafusion/expr/src/udwf.rs b/datafusion/expr/src/udwf.rs
index 9e6d963ccf..aa754a5708 100644
--- a/datafusion/expr/src/udwf.rs
+++ b/datafusion/expr/src/udwf.rs
@@ -17,6 +17,7 @@
//! [`WindowUDF`]: User Defined Window Functions
+use arrow::compute::SortOptions;
use std::hash::{DefaultHasher, Hash, Hasher};
use std::{
any::Any,
@@ -176,6 +177,21 @@ impl WindowUDF {
pub fn partition_evaluator_factory(&self) -> Result<Box<dyn
PartitionEvaluator>> {
self.inner.partition_evaluator()
}
+
+ /// Returns if column values are nullable for this window function.
+ ///
+ /// See [`WindowUDFImpl::nullable`] for more details.
+ pub fn nullable(&self) -> bool {
+ self.inner.nullable()
+ }
+
+ /// Returns custom result ordering introduced by this window function
+ /// which is used to update ordering equivalences.
+ ///
+ /// See [`WindowUDFImpl::sort_options`] for more details.
+ pub fn sort_options(&self) -> Option<SortOptions> {
+ self.inner.sort_options()
+ }
}
impl<F> From<F> for WindowUDF
@@ -319,6 +335,24 @@ pub trait WindowUDFImpl: Debug + Send + Sync {
self.signature().hash(hasher);
hasher.finish()
}
+
+ /// Allows customizing nullable of column for this window UDF.
+ ///
+ /// By default, the final result of evaluating the window UDF is
+ /// allowed to have null values. But if that is not the case then
+ /// it can be customized in the window UDF implementation.
+ fn nullable(&self) -> bool {
+ true
+ }
+
+ /// Allows the window UDF to define a custom result ordering.
+ ///
+ /// By default, a window UDF doesn't introduce an ordering.
+ /// But when specified by a window UDF this is used to update
+ /// ordering equivalences.
+ fn sort_options(&self) -> Option<SortOptions> {
+ None
+ }
}
/// WindowUDF that adds an alias to the underlying function. It is better to
diff --git a/datafusion/expr/src/window_function.rs
b/datafusion/expr/src/window_function.rs
index 5e81464d39..a80718147c 100644
--- a/datafusion/expr/src/window_function.rs
+++ b/datafusion/expr/src/window_function.rs
@@ -19,14 +19,6 @@ use datafusion_common::ScalarValue;
use crate::{expr::WindowFunction, BuiltInWindowFunction, Expr, Literal};
-/// Create an expression to represent the `row_number` window function
-pub fn row_number() -> Expr {
- Expr::WindowFunction(WindowFunction::new(
- BuiltInWindowFunction::RowNumber,
- vec![],
- ))
-}
-
/// Create an expression to represent the `rank` window function
pub fn rank() -> Expr {
Expr::WindowFunction(WindowFunction::new(BuiltInWindowFunction::Rank,
vec![]))
diff --git a/datafusion/sql/Cargo.toml b/datafusion/functions-window/Cargo.toml
similarity index 62%
copy from datafusion/sql/Cargo.toml
copy to datafusion/functions-window/Cargo.toml
index c4ae3a8134..94dd421284 100644
--- a/datafusion/sql/Cargo.toml
+++ b/datafusion/functions-window/Cargo.toml
@@ -16,9 +16,9 @@
# under the License.
[package]
-name = "datafusion-sql"
-description = "DataFusion SQL Query Planner"
-keywords = ["datafusion", "sql", "parser", "planner"]
+name = "datafusion-functions-window"
+description = "Window function packages for the DataFusion query engine"
+keywords = ["datafusion", "logical", "plan", "expressions"]
readme = "README.md"
version = { workspace = true }
edition = { workspace = true }
@@ -32,29 +32,16 @@ rust-version = { workspace = true }
workspace = true
[lib]
-name = "datafusion_sql"
+name = "datafusion_functions_window"
path = "src/lib.rs"
-[features]
-default = ["unicode_expressions", "unparser"]
-unicode_expressions = []
-unparser = []
+# See more keys and their definitions at
https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
-arrow = { workspace = true }
-arrow-array = { workspace = true }
-arrow-schema = { workspace = true }
-datafusion-common = { workspace = true, default-features = true }
+datafusion-common = { workspace = true }
datafusion-expr = { workspace = true }
+datafusion-physical-expr-common = { workspace = true }
log = { workspace = true }
-regex = { workspace = true }
-sqlparser = { workspace = true }
-strum = { version = "0.26.1", features = ["derive"] }
[dev-dependencies]
-ctor = { workspace = true }
-datafusion-functions = { workspace = true, default-features = true }
-datafusion-functions-aggregate = { workspace = true }
-env_logger = { workspace = true }
-paste = "^1.0"
-rstest = { workspace = true }
+arrow = { workspace = true }
diff --git a/datafusion/functions-window/README.md
b/datafusion/functions-window/README.md
new file mode 100644
index 0000000000..18590983ca
--- /dev/null
+++ b/datafusion/functions-window/README.md
@@ -0,0 +1,26 @@
+<!---
+ 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.
+-->
+
+# DataFusion Window Function Library
+
+[DataFusion][df] is an extensible query execution framework, written in Rust,
that uses Apache Arrow as its in-memory format.
+
+This crate contains user-defined window functions.
+
+[df]: https://crates.io/crates/datafusion
diff --git a/datafusion/functions-window/src/lib.rs
b/datafusion/functions-window/src/lib.rs
new file mode 100644
index 0000000000..790a500f1f
--- /dev/null
+++ b/datafusion/functions-window/src/lib.rs
@@ -0,0 +1,58 @@
+// 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.
+
+//! Window Function packages for [DataFusion].
+//!
+//! This crate contains a collection of various window function packages for
DataFusion,
+//! implemented using the extension API.
+//!
+//! [DataFusion]: https://crates.io/crates/datafusion
+//!
+use std::sync::Arc;
+
+use log::debug;
+
+use datafusion_expr::registry::FunctionRegistry;
+use datafusion_expr::WindowUDF;
+
+pub mod row_number;
+
+/// Fluent-style API for creating `Expr`s
+pub mod expr_fn {
+ pub use super::row_number::row_number;
+}
+
+/// Returns all default window functions
+pub fn all_default_window_functions() -> Vec<Arc<WindowUDF>> {
+ vec![row_number::row_number_udwf()]
+}
+/// Registers all enabled packages with a [`FunctionRegistry`]
+pub fn register_all(
+ registry: &mut dyn FunctionRegistry,
+) -> datafusion_common::Result<()> {
+ let functions: Vec<Arc<WindowUDF>> = all_default_window_functions();
+
+ functions.into_iter().try_for_each(|fun| {
+ let existing_udwf = registry.register_udwf(fun)?;
+ if let Some(existing_udwf) = existing_udwf {
+ debug!("Overwrite existing UDWF: {}", existing_udwf.name());
+ }
+ Ok(()) as datafusion_common::Result<()>
+ })?;
+
+ Ok(())
+}
diff --git a/datafusion/functions-window/src/row_number.rs
b/datafusion/functions-window/src/row_number.rs
new file mode 100644
index 0000000000..08d7111824
--- /dev/null
+++ b/datafusion/functions-window/src/row_number.rs
@@ -0,0 +1,183 @@
+// 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 `row_number` that can evaluated at runtime
during query execution
+
+use std::any::Any;
+use std::fmt::Debug;
+use std::ops::Range;
+
+use datafusion_common::arrow::array::ArrayRef;
+use datafusion_common::arrow::array::UInt64Array;
+use datafusion_common::arrow::compute::SortOptions;
+use datafusion_common::arrow::datatypes::DataType;
+use datafusion_common::{Result, ScalarValue};
+use datafusion_expr::expr::WindowFunction;
+use datafusion_expr::{Expr, PartitionEvaluator, Signature, Volatility,
WindowUDFImpl};
+
+/// Create a [`WindowFunction`](Expr::WindowFunction) expression for
+/// `row_number` user-defined window function.
+pub fn row_number(args: Vec<Expr>) -> Expr {
+ Expr::WindowFunction(WindowFunction::new(row_number_udwf(), args))
+}
+
+/// Singleton instance of `row_number`, ensures the UDWF is only created once.
+#[allow(non_upper_case_globals)]
+static STATIC_RowNumber:
std::sync::OnceLock<std::sync::Arc<datafusion_expr::WindowUDF>> =
+ std::sync::OnceLock::new();
+
+/// Returns a [`WindowUDF`](datafusion_expr::WindowUDF) for `row_number`
+/// user-defined window function.
+pub fn row_number_udwf() -> std::sync::Arc<datafusion_expr::WindowUDF> {
+ STATIC_RowNumber
+ .get_or_init(|| {
+
std::sync::Arc::new(datafusion_expr::WindowUDF::from(RowNumber::default()))
+ })
+ .clone()
+}
+
+/// row_number expression
+#[derive(Debug)]
+struct RowNumber {
+ signature: Signature,
+}
+
+impl RowNumber {
+ /// Create a new `row_number` function
+ fn new() -> Self {
+ Self {
+ signature: Signature::any(0, Volatility::Immutable),
+ }
+ }
+}
+
+impl Default for RowNumber {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl WindowUDFImpl for RowNumber {
+ fn as_any(&self) -> &dyn Any {
+ self
+ }
+
+ fn name(&self) -> &str {
+ "row_number"
+ }
+
+ fn signature(&self) -> &Signature {
+ &self.signature
+ }
+
+ fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+ Ok(DataType::UInt64)
+ }
+
+ fn partition_evaluator(&self) -> Result<Box<dyn PartitionEvaluator>> {
+ Ok(Box::<NumRowsEvaluator>::default())
+ }
+
+ fn nullable(&self) -> bool {
+ false
+ }
+
+ fn sort_options(&self) -> Option<SortOptions> {
+ Some(SortOptions {
+ descending: false,
+ nulls_first: false,
+ })
+ }
+}
+
+/// State for the `row_number` built-in window function.
+#[derive(Debug, Default)]
+struct NumRowsEvaluator {
+ n_rows: usize,
+}
+
+impl PartitionEvaluator for NumRowsEvaluator {
+ fn is_causal(&self) -> bool {
+ // The row_number function doesn't need "future" values to emit
results:
+ true
+ }
+
+ fn evaluate_all(
+ &mut self,
+ _values: &[ArrayRef],
+ num_rows: usize,
+ ) -> Result<ArrayRef> {
+ Ok(std::sync::Arc::new(UInt64Array::from_iter_values(
+ 1..(num_rows as u64) + 1,
+ )))
+ }
+
+ fn evaluate(
+ &mut self,
+ _values: &[ArrayRef],
+ _range: &Range<usize>,
+ ) -> Result<ScalarValue> {
+ self.n_rows += 1;
+ Ok(ScalarValue::UInt64(Some(self.n_rows as u64)))
+ }
+
+ fn supports_bounded_execution(&self) -> bool {
+ true
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use std::sync::Arc;
+
+ use datafusion_common::arrow::array::{Array, BooleanArray};
+ use datafusion_common::cast::as_uint64_array;
+
+ use super::*;
+
+ #[test]
+ fn row_number_all_null() -> Result<()> {
+ let values: ArrayRef = Arc::new(BooleanArray::from(vec![
+ None, None, None, None, None, None, None, None,
+ ]));
+ let num_rows = values.len();
+
+ let actual = RowNumber::default()
+ .partition_evaluator()?
+ .evaluate_all(&[values], num_rows)?;
+ let actual = as_uint64_array(&actual)?;
+
+ assert_eq!(vec![1, 2, 3, 4, 5, 6, 7, 8], *actual.values());
+ Ok(())
+ }
+
+ #[test]
+ fn row_number_all_values() -> Result<()> {
+ let values: ArrayRef = Arc::new(BooleanArray::from(vec![
+ true, false, true, false, false, true, false, true,
+ ]));
+ let num_rows = values.len();
+
+ let actual = RowNumber::default()
+ .partition_evaluator()?
+ .evaluate_all(&[values], num_rows)?;
+ let actual = as_uint64_array(&actual)?;
+
+ assert_eq!(vec![1, 2, 3, 4, 5, 6, 7, 8], *actual.values());
+ Ok(())
+ }
+}
diff --git a/datafusion/physical-expr/src/expressions/mod.rs
b/datafusion/physical-expr/src/expressions/mod.rs
index 9e65889d87..87d8f04a68 100644
--- a/datafusion/physical-expr/src/expressions/mod.rs
+++ b/datafusion/physical-expr/src/expressions/mod.rs
@@ -40,7 +40,6 @@ pub use crate::window::lead_lag::{lag, lead, WindowShift};
pub use crate::window::nth_value::NthValue;
pub use crate::window::ntile::Ntile;
pub use crate::window::rank::{dense_rank, percent_rank, rank, Rank, RankType};
-pub use crate::window::row_number::RowNumber;
pub use crate::PhysicalSortExpr;
pub use binary::{binary, BinaryExpr};
diff --git a/datafusion/physical-expr/src/window/mod.rs
b/datafusion/physical-expr/src/window/mod.rs
index 644edae36c..2aeb053331 100644
--- a/datafusion/physical-expr/src/window/mod.rs
+++ b/datafusion/physical-expr/src/window/mod.rs
@@ -23,7 +23,6 @@ pub(crate) mod lead_lag;
pub(crate) mod nth_value;
pub(crate) mod ntile;
pub(crate) mod rank;
-pub(crate) mod row_number;
mod sliding_aggregate;
mod window_expr;
diff --git a/datafusion/physical-expr/src/window/row_number.rs
b/datafusion/physical-expr/src/window/row_number.rs
deleted file mode 100644
index 0a1255018d..0000000000
--- a/datafusion/physical-expr/src/window/row_number.rs
+++ /dev/null
@@ -1,166 +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 `row_number` that can evaluated at runtime
during query execution
-
-use crate::expressions::Column;
-use crate::window::window_expr::NumRowsState;
-use crate::window::BuiltInWindowFunctionExpr;
-use crate::{PhysicalExpr, PhysicalSortExpr};
-
-use arrow::array::{ArrayRef, UInt64Array};
-use arrow::datatypes::{DataType, Field};
-use arrow_schema::{SchemaRef, SortOptions};
-use datafusion_common::{Result, ScalarValue};
-use datafusion_expr::PartitionEvaluator;
-
-use std::any::Any;
-use std::ops::Range;
-use std::sync::Arc;
-
-/// row_number expression
-#[derive(Debug)]
-pub struct RowNumber {
- name: String,
- /// Output data type
- data_type: DataType,
-}
-
-impl RowNumber {
- /// Create a new ROW_NUMBER function
- pub fn new(name: impl Into<String>, data_type: &DataType) -> Self {
- Self {
- name: name.into(),
- data_type: data_type.clone(),
- }
- }
-}
-
-impl BuiltInWindowFunctionExpr for RowNumber {
- /// 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 get_result_ordering(&self, schema: &SchemaRef) ->
Option<PhysicalSortExpr> {
- // The built-in ROW_NUMBER window function introduces a new ordering:
- schema.column_with_name(self.name()).map(|(idx, field)| {
- let expr = Arc::new(Column::new(field.name(), idx));
- let options = SortOptions {
- descending: false,
- nulls_first: false,
- }; // ASC, NULLS LAST
- PhysicalSortExpr { expr, options }
- })
- }
-
- fn create_evaluator(&self) -> Result<Box<dyn PartitionEvaluator>> {
- Ok(Box::<NumRowsEvaluator>::default())
- }
-}
-
-#[derive(Default, Debug)]
-pub(crate) struct NumRowsEvaluator {
- state: NumRowsState,
-}
-
-impl PartitionEvaluator for NumRowsEvaluator {
- fn is_causal(&self) -> bool {
- // The ROW_NUMBER function doesn't need "future" values to emit
results:
- true
- }
-
- /// evaluate window function result inside given range
- fn evaluate(
- &mut self,
- _values: &[ArrayRef],
- _range: &Range<usize>,
- ) -> Result<ScalarValue> {
- self.state.n_rows += 1;
- Ok(ScalarValue::UInt64(Some(self.state.n_rows as u64)))
- }
-
- fn evaluate_all(
- &mut self,
- _values: &[ArrayRef],
- num_rows: usize,
- ) -> Result<ArrayRef> {
- Ok(Arc::new(UInt64Array::from_iter_values(
- 1..(num_rows as u64) + 1,
- )))
- }
-
- fn supports_bounded_execution(&self) -> bool {
- true
- }
-}
-
-#[cfg(test)]
-mod tests {
- use super::*;
- use arrow::{array::*, datatypes::*};
- use datafusion_common::cast::as_uint64_array;
-
- #[test]
- fn row_number_all_null() -> Result<()> {
- let arr: ArrayRef = Arc::new(BooleanArray::from(vec![
- None, None, None, None, None, None, None, None,
- ]));
- let schema = Schema::new(vec![Field::new("arr", DataType::Boolean,
true)]);
- let batch = RecordBatch::try_new(Arc::new(schema), vec![arr])?;
- let row_number = RowNumber::new("row_number".to_owned(),
&DataType::UInt64);
- let values = row_number.evaluate_args(&batch)?;
- let result = row_number
- .create_evaluator()?
- .evaluate_all(&values, batch.num_rows())?;
- let result = as_uint64_array(&result)?;
- let result = result.values();
- assert_eq!(vec![1, 2, 3, 4, 5, 6, 7, 8], *result);
- Ok(())
- }
-
- #[test]
- fn row_number_all_values() -> Result<()> {
- let arr: ArrayRef = Arc::new(BooleanArray::from(vec![
- true, false, true, false, false, true, false, true,
- ]));
- let schema = Schema::new(vec![Field::new("arr", DataType::Boolean,
false)]);
- let batch = RecordBatch::try_new(Arc::new(schema), vec![arr])?;
- let row_number = RowNumber::new("row_number".to_owned(),
&DataType::UInt64);
- let values = row_number.evaluate_args(&batch)?;
- let result = row_number
- .create_evaluator()?
- .evaluate_all(&values, batch.num_rows())?;
- let result = as_uint64_array(&result)?;
- let result = result.values();
- assert_eq!(vec![1, 2, 3, 4, 5, 6, 7, 8], *result);
- Ok(())
- }
-}
diff --git a/datafusion/physical-expr/src/window/window_expr.rs
b/datafusion/physical-expr/src/window/window_expr.rs
index 7020f7f5cf..8f6f78df8c 100644
--- a/datafusion/physical-expr/src/window/window_expr.rs
+++ b/datafusion/physical-expr/src/window/window_expr.rs
@@ -543,12 +543,6 @@ pub struct RankState {
pub n_rank: usize,
}
-/// State for the 'ROW_NUMBER' built-in window function.
-#[derive(Debug, Clone, Default)]
-pub struct NumRowsState {
- pub n_rows: usize,
-}
-
/// Tag to differentiate special use cases of the NTH_VALUE built-in window
function.
#[derive(Debug, Copy, Clone)]
pub enum NthValueKind {
diff --git a/datafusion/physical-plan/src/windows/mod.rs
b/datafusion/physical-plan/src/windows/mod.rs
index 03090faf3e..154beb79f7 100644
--- a/datafusion/physical-plan/src/windows/mod.rs
+++ b/datafusion/physical-plan/src/windows/mod.rs
@@ -23,7 +23,7 @@ use std::sync::Arc;
use crate::{
expressions::{
cume_dist, dense_rank, lag, lead, percent_rank, rank, Literal,
NthValue, Ntile,
- PhysicalSortExpr, RowNumber,
+ PhysicalSortExpr,
},
ExecutionPlan, ExecutionPlanProperties, InputOrderMode, PhysicalExpr,
};
@@ -49,6 +49,7 @@ mod bounded_window_agg_exec;
mod window_agg_exec;
pub use bounded_window_agg_exec::BoundedWindowAggExec;
+use datafusion_physical_expr::expressions::Column;
pub use datafusion_physical_expr::window::{
BuiltInWindowExpr, PlainAggregateWindowExpr, WindowExpr,
};
@@ -218,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::RowNumber => Arc::new(RowNumber::new(name,
out_data_type)),
BuiltInWindowFunction::Rank => Arc::new(rank(name, out_data_type)),
BuiltInWindowFunction::DenseRank => Arc::new(dense_rank(name,
out_data_type)),
BuiltInWindowFunction::PercentRank => Arc::new(percent_rank(name,
out_data_type)),
@@ -356,8 +356,11 @@ impl BuiltInWindowFunctionExpr for WindowUDFExpr {
}
fn field(&self) -> Result<Field> {
- let nullable = true;
- Ok(Field::new(&self.name, self.data_type.clone(), nullable))
+ Ok(Field::new(
+ &self.name,
+ self.data_type.clone(),
+ self.fun.nullable(),
+ ))
}
fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>> {
@@ -375,6 +378,16 @@ impl BuiltInWindowFunctionExpr for WindowUDFExpr {
fn reverse_expr(&self) -> Option<Arc<dyn BuiltInWindowFunctionExpr>> {
None
}
+
+ fn get_result_ordering(&self, schema: &SchemaRef) ->
Option<PhysicalSortExpr> {
+ self.fun
+ .sort_options()
+ .zip(schema.column_with_name(self.name()))
+ .map(|(options, (idx, field))| {
+ let expr = Arc::new(Column::new(field.name(), idx));
+ PhysicalSortExpr { expr, options }
+ })
+ }
}
pub(crate) fn calc_requirements<
diff --git a/datafusion/proto/proto/datafusion.proto
b/datafusion/proto/proto/datafusion.proto
index 819130b08e..acf540d444 100644
--- a/datafusion/proto/proto/datafusion.proto
+++ b/datafusion/proto/proto/datafusion.proto
@@ -481,7 +481,8 @@ message ScalarUDFExprNode {
}
enum BuiltInWindowFunction {
- ROW_NUMBER = 0;
+ UNSPECIFIED = 0; //
https://protobuf.dev/programming-guides/dos-donts/#unspecified-enum
+ // ROW_NUMBER = 0;
RANK = 1;
DENSE_RANK = 2;
PERCENT_RANK = 3;
diff --git a/datafusion/proto/src/generated/pbjson.rs
b/datafusion/proto/src/generated/pbjson.rs
index 521a0d90c1..489b6c6753 100644
--- a/datafusion/proto/src/generated/pbjson.rs
+++ b/datafusion/proto/src/generated/pbjson.rs
@@ -1659,7 +1659,7 @@ impl serde::Serialize for BuiltInWindowFunction {
S: serde::Serializer,
{
let variant = match self {
- Self::RowNumber => "ROW_NUMBER",
+ Self::Unspecified => "UNSPECIFIED",
Self::Rank => "RANK",
Self::DenseRank => "DENSE_RANK",
Self::PercentRank => "PERCENT_RANK",
@@ -1681,7 +1681,7 @@ impl<'de> serde::Deserialize<'de> for
BuiltInWindowFunction {
D: serde::Deserializer<'de>,
{
const FIELDS: &[&str] = &[
- "ROW_NUMBER",
+ "UNSPECIFIED",
"RANK",
"DENSE_RANK",
"PERCENT_RANK",
@@ -1732,7 +1732,7 @@ impl<'de> serde::Deserialize<'de> for
BuiltInWindowFunction {
E: serde::de::Error,
{
match value {
- "ROW_NUMBER" => Ok(BuiltInWindowFunction::RowNumber),
+ "UNSPECIFIED" => Ok(BuiltInWindowFunction::Unspecified),
"RANK" => Ok(BuiltInWindowFunction::Rank),
"DENSE_RANK" => Ok(BuiltInWindowFunction::DenseRank),
"PERCENT_RANK" => Ok(BuiltInWindowFunction::PercentRank),
diff --git a/datafusion/proto/src/generated/prost.rs
b/datafusion/proto/src/generated/prost.rs
index 070c9b31d3..c98c950d35 100644
--- a/datafusion/proto/src/generated/prost.rs
+++ b/datafusion/proto/src/generated/prost.rs
@@ -1919,7 +1919,9 @@ pub struct PartitionStats {
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord,
::prost::Enumeration)]
#[repr(i32)]
pub enum BuiltInWindowFunction {
- RowNumber = 0,
+ /// <https://protobuf.dev/programming-guides/dos-donts/#unspecified-enum>
+ Unspecified = 0,
+ /// ROW_NUMBER = 0;
Rank = 1,
DenseRank = 2,
PercentRank = 3,
@@ -1938,7 +1940,7 @@ impl BuiltInWindowFunction {
/// (if the ProtoBuf definition does not change) and safe for programmatic
use.
pub fn as_str_name(&self) -> &'static str {
match self {
- BuiltInWindowFunction::RowNumber => "ROW_NUMBER",
+ BuiltInWindowFunction::Unspecified => "UNSPECIFIED",
BuiltInWindowFunction::Rank => "RANK",
BuiltInWindowFunction::DenseRank => "DENSE_RANK",
BuiltInWindowFunction::PercentRank => "PERCENT_RANK",
@@ -1954,7 +1956,7 @@ impl BuiltInWindowFunction {
/// Creates an enum from field names used in the ProtoBuf definition.
pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
match value {
- "ROW_NUMBER" => Some(Self::RowNumber),
+ "UNSPECIFIED" => Some(Self::Unspecified),
"RANK" => Some(Self::Rank),
"DENSE_RANK" => Some(Self::DenseRank),
"PERCENT_RANK" => Some(Self::PercentRank),
diff --git a/datafusion/proto/src/logical_plan/from_proto.rs
b/datafusion/proto/src/logical_plan/from_proto.rs
index 6cbea5f0cf..b74237b528 100644
--- a/datafusion/proto/src/logical_plan/from_proto.rs
+++ b/datafusion/proto/src/logical_plan/from_proto.rs
@@ -141,7 +141,7 @@ impl From<&protobuf::StringifiedPlan> for StringifiedPlan {
impl From<protobuf::BuiltInWindowFunction> for BuiltInWindowFunction {
fn from(built_in_function: protobuf::BuiltInWindowFunction) -> Self {
match built_in_function {
- protobuf::BuiltInWindowFunction::RowNumber => Self::RowNumber,
+ protobuf::BuiltInWindowFunction::Unspecified => todo!(),
protobuf::BuiltInWindowFunction::Rank => Self::Rank,
protobuf::BuiltInWindowFunction::PercentRank => Self::PercentRank,
protobuf::BuiltInWindowFunction::DenseRank => Self::DenseRank,
diff --git a/datafusion/proto/src/logical_plan/to_proto.rs
b/datafusion/proto/src/logical_plan/to_proto.rs
index c7361c89c3..bb7bf84a33 100644
--- a/datafusion/proto/src/logical_plan/to_proto.rs
+++ b/datafusion/proto/src/logical_plan/to_proto.rs
@@ -120,7 +120,6 @@ impl From<&BuiltInWindowFunction> for
protobuf::BuiltInWindowFunction {
BuiltInWindowFunction::Ntile => Self::Ntile,
BuiltInWindowFunction::CumeDist => Self::CumeDist,
BuiltInWindowFunction::PercentRank => Self::PercentRank,
- BuiltInWindowFunction::RowNumber => Self::RowNumber,
BuiltInWindowFunction::Rank => Self::Rank,
BuiltInWindowFunction::Lag => Self::Lag,
BuiltInWindowFunction::Lead => Self::Lead,
diff --git a/datafusion/proto/src/physical_plan/to_proto.rs
b/datafusion/proto/src/physical_plan/to_proto.rs
index 57cd22a99a..7949a457f4 100644
--- a/datafusion/proto/src/physical_plan/to_proto.rs
+++ b/datafusion/proto/src/physical_plan/to_proto.rs
@@ -25,7 +25,7 @@ use datafusion::physical_expr::{PhysicalSortExpr,
ScalarFunctionExpr};
use datafusion::physical_plan::expressions::{
BinaryExpr, CaseExpr, CastExpr, Column, CumeDist, InListExpr,
IsNotNullExpr,
IsNullExpr, Literal, NegativeExpr, NotExpr, NthValue, Ntile, Rank,
RankType,
- RowNumber, TryCastExpr, WindowShift,
+ TryCastExpr, WindowShift,
};
use datafusion::physical_plan::udaf::AggregateFunctionExpr;
use datafusion::physical_plan::windows::{BuiltInWindowExpr,
PlainAggregateWindowExpr};
@@ -117,9 +117,8 @@ 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::<RowNumber>().is_some() {
- protobuf::BuiltInWindowFunction::RowNumber
- } else if let Some(rank_expr) =
built_in_fn_expr.downcast_ref::<Rank>() {
+ let builtin_fn = if let Some(rank_expr) =
built_in_fn_expr.downcast_ref::<Rank>()
+ {
match rank_expr.get_type() {
RankType::Basic => protobuf::BuiltInWindowFunction::Rank,
RankType::Dense => protobuf::BuiltInWindowFunction::DenseRank,
diff --git a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
index eb7cc5c4b9..4b20f14dde 100644
--- a/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
+++ b/datafusion/proto/tests/cases/roundtrip_logical_plan.rs
@@ -47,6 +47,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::row_number::row_number;
use datafusion::prelude::*;
use datafusion::test_util::{TestTableFactory, TestTableProvider};
use datafusion_common::config::TableOptions;
@@ -903,6 +904,7 @@ async fn roundtrip_expr_api() -> Result<()> {
vec![lit(1), lit(2), lit(3)],
vec![lit(10), lit(20), lit(30)],
),
+ row_number(vec![col("a")]),
];
// ensure expressions created with the expr api can be round tripped
diff --git a/datafusion/sql/Cargo.toml b/datafusion/sql/Cargo.toml
index c4ae3a8134..5c4b83fe38 100644
--- a/datafusion/sql/Cargo.toml
+++ b/datafusion/sql/Cargo.toml
@@ -55,6 +55,7 @@ strum = { version = "0.26.1", features = ["derive"] }
ctor = { workspace = true }
datafusion-functions = { workspace = true, default-features = true }
datafusion-functions-aggregate = { workspace = true }
+datafusion-functions-window = { workspace = true }
env_logger = { workspace = true }
paste = "^1.0"
rstest = { workspace = true }
diff --git a/datafusion/sql/src/unparser/expr.rs
b/datafusion/sql/src/unparser/expr.rs
index 39511ea4d0..c941d7098d 100644
--- a/datafusion/sql/src/unparser/expr.rs
+++ b/datafusion/sql/src/unparser/expr.rs
@@ -1531,6 +1531,7 @@ mod tests {
use datafusion_expr::{interval_month_day_nano_lit, ExprFunctionExt};
use datafusion_functions_aggregate::count::count_udaf;
use datafusion_functions_aggregate::expr_fn::sum;
+ use datafusion_functions_window::row_number::row_number_udwf;
use crate::unparser::dialect::{CustomDialect, CustomDialectBuilder};
@@ -1793,16 +1794,14 @@ mod tests {
),
(
Expr::WindowFunction(WindowFunction {
- fun: WindowFunctionDefinition::BuiltInWindowFunction(
- datafusion_expr::BuiltInWindowFunction::RowNumber,
- ),
+ fun:
WindowFunctionDefinition::WindowUDF(row_number_udwf()),
args: vec![col("col")],
partition_by: vec![],
order_by: vec![],
window_frame: WindowFrame::new(None),
null_treatment: None,
}),
- r#"ROW_NUMBER(col) OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND
UNBOUNDED FOLLOWING)"#,
+ r#"row_number(col) OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND
UNBOUNDED FOLLOWING)"#,
),
(
Expr::WindowFunction(WindowFunction {
diff --git a/datafusion/sqllogictest/test_files/functions.slt
b/datafusion/sqllogictest/test_files/functions.slt
index cb592fdda0..074e4ef834 100644
--- a/datafusion/sqllogictest/test_files/functions.slt
+++ b/datafusion/sqllogictest/test_files/functions.slt
@@ -587,7 +587,7 @@ statement error
SELECT v1, v2, SUMM(v2) OVER(ORDER BY v1) from test;
# Window function
-statement error Did you mean 'ROW_NUMBER'?
+statement error Did you mean 'row_number'?
SELECT v1, v2, ROWNUMBER() OVER(ORDER BY v1) from test;
statement ok
diff --git a/datafusion/sqllogictest/test_files/group_by.slt
b/datafusion/sqllogictest/test_files/group_by.slt
index 3d78bd06c3..efcc0e5b0f 100644
--- a/datafusion/sqllogictest/test_files/group_by.slt
+++ b/datafusion/sqllogictest/test_files/group_by.slt
@@ -3861,8 +3861,8 @@ logical_plan
06)----------Inner Join: l.d = r.d Filter: CAST(l.a AS Int64) >= CAST(r.a AS
Int64) - Int64(10)
07)------------SubqueryAlias: l
08)--------------TableScan: multiple_ordered_table projection=[a, d]
-09)------------Projection: r.a, r.d, ROW_NUMBER() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS row_n
-10)--------------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+09)------------Projection: r.a, r.d, row_number() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS row_n
+10)--------------WindowAggr: windowExpr=[[row_number() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
11)----------------SubqueryAlias: r
12)------------------TableScan: multiple_ordered_table projection=[a, d]
physical_plan
@@ -3871,8 +3871,8 @@ physical_plan
03)----CoalesceBatchesExec: target_batch_size=2
04)------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d@1, d@1)],
filter=CAST(a@0 AS Int64) >= CAST(a@1 AS Int64) - 10, projection=[a@0, d@1,
row_n@4]
05)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d],
output_ordering=[a@0 ASC NULLS LAST], has_header=true
-06)--------ProjectionExec: expr=[a@0 as a, d@1 as d, ROW_NUMBER() ORDER BY
[r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as
row_n]
-07)----------BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field { name:
"ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0,
dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range,
start_bound: Preceding(Int32(NULL)), end_bound: CurrentRow, is_causal: false
}], mode=[Sorted]
+06)--------ProjectionExec: expr=[a@0 as a, d@1 as d, row_number() ORDER BY
[r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as
row_n]
+07)----------BoundedWindowAggExec: wdw=[row_number() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field { name:
"row_number() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0,
dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range,
start_bound: Preceding(Int32(NULL)), end_bound: CurrentRow, is_causal: false
}], mode=[Sorted]
08)------------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d],
output_ordering=[a@0 ASC NULLS LAST], has_header=true
# reset partition number to 8.
diff --git a/datafusion/sqllogictest/test_files/joins.slt
b/datafusion/sqllogictest/test_files/joins.slt
index 441ccb7d99..fa0d4cbc49 100644
--- a/datafusion/sqllogictest/test_files/joins.slt
+++ b/datafusion/sqllogictest/test_files/joins.slt
@@ -3235,8 +3235,8 @@ logical_plan
01)Sort: l_table.rn1 ASC NULLS LAST
02)--Inner Join: l_table.a = r_table.a
03)----SubqueryAlias: l_table
-04)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
-05)--------WindowAggr: windowExpr=[[ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
+04)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
+05)--------WindowAggr: windowExpr=[[row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
06)----------TableScan: annotated_data projection=[a0, a, b, c, d]
07)----SubqueryAlias: r_table
08)------TableScan: annotated_data projection=[a0, a, b, c, d]
@@ -3246,8 +3246,8 @@ physical_plan
03)----CoalesceBatchesExec: target_batch_size=2
04)------RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=2,
preserve_order=true, sort_exprs=a@1 ASC,b@2 ASC NULLS LAST,c@3 ASC NULLS
LAST,rn1@5 ASC NULLS LAST
05)--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-06)----------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c,
d@4 as d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@5 as rn1]
-07)------------BoundedWindowAggExec: wdw=[ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "ROW_NUMBER() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
+06)----------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c,
d@4 as d, row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@5 as rn1]
+07)------------BoundedWindowAggExec: wdw=[row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "row_number() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
08)--------------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a,
b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST],
has_header=true
09)----CoalesceBatchesExec: target_batch_size=2
10)------RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=2,
preserve_order=true, sort_exprs=a@1 ASC,b@2 ASC NULLS LAST,c@3 ASC NULLS LAST
@@ -3271,8 +3271,8 @@ logical_plan
03)----SubqueryAlias: l_table
04)------TableScan: annotated_data projection=[a0, a, b, c, d]
05)----SubqueryAlias: r_table
-06)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
-07)--------WindowAggr: windowExpr=[[ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
+06)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
+07)--------WindowAggr: windowExpr=[[row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
08)----------TableScan: annotated_data projection=[a0, a, b, c, d]
physical_plan
01)SortPreservingMergeExec: [rn1@10 ASC NULLS LAST]
@@ -3284,8 +3284,8 @@ physical_plan
07)----CoalesceBatchesExec: target_batch_size=2
08)------RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=2,
preserve_order=true, sort_exprs=a@1 ASC,b@2 ASC NULLS LAST,c@3 ASC NULLS
LAST,rn1@5 ASC NULLS LAST
09)--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-10)----------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c,
d@4 as d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@5 as rn1]
-11)------------BoundedWindowAggExec: wdw=[ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "ROW_NUMBER() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
+10)----------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c,
d@4 as d, row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@5 as rn1]
+11)------------BoundedWindowAggExec: wdw=[row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "row_number() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
12)--------------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a,
b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST],
has_header=true
statement ok
@@ -3308,12 +3308,12 @@ logical_plan
01)Sort: l_table.a ASC NULLS FIRST, l_table.b ASC NULLS LAST, l_table.c ASC
NULLS LAST, r_table.rn1 ASC NULLS LAST
02)--Inner Join: l_table.a = r_table.a
03)----SubqueryAlias: l_table
-04)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
-05)--------WindowAggr: windowExpr=[[ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
+04)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
+05)--------WindowAggr: windowExpr=[[row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
06)----------TableScan: annotated_data projection=[a0, a, b, c, d]
07)----SubqueryAlias: r_table
-08)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
-09)--------WindowAggr: windowExpr=[[ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
+08)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
+09)--------WindowAggr: windowExpr=[[row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
10)----------TableScan: annotated_data projection=[a0, a, b, c, d]
physical_plan
01)SortPreservingMergeExec: [a@1 ASC,b@2 ASC NULLS LAST,c@3 ASC NULLS
LAST,rn1@11 ASC NULLS LAST]
@@ -3323,15 +3323,15 @@ physical_plan
05)--------CoalesceBatchesExec: target_batch_size=2
06)----------RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=2
07)------------RepartitionExec: partitioning=RoundRobinBatch(2),
input_partitions=1
-08)--------------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as
c, d@4 as d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@5 as rn1]
-09)----------------BoundedWindowAggExec: wdw=[ROW_NUMBER() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "ROW_NUMBER()
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
+08)--------------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as
c, d@4 as d, row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@5 as rn1]
+09)----------------BoundedWindowAggExec: wdw=[row_number() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "row_number()
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
10)------------------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a,
b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST],
has_header=true
11)------SortExec: expr=[a@1 ASC], preserve_partitioning=[true]
12)--------CoalesceBatchesExec: target_batch_size=2
13)----------RepartitionExec: partitioning=Hash([a@1], 2), input_partitions=2
14)------------RepartitionExec: partitioning=RoundRobinBatch(2),
input_partitions=1
-15)--------------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as
c, d@4 as d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@5 as rn1]
-16)----------------BoundedWindowAggExec: wdw=[ROW_NUMBER() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "ROW_NUMBER()
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
+15)--------------ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as
c, d@4 as d, row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@5 as rn1]
+16)----------------BoundedWindowAggExec: wdw=[row_number() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "row_number()
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
17)------------------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a,
b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST],
has_header=true
statement ok
@@ -3358,15 +3358,15 @@ logical_plan
03)----SubqueryAlias: l_table
04)------TableScan: annotated_data projection=[a0, a, b, c, d]
05)----SubqueryAlias: r_table
-06)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
-07)--------WindowAggr: windowExpr=[[ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
+06)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
+07)--------WindowAggr: windowExpr=[[row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
08)----------TableScan: annotated_data projection=[a0, a, b, c, d]
physical_plan
01)CoalesceBatchesExec: target_batch_size=2
02)--HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@1, a@1)]
03)----CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a,
b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST],
has_header=true
-04)----ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c, d@4 as
d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@5 as
rn1]
-05)------BoundedWindowAggExec: wdw=[ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "ROW_NUMBER() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
+04)----ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c, d@4 as
d, row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@5 as
rn1]
+05)------BoundedWindowAggExec: wdw=[row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "row_number() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
06)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a,
b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST],
has_header=true
# hash join should propagate ordering equivalence of the right side for RIGHT
ANTI join.
@@ -3385,15 +3385,15 @@ logical_plan
03)----SubqueryAlias: l_table
04)------TableScan: annotated_data projection=[a]
05)----SubqueryAlias: r_table
-06)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
-07)--------WindowAggr: windowExpr=[[ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
+06)------Projection: annotated_data.a0, annotated_data.a, annotated_data.b,
annotated_data.c, annotated_data.d, row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING AS rn1
+07)--------WindowAggr: windowExpr=[[row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING]]
08)----------TableScan: annotated_data projection=[a0, a, b, c, d]
physical_plan
01)CoalesceBatchesExec: target_batch_size=2
02)--HashJoinExec: mode=CollectLeft, join_type=RightAnti, on=[(a@0, a@1)]
03)----CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a],
output_ordering=[a@0 ASC], has_header=true
-04)----ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c, d@4 as
d, ROW_NUMBER() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@5 as
rn1]
-05)------BoundedWindowAggExec: wdw=[ROW_NUMBER() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "ROW_NUMBER() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
+04)----ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c, d@4 as
d, row_number() ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@5 as
rn1]
+05)------BoundedWindowAggExec: wdw=[row_number() ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "row_number() ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode=[Sorted]
06)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a0, a,
b, c, d], output_ordering=[a@1 ASC, b@2 ASC NULLS LAST, c@3 ASC NULLS LAST],
has_header=true
query TT
@@ -3457,8 +3457,8 @@ logical_plan
06)----------Inner Join: l.d = r.d Filter: CAST(l.a AS Int64) >= CAST(r.a AS
Int64) - Int64(10)
07)------------SubqueryAlias: l
08)--------------TableScan: multiple_ordered_table projection=[a, d]
-09)------------Projection: r.a, r.d, ROW_NUMBER() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS row_n
-10)--------------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+09)------------Projection: r.a, r.d, row_number() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS row_n
+10)--------------WindowAggr: windowExpr=[[row_number() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
11)----------------SubqueryAlias: r
12)------------------TableScan: multiple_ordered_table projection=[a, d]
physical_plan
@@ -3467,8 +3467,8 @@ physical_plan
03)----CoalesceBatchesExec: target_batch_size=2
04)------HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(d@1, d@1)],
filter=CAST(a@0 AS Int64) >= CAST(a@1 AS Int64) - 10, projection=[a@0, d@1,
row_n@4]
05)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d],
output_ordering=[a@0 ASC NULLS LAST], has_header=true
-06)--------ProjectionExec: expr=[a@0 as a, d@1 as d, ROW_NUMBER() ORDER BY
[r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as
row_n]
-07)----------BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field { name:
"ROW_NUMBER() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0,
dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range,
start_bound: Preceding(Int32(NULL)), end_bound: CurrentRow, is_causal: false
}], mode=[Sorted]
+06)--------ProjectionExec: expr=[a@0 as a, d@1 as d, row_number() ORDER BY
[r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as
row_n]
+07)----------BoundedWindowAggExec: wdw=[row_number() ORDER BY [r.a ASC NULLS
LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field { name:
"row_number() ORDER BY [r.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0,
dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range,
start_bound: Preceding(Int32(NULL)), end_bound: CurrentRow, is_causal: false
}], mode=[Sorted]
08)------------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_2.csv]]}, projection=[a, d],
output_ordering=[a@0 ASC NULLS LAST], has_header=true
# run query above in multiple partitions
diff --git a/datafusion/sqllogictest/test_files/window.slt
b/datafusion/sqllogictest/test_files/window.slt
index f56ac414a3..0bf7a8a1eb 100644
--- a/datafusion/sqllogictest/test_files/window.slt
+++ b/datafusion/sqllogictest/test_files/window.slt
@@ -342,8 +342,8 @@ logical_plan
03)----Aggregate: groupBy=[[d.b]], aggr=[[max(d.a), max(d.seq)]]
04)------SubqueryAlias: d
05)--------SubqueryAlias: _data2
-06)----------Projection: ROW_NUMBER() PARTITION BY [s.b] ORDER BY [s.a ASC
NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS seq, s.a, s.b
-07)------------WindowAggr: windowExpr=[[ROW_NUMBER() PARTITION BY [s.b] ORDER
BY [s.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
+06)----------Projection: row_number() PARTITION BY [s.b] ORDER BY [s.a ASC
NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS seq, s.a, s.b
+07)------------WindowAggr: windowExpr=[[row_number() PARTITION BY [s.b] ORDER
BY [s.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW]]
08)--------------SubqueryAlias: s
09)----------------SubqueryAlias: _sample_data
10)------------------Union
@@ -359,8 +359,8 @@ physical_plan
01)SortPreservingMergeExec: [b@0 ASC NULLS LAST]
02)--ProjectionExec: expr=[b@0 as b, max(d.a)@1 as max_a, max(d.seq)@2 as
max(d.seq)]
03)----AggregateExec: mode=SinglePartitioned, gby=[b@2 as b], aggr=[max(d.a),
max(d.seq)], ordering_mode=Sorted
-04)------ProjectionExec: expr=[ROW_NUMBER() PARTITION BY [s.b] ORDER BY [s.a
ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as seq, a@0
as a, b@1 as b]
-05)--------BoundedWindowAggExec: wdw=[ROW_NUMBER() PARTITION BY [s.b] ORDER BY
[s.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW:
Ok(Field { name: "ROW_NUMBER() PARTITION BY [s.b] ORDER BY [s.a ASC NULLS LAST]
RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Range, start_bound: Preceding(Int64(NULL)), end_bound:
CurrentRow, is_causal: false }], mode [...]
+04)------ProjectionExec: expr=[row_number() PARTITION BY [s.b] ORDER BY [s.a
ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@2 as seq, a@0
as a, b@1 as b]
+05)--------BoundedWindowAggExec: wdw=[row_number() PARTITION BY [s.b] ORDER BY
[s.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW:
Ok(Field { name: "row_number() PARTITION BY [s.b] ORDER BY [s.a ASC NULLS LAST]
RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Range, start_bound: Preceding(Int64(NULL)), end_bound:
CurrentRow, is_causal: false }], mode [...]
06)----------SortExec: expr=[b@1 ASC NULLS LAST,a@0 ASC NULLS LAST],
preserve_partitioning=[true]
07)------------CoalesceBatchesExec: target_batch_size=8192
08)--------------RepartitionExec: partitioning=Hash([b@1], 4),
input_partitions=4
@@ -1419,17 +1419,17 @@ EXPLAIN SELECT
LIMIT 5
----
logical_plan
-01)Projection: aggregate_test_100.c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING
AS rn1, ROW_NUMBER() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] ROWS
BETWEEN 1 PRECEDING AND 5 FOLLOWING AS rn2
+01)Projection: aggregate_test_100.c9, row_number() ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING
AS rn1, row_number() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] ROWS
BETWEEN 1 PRECEDING AND 5 FOLLOWING AS rn2
02)--Limit: skip=0, fetch=5
-03)----WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY [aggregate_test_100.c9
ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING]]
-04)------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING]]
+03)----WindowAggr: windowExpr=[[row_number() ORDER BY [aggregate_test_100.c9
ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING]]
+04)------WindowAggr: windowExpr=[[row_number() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING]]
05)--------TableScan: aggregate_test_100 projection=[c9]
physical_plan
-01)ProjectionExec: expr=[c9@0 as c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5
FOLLOWING@2 as rn1, ROW_NUMBER() ORDER BY [aggregate_test_100.c9 DESC NULLS
FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING@1 as rn2]
+01)ProjectionExec: expr=[c9@0 as c9, row_number() ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5
FOLLOWING@2 as rn1, row_number() ORDER BY [aggregate_test_100.c9 DESC NULLS
FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING@1 as rn2]
02)--GlobalLimitExec: skip=0, fetch=5
-03)----BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [aggregate_test_100.c9
ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING: Ok(Field { name:
"ROW_NUMBER() ORDER BY [aggregate_test_100.c9 ASC NULLS LAST] ROWS BETWEEN 1
PRECEDING AND 5 FOLLOWING", data_type: UInt64, nullable: false, dict_id: 0,
dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Rows,
start_bound: Preceding(UInt64(1)), end_bound: Following(UInt64(5)), is_causal:
false }], mode=[Sorted]
+03)----BoundedWindowAggExec: wdw=[row_number() ORDER BY [aggregate_test_100.c9
ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING: Ok(Field { name:
"row_number() ORDER BY [aggregate_test_100.c9 ASC NULLS LAST] ROWS BETWEEN 1
PRECEDING AND 5 FOLLOWING", data_type: UInt64, nullable: false, dict_id: 0,
dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Rows,
start_bound: Preceding(UInt64(1)), end_bound: Following(UInt64(5)), is_causal:
false }], mode=[Sorted]
04)------SortExec: expr=[c9@0 ASC NULLS LAST], preserve_partitioning=[false]
-05)--------BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5
FOLLOWING: Ok(Field { name: "ROW_NUMBER() ORDER BY [aggregate_test_100.c9 DESC
NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(1)), end_bound:
Following(UInt64(5)), is_causal: false }], mode=[Sorted]
+05)--------BoundedWindowAggExec: wdw=[row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5
FOLLOWING: Ok(Field { name: "row_number() ORDER BY [aggregate_test_100.c9 DESC
NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(1)), end_bound:
Following(UInt64(5)), is_causal: false }], mode=[Sorted]
06)----------SortExec: expr=[c9@0 DESC], preserve_partitioning=[false]
07)------------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c9],
has_header=true
@@ -1460,18 +1460,18 @@ EXPLAIN SELECT
LIMIT 5
----
logical_plan
-01)Projection: aggregate_test_100.c9, sum(aggregate_test_100.c9) ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST, aggregate_test_100.c1 ASC NULLS LAST,
aggregate_test_100.c2 ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING
AS sum1, sum(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c9 DESC NULLS
FIRST, aggregate_test_100.c1 DESC NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5
FOLLOWING AS sum2, ROW_NUMBER() ORDER BY [aggregate_test_100.c9 DESC NULLS
FIRST] ROWS BETWEEN 1 PRECEDIN [...]
+01)Projection: aggregate_test_100.c9, sum(aggregate_test_100.c9) ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST, aggregate_test_100.c1 ASC NULLS LAST,
aggregate_test_100.c2 ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING
AS sum1, sum(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c9 DESC NULLS
FIRST, aggregate_test_100.c1 DESC NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5
FOLLOWING AS sum2, row_number() ORDER BY [aggregate_test_100.c9 DESC NULLS
FIRST] ROWS BETWEEN 1 PRECEDIN [...]
02)--Limit: skip=0, fetch=5
03)----WindowAggr: windowExpr=[[sum(aggregate_test_100.c9) ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST, aggregate_test_100.c1 ASC NULLS LAST,
aggregate_test_100.c2 ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING]]
-04)------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING]]
+04)------WindowAggr: windowExpr=[[row_number() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING]]
05)--------WindowAggr: windowExpr=[[sum(aggregate_test_100.c9) ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST, aggregate_test_100.c1 DESC NULLS
FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING]]
06)----------TableScan: aggregate_test_100 projection=[c1, c2, c9]
physical_plan
-01)ProjectionExec: expr=[c9@2 as c9, sum(aggregate_test_100.c9) ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST, aggregate_test_100.c1 ASC NULLS LAST,
aggregate_test_100.c2 ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5
FOLLOWING@5 as sum1, sum(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST, aggregate_test_100.c1 DESC NULLS FIRST] ROWS BETWEEN 1
PRECEDING AND 5 FOLLOWING@3 as sum2, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] ROWS BETWEEN 1 PRECE [...]
+01)ProjectionExec: expr=[c9@2 as c9, sum(aggregate_test_100.c9) ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST, aggregate_test_100.c1 ASC NULLS LAST,
aggregate_test_100.c2 ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5
FOLLOWING@5 as sum1, sum(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST, aggregate_test_100.c1 DESC NULLS FIRST] ROWS BETWEEN 1
PRECEDING AND 5 FOLLOWING@3 as sum2, row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] ROWS BETWEEN 1 PRECE [...]
02)--GlobalLimitExec: skip=0, fetch=5
03)----BoundedWindowAggExec: wdw=[sum(aggregate_test_100.c9) ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST, aggregate_test_100.c1 ASC NULLS LAST,
aggregate_test_100.c2 ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING:
Ok(Field { name: "sum(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c9
ASC NULLS LAST, aggregate_test_100.c1 ASC NULLS LAST, aggregate_test_100.c2 ASC
NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING", data_type: UInt64,
nullable: true, dict_id: 0, dict_i [...]
04)------SortExec: expr=[c9@2 ASC NULLS LAST,c1@0 ASC NULLS LAST,c2@1 ASC
NULLS LAST], preserve_partitioning=[false]
-05)--------BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5
FOLLOWING: Ok(Field { name: "ROW_NUMBER() ORDER BY [aggregate_test_100.c9 DESC
NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(1)), end_bound:
Following(UInt64(5)), is_causal: false }], mode=[Sorted]
+05)--------BoundedWindowAggExec: wdw=[row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5
FOLLOWING: Ok(Field { name: "row_number() ORDER BY [aggregate_test_100.c9 DESC
NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(1)), end_bound:
Following(UInt64(5)), is_causal: false }], mode=[Sorted]
06)----------BoundedWindowAggExec: wdw=[sum(aggregate_test_100.c9) ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST, aggregate_test_100.c1 DESC NULLS
FIRST] ROWS BETWEEN 1 PRECEDING AND 5 FOLLOWING: Ok(Field { name:
"sum(aggregate_test_100.c9) ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST,
aggregate_test_100.c1 DESC NULLS FIRST] ROWS BETWEEN 1 PRECEDING AND 5
FOLLOWING", data_type: UInt64, nullable: true, dict_id: 0, dict_is_ordered:
false, metadata: {} }), frame: WindowFrame { units: [...]
07)------------SortExec: expr=[c9@2 DESC,c1@0 DESC],
preserve_partitioning=[false]
08)--------------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c1,
c2, c9], has_header=true
@@ -1863,13 +1863,13 @@ EXPLAIN SELECT c1, ROW_NUMBER() OVER (PARTITION BY c1)
as rn1 FROM aggregate_tes
----
logical_plan
01)Sort: aggregate_test_100.c1 ASC NULLS LAST
-02)--Projection: aggregate_test_100.c1, ROW_NUMBER() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING AS rn1
-03)----WindowAggr: windowExpr=[[ROW_NUMBER() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING]]
+02)--Projection: aggregate_test_100.c1, row_number() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING AS rn1
+03)----WindowAggr: windowExpr=[[row_number() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING]]
04)------TableScan: aggregate_test_100 projection=[c1]
physical_plan
01)SortPreservingMergeExec: [c1@0 ASC NULLS LAST]
-02)--ProjectionExec: expr=[c1@0 as c1, ROW_NUMBER() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@1 as rn1]
-03)----BoundedWindowAggExec: wdw=[ROW_NUMBER() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING: Ok(Field { name: "ROW_NUMBER() PARTITION BY [aggregate_test_100.c1]
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode= [...]
+02)--ProjectionExec: expr=[c1@0 as c1, row_number() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@1 as rn1]
+03)----BoundedWindowAggExec: wdw=[row_number() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING: Ok(Field { name: "row_number() PARTITION BY [aggregate_test_100.c1]
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode= [...]
04)------SortExec: expr=[c1@0 ASC NULLS LAST], preserve_partitioning=[true]
05)--------CoalesceBatchesExec: target_batch_size=4096
06)----------RepartitionExec: partitioning=Hash([c1@0], 2), input_partitions=2
@@ -1992,13 +1992,13 @@ EXPLAIN SELECT c1, ROW_NUMBER() OVER (PARTITION BY c1)
as rn1 FROM aggregate_tes
----
logical_plan
01)Sort: aggregate_test_100.c1 ASC NULLS LAST
-02)--Projection: aggregate_test_100.c1, ROW_NUMBER() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING AS rn1
-03)----WindowAggr: windowExpr=[[ROW_NUMBER() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING]]
+02)--Projection: aggregate_test_100.c1, row_number() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING AS rn1
+03)----WindowAggr: windowExpr=[[row_number() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING]]
04)------TableScan: aggregate_test_100 projection=[c1]
physical_plan
01)SortPreservingMergeExec: [c1@0 ASC NULLS LAST,rn1@1 ASC NULLS LAST]
-02)--ProjectionExec: expr=[c1@0 as c1, ROW_NUMBER() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@1 as rn1]
-03)----BoundedWindowAggExec: wdw=[ROW_NUMBER() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING: Ok(Field { name: "ROW_NUMBER() PARTITION BY [aggregate_test_100.c1]
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode= [...]
+02)--ProjectionExec: expr=[c1@0 as c1, row_number() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING@1 as rn1]
+03)----BoundedWindowAggExec: wdw=[row_number() PARTITION BY
[aggregate_test_100.c1] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED
FOLLOWING: Ok(Field { name: "row_number() PARTITION BY [aggregate_test_100.c1]
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING", data_type: UInt64,
nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame:
WindowFrame { units: Rows, start_bound: Preceding(UInt64(NULL)), end_bound:
Following(UInt64(NULL)), is_causal: false }], mode= [...]
04)------SortExec: expr=[c1@0 ASC NULLS LAST], preserve_partitioning=[true]
05)--------CoalesceBatchesExec: target_batch_size=4096
06)----------RepartitionExec: partitioning=Hash([c1@0], 2), input_partitions=2
@@ -2249,13 +2249,13 @@ logical_plan
01)Limit: skip=0, fetch=5
02)--Sort: rn1 ASC NULLS LAST, fetch=5
03)----Sort: aggregate_test_100.c9 ASC NULLS LAST
-04)------Projection: aggregate_test_100.c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS rn1
-05)--------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
+04)------Projection: aggregate_test_100.c9, row_number() ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS rn1
+05)--------WindowAggr: windowExpr=[[row_number() ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
06)----------TableScan: aggregate_test_100 projection=[c9]
physical_plan
-01)ProjectionExec: expr=[c9@0 as c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 as rn1]
+01)ProjectionExec: expr=[c9@0 as c9, row_number() ORDER BY
[aggregate_test_100.c9 ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 as rn1]
02)--GlobalLimitExec: skip=0, fetch=5
-03)----BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [aggregate_test_100.c9
ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "ROW_NUMBER() ORDER BY [aggregate_test_100.c9 ASC NULLS LAST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode=[Sorted]
+03)----BoundedWindowAggExec: wdw=[row_number() ORDER BY [aggregate_test_100.c9
ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "row_number() ORDER BY [aggregate_test_100.c9 ASC NULLS LAST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode=[Sorted]
04)------SortExec: expr=[c9@0 ASC NULLS LAST], preserve_partitioning=[false]
05)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c9],
has_header=true
@@ -2288,13 +2288,13 @@ logical_plan
01)Limit: skip=0, fetch=5
02)--Sort: rn1 ASC NULLS LAST, fetch=5
03)----Sort: aggregate_test_100.c9 DESC NULLS FIRST
-04)------Projection: aggregate_test_100.c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS rn1
-05)--------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
+04)------Projection: aggregate_test_100.c9, row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS rn1
+05)--------WindowAggr: windowExpr=[[row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
06)----------TableScan: aggregate_test_100 projection=[c9]
physical_plan
-01)ProjectionExec: expr=[c9@0 as c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 as rn1]
+01)ProjectionExec: expr=[c9@0 as c9, row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 as rn1]
02)--GlobalLimitExec: skip=0, fetch=5
-03)----BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "ROW_NUMBER() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode= [...]
+03)----BoundedWindowAggExec: wdw=[row_number() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "row_number() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode= [...]
04)------SortExec: expr=[c9@0 DESC], preserve_partitioning=[false]
05)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c9],
has_header=true
@@ -2327,13 +2327,13 @@ logical_plan
01)Limit: skip=0, fetch=5
02)--Sort: rn1 DESC NULLS FIRST, fetch=5
03)----Sort: aggregate_test_100.c9 DESC NULLS FIRST
-04)------Projection: aggregate_test_100.c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS rn1
-05)--------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
+04)------Projection: aggregate_test_100.c9, row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS rn1
+05)--------WindowAggr: windowExpr=[[row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
06)----------TableScan: aggregate_test_100 projection=[c9]
physical_plan
01)SortExec: TopK(fetch=5), expr=[rn1@1 DESC], preserve_partitioning=[false]
-02)--ProjectionExec: expr=[c9@0 as c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 as rn1]
-03)----BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "ROW_NUMBER() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode= [...]
+02)--ProjectionExec: expr=[c9@0 as c9, row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 as rn1]
+03)----BoundedWindowAggExec: wdw=[row_number() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "row_number() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode= [...]
04)------SortExec: expr=[c9@0 DESC], preserve_partitioning=[false]
05)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c9],
has_header=true
@@ -2369,13 +2369,13 @@ logical_plan
01)Limit: skip=0, fetch=5
02)--Sort: rn1 ASC NULLS LAST, aggregate_test_100.c9 ASC NULLS LAST, fetch=5
03)----Sort: aggregate_test_100.c9 DESC NULLS FIRST
-04)------Projection: aggregate_test_100.c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS rn1
-05)--------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
+04)------Projection: aggregate_test_100.c9, row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS rn1
+05)--------WindowAggr: windowExpr=[[row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
06)----------TableScan: aggregate_test_100 projection=[c9]
physical_plan
01)SortExec: TopK(fetch=5), expr=[rn1@1 ASC NULLS LAST,c9@0 ASC NULLS LAST],
preserve_partitioning=[false]
-02)--ProjectionExec: expr=[c9@0 as c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 as rn1]
-03)----BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "ROW_NUMBER() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode= [...]
+02)--ProjectionExec: expr=[c9@0 as c9, row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 as rn1]
+03)----BoundedWindowAggExec: wdw=[row_number() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "row_number() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode= [...]
04)------SortExec: expr=[c9@0 DESC], preserve_partitioning=[false]
05)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c9],
has_header=true
@@ -2422,13 +2422,13 @@ logical_plan
01)Limit: skip=0, fetch=5
02)--Sort: rn1 ASC NULLS LAST, aggregate_test_100.c9 DESC NULLS FIRST, fetch=5
03)----Sort: aggregate_test_100.c9 DESC NULLS FIRST
-04)------Projection: aggregate_test_100.c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS rn1
-05)--------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
+04)------Projection: aggregate_test_100.c9, row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS rn1
+05)--------WindowAggr: windowExpr=[[row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
06)----------TableScan: aggregate_test_100 projection=[c9]
physical_plan
-01)ProjectionExec: expr=[c9@0 as c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 as rn1]
+01)ProjectionExec: expr=[c9@0 as c9, row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 as rn1]
02)--GlobalLimitExec: skip=0, fetch=5
-03)----BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "ROW_NUMBER() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode= [...]
+03)----BoundedWindowAggExec: wdw=[row_number() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "row_number() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode= [...]
04)------SortExec: expr=[c9@0 DESC], preserve_partitioning=[false]
05)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c9],
has_header=true
@@ -2446,13 +2446,13 @@ logical_plan
01)Limit: skip=0, fetch=5
02)--Sort: rn1 ASC NULLS LAST, CAST(aggregate_test_100.c9 AS Int32) +
aggregate_test_100.c5 DESC NULLS FIRST, fetch=5
03)----Sort: CAST(aggregate_test_100.c9 AS Int32) + aggregate_test_100.c5 DESC
NULLS FIRST
-04)------Projection: aggregate_test_100.c5, aggregate_test_100.c9,
ROW_NUMBER() ORDER BY [aggregate_test_100.c9 + aggregate_test_100.c5 DESC NULLS
FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS rn1
-05)--------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY
[CAST(aggregate_test_100.c9 AS Int32) + aggregate_test_100.c5 DESC NULLS FIRST]
RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 + aggregate_test_100.c5 DESC NULLS FIRST] RANGE BETWEEN
UNBOUNDED PRECEDING AND CURRENT ROW]]
+04)------Projection: aggregate_test_100.c5, aggregate_test_100.c9,
row_number() ORDER BY [aggregate_test_100.c9 + aggregate_test_100.c5 DESC NULLS
FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS rn1
+05)--------WindowAggr: windowExpr=[[row_number() ORDER BY
[CAST(aggregate_test_100.c9 AS Int32) + aggregate_test_100.c5 DESC NULLS FIRST]
RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW AS row_number() ORDER BY
[aggregate_test_100.c9 + aggregate_test_100.c5 DESC NULLS FIRST] RANGE BETWEEN
UNBOUNDED PRECEDING AND CURRENT ROW]]
06)----------TableScan: aggregate_test_100 projection=[c5, c9]
physical_plan
-01)ProjectionExec: expr=[c5@0 as c5, c9@1 as c9, ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 + aggregate_test_100.c5 DESC NULLS FIRST] RANGE BETWEEN
UNBOUNDED PRECEDING AND CURRENT ROW@2 as rn1]
+01)ProjectionExec: expr=[c5@0 as c5, c9@1 as c9, row_number() ORDER BY
[aggregate_test_100.c9 + aggregate_test_100.c5 DESC NULLS FIRST] RANGE BETWEEN
UNBOUNDED PRECEDING AND CURRENT ROW@2 as rn1]
02)--GlobalLimitExec: skip=0, fetch=5
-03)----BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [aggregate_test_100.c9
+ aggregate_test_100.c5 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW: Ok(Field { name: "ROW_NUMBER() ORDER BY [aggregate_test_100.c9 +
aggregate_test_100.c5 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered:
false, metadata: {} }), frame: WindowFrame { units: Range, start_bound:
Preceding(Int32(NULL)), en [...]
+03)----BoundedWindowAggExec: wdw=[row_number() ORDER BY [aggregate_test_100.c9
+ aggregate_test_100.c5 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW: Ok(Field { name: "row_number() ORDER BY [aggregate_test_100.c9 +
aggregate_test_100.c5 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered:
false, metadata: {} }), frame: WindowFrame { units: Range, start_bound:
Preceding(Int32(NULL)), en [...]
04)------SortExec: expr=[CAST(c9@1 AS Int32) + c5@0 DESC],
preserve_partitioning=[false]
05)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c5,
c9], has_header=true
@@ -2469,13 +2469,13 @@ logical_plan
01)Limit: skip=0, fetch=5
02)--Sort: rn1 ASC NULLS LAST, fetch=5
03)----Sort: aggregate_test_100.c9 DESC NULLS FIRST
-04)------Projection: aggregate_test_100.c9, CAST(ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS Int64) AS rn1
-05)--------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
+04)------Projection: aggregate_test_100.c9, CAST(row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW AS Int64) AS rn1
+05)--------WindowAggr: windowExpr=[[row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW]]
06)----------TableScan: aggregate_test_100 projection=[c9]
physical_plan
-01)ProjectionExec: expr=[c9@0 as c9, CAST(ROW_NUMBER() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 AS Int64) as rn1]
+01)ProjectionExec: expr=[c9@0 as c9, CAST(row_number() ORDER BY
[aggregate_test_100.c9 DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND
CURRENT ROW@1 AS Int64) as rn1]
02)--GlobalLimitExec: skip=0, fetch=5
-03)----BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "ROW_NUMBER() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode= [...]
+03)----BoundedWindowAggExec: wdw=[row_number() ORDER BY [aggregate_test_100.c9
DESC NULLS FIRST] RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW: Ok(Field {
name: "row_number() ORDER BY [aggregate_test_100.c9 DESC NULLS FIRST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW", data_type: UInt64, nullable:
false, dict_id: 0, dict_is_ordered: false, metadata: {} }), frame: WindowFrame
{ units: Range, start_bound: Preceding(UInt64(NULL)), end_bound: CurrentRow,
is_causal: false }], mode= [...]
04)------SortExec: expr=[c9@0 DESC], preserve_partitioning=[false]
05)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100.csv]]}, projection=[c9],
has_header=true
@@ -2661,14 +2661,14 @@ EXPLAIN SELECT
logical_plan
01)Limit: skip=0, fetch=5
02)--Sort: annotated_data_finite.ts DESC NULLS FIRST, fetch=5
-03)----Projection: annotated_data_finite.ts,
first_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING AS fv1,
first_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING AS fv2,
last_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING AS lv1,
last_value(ann [...]
-04)------WindowAggr: windowExpr=[[first_value(annotated_data_finite.inc_col)
ORDER BY [annotated_data_finite.ts ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING
AND 1 FOLLOWING, first_value(annotated_data_finite.inc_col) ORDER BY
[annotated_data_finite.ts ASC NULLS LAST] ROWS BETWEEN 10 PRECEDING AND 1
FOLLOWING, last_value(annotated_data_finite.inc_col) ORDER BY
[annotated_data_finite.ts ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING AND 1
FOLLOWING, last_value(annotated_data_finite.inc_col) ORDE [...]
+03)----Projection: annotated_data_finite.ts,
first_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING AS fv1,
first_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING AS fv2,
last_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING AS lv1,
last_value(ann [...]
+04)------WindowAggr: windowExpr=[[first_value(annotated_data_finite.inc_col)
ORDER BY [annotated_data_finite.ts ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING
AND 1 FOLLOWING, first_value(annotated_data_finite.inc_col) ORDER BY
[annotated_data_finite.ts ASC NULLS LAST] ROWS BETWEEN 10 PRECEDING AND 1
FOLLOWING, last_value(annotated_data_finite.inc_col) ORDER BY
[annotated_data_finite.ts ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING AND 1
FOLLOWING, last_value(annotated_data_finite.inc_col) ORDE [...]
05)--------WindowAggr: windowExpr=[[first_value(annotated_data_finite.inc_col)
ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING
AND 1 FOLLOWING, first_value(annotated_data_finite.inc_col) ORDER BY
[annotated_data_finite.ts DESC NULLS FIRST] ROWS BETWEEN 10 PRECEDING AND 1
FOLLOWING, last_value(annotated_data_finite.inc_col) ORDER BY
[annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING AND 1
FOLLOWING, last_value(annotated_data_finite.inc_c [...]
06)----------TableScan: annotated_data_finite projection=[ts, inc_col]
physical_plan
01)SortExec: TopK(fetch=5), expr=[ts@0 DESC], preserve_partitioning=[false]
-02)--ProjectionExec: expr=[ts@0 as ts,
first_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING@10 as fv1,
first_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING@11 as fv2,
last_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING@12 as lv1,
last_value( [...]
-03)----BoundedWindowAggExec: wdw=[first_value(annotated_data_finite.inc_col)
ORDER BY [annotated_data_finite.ts ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING
AND 1 FOLLOWING: Ok(Field { name: "first_value(annotated_data_finite.inc_col)
ORDER BY [annotated_data_finite.ts ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING
AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0,
dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range,
start_bound: Preceding(Int32(10)), end_bo [...]
+02)--ProjectionExec: expr=[ts@0 as ts,
first_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING@10 as fv1,
first_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] ROWS BETWEEN 10 PRECEDING AND 1 FOLLOWING@11 as fv2,
last_value(annotated_data_finite.inc_col) ORDER BY [annotated_data_finite.ts
ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING AND 1 FOLLOWING@12 as lv1,
last_value( [...]
+03)----BoundedWindowAggExec: wdw=[first_value(annotated_data_finite.inc_col)
ORDER BY [annotated_data_finite.ts ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING
AND 1 FOLLOWING: Ok(Field { name: "first_value(annotated_data_finite.inc_col)
ORDER BY [annotated_data_finite.ts ASC NULLS LAST] RANGE BETWEEN 10 PRECEDING
AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0,
dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range,
start_bound: Preceding(Int32(10)), end_bo [...]
04)------BoundedWindowAggExec: wdw=[first_value(annotated_data_finite.inc_col)
ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING
AND 1 FOLLOWING: Ok(Field { name: "first_value(annotated_data_finite.inc_col)
ORDER BY [annotated_data_finite.ts DESC NULLS FIRST] RANGE BETWEEN 10 PRECEDING
AND 1 FOLLOWING", data_type: Int32, nullable: true, dict_id: 0,
dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range,
start_bound: Preceding(Int32(1)), e [...]
05)--------CsvExec: file_groups={1 group:
[[WORKSPACE_ROOT/datafusion/core/tests/data/window_1.csv]]}, projection=[ts,
inc_col], output_ordering=[ts@0 ASC NULLS LAST], has_header=true
@@ -3124,15 +3124,15 @@ logical_plan
02)--Filter: rn1 < UInt64(50)
03)----Limit: skip=0, fetch=5
04)------Sort: rn1 ASC NULLS LAST, fetch=5
-05)--------Projection: annotated_data_infinite2.a0,
annotated_data_infinite2.a, annotated_data_infinite2.b,
annotated_data_infinite2.c, annotated_data_infinite2.d, ROW_NUMBER() ORDER BY
[annotated_data_infinite2.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW AS rn1
-06)----------WindowAggr: windowExpr=[[ROW_NUMBER() ORDER BY
[annotated_data_infinite2.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW]]
+05)--------Projection: annotated_data_infinite2.a0,
annotated_data_infinite2.a, annotated_data_infinite2.b,
annotated_data_infinite2.c, annotated_data_infinite2.d, row_number() ORDER BY
[annotated_data_infinite2.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW AS rn1
+06)----------WindowAggr: windowExpr=[[row_number() ORDER BY
[annotated_data_infinite2.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW]]
07)------------TableScan: annotated_data_infinite2 projection=[a0, a, b, c, d]
physical_plan
01)CoalesceBatchesExec: target_batch_size=4096
02)--FilterExec: rn1@5 < 50
-03)----ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c, d@4 as
d, ROW_NUMBER() ORDER BY [annotated_data_infinite2.a ASC NULLS LAST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@5 as rn1]
+03)----ProjectionExec: expr=[a0@0 as a0, a@1 as a, b@2 as b, c@3 as c, d@4 as
d, row_number() ORDER BY [annotated_data_infinite2.a ASC NULLS LAST] RANGE
BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW@5 as rn1]
04)------GlobalLimitExec: skip=0, fetch=5
-05)--------BoundedWindowAggExec: wdw=[ROW_NUMBER() ORDER BY
[annotated_data_infinite2.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW: Ok(Field { name: "ROW_NUMBER() ORDER BY
[annotated_data_infinite2.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0,
dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range,
start_bound: Preceding(Int32(NULL)), end_bound: CurrentRow, is_causal: false
[...]
+05)--------BoundedWindowAggExec: wdw=[row_number() ORDER BY
[annotated_data_infinite2.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW: Ok(Field { name: "row_number() ORDER BY
[annotated_data_infinite2.a ASC NULLS LAST] RANGE BETWEEN UNBOUNDED PRECEDING
AND CURRENT ROW", data_type: UInt64, nullable: false, dict_id: 0,
dict_is_ordered: false, metadata: {} }), frame: WindowFrame { units: Range,
start_bound: Preceding(Int32(NULL)), end_bound: CurrentRow, is_causal: false
[...]
06)----------StreamingTableExec: partition_sizes=1, projection=[a0, a, b, c,
d], infinite_source=true, output_ordering=[a@1 ASC NULLS LAST, b@2 ASC NULLS
LAST, c@3 ASC NULLS LAST]
# this is a negative test for asserting that window functions (other than
ROW_NUMBER)
@@ -4147,13 +4147,13 @@ query TT
EXPLAIN select ROW_NUMBER() over (partition by a) from (select * from a where
a = 1);
----
logical_plan
-01)Projection: ROW_NUMBER() PARTITION BY [a.a] ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING
-02)--WindowAggr: windowExpr=[[ROW_NUMBER() PARTITION BY [a.a] ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]]
+01)Projection: row_number() PARTITION BY [a.a] ROWS BETWEEN UNBOUNDED
PRECEDING AND UNBOUNDED FOLLOWING
+02)--WindowAggr: windowExpr=[[row_number() PARTITION BY [a.a] ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]]
03)----Filter: a.a = Int64(1)
04)------TableScan: a projection=[a]
physical_plan
-01)ProjectionExec: expr=[ROW_NUMBER() PARTITION BY [a.a] ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@1 as ROW_NUMBER() PARTITION BY
[a.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]
-02)--BoundedWindowAggExec: wdw=[ROW_NUMBER() PARTITION BY [a.a] ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "ROW_NUMBER()
PARTITION BY [a.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING",
data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false,
metadata: {} }), frame: WindowFrame { units: Rows, start_bound:
Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false
}], mode=[Sorted]
+01)ProjectionExec: expr=[row_number() PARTITION BY [a.a] ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING@1 as row_number() PARTITION BY
[a.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING]
+02)--BoundedWindowAggExec: wdw=[row_number() PARTITION BY [a.a] ROWS BETWEEN
UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING: Ok(Field { name: "row_number()
PARTITION BY [a.a] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING",
data_type: UInt64, nullable: false, dict_id: 0, dict_is_ordered: false,
metadata: {} }), frame: WindowFrame { units: Rows, start_bound:
Preceding(UInt64(NULL)), end_bound: Following(UInt64(NULL)), is_causal: false
}], mode=[Sorted]
03)----CoalesceBatchesExec: target_batch_size=4096
04)------RepartitionExec: partitioning=Hash([a@0], 2), input_partitions=2
05)--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
