diff --git a/Cargo.lock b/Cargo.lock index 41d8a827cc..84e3160be7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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", diff --git a/Cargo.toml b/Cargo.toml index 48237d0f49..8bf30d4b31 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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" diff --git a/src/service/search/datafusion/distributed_plan/empty_exec.rs b/src/service/search/datafusion/distributed_plan/empty_exec.rs index 4cbca063d4..a83151abca 100644 --- a/src/service/search/datafusion/distributed_plan/empty_exec.rs +++ b/src/service/search/datafusion/distributed_plan/empty_exec.rs @@ -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), diff --git a/src/service/search/datafusion/exec.rs b/src/service/search/datafusion/exec.rs index 83d96cd206..9552c8ac2c 100644 --- a/src/service/search/datafusion/exec.rs +++ b/src/service/search/datafusion/exec.rs @@ -223,7 +223,7 @@ pub async fn create_runtime_env(memory_limit: usize) -> Result { } super::MemoryPoolType::None => {} }; - RuntimeEnv::new(rn_config) + RuntimeEnv::try_new(rn_config) } pub async fn prepare_datafusion_context( diff --git a/src/service/search/datafusion/optimizer/add_sort_and_limit.rs b/src/service/search/datafusion/optimizer/add_sort_and_limit.rs index 072c7da992..f3b0be153d 100644 --- a/src/service/search/datafusion/optimizer/add_sort_and_limit.rs +++ b/src/service/search/datafusion/optimizer/add_sort_and_limit.rs @@ -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, 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>) -> usize { + match expr { + Some(expr) => match expr.as_ref() { + Expr::Literal(ScalarValue::Int64(Some(value))) => *value as usize, + _ => 0, + }, + _ => 0, + } +} + struct ChangeTableScanSchema {} impl ChangeTableScanSchema { diff --git a/src/service/search/datafusion/optimizer/add_timestamp.rs b/src/service/search/datafusion/optimizer/add_timestamp.rs index 7a281e96a4..25a325d8e8 100644 --- a/src/service/search/datafusion/optimizer/add_timestamp.rs +++ b/src/service/search/datafusion/optimizer/add_timestamp.rs @@ -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 () 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)), diff --git a/src/service/search/datafusion/optimizer/join_reorder.rs b/src/service/search/datafusion/optimizer/join_reorder.rs index 7eeb451246..d174f98773 100644 --- a/src/service/search/datafusion/optimizer/join_reorder.rs +++ b/src/service/search/datafusion/optimizer/join_reorder.rs @@ -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(); diff --git a/src/service/search/datafusion/optimizer/mod.rs b/src/service/search/datafusion/optimizer/mod.rs index 14c404a194..a969855104 100644 --- a/src/service/search/datafusion/optimizer/mod.rs +++ b/src/service/search/datafusion/optimizer/mod.rs @@ -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, } @@ -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(); diff --git a/src/service/search/datafusion/table_provider/empty_table.rs b/src/service/search/datafusion/table_provider/empty_table.rs index 449b91bc63..c78885f07f 100644 --- a/src/service/search/datafusion/table_provider/empty_table.rs +++ b/src/service/search/datafusion/table_provider/empty_table.rs @@ -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, diff --git a/src/service/search/datafusion/table_provider/memtable.rs b/src/service/search/datafusion/table_provider/memtable.rs index becce65d61..d2f6a05586 100644 --- a/src/service/search/datafusion/table_provider/memtable.rs +++ b/src/service/search/datafusion/table_provider/memtable.rs @@ -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, @@ -115,15 +116,6 @@ impl TableProvider for NewMemTable { }) } - async fn insert_into( - &self, - state: &dyn Session, - input: Arc, - overwrite: bool, - ) -> Result> { - 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) -> Arc { 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, diff --git a/src/service/search/datafusion/table_provider/mod.rs b/src/service/search/datafusion/table_provider/mod.rs index d068384790..e9a1b4c9ca 100644 --- a/src/service/search/datafusion/table_provider/mod.rs +++ b/src/service/search/datafusion/table_provider/mod.rs @@ -67,6 +67,7 @@ mod helpers; pub mod memtable; pub mod uniontable; +#[derive(Debug)] pub(crate) struct NewListingTable { table_paths: Vec, /// File fields only @@ -441,7 +442,7 @@ fn create_ordering(schema: &Schema, sort_order: &[Vec]) -> Result match expressions::col(&col.name, schema) { diff --git a/src/service/search/datafusion/table_provider/uniontable.rs b/src/service/search/datafusion/table_provider/uniontable.rs index 7a61f38afb..a077708614 100644 --- a/src/service/search/datafusion/table_provider/uniontable.rs +++ b/src/service/search/datafusion/table_provider/uniontable.rs @@ -25,6 +25,7 @@ use datafusion::{ physical_plan::{empty::EmptyExec, union::UnionExec, ExecutionPlan}, }; +#[derive(Debug)] pub(crate) struct NewUnionTable { schema: SchemaRef, tables: Vec>, diff --git a/src/service/search/datafusion/udf/arr_descending_udf.rs b/src/service/search/datafusion/udf/arr_descending_udf.rs index a4c79c0006..85b4d93d31 100644 --- a/src/service/search/datafusion/udf/arr_descending_udf.rs +++ b/src/service/search/datafusion/udf/arr_descending_udf.rs @@ -39,7 +39,7 @@ pub(crate) static ARR_DESCENDING_UDF: Lazy = Lazy::new(|| { // expects three string vec![DataType::Utf8], // returns string - Arc::new(DataType::Utf8), + DataType::Utf8, Volatility::Immutable, Arc::new(arr_descending_impl), ) diff --git a/src/service/search/datafusion/udf/arrcount_udf.rs b/src/service/search/datafusion/udf/arrcount_udf.rs index 789ab66a0d..31497a592e 100644 --- a/src/service/search/datafusion/udf/arrcount_udf.rs +++ b/src/service/search/datafusion/udf/arrcount_udf.rs @@ -37,7 +37,7 @@ pub(crate) static ARR_COUNT_UDF: Lazy = 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), ) diff --git a/src/service/search/datafusion/udf/arrindex_udf.rs b/src/service/search/datafusion/udf/arrindex_udf.rs index e6411b6c1b..6641a21d7f 100644 --- a/src/service/search/datafusion/udf/arrindex_udf.rs +++ b/src/service/search/datafusion/udf/arrindex_udf.rs @@ -40,7 +40,7 @@ pub(crate) static ARR_INDEX_UDF: Lazy = 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), ) diff --git a/src/service/search/datafusion/udf/arrjoin_udf.rs b/src/service/search/datafusion/udf/arrjoin_udf.rs index 93aae5253a..40367a4566 100644 --- a/src/service/search/datafusion/udf/arrjoin_udf.rs +++ b/src/service/search/datafusion/udf/arrjoin_udf.rs @@ -37,7 +37,7 @@ pub(crate) static ARR_JOIN_UDF: Lazy = 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), ) diff --git a/src/service/search/datafusion/udf/arrsort_udf.rs b/src/service/search/datafusion/udf/arrsort_udf.rs index ef474ddf7f..e4a87abce5 100644 --- a/src/service/search/datafusion/udf/arrsort_udf.rs +++ b/src/service/search/datafusion/udf/arrsort_udf.rs @@ -39,7 +39,7 @@ pub(crate) static ARR_SORT_UDF: Lazy = 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), ) diff --git a/src/service/search/datafusion/udf/arrzip_udf.rs b/src/service/search/datafusion/udf/arrzip_udf.rs index 708f0f6b10..6f4135b2dc 100644 --- a/src/service/search/datafusion/udf/arrzip_udf.rs +++ b/src/service/search/datafusion/udf/arrzip_udf.rs @@ -37,7 +37,7 @@ pub(crate) static ARR_ZIP_UDF: Lazy = 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), ) diff --git a/src/service/search/datafusion/udf/cast_to_arr_udf.rs b/src/service/search/datafusion/udf/cast_to_arr_udf.rs index 21e51ab482..132e151b54 100644 --- a/src/service/search/datafusion/udf/cast_to_arr_udf.rs +++ b/src/service/search/datafusion/udf/cast_to_arr_udf.rs @@ -37,11 +37,7 @@ pub(crate) static CAST_TO_ARR_UDF: Lazy = 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), ) diff --git a/src/service/search/datafusion/udf/date_format_udf.rs b/src/service/search/datafusion/udf/date_format_udf.rs index 1a45491175..0445163dac 100644 --- a/src/service/search/datafusion/udf/date_format_udf.rs +++ b/src/service/search/datafusion/udf/date_format_udf.rs @@ -40,7 +40,7 @@ pub(crate) static DATE_FORMAT_UDF: Lazy = 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), ) diff --git a/src/service/search/datafusion/udf/match_udf.rs b/src/service/search/datafusion/udf/match_udf.rs index fb4dc5f211..c96641fb4d 100644 --- a/src/service/search/datafusion/udf/match_udf.rs +++ b/src/service/search/datafusion/udf/match_udf.rs @@ -36,7 +36,7 @@ pub(crate) static MATCH_UDF: Lazy = 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 = Lazy::new(|| { // expects two string vec![DataType::Utf8, DataType::Utf8], // returns boolean - Arc::new(DataType::Boolean), + DataType::Boolean, Volatility::Stable, match_expr_impl(true), ) diff --git a/src/service/search/datafusion/udf/regexp_udf.rs b/src/service/search/datafusion/udf/regexp_udf.rs index 9638b74430..213ebc3235 100644 --- a/src/service/search/datafusion/udf/regexp_udf.rs +++ b/src/service/search/datafusion/udf/regexp_udf.rs @@ -43,7 +43,7 @@ pub(crate) static REGEX_MATCH_UDF: Lazy = 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 = 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), ) diff --git a/src/service/search/datafusion/udf/spath_udf.rs b/src/service/search/datafusion/udf/spath_udf.rs index c442869ee4..39ecfe12ad 100644 --- a/src/service/search/datafusion/udf/spath_udf.rs +++ b/src/service/search/datafusion/udf/spath_udf.rs @@ -37,7 +37,7 @@ pub(crate) static SPATH_UDF: Lazy = 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), ) diff --git a/src/service/search/datafusion/udf/string_to_array_v2_udf.rs b/src/service/search/datafusion/udf/string_to_array_v2_udf.rs index 8a5c3fa46a..abad2e8b45 100644 --- a/src/service/search/datafusion/udf/string_to_array_v2_udf.rs +++ b/src/service/search/datafusion/udf/string_to_array_v2_udf.rs @@ -35,11 +35,7 @@ pub(crate) static STRING_TO_ARRAY_V2_UDF: Lazy = 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), ) diff --git a/src/service/search/datafusion/udf/time_range_udf.rs b/src/service/search/datafusion/udf/time_range_udf.rs index 941604aa13..d572e9eab7 100644 --- a/src/service/search/datafusion/udf/time_range_udf.rs +++ b/src/service/search/datafusion/udf/time_range_udf.rs @@ -38,7 +38,7 @@ pub(crate) static TIME_RANGE_UDF: Lazy = 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), ) diff --git a/src/service/search/datafusion/udf/to_arr_string_udf.rs b/src/service/search/datafusion/udf/to_arr_string_udf.rs index 3cfd5bee1c..6215d1bb65 100644 --- a/src/service/search/datafusion/udf/to_arr_string_udf.rs +++ b/src/service/search/datafusion/udf/to_arr_string_udf.rs @@ -41,7 +41,7 @@ pub(crate) static TO_ARR_STRING: Lazy = Lazy::new(|| { DataType::Utf8, true, )))], - Arc::new(DataType::Utf8), + DataType::Utf8, Volatility::Immutable, Arc::new(to_arr_string_impl), ) diff --git a/src/service/search/datafusion/udf/transform_udf.rs b/src/service/search/datafusion/udf/transform_udf.rs index e16dda9347..43803acd2b 100644 --- a/src/service/search/datafusion/udf/transform_udf.rs +++ b/src/service/search/datafusion/udf/transform_udf.rs @@ -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, )