Skip to content

Commit 1dd5354

Browse files
jayzhan211jayzhan211
authored
Move AggregateExpr, PhysicalExpr and PhysicalSortExpr to physical-expr-core (#9926)
* move PhysicalExpr Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * cleanup Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * move physical sort Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * cleanup dependencies Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * add readme Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * disable doc test Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * move column Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * fmt Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * move aggregatexp Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * move other two utils Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * license Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * switch to ignore Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * move reverse order Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * rename to common Signed-off-by: jayzhan211 <jayzhan211@gmail.com> * cleanup Signed-off-by: jayzhan211 <jayzhan211@gmail.com> --------- Signed-off-by: jayzhan211 <jayzhan211@gmail.com>
1 parent 701abf7 commit 1dd5354

23 files changed

Lines changed: 896 additions & 621 deletions

File tree

‎Cargo.toml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ members = [
2626
"datafusion/functions",
2727
"datafusion/functions-array",
2828
"datafusion/optimizer",
29+
"datafusion/physical-expr-common",
2930
"datafusion/physical-expr",
3031
"datafusion/physical-plan",
3132
"datafusion/proto",
@@ -80,6 +81,7 @@ datafusion-functions = { path = "datafusion/functions", version = "37.0.0" }
8081
datafusion-functions-array = { path = "datafusion/functions-array", version = "37.0.0" }
8182
datafusion-optimizer = { path = "datafusion/optimizer", version = "37.0.0", default-features = false }
8283
datafusion-physical-expr = { path = "datafusion/physical-expr", version = "37.0.0", default-features = false }
84+
datafusion-physical-expr-common = { path = "datafusion/physical-expr-common", version = "37.0.0", default-features = false }
8385
datafusion-physical-plan = { path = "datafusion/physical-plan", version = "37.0.0" }
8486
datafusion-proto = { path = "datafusion/proto", version = "37.0.0" }
8587
datafusion-sql = { path = "datafusion/sql", version = "37.0.0" }

‎datafusion-cli/Cargo.lock‎

Lines changed: 10 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎datafusion/physical-expr-common/Cargo.toml‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one
2+
# or more contributor license agreements. See the NOTICE file
3+
# distributed with this work for additional information
4+
# regarding copyright ownership. The ASF licenses this file
5+
# to you under the Apache License, Version 2.0 (the
6+
# "License"); you may not use this file except in compliance
7+
# with the License. You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing,
12+
# software distributed under the License is distributed on an
13+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
# KIND, either express or implied. See the License for the
15+
# specific language governing permissions and limitations
16+
# under the License.
17+
18+
[package]
19+
name = "datafusion-physical-expr-common"
20+
description = "Common functionality of physical expression for DataFusion query engine"
21+
keywords = ["arrow", "query", "sql"]
22+
readme = "README.md"
23+
version = { workspace = true }
24+
edition = { workspace = true }
25+
homepage = { workspace = true }
26+
repository = { workspace = true }
27+
license = { workspace = true }
28+
authors = { workspace = true }
29+
rust-version = { workspace = true }
30+
31+
[lib]
32+
name = "datafusion_physical_expr_common"
33+
path = "src/lib.rs"
34+
35+
[dependencies]
36+
arrow = { workspace = true }
37+
datafusion-common = { workspace = true, default-features = true }
38+
datafusion-expr = { workspace = true }

‎datafusion/physical-expr-common/README.md‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
<!---
2+
Licensed to the Apache Software Foundation (ASF) under one
3+
or more contributor license agreements. See the NOTICE file
4+
distributed with this work for additional information
5+
regarding copyright ownership. The ASF licenses this file
6+
to you under the Apache License, Version 2.0 (the
7+
"License"); you may not use this file except in compliance
8+
with the License. You may obtain a copy of the License at
9+
10+
http://www.apache.org/licenses/LICENSE-2.0
11+
12+
Unless required by applicable law or agreed to in writing,
13+
software distributed under the License is distributed on an
14+
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
KIND, either express or implied. See the License for the
16+
specific language governing permissions and limitations
17+
under the License.
18+
-->
19+
20+
# DataFusion Core Physical Expressions
21+
22+
[DataFusion][df] is an extensible query execution framework, written in Rust, that uses Apache Arrow as its in-memory format.
23+
24+
This crate is a submodule of DataFusion that provides shared APIs for implementing
25+
physical expressions such as `PhysicalExpr` and `PhysicalSortExpr`.
26+
27+
[df]: https://crates.io/crates/datafusion

‎datafusion/physical-expr-common/src/aggregate/mod.rs‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,102 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
pub mod utils;
19+
20+
use std::any::Any;
21+
use std::fmt::Debug;
22+
use std::sync::Arc;
23+
24+
use crate::physical_expr::PhysicalExpr;
25+
use crate::sort_expr::PhysicalSortExpr;
26+
27+
use arrow::datatypes::Field;
28+
use datafusion_common::{not_impl_err, Result};
29+
use datafusion_expr::{Accumulator, GroupsAccumulator};
30+
31+
/// An aggregate expression that:
32+
/// * knows its resulting field
33+
/// * knows how to create its accumulator
34+
/// * knows its accumulator's state's field
35+
/// * knows the expressions from whose its accumulator will receive values
36+
///
37+
/// Any implementation of this trait also needs to implement the
38+
/// `PartialEq<dyn Any>` to allows comparing equality between the
39+
/// trait objects.
40+
pub trait AggregateExpr: Send + Sync + Debug + PartialEq<dyn Any> {
41+
/// Returns the aggregate expression as [`Any`] so that it can be
42+
/// downcast to a specific implementation.
43+
fn as_any(&self) -> &dyn Any;
44+
45+
/// the field of the final result of this aggregation.
46+
fn field(&self) -> Result<Field>;
47+
48+
/// the accumulator used to accumulate values from the expressions.
49+
/// the accumulator expects the same number of arguments as `expressions` and must
50+
/// return states with the same description as `state_fields`
51+
fn create_accumulator(&self) -> Result<Box<dyn Accumulator>>;
52+
53+
/// the fields that encapsulate the Accumulator's state
54+
/// the number of fields here equals the number of states that the accumulator contains
55+
fn state_fields(&self) -> Result<Vec<Field>>;
56+
57+
/// expressions that are passed to the Accumulator.
58+
/// Single-column aggregations such as `sum` return a single value, others (e.g. `cov`) return many.
59+
fn expressions(&self) -> Vec<Arc<dyn PhysicalExpr>>;
60+
61+
/// Order by requirements for the aggregate function
62+
/// By default it is `None` (there is no requirement)
63+
/// Order-sensitive aggregators, such as `FIRST_VALUE(x ORDER BY y)` should implement this
64+
fn order_bys(&self) -> Option<&[PhysicalSortExpr]> {
65+
None
66+
}
67+
68+
/// Human readable name such as `"MIN(c2)"`. The default
69+
/// implementation returns placeholder text.
70+
fn name(&self) -> &str {
71+
"AggregateExpr: default name"
72+
}
73+
74+
/// If the aggregate expression has a specialized
75+
/// [`GroupsAccumulator`] implementation. If this returns true,
76+
/// `[Self::create_groups_accumulator`] will be called.
77+
fn groups_accumulator_supported(&self) -> bool {
78+
false
79+
}
80+
81+
/// Return a specialized [`GroupsAccumulator`] that manages state
82+
/// for all groups.
83+
///
84+
/// For maximum performance, a [`GroupsAccumulator`] should be
85+
/// implemented in addition to [`Accumulator`].
86+
fn create_groups_accumulator(&self) -> Result<Box<dyn GroupsAccumulator>> {
87+
not_impl_err!("GroupsAccumulator hasn't been implemented for {self:?} yet")
88+
}
89+
90+
/// Construct an expression that calculates the aggregate in reverse.
91+
/// Typically the "reverse" expression is itself (e.g. SUM, COUNT).
92+
/// For aggregates that do not support calculation in reverse,
93+
/// returns None (which is the default value).
94+
fn reverse_expr(&self) -> Option<Arc<dyn AggregateExpr>> {
95+
None
96+
}
97+
98+
/// Creates accumulator implementation that supports retract
99+
fn create_sliding_accumulator(&self) -> Result<Box<dyn Accumulator>> {
100+
not_impl_err!("Retractable Accumulator hasn't been implemented for {self:?} yet")
101+
}
102+
}

‎datafusion/physical-expr-common/src/aggregate/utils.rs‎

Lines changed: 69 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,69 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
use std::{any::Any, sync::Arc};
19+
20+
use arrow::{
21+
compute::SortOptions,
22+
datatypes::{DataType, Field},
23+
};
24+
25+
use crate::sort_expr::PhysicalSortExpr;
26+
27+
use super::AggregateExpr;
28+
29+
/// Downcast a `Box<dyn AggregateExpr>` or `Arc<dyn AggregateExpr>`
30+
/// and return the inner trait object as [`Any`] so
31+
/// that it can be downcast to a specific implementation.
32+
///
33+
/// This method is used when implementing the `PartialEq<dyn Any>`
34+
/// for [`AggregateExpr`] aggregation expressions and allows comparing the equality
35+
/// between the trait objects.
36+
pub fn down_cast_any_ref(any: &dyn Any) -> &dyn Any {
37+
if let Some(obj) = any.downcast_ref::<Arc<dyn AggregateExpr>>() {
38+
obj.as_any()
39+
} else if let Some(obj) = any.downcast_ref::<Box<dyn AggregateExpr>>() {
40+
obj.as_any()
41+
} else {
42+
any
43+
}
44+
}
45+
46+
/// Construct corresponding fields for lexicographical ordering requirement expression
47+
pub fn ordering_fields(
48+
ordering_req: &[PhysicalSortExpr],
49+
// Data type of each expression in the ordering requirement
50+
data_types: &[DataType],
51+
) -> Vec<Field> {
52+
ordering_req
53+
.iter()
54+
.zip(data_types.iter())
55+
.map(|(sort_expr, dtype)| {
56+
Field::new(
57+
sort_expr.expr.to_string().as_str(),
58+
dtype.clone(),
59+
// Multi partitions may be empty hence field should be nullable.
60+
true,
61+
)
62+
})
63+
.collect()
64+
}
65+
66+
/// Selects the sort option attribute from all the given `PhysicalSortExpr`s.
67+
pub fn get_sort_options(ordering_req: &[PhysicalSortExpr]) -> Vec<SortOptions> {
68+
ordering_req.iter().map(|item| item.options).collect()
69+
}

0 commit comments

Comments
 (0)