starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-05 00:09:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: upgrade datafusion to v43 (#5090)

Browse Source
This commit is contained in:
Huaijin
2024-11-18 18:02:12 +08:00
committed by GitHub
parent 7a689c46c9
commit f99d45d2ef
29 changed files with 148 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+66 -50
View File
@@ -2918,9 +2918,9 @@ checksum = "e8566979429cf69b49a5c740c60791108e86440e8be149bbea4fe54d2c32d6e2"
[[package]]
name = "datafusion"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3e8053b4cedc24eb158e4c041b38cfa0677ef5f4a7ccaa31ee5dcad61dd7aa54"
checksum = "cbba0799cf6913b456ed07a94f0f3b6e12c62a5d88b10809e2284a0f2b915c05"
dependencies = [
"ahash 0.8.11",
"arrow",
@@ -2975,9 +2975,9 @@ dependencies = [
[[package]]
name = "datafusion-catalog"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7d95efedb3a32f6f74df5bb8fda7b69fb9babe80e92137f25de6ddb15e8e8801"
checksum = "7493c5c2d40eec435b13d92e5703554f4efc7059451fcb8d3a79580ff0e45560"
dependencies = [
"arrow-schema",
"async-trait",
@@ -2990,9 +2990,9 @@ dependencies = [
[[package]]
name = "datafusion-common"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b7d766e0d3dec01a0ab70b1b31678c286cddc0bd7afc9bd82504a1d9a70a7d94"
checksum = "24953049ebbd6f8964f91f60aa3514e121b5e81e068e33b60e77815ab369b25c"
dependencies = [
"ahash 0.8.11",
"arrow",
@@ -3002,6 +3002,7 @@ dependencies = [
"chrono",
"half",
"hashbrown 0.14.5",
"indexmap 2.2.6",
"instant",
"libc",
"num_cpus",
@@ -3014,9 +3015,9 @@ dependencies = [
[[package]]
name = "datafusion-common-runtime"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4e55db6df319f9e7cf366d0d4ffae793c863823421b2f2b7314a0fefd8e8c11a"
checksum = "f06df4ef76872e11c924d3c814fd2a8dd09905ed2e2195f71c857d78abd19685"
dependencies = [
"log",
"tokio",
@@ -3024,9 +3025,9 @@ dependencies = [
[[package]]
name = "datafusion-execution"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6d0c6dc013f955c382438a78fa3de8b0a8bf7b1a4cda5bc46335fe445ff3ff1a"
checksum = "6bbdcb628d690f3ce5fea7de81642b514486d58ff9779a51f180a69a4eadb361"
dependencies = [
"arrow",
"chrono",
@@ -3045,9 +3046,9 @@ dependencies = [
[[package]]
name = "datafusion-expr"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f31405c0bb854451d755b224d41dc466a8f7fd36f8c041c29d2d8432bd0c08c"
checksum = "8036495980e3131f706b7d33ab00b4492d73dc714e3cb74d11b50f9602a73246"
dependencies = [
"ahash 0.8.11",
"arrow",
@@ -3057,7 +3058,9 @@ dependencies = [
"datafusion-common",
"datafusion-expr-common",
"datafusion-functions-aggregate-common",
"datafusion-functions-window-common",
"datafusion-physical-expr-common",
"indexmap 2.2.6",
"paste",
"serde_json",
"sqlparser",
@@ -3067,20 +3070,21 @@ dependencies = [
[[package]]
name = "datafusion-expr-common"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ebc8266b6627c8264c87bc7c82564e3d89ed5f0f9943b49a30dac1f1ac12e4c0"
checksum = "4da0f3cb4669f9523b403d6b5a0ec85023e0ab3bf0183afd1517475b3e64fdd2"
dependencies = [
"arrow",
"datafusion-common",
"itertools 0.13.0",
"paste",
]
[[package]]
name = "datafusion-functions"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5712668780bc43666ecd10acd188b7df58e2a5501d4dbbd972bf209f1790138b"
checksum = "f52c4012648b34853e40a2c6bcaa8772f837831019b68aca384fb38436dba162"
dependencies = [
"arrow",
"arrow-buffer",
@@ -3105,9 +3109,9 @@ dependencies = [
[[package]]
name = "datafusion-functions-aggregate"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8ec138af6b7482fb726f1bfeec010fc063b9614594c36a1051a4d3b365ba6a5f"
checksum = "e5b8bb624597ba28ed7446df4a9bd7c7a7bde7c578b6b527da3f47371d5f6741"
dependencies = [
"ahash 0.8.11",
"arrow",
@@ -3119,16 +3123,16 @@ dependencies = [
"datafusion-physical-expr",
"datafusion-physical-expr-common",
"half",
"indexmap 2.2.6",
"log",
"paste",
"sqlparser",
]
[[package]]
name = "datafusion-functions-aggregate-common"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "564499c6bdd3ab9f76c7ad727e858bc6791e4de6c1a484d21d2bf49daaa658d6"
checksum = "6fb06208fc470bc8cf1ce2d9a1159d42db591f2c7264a8c1776b53ad8f675143"
dependencies = [
"ahash 0.8.11",
"arrow",
@@ -3140,9 +3144,9 @@ dependencies = [
[[package]]
name = "datafusion-functions-json"
version = "0.42.0"
version = "0.43.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "532feb5c208fd1708f4d93b1984fb7a7ed678a9f0e6f799af97118d7c4e863a1"
checksum = "744cf7ae121977c453586f3f098239e50da6b0cfcf2de3ccb9338a4896f97dc0"
dependencies = [
"datafusion",
"jiter",
@@ -3152,9 +3156,9 @@ dependencies = [
[[package]]
name = "datafusion-functions-nested"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9b55ea2221ae1c1e37d524f8330f763dcdc205edb74fe5f54cbdea475c17fd18"
checksum = "fca25bbb87323716d05e54114666e942172ccca23c5a507e9c7851db6e965317"
dependencies = [
"arrow",
"arrow-array",
@@ -3175,21 +3179,34 @@ dependencies = [
[[package]]
name = "datafusion-functions-window"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6932996c4407ee1ebf23ffd706e982729cb9b6f7a31a281abac51fe524c3a049"
checksum = "5ae23356c634e54c59f7c51acb7a5b9f6240ffb2cf997049a1a24a8a88598dbe"
dependencies = [
"datafusion-common",
"datafusion-expr",
"datafusion-functions-window-common",
"datafusion-physical-expr",
"datafusion-physical-expr-common",
"log",
"paste",
]
[[package]]
name = "datafusion-functions-window-common"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d4b3d6ff7794acea026de36007077a06b18b89e4f9c3fea7f2215f9f7dd9059b"
dependencies = [
"datafusion-common",
"datafusion-physical-expr-common",
]
[[package]]
name = "datafusion-optimizer"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f7d8afa1eb44e2f00cc8d82b88803e456a681474b8877ceecc04e9517d5c843c"
checksum = "bec6241eb80c595fa0e1a8a6b69686b5cf3bd5fdacb8319582a0943b0bd788aa"
dependencies = [
"arrow",
"async-trait",
@@ -3207,9 +3224,9 @@ dependencies = [
[[package]]
name = "datafusion-physical-expr"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "570666d84df483473626fab4e69997d048b40d0e7c67c540299714f244d99e73"
checksum = "3370357b8fc75ec38577700644e5d1b0bc78f38babab99c0b8bd26bafb3e4335"
dependencies = [
"ahash 0.8.11",
"arrow",
@@ -3218,30 +3235,26 @@ dependencies = [
"arrow-ord",
"arrow-schema",
"arrow-string",
"base64 0.22.1",
"chrono",
"datafusion-common",
"datafusion-execution",
"datafusion-expr",
"datafusion-expr-common",
"datafusion-functions-aggregate-common",
"datafusion-physical-expr-common",
"half",
"hashbrown 0.14.5",
"hex",
"indexmap 2.2.6",
"itertools 0.13.0",
"log",
"paste",
"petgraph",
"regex",
]
[[package]]
name = "datafusion-physical-expr-common"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3746cbdfb32d67399dcaad17042e419ac6da454a7e38ff098aa2fbf0a7388982"
checksum = "b8b7734d94bf2fa6f6e570935b0ddddd8421179ce200065be97874e13d46a47b"
dependencies = [
"ahash 0.8.11",
"arrow",
@@ -3253,13 +3266,15 @@ dependencies = [
[[package]]
name = "datafusion-physical-optimizer"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "696f06e79d44f7c50f57cea23493881d86d9d9647884d38ce467c7f75c13e286"
checksum = "7eee8c479522df21d7b395640dff88c5ed05361852dce6544d7c98e9dbcebffe"
dependencies = [
"arrow",
"arrow-schema",
"datafusion-common",
"datafusion-execution",
"datafusion-expr-common",
"datafusion-physical-expr",
"datafusion-physical-plan",
"itertools 0.13.0",
@@ -3267,9 +3282,9 @@ dependencies = [
[[package]]
name = "datafusion-physical-plan"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "04e1d084224023e09cdea14d01ded0f2092c319c7b4594ebc821283b9c7c4a35"
checksum = "17e1fc2e2c239d14e8556f2622b19a726bf6bc6962cc00c71fc52626274bee24"
dependencies = [
"ahash 0.8.11",
"arrow",
@@ -3283,8 +3298,8 @@ dependencies = [
"datafusion-common-runtime",
"datafusion-execution",
"datafusion-expr",
"datafusion-functions-aggregate",
"datafusion-functions-aggregate-common",
"datafusion-functions-window-common",
"datafusion-physical-expr",
"datafusion-physical-expr-common",
"futures",
@@ -3302,9 +3317,9 @@ dependencies = [
[[package]]
name = "datafusion-proto"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "927a3dd9c89bc23c74677e25a68791b0e5c92730eebc0bab84bce7b347ed1835"
checksum = "f730f7fc5a20134d4e5ecdf7bbf392002ac58163d58423ea28a702dc077b06e1"
dependencies = [
"arrow",
"chrono",
@@ -3318,9 +3333,9 @@ dependencies = [
[[package]]
name = "datafusion-proto-common"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "59e109a2b038227b0fb3c240b7694d0b5acbb1d66617a7852bc73fb69897439e"
checksum = "12c225fe49e4f943e35446b263613ada7a9e9f8d647544e6b07037b9803567df"
dependencies = [
"arrow",
"chrono",
@@ -3331,15 +3346,16 @@ dependencies = [
[[package]]
name = "datafusion-sql"
version = "42.1.0"
version = "43.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c105148357dcbd9e4c97eada2930a59f7923215461d9f47de6e76edd60eab2d5"
checksum = "63e3a4ed41dbee20a5d947a59ca035c225d67dc9cbe869c10f66dcdf25e7ce51"
dependencies = [
"arrow",
"arrow-array",
"arrow-schema",
"datafusion-common",
"datafusion-expr",
"indexmap 2.2.6",
"log",
"regex",
"sqlparser",
@@ -8476,9 +8492,9 @@ dependencies = [
[[package]]
name = "sqlparser"
version = "0.50.0"
version = "0.51.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b2e5b515a2bd5168426033e9efbfd05500114833916f1d5c268f938b4ee130ac"
checksum = "5fe11944a61da0da3f592e19a45ebe5ab92dc14a779907ff1f08fbb797bfefc7"
dependencies = [
"log",
"serde",
Cargo.toml
+10 -10
View File
@@ -87,7 +87,7 @@ dashmap.workspace = true
datafusion.workspace = true
datafusion-proto.workspace = true
datafusion-functions-aggregate-common.workspace = true
datafusion-functions-json = "0.42.0"
datafusion-functions-json = "0.43.0"
arrow.workspace = true
arrow-flight.workspace = true
arrow-schema.workspace = true
@@ -233,16 +233,16 @@ chromiumoxide = { git = "https://github.com/mattsse/chromiumoxide", features = [
chrono = { version = "0.4", default-features = false, features = ["clock"] }
cityhasher = { version = "0.1", default-features = false }
dashmap = { version = "6.0", features = ["serde"] }
datafusion = "42.0.0"
datafusion-proto = "42.0.0"
datafusion-functions-aggregate-common = "42.0.0"
arrow = { version = "53.0.0", features = ["ipc_compression", "prettyprint"] }
arrow-flight = "53.0.0"
arrow-json = "53.0.0"
arrow-schema = { version = "53.0.0", features = ["serde"] }
parquet = { version = "53.0.0", features = ["arrow", "async", "object_store"] }
datafusion = "43.0.0"
datafusion-proto = "43.0.0"
datafusion-functions-aggregate-common = "43.0.0"
arrow = { version = "53.2.0", features = ["ipc_compression", "prettyprint"] }
arrow-flight = "53.2.0"
arrow-json = "53.2.0"
arrow-schema = { version = "53.2.0", features = ["serde"] }
parquet = { version = "53.2.0", features = ["arrow", "async", "object_store"] }
object_store = { version = "0.11.0", features = ["aws", "azure", "gcp"] }
sqlparser = { version = "0.50.0", features = ["serde", "visitor"] }
sqlparser = { version = "0.51.0", features = ["serde", "visitor"] }
dotenv_config = "0.2.1"
dotenvy = "0.15.7"
env_logger = "0.10"
src/service/search/datafusion/distributed_plan/empty_exec.rs
+3 -3
View File
@@ -21,7 +21,7 @@ use datafusion::{
arrow::{array::RecordBatch, datatypes::SchemaRef},
common::{internal_err, Result, Statistics},
execution::{SendableRecordBatchStream, TaskContext},
physical_expr::{EquivalenceProperties, Partitioning, PhysicalSortExpr},
physical_expr::{EquivalenceProperties, LexOrdering, Partitioning, PhysicalSortExpr},
physical_plan::{
common, expressions::Column, memory::MemoryStream, DisplayAs, DisplayFormatType,
ExecutionMode, ExecutionPlan, PlanProperties,
@@ -95,13 +95,13 @@ impl NewEmptyExec {
} else {
match index {
Ok(index) => {
let ordering = vec![vec![PhysicalSortExpr {
let ordering = vec![LexOrdering::new(vec![PhysicalSortExpr {
expr: Arc::new(Column::new(&get_config().common.column_timestamp, index)),
options: SortOptions {
descending: true,
nulls_first: false,
},
}]];
}])];
EquivalenceProperties::new_with_orderings(schema, &ordering)
}
Err(_) => EquivalenceProperties::new(schema),
src/service/search/datafusion/exec.rs
+1 -1
View File
@@ -223,7 +223,7 @@ pub async fn create_runtime_env(memory_limit: usize) -> Result<RuntimeEnv> {
}
super::MemoryPoolType::None => {}
};
RuntimeEnv::new(rn_config)
RuntimeEnv::try_new(rn_config)
}
pub async fn prepare_datafusion_context(
src/service/search/datafusion/optimizer/add_sort_and_limit.rs
+28 -8
View File
@@ -25,10 +25,12 @@ use datafusion::{
},
logical_expr::{col, Limit, LogicalPlan, Projection, Sort, SortExpr, TableScan},
optimizer::{optimizer::ApplyOrder, OptimizerConfig, OptimizerRule},
prelude::Expr,
scalar::ScalarValue,
};
/// Optimization rule that add sort and limit to table scan
#[derive(Default)]
#[derive(Default, Debug)]
pub struct AddSortAndLimitRule {
limit: usize,
offset: usize,
@@ -77,8 +79,9 @@ impl OptimizerRule for AddSortAndLimitRule {
(Transformed::no(LogicalPlan::Limit(limit)), None)
} else {
// the add sort plan should reflect the limit
let fetch = limit.fetch.unwrap() + limit.skip;
let (sort, schema) = generate_sort_plan(limit.input.clone(), fetch);
let fetch = get_int_from_expr(&limit.fetch);
let skip = get_int_from_expr(&limit.skip);
let (sort, schema) = generate_sort_plan(limit.input.clone(), fetch + skip);
limit.input = Arc::new(sort);
(Transformed::yes(LogicalPlan::Limit(limit)), schema)
}
@@ -131,7 +134,6 @@ fn is_complex_query(plan: &LogicalPlan) -> bool {
plan,
LogicalPlan::Aggregate(_)
| LogicalPlan::Join(_)
| LogicalPlan::CrossJoin(_)
| LogicalPlan::Distinct(_)
| LogicalPlan::RecursiveQuery(_)
| LogicalPlan::SubqueryAlias(_)
@@ -142,8 +144,12 @@ fn is_complex_query(plan: &LogicalPlan) -> bool {
fn generate_limit_plan(input: Arc<LogicalPlan>, limit: usize, skip: usize) -> LogicalPlan {
LogicalPlan::Limit(Limit {
skip,
fetch: Some(limit),
skip: Some(Box::new(Expr::Literal(ScalarValue::Int64(Some(
skip as i64,
))))),
fetch: Some(Box::new(Expr::Literal(ScalarValue::Int64(Some(
limit as i64,
))))),
input,
})
}
@@ -195,14 +201,28 @@ fn generate_limit_and_sort_plan(
let (sort, schema) = generate_sort_plan(input, limit + skip);
(
LogicalPlan::Limit(Limit {
skip,
fetch: Some(limit),
skip: Some(Box::new(Expr::Literal(ScalarValue::Int64(Some(
skip as i64,
))))),
fetch: Some(Box::new(Expr::Literal(ScalarValue::Int64(Some(
limit as i64,
))))),
input: Arc::new(sort),
}),
schema,
)
}
fn get_int_from_expr(expr: &Option<Box<Expr>>) -> usize {
match expr {
Some(expr) => match expr.as_ref() {
Expr::Literal(ScalarValue::Int64(Some(value))) => *value as usize,
_ => 0,
},
_ => 0,
}
}
struct ChangeTableScanSchema {}
impl ChangeTableScanSchema {
src/service/search/datafusion/optimizer/add_timestamp.rs
+6 -4
View File
@@ -25,7 +25,7 @@ use datafusion::{
/// Optimization rule that add _timestamp constraint to table scan
///
/// Note: should apply before push down filter rule
#[derive(Default)]
#[derive(Default, Debug)]
pub struct AddTimestampRule {
filter: Expr,
}
@@ -195,10 +195,10 @@ mod tests {
let expected = "Projection: test.name\
\n Filter: test.name IN (<subquery>) AND test.id = UInt32(1) AND test.id < UInt32(30)\
\n Subquery:\
\n TableScan: sq\
\n Filter: _timestamp >= Int64(0) AND _timestamp < Int64(5)\
\n TableScan: sq\
\n Filter: _timestamp >= Int64(0) AND _timestamp < Int64(5)\
\n TableScan: test";
// after DecorrelatePredicateSubquery, the plan look like below
// let expected = "Projection: test.name
// \n Filter: test.id = UInt32(1) AND test.id < UInt32(30)
@@ -308,7 +308,9 @@ mod tests {
let state = SessionStateBuilder::new()
.with_config(SessionConfig::new())
.with_runtime_env(Arc::new(RuntimeEnv::new(RuntimeConfig::default()).unwrap()))
.with_runtime_env(Arc::new(
RuntimeEnv::try_new(RuntimeConfig::default()).unwrap(),
))
.with_default_features()
.with_optimizer_rules(vec![
Arc::new(AddTimestampRule::new(2, 4)),
src/service/search/datafusion/optimizer/join_reorder.rs
+4 -2
View File
@@ -28,7 +28,7 @@ use datafusion::{
},
};
#[derive(Default)]
#[derive(Default, Debug)]
pub struct JoinReorderRule;
impl JoinReorderRule {
@@ -141,7 +141,9 @@ mod tests {
let state = SessionStateBuilder::new()
.with_config(SessionConfig::new().with_target_partitions(12))
.with_runtime_env(Arc::new(RuntimeEnv::new(RuntimeConfig::default()).unwrap()))
.with_runtime_env(Arc::new(
RuntimeEnv::try_new(RuntimeConfig::default()).unwrap(),
))
.with_default_features()
.with_physical_optimizer_rule(Arc::new(JoinReorderRule::new()))
.build();
src/service/search/datafusion/optimizer/mod.rs
-2
View File
@@ -28,7 +28,6 @@ use datafusion::optimizer::{
filter_null_join_keys::FilterNullJoinKeys, optimize_projections::OptimizeProjections,
propagate_empty_relation::PropagateEmptyRelation, push_down_filter::PushDownFilter,
push_down_limit::PushDownLimit, replace_distinct_aggregate::ReplaceDistinctWithAggregate,
rewrite_disjunctive_predicate::RewriteDisjunctivePredicate,
scalar_subquery_to_join::ScalarSubqueryToJoin, simplify_expressions::SimplifyExpressions,
single_distinct_to_groupby::SingleDistinctToGroupBy,
unwrap_cast_in_comparison::UnwrapCastInComparison, OptimizerRule,
@@ -91,7 +90,6 @@ pub fn generate_optimizer_rules(sql: &Sql) -> Vec<Arc<dyn OptimizerRule + Send +
// run it again after running the optimizations that potentially converted
// subqueries to joins
rules.push(Arc::new(SimplifyExpressions::new()));
rules.push(Arc::new(RewriteDisjunctivePredicate::new()));
rules.push(Arc::new(EliminateDuplicatedExpr::new()));
rules.push(Arc::new(EliminateFilter::new()));
rules.push(Arc::new(EliminateCrossJoin::new()));
src/service/search/datafusion/optimizer/rewrite_histogram.rs
+1 -1
View File
@@ -36,7 +36,7 @@ use crate::service::search::{
};
/// Optimization rule that rewrite histogram to date_bin()
#[derive(Default)]
#[derive(Default, Debug)]
pub struct RewriteHistogram {
start_time: i64,
end_time: i64,
src/service/search/datafusion/optimizer/rewrite_match.rs
+4 -2
View File
@@ -30,7 +30,7 @@ use crate::service::search::datafusion::udf::match_all_udf::{
};
/// Optimization rule that rewrite match_all() to str_match()
#[derive(Default)]
#[derive(Default, Debug)]
pub struct RewriteMatch {
fields: Vec<String>,
}
@@ -262,7 +262,9 @@ mod tests {
let state = SessionStateBuilder::new()
.with_config(SessionConfig::new())
.with_runtime_env(Arc::new(RuntimeEnv::new(RuntimeConfig::default()).unwrap()))
.with_runtime_env(Arc::new(
RuntimeEnv::try_new(RuntimeConfig::default()).unwrap(),
))
.with_default_features()
.with_optimizer_rules(vec![Arc::new(RewriteMatch::new(fields.clone()))])
.build();
src/service/search/datafusion/table_provider/empty_table.rs
+1
View File
@@ -30,6 +30,7 @@ use crate::service::search::datafusion::distributed_plan::empty_exec::NewEmptyEx
/// An empty plan that is useful for testing and generating plans
/// without mapping them to actual data.
#[derive(Debug)]
pub struct NewEmptyTable {
name: String,
schema: SchemaRef,
src/service/search/datafusion/table_provider/memtable.rs
+4 -12
View File
@@ -23,7 +23,7 @@ use datafusion::{
common::{project_schema, Constraints, Result},
datasource::{MemTable, TableProvider},
logical_expr::{Expr, TableType},
physical_expr::PhysicalSortExpr,
physical_expr::{LexOrdering, PhysicalSortExpr},
physical_plan::{
expressions::{CastExpr, Column},
projection::ProjectionExec,
@@ -33,6 +33,7 @@ use datafusion::{
};
use hashbrown::HashMap;
#[derive(Debug)]
pub(crate) struct NewMemTable {
mem_table: MemTable,
diff_rules: HashMap<String, DataType>,
@@ -115,15 +116,6 @@ impl TableProvider for NewMemTable {
})
}
async fn insert_into(
&self,
state: &dyn Session,
input: Arc<dyn ExecutionPlan>,
overwrite: bool,
) -> Result<Arc<dyn ExecutionPlan>> {
self.mem_table.insert_into(state, input, overwrite).await
}
fn get_column_default(&self, column: &str) -> Option<&Expr> {
self.mem_table.get_column_default(column)
}
@@ -135,13 +127,13 @@ fn wrap_sort(exec: Arc<dyn ExecutionPlan>) -> Arc<dyn ExecutionPlan> {
let index = exec.schema().index_of(&column_timestamp);
(match index {
Ok(index) => {
let ordering = vec![PhysicalSortExpr {
let ordering = LexOrdering::new(vec![PhysicalSortExpr {
expr: Arc::new(Column::new(&column_timestamp, index)),
options: SortOptions {
descending: true,
nulls_first: false,
},
}];
}]);
Arc::new(SortExec::new(ordering, exec))
}
Err(_) => exec,
src/service/search/datafusion/table_provider/mod.rs
+2 -1
View File
@@ -67,6 +67,7 @@ mod helpers;
pub mod memtable;
pub mod uniontable;
#[derive(Debug)]
pub(crate) struct NewListingTable {
table_paths: Vec<ListingTableUrl>,
/// File fields only
@@ -441,7 +442,7 @@ fn create_ordering(schema: &Schema, sort_order: &[Vec<SortExpr>]) -> Result<Vec<
for exprs in sort_order {
// Construct PhysicalSortExpr objects from Expr objects:
let mut sort_exprs = vec![];
let mut sort_exprs = LexOrdering::default();
for sort in exprs {
match &sort.expr {
Expr::Column(col) => match expressions::col(&col.name, schema) {
src/service/search/datafusion/table_provider/uniontable.rs
+1
View File
@@ -25,6 +25,7 @@ use datafusion::{
physical_plan::{empty::EmptyExec, union::UnionExec, ExecutionPlan},
};
#[derive(Debug)]
pub(crate) struct NewUnionTable {
schema: SchemaRef,
tables: Vec<Arc<dyn TableProvider>>,
src/service/search/datafusion/udf/arr_descending_udf.rs
+1 -1
View File
@@ -39,7 +39,7 @@ pub(crate) static ARR_DESCENDING_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
// expects three string
vec![DataType::Utf8],
// returns string
Arc::new(DataType::Utf8),
DataType::Utf8,
Volatility::Immutable,
Arc::new(arr_descending_impl),
)
src/service/search/datafusion/udf/arrcount_udf.rs
+1 -1
View File
@@ -37,7 +37,7 @@ pub(crate) static ARR_COUNT_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
// expects two string - the field and the delimiter
vec![DataType::Utf8],
// returns string
Arc::new(DataType::UInt64),
DataType::UInt64,
Volatility::Immutable,
Arc::new(arr_count_impl),
)
src/service/search/datafusion/udf/arrindex_udf.rs
+1 -1
View File
@@ -40,7 +40,7 @@ pub(crate) static ARR_INDEX_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
// expects a string (the array field) and two integers - start index and end index
vec![DataType::Utf8, DataType::Int64, DataType::Int64],
// returns string
Arc::new(DataType::Utf8),
DataType::Utf8,
Volatility::Immutable,
Arc::new(arr_index_impl),
)
src/service/search/datafusion/udf/arrjoin_udf.rs
+1 -1
View File
@@ -37,7 +37,7 @@ pub(crate) static ARR_JOIN_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
// expects two string - the field and the delimiter
vec![DataType::Utf8, DataType::Utf8],
// returns string
Arc::new(DataType::Utf8),
DataType::Utf8,
Volatility::Immutable,
Arc::new(arr_join_impl),
)
src/service/search/datafusion/udf/arrsort_udf.rs
+1 -1
View File
@@ -39,7 +39,7 @@ pub(crate) static ARR_SORT_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
// expects one string - the array field
vec![DataType::Utf8],
// returns string
Arc::new(DataType::Utf8),
DataType::Utf8,
Volatility::Immutable,
Arc::new(arr_sort_impl),
)
src/service/search/datafusion/udf/arrzip_udf.rs
+1 -1
View File
@@ -37,7 +37,7 @@ pub(crate) static ARR_ZIP_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
// expects three string - field1, field2 and the delim
vec![DataType::Utf8, DataType::Utf8, DataType::Utf8],
// returns string
Arc::new(DataType::Utf8),
DataType::Utf8,
Volatility::Immutable,
Arc::new(arr_zip_impl),
)
src/service/search/datafusion/udf/cast_to_arr_udf.rs
+1 -5
View File
@@ -37,11 +37,7 @@ pub(crate) static CAST_TO_ARR_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
CAST_TO_ARR_UDF_NAME,
// takes one argument: the field
vec![DataType::Utf8],
Arc::new(DataType::List(Arc::new(Field::new(
"item",
DataType::Utf8,
true,
)))),
DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
Volatility::Immutable,
Arc::new(cast_to_arr_impl),
)
src/service/search/datafusion/udf/date_format_udf.rs
+1 -1
View File
@@ -40,7 +40,7 @@ pub(crate) static DATE_FORMAT_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
// expects three string
vec![DataType::Int64, DataType::Utf8, DataType::Utf8],
// returns string
Arc::new(DataType::Utf8),
DataType::Utf8,
Volatility::Immutable,
Arc::new(date_format_expr_impl),
)
src/service/search/datafusion/udf/match_udf.rs
+2 -2
View File
@@ -36,7 +36,7 @@ pub(crate) static MATCH_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
// expects two string
vec![DataType::Utf8, DataType::Utf8],
// returns boolean
Arc::new(DataType::Boolean),
DataType::Boolean,
Volatility::Stable,
match_expr_impl(false),
)
@@ -49,7 +49,7 @@ pub(crate) static MATCH_IGNORE_CASE_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
// expects two string
vec![DataType::Utf8, DataType::Utf8],
// returns boolean
Arc::new(DataType::Boolean),
DataType::Boolean,
Volatility::Stable,
match_expr_impl(true),
)
src/service/search/datafusion/udf/regexp_udf.rs
+2 -2
View File
@@ -43,7 +43,7 @@ pub(crate) static REGEX_MATCH_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
super::REGEX_MATCH_UDF_NAME,
// takes two arguments: regex, pattern
vec![DataType::Utf8, DataType::Utf8],
Arc::new(DataType::Boolean),
DataType::Boolean,
Volatility::Stable,
regex_match_expr_impl(true),
)
@@ -55,7 +55,7 @@ pub(crate) static REGEX_NOT_MATCH_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
super::REGEX_NOT_MATCH_UDF_NAME,
// takes two arguments: regex, pattern
vec![DataType::Utf8, DataType::Utf8],
Arc::new(DataType::Boolean),
DataType::Boolean,
Volatility::Stable,
regex_match_expr_impl(false),
)
src/service/search/datafusion/udf/spath_udf.rs
+1 -1
View File
@@ -37,7 +37,7 @@ pub(crate) static SPATH_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
// expects two string - the field and the delimiter
vec![DataType::Utf8, DataType::Utf8],
// returns string
Arc::new(DataType::Utf8),
DataType::Utf8,
Volatility::Immutable,
Arc::new(spath_impl),
)
src/service/search/datafusion/udf/string_to_array_v2_udf.rs
+1 -5
View File
@@ -35,11 +35,7 @@ pub(crate) static STRING_TO_ARRAY_V2_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
STRING_TO_ARRAY_V2_UDF_NAME,
// takes two arguments: regex, pattern
vec![DataType::Utf8, DataType::Utf8],
Arc::new(DataType::List(Arc::new(Field::new(
"item",
DataType::Utf8,
true,
)))),
DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))),
Volatility::Immutable,
Arc::new(string_to_array_v2_impl),
)
src/service/search/datafusion/udf/time_range_udf.rs
+1 -1
View File
@@ -38,7 +38,7 @@ pub(crate) static TIME_RANGE_UDF: Lazy<ScalarUDF> = Lazy::new(|| {
// expects three string
vec![DataType::Int64, DataType::Utf8, DataType::Utf8],
// returns boolean
Arc::new(DataType::Boolean),
DataType::Boolean,
Volatility::Immutable,
Arc::new(time_range_expr_impl),
)
src/service/search/datafusion/udf/to_arr_string_udf.rs
+1 -1
View File
@@ -41,7 +41,7 @@ pub(crate) static TO_ARR_STRING: Lazy<ScalarUDF> = Lazy::new(|| {
DataType::Utf8,
true,
)))],
Arc::new(DataType::Utf8),
DataType::Utf8,
Volatility::Immutable,
Arc::new(to_arr_string_impl),
)
src/service/search/datafusion/udf/transform_udf.rs
+1 -1
View File
@@ -40,7 +40,7 @@ fn create_user_df(fn_name: &str, num_args: u8, pow_scalar: FnType) -> ScalarUDF
create_udf(
fn_name,
input_vec,
Arc::new(DataType::Utf8),
DataType::Utf8,
Volatility::Immutable,
pow_scalar,
)