You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@arrow.apache.org by av...@apache.org on 2023/02/26 02:05:05 UTC
[arrow-datafusion] branch main updated: rebase (#5367)
This is an automated email from the ASF dual-hosted git repository.
avantgardner pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new fad360df0 rebase (#5367)
fad360df0 is described below
commit fad360df0132a2fcb264a7c07b2b02f0b1dfc644
Author: Brent Gardner <bg...@squarelabs.net>
AuthorDate: Sat Feb 25 19:04:59 2023 -0700
rebase (#5367)
---
datafusion-cli/Cargo.lock | 90 ++++++++++------------------
datafusion/expr/src/logical_plan/builder.rs | 4 +-
datafusion/expr/src/logical_plan/plan.rs | 15 +++--
datafusion/expr/src/table_source.rs | 17 +++++-
datafusion/optimizer/Cargo.toml | 1 +
datafusion/optimizer/src/push_down_filter.rs | 79 ++++++++++++++++--------
6 files changed, 115 insertions(+), 91 deletions(-)
diff --git a/datafusion-cli/Cargo.lock b/datafusion-cli/Cargo.lock
index 9590e5ffc..3bcf95354 100644
--- a/datafusion-cli/Cargo.lock
+++ b/datafusion-cli/Cargo.lock
@@ -382,18 +382,6 @@ dependencies = [
"alloc-stdlib",
]
-[[package]]
-name = "bstr"
-version = "0.2.17"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba3569f383e8f1598449f1a423e72e99569137b47740b1da11ef19af3d5c3223"
-dependencies = [
- "lazy_static",
- "memchr",
- "regex-automata",
- "serde",
-]
-
[[package]]
name = "bumpalo"
version = "3.12.0"
@@ -602,13 +590,12 @@ dependencies = [
[[package]]
name = "csv"
-version = "1.1.6"
+version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "22813a6dc45b335f9bade10bf7271dc477e81113e89eb251a0bc2a8a81c536e1"
+checksum = "af91f40b7355f82b0a891f50e70399475945bb0b0da4f1700ce60761c9d3e359"
dependencies = [
- "bstr",
"csv-core",
- "itoa 0.4.8",
+ "itoa",
"ryu",
"serde",
]
@@ -624,9 +611,9 @@ dependencies = [
[[package]]
name = "cxx"
-version = "1.0.89"
+version = "1.0.91"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bc831ee6a32dd495436e317595e639a587aa9907bef96fe6e6abc290ab6204e9"
+checksum = "86d3488e7665a7a483b57e25bdd90d0aeb2bc7608c8d0346acf2ad3f1caf1d62"
dependencies = [
"cc",
"cxxbridge-flags",
@@ -636,9 +623,9 @@ dependencies = [
[[package]]
name = "cxx-build"
-version = "1.0.89"
+version = "1.0.91"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "94331d54f1b1a8895cd81049f7eaaaef9d05a7dcb4d1fd08bf3ff0806246789d"
+checksum = "48fcaf066a053a41a81dfb14d57d99738b767febb8b735c3016e469fac5da690"
dependencies = [
"cc",
"codespan-reporting",
@@ -651,15 +638,15 @@ dependencies = [
[[package]]
name = "cxxbridge-flags"
-version = "1.0.89"
+version = "1.0.91"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "48dcd35ba14ca9b40d6e4b4b39961f23d835dbb8eed74565ded361d93e1feb8a"
+checksum = "a2ef98b8b717a829ca5603af80e1f9e2e48013ab227b68ef37872ef84ee479bf"
[[package]]
name = "cxxbridge-macro"
-version = "1.0.89"
+version = "1.0.91"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "81bbeb29798b407ccd82a3324ade1a7286e0d29851475990b612670f6f5124d2"
+checksum = "086c685979a698443656e5cf7856c95c642295a38599f12fb1ff76fb28d19892"
dependencies = [
"proc-macro2",
"quote",
@@ -776,6 +763,7 @@ dependencies = [
"datafusion-expr",
"datafusion-physical-expr",
"hashbrown 0.13.2",
+ "itertools",
"log",
"regex-syntax",
]
@@ -955,9 +943,9 @@ dependencies = [
[[package]]
name = "fastrand"
-version = "1.8.0"
+version = "1.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a7a407cfaa3385c4ae6b23e84623d48c2798d06e3e6a1878f7f59f17b3f86499"
+checksum = "e51093e27b0797c359783294ca4f0a911c270184cb10f85783b118614a1501be"
dependencies = [
"instant",
]
@@ -1194,13 +1182,13 @@ dependencies = [
[[package]]
name = "http"
-version = "0.2.8"
+version = "0.2.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "75f43d41e26995c17e71ee126451dd3941010b0514a81a9d11f3b341debc2399"
+checksum = "bd6effc99afb63425aff9b05836f029929e345a6148a14b7ecd5ab67af944482"
dependencies = [
"bytes",
"fnv",
- "itoa 1.0.5",
+ "itoa",
]
[[package]]
@@ -1247,7 +1235,7 @@ dependencies = [
"http-body",
"httparse",
"httpdate",
- "itoa 1.0.5",
+ "itoa",
"pin-project-lite",
"socket2",
"tokio",
@@ -1353,12 +1341,6 @@ dependencies = [
"either",
]
-[[package]]
-name = "itoa"
-version = "0.4.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b71991ff56294aa922b450139ee08b3bfc70982c6b2c7562771375cf73542dd4"
-
[[package]]
name = "itoa"
version = "1.0.5"
@@ -1581,14 +1563,14 @@ dependencies = [
[[package]]
name = "mio"
-version = "0.8.5"
+version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e5d732bc30207a6423068df043e3d02e0735b155ad7ce1a6f76fe2baa5b158de"
+checksum = "5b9d9a46eff5b4ff64b45a9e316a6d1e0bc719ef429cbec4dc630684212bfdf9"
dependencies = [
"libc",
"log",
"wasi",
- "windows-sys 0.42.0",
+ "windows-sys 0.45.0",
]
[[package]]
@@ -1729,9 +1711,9 @@ dependencies = [
[[package]]
name = "once_cell"
-version = "1.17.0"
+version = "1.17.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6f61fba1741ea2b3d6a1e3178721804bb716a68a6aeba1149b5d52e3d464ea66"
+checksum = "b7e5500299e16ebb147ae15a00a942af264cf3688f47923b8fc2cd5858f23ad3"
[[package]]
name = "ordered-float"
@@ -1969,12 +1951,6 @@ dependencies = [
"regex-syntax",
]
-[[package]]
-name = "regex-automata"
-version = "0.1.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6c230d73fb8d8c1b9c0b3135c5142a8acee3a0558fb8db5cf1cb65f8d7862132"
-
[[package]]
name = "regex-syntax"
version = "0.6.28"
@@ -2194,7 +2170,7 @@ version = "1.0.93"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cad406b69c91885b5107daf2c29572f6c8cdb3c66826821e286c533490c0bc76"
dependencies = [
- "itoa 1.0.5",
+ "itoa",
"ryu",
"serde",
]
@@ -2206,7 +2182,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd"
dependencies = [
"form_urlencoded",
- "itoa 1.0.5",
+ "itoa",
"ryu",
"serde",
]
@@ -2224,9 +2200,9 @@ dependencies = [
[[package]]
name = "slab"
-version = "0.4.7"
+version = "0.4.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4614a76b2a8be0058caa9dbbaf66d988527d86d003c11a94fbd335d7661edcef"
+checksum = "6528351c9bc8ab22353f9d776db39a20288e8d6c37ef8cfe3317cf875eecfc2d"
dependencies = [
"autocfg",
]
@@ -2347,9 +2323,9 @@ checksum = "6bdef32e8150c2a081110b42772ffe7d7c9032b606bc226c8260fd97e0976601"
[[package]]
name = "syn"
-version = "1.0.107"
+version = "1.0.108"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1f4064b5b16e03ae50984a5a8ed5d4f8803e6bc1fd170a3cda91a1be4b18e3f5"
+checksum = "d56e159d99e6c2b93995d171050271edb50ecc5288fbc7cc17de8fdce4e58c14"
dependencies = [
"proc-macro2",
"quote",
@@ -2483,9 +2459,9 @@ dependencies = [
[[package]]
name = "tokio-stream"
-version = "0.1.11"
+version = "0.1.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d660770404473ccd7bc9f8b28494a811bc18542b915c0855c51e8f419d5223ce"
+checksum = "8fb52b74f05dbf495a8fba459fdc331812b96aa086d9eb78101fa0d4569c3313"
dependencies = [
"futures-core",
"pin-project-lite",
@@ -2494,9 +2470,9 @@ dependencies = [
[[package]]
name = "tokio-util"
-version = "0.7.6"
+version = "0.7.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bc6a3b08b64e6dfad376fa2432c7b1f01522e37a623c3050bc95db2d3ff21583"
+checksum = "5427d89453009325de0d8f342c9490009f76e999cb7672d77e46267448f7e6b2"
dependencies = [
"bytes",
"futures-core",
diff --git a/datafusion/expr/src/logical_plan/builder.rs b/datafusion/expr/src/logical_plan/builder.rs
index 8087d95be..e6ff15244 100644
--- a/datafusion/expr/src/logical_plan/builder.rs
+++ b/datafusion/expr/src/logical_plan/builder.rs
@@ -451,9 +451,9 @@ impl LogicalPlanBuilder {
.map(|col| col.flat_name())
.collect::<String>();
- return Err(DataFusionError::Plan(format!(
+ Err(DataFusionError::Plan(format!(
"For SELECT DISTINCT, ORDER BY expressions {missing_col_names} must appear in select list",
- )));
+ )))
}
/// Apply a sort
diff --git a/datafusion/expr/src/logical_plan/plan.rs b/datafusion/expr/src/logical_plan/plan.rs
index d86a44e5d..9521ae602 100644
--- a/datafusion/expr/src/logical_plan/plan.rs
+++ b/datafusion/expr/src/logical_plan/plan.rs
@@ -962,10 +962,13 @@ impl LogicalPlan {
let mut full_filter = vec![];
let mut partial_filter = vec![];
let mut unsupported_filters = vec![];
+ let filters: Vec<&Expr> = filters.iter().collect();
- filters.iter().for_each(|x| {
- if let Ok(t) = source.supports_filter_pushdown(x) {
- match t {
+ if let Ok(results) =
+ source.supports_filters_pushdown(&filters)
+ {
+ filters.iter().zip(results.iter()).for_each(
+ |(x, res)| match res {
TableProviderFilterPushDown::Exact => {
full_filter.push(x)
}
@@ -975,9 +978,9 @@ impl LogicalPlan {
TableProviderFilterPushDown::Unsupported => {
unsupported_filters.push(x)
}
- }
- }
- });
+ },
+ );
+ }
if !full_filter.is_empty() {
write!(f, ", full_filters={full_filter:?}")?;
diff --git a/datafusion/expr/src/table_source.rs b/datafusion/expr/src/table_source.rs
index 10984f779..5954335c6 100644
--- a/datafusion/expr/src/table_source.rs
+++ b/datafusion/expr/src/table_source.rs
@@ -17,6 +17,7 @@
use crate::{Expr, LogicalPlan};
use arrow::datatypes::SchemaRef;
+use datafusion_common::Result;
use std::any::Any;
///! Table source
@@ -70,13 +71,27 @@ pub trait TableSource: Sync + Send {
/// Tests whether the table provider can make use of a filter expression
/// to optimise data retrieval.
+ #[deprecated(since = "20.0.0", note = "use supports_filters_pushdown instead")]
fn supports_filter_pushdown(
&self,
_filter: &Expr,
- ) -> datafusion_common::Result<TableProviderFilterPushDown> {
+ ) -> Result<TableProviderFilterPushDown> {
Ok(TableProviderFilterPushDown::Unsupported)
}
+ /// Tests whether the table provider can make use of any or all filter expressions
+ /// to optimise data retrieval.
+ #[allow(deprecated)]
+ fn supports_filters_pushdown(
+ &self,
+ filters: &[&Expr],
+ ) -> Result<Vec<TableProviderFilterPushDown>> {
+ filters
+ .iter()
+ .map(|f| self.supports_filter_pushdown(f))
+ .collect()
+ }
+
/// Get the Logical plan of this table provider, if available.
fn get_logical_plan(&self) -> Option<&LogicalPlan> {
None
diff --git a/datafusion/optimizer/Cargo.toml b/datafusion/optimizer/Cargo.toml
index ae9cd3d32..3553a56e4 100644
--- a/datafusion/optimizer/Cargo.toml
+++ b/datafusion/optimizer/Cargo.toml
@@ -44,6 +44,7 @@ datafusion-common = { path = "../common", version = "19.0.0" }
datafusion-expr = { path = "../expr", version = "19.0.0" }
datafusion-physical-expr = { path = "../physical-expr", version = "19.0.0" }
hashbrown = { version = "0.13", features = ["raw"] }
+itertools = "0.10"
log = "^0.4"
regex-syntax = "0.6.28"
diff --git a/datafusion/optimizer/src/push_down_filter.rs b/datafusion/optimizer/src/push_down_filter.rs
index 4369b025f..6d8db2043 100644
--- a/datafusion/optimizer/src/push_down_filter.rs
+++ b/datafusion/optimizer/src/push_down_filter.rs
@@ -27,6 +27,7 @@ use datafusion_expr::{
utils::from_plan,
BinaryExpr, Expr, Filter, Operator, TableProviderFilterPushDown,
};
+use itertools::Itertools;
use std::collections::{HashMap, HashSet};
use std::iter::once;
use std::sync::Arc;
@@ -695,30 +696,27 @@ impl OptimizerRule for PushDownFilter {
push_down_all_join(predicates, &filter.input, left, right, vec![])?
}
LogicalPlan::TableScan(scan) => {
- let mut new_scan_filters = scan.filters.clone();
- let mut new_predicate = vec![];
-
- let filter_predicates =
- utils::split_conjunction_owned(filter.predicate.clone());
-
- for filter_expr in &filter_predicates {
- let (preserve_filter_node, add_to_provider) =
- match scan.source.supports_filter_pushdown(filter_expr)? {
- TableProviderFilterPushDown::Unsupported => (true, false),
- TableProviderFilterPushDown::Inexact => (true, true),
- TableProviderFilterPushDown::Exact => (false, true),
- };
- if preserve_filter_node {
- new_predicate.push(filter_expr.clone());
- }
- if add_to_provider {
- // avoid reduplicated filter expr.
- if new_scan_filters.contains(filter_expr) {
- continue;
- }
- new_scan_filters.push(filter_expr.clone());
- }
- }
+ let filter_predicates = split_conjunction(&filter.predicate);
+ let results = scan
+ .source
+ .supports_filters_pushdown(filter_predicates.as_slice())?;
+ let zip = filter_predicates.iter().zip(results.into_iter());
+
+ let new_scan_filters = zip
+ .clone()
+ .filter(|(_, res)| res != &TableProviderFilterPushDown::Unsupported)
+ .map(|(pred, _)| *pred);
+ let new_scan_filters: Vec<Expr> = scan
+ .filters
+ .iter()
+ .chain(new_scan_filters)
+ .unique()
+ .cloned()
+ .collect();
+ let new_predicate: Vec<Expr> = zip
+ .filter(|(_, res)| res != &TableProviderFilterPushDown::Exact)
+ .map(|(pred, _)| (*pred).clone())
+ .collect();
let new_scan = LogicalPlan::TableScan(TableScan {
source: scan.source.clone(),
@@ -1903,7 +1901,7 @@ mod tests {
fn supports_filter_pushdown(
&self,
- _: &Expr,
+ _e: &Expr,
) -> Result<TableProviderFilterPushDown> {
Ok(self.filter_support.clone())
}
@@ -2013,6 +2011,37 @@ mod tests {
assert_optimized_plan_eq(&plan, expected)
}
+ #[test]
+ fn multi_combined_filter_exact() -> Result<()> {
+ let test_provider = PushDownProvider {
+ filter_support: TableProviderFilterPushDown::Exact,
+ };
+
+ let table_scan = LogicalPlan::TableScan(TableScan {
+ table_name: "test".to_string(),
+ filters: vec![],
+ projected_schema: Arc::new(DFSchema::try_from(
+ (*test_provider.schema()).clone(),
+ )?),
+ projection: Some(vec![0]),
+ source: Arc::new(test_provider),
+ fetch: None,
+ });
+
+ let plan = LogicalPlanBuilder::from(table_scan)
+ .filter(and(col("a").eq(lit(10i64)), col("b").gt(lit(11i64))))?
+ .project(vec![col("a"), col("b")])?
+ .build()?;
+
+ let expected = r#"
+Projection: a, b
+ TableScan: test projection=[a], full_filters=[a = Int64(10), b > Int64(11)]
+ "#
+ .trim();
+
+ assert_optimized_plan_eq(&plan, expected)
+ }
+
#[test]
fn test_filter_with_alias() -> Result<()> {
// in table scan the true col name is 'test.a',