From 4d59f10829483c5f07a07d6de4c563a779eea7db Mon Sep 17 00:00:00 2001 From: osipovartem Date: Tue, 1 Sep 2026 23:21:05 +0300 Subject: [PATCH 01/15] Expose input schema to aggregate expression planners --- datafusion/expr/src/planner.rs | 13 +++++++++ datafusion/sql/src/expr/function.rs | 2 +- datafusion/sql/tests/common/mod.rs | 37 +++++++++++++++++++++++-- datafusion/sql/tests/sql_integration.rs | 27 +++++++++++++++++- 4 files changed, 75 insertions(+), 4 deletions(-) diff --git a/datafusion/expr/src/planner.rs b/datafusion/expr/src/planner.rs index b3ac64eede717..99cf5bef19fbf 100644 --- a/datafusion/expr/src/planner.rs +++ b/datafusion/expr/src/planner.rs @@ -273,6 +273,19 @@ pub trait ExprPlanner: Debug + Send + Sync { Ok(PlannerResult::Original(expr)) } + /// Plans aggregate functions with access to the input schema. + /// + /// The default implementation delegates to [`Self::plan_aggregate`] so existing planners do + /// not need to change. Planners that need to resolve schema-dependent arguments, such as a + /// qualified wildcard, can override this method instead. + fn plan_aggregate_with_schema( + &self, + expr: RawAggregateExpr, + _schema: &DFSchema, + ) -> Result> { + self.plan_aggregate(expr) + } + /// Plans window functions, such as `COUNT()` /// /// Returns original expression arguments if not possible diff --git a/datafusion/sql/src/expr/function.rs b/datafusion/sql/src/expr/function.rs index e6bee31fbf106..4269263dea1d9 100644 --- a/datafusion/sql/src/expr/function.rs +++ b/datafusion/sql/src/expr/function.rs @@ -839,7 +839,7 @@ impl SqlToRel<'_, S> { null_treatment, }; for planner in self.context_provider.get_expr_planners().iter() { - match planner.plan_aggregate(aggregate_expr)? { + match planner.plan_aggregate_with_schema(aggregate_expr, schema)? { PlannerResult::Planned(expr) => return Ok(expr), PlannerResult::Original(expr) => aggregate_expr = expr, } diff --git a/datafusion/sql/tests/common/mod.rs b/datafusion/sql/tests/common/mod.rs index e7c819bbf64a6..a9e08440c318c 100644 --- a/datafusion/sql/tests/common/mod.rs +++ b/datafusion/sql/tests/common/mod.rs @@ -25,8 +25,10 @@ use arrow::datatypes::*; use datafusion_common::config::ConfigOptions; use datafusion_common::datatype::DataTypeExt; use datafusion_common::file_options::file_type::FileType; -use datafusion_common::{DFSchema, GetExt, Result, TableReference, plan_err}; -use datafusion_expr::planner::{ExprPlanner, PlannerResult, TypePlanner}; +use datafusion_common::{Column, DFSchema, GetExt, Result, TableReference, plan_err}; +use datafusion_expr::planner::{ + ExprPlanner, PlannerResult, RawAggregateExpr, TypePlanner, +}; use datafusion_expr::{ AggregateUDF, Expr, HigherOrderUDF, ScalarUDF, TableSource, WindowUDF, }; @@ -415,3 +417,34 @@ impl ExprPlanner for CustomExprPlanner { Ok(PlannerResult::Planned(make_array(exprs))) } } + +#[derive(Debug)] +pub struct QualifiedWildcardCountPlanner; + +impl ExprPlanner for QualifiedWildcardCountPlanner { + fn plan_aggregate_with_schema( + &self, + mut expr: RawAggregateExpr, + schema: &DFSchema, + ) -> Result> { + #[expect(deprecated)] + let Some(Expr::Wildcard { + qualifier: Some(qualifier), + .. + }) = expr.args.first() + else { + return Ok(PlannerResult::Original(expr)); + }; + + if expr.func.name() != "count" || expr.args.len() != 1 { + return Ok(PlannerResult::Original(expr)); + } + + expr.args = schema + .fields_indices_with_qualified(qualifier) + .into_iter() + .map(|index| Expr::Column(Column::from(schema.qualified_field(index)))) + .collect(); + Ok(PlannerResult::Original(expr)) + } +} diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index 08a95381b32c8..39a17024fb632 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -43,7 +43,9 @@ use datafusion_sql::{ planner::{NullOrdering, ParserOptions, PlannerContext, SqlToRel}, }; -use crate::common::{CustomExprPlanner, CustomTypePlanner, MockSessionState}; +use crate::common::{ + CustomExprPlanner, CustomTypePlanner, MockSessionState, QualifiedWildcardCountPlanner, +}; use datafusion_functions::core::planner::CoreFunctionPlanner; use datafusion_functions_aggregate::{ approx_median::approx_median_udaf, @@ -2155,6 +2157,29 @@ fn select_count_column() { ); } +#[test] +fn aggregate_expr_planner_can_resolve_qualified_wildcard_from_schema() { + let state = + mock_session_state().with_expr_planner(Arc::new(QualifiedWildcardCountPlanner)); + let plan = logical_plan_from_state( + "SELECT count(p.*) FROM person AS p", + &GenericDialect {}, + ParserOptions::default(), + state, + ) + .unwrap(); + + assert_snapshot!( + plan, + @r" + Projection: count(p.id,p.first_name,p.last_name,p.age,p.state,p.salary,p.birth_date,p.😀) + Aggregate: groupBy=[[]], aggr=[[count(p.id, p.first_name, p.last_name, p.age, p.state, p.salary, p.birth_date, p.😀)]] + SubqueryAlias: p + TableScan: person + " + ); +} + #[test] fn select_approx_median() { let sql = "SELECT approx_median(age) FROM person"; From 8461571125c601702345b27a53d67d7120592579 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Tue, 1 Sep 2026 23:56:59 +0300 Subject: [PATCH 02/15] Expose input schema to scalar expression planners --- datafusion/expr/src/planner.rs | 18 ++++++++++ datafusion/expr/src/utils.rs | 37 +++++++++++++++++--- datafusion/sql/src/expr/function.rs | 37 ++++++++++++++++++-- datafusion/sql/tests/common/mod.rs | 30 +++++++++++++++-- datafusion/sql/tests/sql_integration.rs | 45 ++++++++++++++++++++++++- 5 files changed, 158 insertions(+), 9 deletions(-) diff --git a/datafusion/expr/src/planner.rs b/datafusion/expr/src/planner.rs index 99cf5bef19fbf..4658d22325d2f 100644 --- a/datafusion/expr/src/planner.rs +++ b/datafusion/expr/src/planner.rs @@ -154,6 +154,17 @@ pub trait ContextProvider { /// /// [Extending SQL in DataFusion: from ->> to TABLESAMPLE blog]: https://datafusion.apache.org/blog/2026/01/12/extending-sql pub trait ExprPlanner: Debug + Send + Sync { + /// Plans scalar functions, such as `CONCAT(, ...)`, with access to the input schema. + /// + /// Returns the original scalar function if planning is not possible. + fn plan_scalar_with_schema( + &self, + expr: RawScalarExpr, + _schema: &DFSchema, + ) -> Result> { + Ok(PlannerResult::Original(expr)) + } + /// Plan the binary operation between two expressions, returns original /// BinaryExpr if not possible fn plan_binary_op( @@ -343,6 +354,13 @@ pub struct RawAggregateExpr { pub null_treatment: Option, } +/// This structure is used by scalar function expression planners. +#[derive(Debug, Clone)] +pub struct RawScalarExpr { + pub func: Arc, + pub args: Vec, +} + /// This structure is used by `WindowFunctionPlanner` to plan operators with /// custom expressions. #[derive(Debug, Clone)] diff --git a/datafusion/expr/src/utils.rs b/datafusion/expr/src/utils.rs index 7f79c5cf18c4a..646beffbc173d 100644 --- a/datafusion/expr/src/utils.rs +++ b/datafusion/expr/src/utils.rs @@ -449,19 +449,48 @@ pub fn expand_wildcard( wildcard_options: Option<&WildcardOptions>, ) -> Result> { let mut columns_to_skip = exclude_using_columns(plan)?; + columns_to_skip.extend(excluded_columns_from_schema( + schema, + wildcard_options, + None, + )?); + Ok(get_exprs_except_skipped(schema, &columns_to_skip)) +} + +/// Resolves an unqualified wildcard using only the input schema. +/// +/// Unlike [`expand_wildcard`], this helper cannot account for duplicate columns introduced by a +/// join with a `USING` clause. It is intended for schema-aware expression planners that receive a +/// wildcard as a function argument. +pub fn expand_wildcard_from_schema( + schema: &DFSchema, + wildcard_options: Option<&WildcardOptions>, +) -> Result> { + let columns_to_skip = excluded_columns_from_schema(schema, wildcard_options, None)?; + Ok(get_exprs_except_skipped(schema, &columns_to_skip)) +} + +fn excluded_columns_from_schema( + schema: &DFSchema, + wildcard_options: Option<&WildcardOptions>, + qualifier: Option<&TableReference>, +) -> Result> { let excluded_columns = if let Some(WildcardOptions { exclude: opt_exclude, except: opt_except, .. }) = wildcard_options { - get_excluded_columns(opt_exclude.as_ref(), opt_except.as_ref(), schema, None)? + get_excluded_columns( + opt_exclude.as_ref(), + opt_except.as_ref(), + schema, + qualifier, + )? } else { vec![] }; - // Add each excluded `Column` to columns_to_skip - columns_to_skip.extend(excluded_columns); - Ok(get_exprs_except_skipped(schema, &columns_to_skip)) + Ok(excluded_columns.into_iter().collect()) } /// Resolves an `Expr::Wildcard` to a collection of qualified `Expr::Column`'s. diff --git a/datafusion/sql/src/expr/function.rs b/datafusion/sql/src/expr/function.rs index 4269263dea1d9..0b340ff9001dc 100644 --- a/datafusion/sql/src/expr/function.rs +++ b/datafusion/sql/src/expr/function.rs @@ -30,7 +30,7 @@ use datafusion_expr::{ self, HigherOrderFunction, Lambda, NullTreatment, ScalarFunction, Unnest, WildcardOptions, WindowFunction, }, - planner::{PlannerResult, RawAggregateExpr, RawWindowExpr}, + planner::{PlannerResult, RawAggregateExpr, RawScalarExpr, RawWindowExpr}, type_coercion::functions::value_fields_with_higher_order_udf, }; use sqlparser::ast::{ @@ -348,7 +348,19 @@ impl SqlToRel<'_, S> { }; // After resolution, all arguments are positional - let inner = ScalarFunction::new_udf(fm, resolved_args); + let mut scalar_expr = RawScalarExpr { + func: fm, + args: resolved_args, + }; + for planner in self.context_provider.get_expr_planners().iter() { + match planner.plan_scalar_with_schema(scalar_expr, schema)? { + PlannerResult::Planned(expr) => return Ok(expr), + PlannerResult::Original(expr) => scalar_expr = expr, + } + } + + let RawScalarExpr { func, args } = scalar_expr; + let inner = ScalarFunction::new_udf(func, args); if name.eq_ignore_ascii_case(inner.name()) { return Ok(Expr::ScalarFunction(inner)); @@ -1063,6 +1075,27 @@ impl SqlToRel<'_, S> { }; Ok((expr, None)) } + FunctionArg::Unnamed(FunctionArgExpr::WildcardWithOptions(options)) => { + if options.opt_alias.is_some() { + return not_impl_err!("wildcard function argument with AS alias"); + } + if options.opt_replace.is_some() { + return not_impl_err!("wildcard function argument with REPLACE"); + } + + #[expect(deprecated)] + let expr = Expr::Wildcard { + qualifier: None, + options: Box::new(WildcardOptions { + ilike: options.opt_ilike, + exclude: options.opt_exclude, + except: options.opt_except, + replace: None, + rename: options.opt_rename, + }), + }; + Ok((expr, None)) + } FunctionArg::Unnamed(FunctionArgExpr::QualifiedWildcard(object_name)) => { let qualifier = self.object_name_to_table_reference(object_name)?; // Sanity check on qualifier with schema diff --git a/datafusion/sql/tests/common/mod.rs b/datafusion/sql/tests/common/mod.rs index a9e08440c318c..dfe393c1952ac 100644 --- a/datafusion/sql/tests/common/mod.rs +++ b/datafusion/sql/tests/common/mod.rs @@ -27,10 +27,11 @@ use datafusion_common::datatype::DataTypeExt; use datafusion_common::file_options::file_type::FileType; use datafusion_common::{Column, DFSchema, GetExt, Result, TableReference, plan_err}; use datafusion_expr::planner::{ - ExprPlanner, PlannerResult, RawAggregateExpr, TypePlanner, + ExprPlanner, PlannerResult, RawAggregateExpr, RawScalarExpr, TypePlanner, }; +use datafusion_expr::utils::expand_wildcard_from_schema; use datafusion_expr::{ - AggregateUDF, Expr, HigherOrderUDF, ScalarUDF, TableSource, WindowUDF, + AggregateUDF, Expr, ExprSchemable, HigherOrderUDF, ScalarUDF, TableSource, WindowUDF, }; use datafusion_functions_nested::expr_fn::make_array; use datafusion_sql::planner::ContextProvider; @@ -448,3 +449,28 @@ impl ExprPlanner for QualifiedWildcardCountPlanner { Ok(PlannerResult::Original(expr)) } } + +#[derive(Debug)] +pub struct ScalarWildcardPlanner; + +impl ExprPlanner for ScalarWildcardPlanner { + fn plan_scalar_with_schema( + &self, + mut expr: RawScalarExpr, + schema: &DFSchema, + ) -> Result> { + #[expect(deprecated)] + let [Expr::Wildcard { options, .. }] = expr.args.as_slice() else { + return Ok(PlannerResult::Original(expr)); + }; + if expr.func.name() != "concat" { + return Ok(PlannerResult::Original(expr)); + } + + expr.args = expand_wildcard_from_schema(schema, Some(options))? + .into_iter() + .filter(|expr| expr.get_type(schema).is_ok_and(|ty| ty == DataType::Utf8)) + .collect(); + Ok(PlannerResult::Original(expr)) + } +} diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index 39a17024fb632..e5897eaa33fee 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -44,7 +44,8 @@ use datafusion_sql::{ }; use crate::common::{ - CustomExprPlanner, CustomTypePlanner, MockSessionState, QualifiedWildcardCountPlanner, + CustomExprPlanner, CustomTypePlanner, MockSessionState, + QualifiedWildcardCountPlanner, ScalarWildcardPlanner, }; use datafusion_functions::core::planner::CoreFunctionPlanner; use datafusion_functions_aggregate::{ @@ -2180,6 +2181,48 @@ fn aggregate_expr_planner_can_resolve_qualified_wildcard_from_schema() { ); } +#[test] +fn scalar_expr_planner_can_resolve_wildcard_from_schema() { + let state = mock_session_state().with_expr_planner(Arc::new(ScalarWildcardPlanner)); + let plan = logical_plan_from_state( + "SELECT concat(*) FROM person AS p", + &GenericDialect {}, + ParserOptions::default(), + state, + ) + .unwrap(); + + assert_snapshot!( + plan, + @r" + Projection: concat(p.first_name, p.last_name, p.state) + SubqueryAlias: p + TableScan: person + " + ); +} + +#[test] +fn scalar_expr_planner_receives_wildcard_options() { + let state = mock_session_state().with_expr_planner(Arc::new(ScalarWildcardPlanner)); + let plan = logical_plan_from_state( + "SELECT concat(* EXCLUDE first_name) FROM person AS p", + &GenericDialect {}, + ParserOptions::default(), + state, + ) + .unwrap(); + + assert_snapshot!( + plan, + @r" + Projection: concat(p.last_name, p.state) + SubqueryAlias: p + TableScan: person + " + ); +} + #[test] fn select_approx_median() { let sql = "SELECT approx_median(age) FROM person"; From 9c8e6192ac5615951a99d20353d55038c5719389 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 00:44:34 +0300 Subject: [PATCH 03/15] Apply ILIKE when expanding wildcard arguments --- datafusion/expr/src/utils.rs | 34 ++++++++++++++++++++++++- datafusion/sql/tests/sql_integration.rs | 21 +++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/datafusion/expr/src/utils.rs b/datafusion/expr/src/utils.rs index 646beffbc173d..297fbce14c795 100644 --- a/datafusion/expr/src/utils.rs +++ b/datafusion/expr/src/utils.rs @@ -490,7 +490,39 @@ fn excluded_columns_from_schema( } else { vec![] }; - Ok(excluded_columns.into_iter().collect()) + let mut excluded_columns = excluded_columns.into_iter().collect::>(); + if let Some(ilike) = wildcard_options.and_then(|options| options.ilike.as_ref()) { + excluded_columns.extend( + schema + .columns() + .into_iter() + .filter(|column| !matches_ilike(&column.name, &ilike.pattern)), + ); + } + Ok(excluded_columns) +} + +fn matches_ilike(value: &str, pattern: &str) -> bool { + let value = value.to_lowercase().chars().collect::>(); + let pattern = pattern.to_lowercase().chars().collect::>(); + let mut previous = vec![false; value.len() + 1]; + previous[0] = true; + + for pattern_char in pattern { + let mut current = vec![false; value.len() + 1]; + if pattern_char == '%' { + current[0] = previous[0]; + } + for (index, value_char) in value.iter().enumerate() { + current[index + 1] = match pattern_char { + '%' => previous[index + 1] || current[index], + '_' => previous[index], + literal => previous[index] && literal == *value_char, + }; + } + previous = current; + } + previous[value.len()] } /// Resolves an `Expr::Wildcard` to a collection of qualified `Expr::Column`'s. diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index e5897eaa33fee..71e7ee7ad9bdf 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -2223,6 +2223,27 @@ fn scalar_expr_planner_receives_wildcard_options() { ); } +#[test] +fn scalar_expr_planner_applies_wildcard_ilike() { + let state = mock_session_state().with_expr_planner(Arc::new(ScalarWildcardPlanner)); + let plan = logical_plan_from_state( + "SELECT concat(* ILIKE '%name') FROM person AS p", + &GenericDialect {}, + ParserOptions::default(), + state, + ) + .unwrap(); + + assert_snapshot!( + plan, + @r" + Projection: concat(p.first_name, p.last_name) + SubqueryAlias: p + TableScan: person + " + ); +} + #[test] fn select_approx_median() { let sql = "SELECT approx_median(age) FROM person"; From f05ec59424321cab1a1af7921758293a18c99c7d Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 01:09:32 +0300 Subject: [PATCH 04/15] Plan qualified wildcard function options --- Cargo.lock | 26 ++++++------ Cargo.toml | 2 +- datafusion/sql/src/expr/function.rs | 53 ++++++++++++++++++------- datafusion/sql/tests/sql_integration.rs | 22 ++++++++++ 4 files changed, 73 insertions(+), 30 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 62c04d332b98e..943fd94e80a35 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -112,7 +112,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -123,7 +123,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -2801,7 +2801,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -2940,7 +2940,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -4212,7 +4212,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -5373,7 +5373,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -5845,7 +5845,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -5876,8 +5876,7 @@ dependencies = [ [[package]] name = "sqlparser" version = "0.62.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "13c6d1b651dc4edf07eead2a0c6c78016ce971bc2c10da5266861b13f25e7cec" +source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=1a3f48f60802d0f135a968a27c89a3351e78b159#1a3f48f60802d0f135a968a27c89a3351e78b159" dependencies = [ "log", "recursive", @@ -5887,8 +5886,7 @@ dependencies = [ [[package]] name = "sqlparser_derive" version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6dd45d8fc1c79299bfbb7190e42ccbbdf6a5f52e4a6ad98d92357ea965bd289" +source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=1a3f48f60802d0f135a968a27c89a3351e78b159#1a3f48f60802d0f135a968a27c89a3351e78b159" dependencies = [ "proc-macro2", "quote", @@ -5945,7 +5943,7 @@ dependencies = [ "cfg-if", "libc", "psm", - "windows-sys 0.61.2", + "windows-sys 0.60.2", ] [[package]] @@ -6125,7 +6123,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix", - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] @@ -7066,7 +7064,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.52.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 4526fa0c58934..13e26dafdc4ee 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -196,7 +196,7 @@ regex = "1.12" rstest = "0.26.1" serde_json = "1" sha2 = "^0.11.0" -sqlparser = { version = "0.62.0", default-features = false, features = ["std", "visitor"] } +sqlparser = { git = "https://github.com/Embucket/datafusion-sqlparser-rs.git", rev = "1a3f48f60802d0f135a968a27c89a3351e78b159", default-features = false, features = ["std", "visitor"] } stacker = "0.1.24" strum = "0.28.0" strum_macros = "0.28.0" diff --git a/datafusion/sql/src/expr/function.rs b/datafusion/sql/src/expr/function.rs index 0b340ff9001dc..7b1b88424594e 100644 --- a/datafusion/sql/src/expr/function.rs +++ b/datafusion/sql/src/expr/function.rs @@ -36,9 +36,29 @@ use datafusion_expr::{ use sqlparser::ast::{ DuplicateTreatment, Expr as SQLExpr, Function as SQLFunction, FunctionArg, FunctionArgExpr, FunctionArgumentClause, FunctionArgumentList, FunctionArguments, - LambdaFunction, ObjectName, OrderByExpr, Spanned, WindowType, + LambdaFunction, ObjectName, OrderByExpr, Spanned, WildcardAdditionalOptions, + WindowType, }; +fn function_wildcard_options( + options: WildcardAdditionalOptions, +) -> Result> { + if options.opt_alias.is_some() { + return not_impl_err!("wildcard function argument with AS alias"); + } + if options.opt_replace.is_some() { + return not_impl_err!("wildcard function argument with REPLACE"); + } + + Ok(Box::new(WildcardOptions { + ilike: options.opt_ilike, + exclude: options.opt_exclude, + except: options.opt_except, + replace: None, + rename: options.opt_rename, + })) +} + /// Suggest a valid function based on an invalid input function name /// /// Returns `None` if no valid matches are found. This happens when there are no @@ -1076,23 +1096,10 @@ impl SqlToRel<'_, S> { Ok((expr, None)) } FunctionArg::Unnamed(FunctionArgExpr::WildcardWithOptions(options)) => { - if options.opt_alias.is_some() { - return not_impl_err!("wildcard function argument with AS alias"); - } - if options.opt_replace.is_some() { - return not_impl_err!("wildcard function argument with REPLACE"); - } - #[expect(deprecated)] let expr = Expr::Wildcard { qualifier: None, - options: Box::new(WildcardOptions { - ilike: options.opt_ilike, - exclude: options.opt_exclude, - except: options.opt_except, - replace: None, - rename: options.opt_rename, - }), + options: function_wildcard_options(options)?, }; Ok((expr, None)) } @@ -1111,6 +1118,22 @@ impl SqlToRel<'_, S> { }; Ok((expr, None)) } + FunctionArg::Unnamed(FunctionArgExpr::QualifiedWildcardWithOptions( + object_name, + options, + )) => { + let qualifier = self.object_name_to_table_reference(object_name)?; + if schema.fields_indices_with_qualified(&qualifier).is_empty() { + return plan_err!("Invalid qualifier {qualifier}"); + } + + #[expect(deprecated)] + let expr = Expr::Wildcard { + qualifier: qualifier.into(), + options: function_wildcard_options(options)?, + }; + Ok((expr, None)) + } // PostgreSQL dialect uses ExprNamed variant with expression for name FunctionArg::ExprNamed { name: SQLExpr::Identifier(name), diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index 71e7ee7ad9bdf..7855e755e1a7d 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -61,6 +61,7 @@ use insta::{allow_duplicates, assert_snapshot}; use rstest::rstest; use sqlparser::dialect::{ DatabricksDialect, Dialect, GenericDialect, HiveDialect, MySqlDialect, + SnowflakeDialect, }; use sqlparser::parser::Parser; @@ -2244,6 +2245,27 @@ fn scalar_expr_planner_applies_wildcard_ilike() { ); } +#[test] +fn scalar_expr_planner_applies_qualified_wildcard_options() { + let state = mock_session_state().with_expr_planner(Arc::new(ScalarWildcardPlanner)); + let plan = logical_plan_from_state( + "SELECT concat((p.* ILIKE '%name')) FROM person AS p", + &SnowflakeDialect {}, + ParserOptions::default(), + state, + ) + .unwrap(); + + assert_snapshot!( + plan, + @r" + Projection: concat(p.first_name, p.last_name) + SubqueryAlias: p + TableScan: person + " + ); +} + #[test] fn select_approx_median() { let sql = "SELECT approx_median(age) FROM person"; From dd655166d978ee663fe3f5c06bf23c762745a378 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 02:01:29 +0300 Subject: [PATCH 05/15] Add optional decimal literal normalization --- datafusion/common/src/config.rs | 4 ++ .../core/src/execution/session_state.rs | 2 + datafusion/sql/src/expr/value.rs | 60 +++++++++++++++---- datafusion/sql/src/planner.rs | 11 ++++ datafusion/sql/tests/sql_integration.rs | 18 ++++++ docs/source/user-guide/configs.md | 1 + 6 files changed, 85 insertions(+), 11 deletions(-) diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs index d823883e58b5d..bbdbd4f2dcaf6 100644 --- a/datafusion/common/src/config.rs +++ b/datafusion/common/src/config.rs @@ -271,6 +271,10 @@ config_namespace! { /// When set to true, SQL parser will parse float as decimal type pub parse_float_as_decimal: bool, default = false + /// When set to true, insignificant trailing zeros are removed from decimal literals. + /// For example, `10.00` is planned as `DECIMAL(2, 0)` instead of `DECIMAL(4, 2)`. + pub trim_decimal_literal_trailing_zeros: bool, default = false + /// When set to true, SQL parser will normalize ident (convert ident to lowercase when not quoted) pub enable_ident_normalization: bool, default = true diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs index 4c4abdebe9211..ee707162908a4 100644 --- a/datafusion/core/src/execution/session_state.rs +++ b/datafusion/core/src/execution/session_state.rs @@ -581,6 +581,8 @@ impl SessionState { ParserOptions { parse_float_as_decimal: sql_parser_options.parse_float_as_decimal, + trim_decimal_literal_trailing_zeros: sql_parser_options + .trim_decimal_literal_trailing_zeros, enable_ident_normalization: sql_parser_options.enable_ident_normalization, enable_options_value_normalization: sql_parser_options .enable_options_value_normalization, diff --git a/datafusion/sql/src/expr/value.rs b/datafusion/sql/src/expr/value.rs index 1307e917e4251..e1a6965fca590 100644 --- a/datafusion/sql/src/expr/value.rs +++ b/datafusion/sql/src/expr/value.rs @@ -23,7 +23,7 @@ use arrow::datatypes::{ DECIMAL128_MAX_PRECISION, DECIMAL256_MAX_PRECISION, FieldRef, i256, }; use bigdecimal::num_bigint::BigInt; -use bigdecimal::{BigDecimal, Signed, ToPrimitive}; +use bigdecimal::{BigDecimal, Signed, ToPrimitive, Zero}; use datafusion_common::{ DFSchema, DataFusionError, Result, ScalarValue, internal_datafusion_err, not_impl_err, plan_err, @@ -108,7 +108,11 @@ impl SqlToRel<'_, S> { } if self.options.parse_float_as_decimal { - parse_decimal(unsigned_number, negative) + parse_decimal( + unsigned_number, + negative, + self.options.trim_decimal_literal_trailing_zeros, + ) } else { signed_number.parse::().map(lit).map_err(|_| { DataFusionError::from(ParserError(format!( @@ -375,7 +379,11 @@ fn bigint_to_i256(v: &BigInt) -> Option { } } -fn parse_decimal(unsigned_number: &str, negative: bool) -> Result { +fn parse_decimal( + unsigned_number: &str, + negative: bool, + trim_trailing_zeros: bool, +) -> Result { let mut dec = BigDecimal::from_str(unsigned_number).map_err(|e| { DataFusionError::from(ParserError(format!( "Cannot parse {unsigned_number} as BigDecimal: {e}" @@ -384,9 +392,14 @@ fn parse_decimal(unsigned_number: &str, negative: bool) -> Result { if negative { dec = dec.neg(); } - - let digits = dec.digits(); - let (int_val, scale) = dec.into_bigint_and_exponent(); + let (mut int_val, mut scale) = dec.into_bigint_and_exponent(); + if trim_trailing_zeros { + while scale > 0 && (&int_val % 10_u8).is_zero() { + int_val /= 10_u8; + scale -= 1; + } + } + let digits = BigDecimal::new(int_val.clone(), scale).digits(); if scale < i8::MIN as i64 { return not_impl_err!( "Decimal scale {} exceeds the minimum supported scale: {}", @@ -394,7 +407,12 @@ fn parse_decimal(unsigned_number: &str, negative: bool) -> Result { i8::MIN ); } - let precision = if scale > 0 { + let precision = if trim_trailing_zeros && scale > 0 { + // Exact numeric types include the zero before the decimal point in their + // precision. This makes `0.00100` normalize to DECIMAL(4, 3), matching + // engines such as Snowflake. + std::cmp::max(digits, scale.unsigned_abs() + 1) + } else if scale > 0 { // arrow-rs requires the precision to include the positive scale. // See std::cmp::max(digits, scale.unsigned_abs()) @@ -509,28 +527,48 @@ mod tests { ), ]; for (input, expect) in cases { - let output = parse_decimal(input, true).unwrap(); + let output = parse_decimal(input, true, false).unwrap(); assert_eq!( output, Expr::Literal(expect.arithmetic_negate().unwrap(), None) ); - let output = parse_decimal(input, false).unwrap(); + let output = parse_decimal(input, false, false).unwrap(); assert_eq!(output, Expr::Literal(expect, None)); } // scale < i8::MIN assert_eq!( - parse_decimal("1e129", false).unwrap_err().strip_backtrace(), + parse_decimal("1e129", false, false) + .unwrap_err() + .strip_backtrace(), "This feature is not implemented: Decimal scale -129 exceeds the minimum supported scale: -128" ); // Unsupported precision assert_eq!( - parse_decimal(&"1".repeat(77), false) + parse_decimal(&"1".repeat(77), false, false) .unwrap_err() .strip_backtrace(), "This feature is not implemented: Decimal precision 77 exceeds the maximum supported precision: 76" ); } + + #[test] + fn test_parse_decimal_trims_insignificant_trailing_zeros() { + let cases = [ + ("10.00", ScalarValue::Decimal128(Some(10), 2, 0)), + ("10.10", ScalarValue::Decimal128(Some(101), 3, 1)), + ("0.00100", ScalarValue::Decimal128(Some(1), 4, 3)), + ("100.0001", ScalarValue::Decimal128(Some(1_000_001), 7, 4)), + ("1.2300e2", ScalarValue::Decimal128(Some(123), 3, 0)), + ]; + + for (input, expected) in cases { + assert_eq!( + parse_decimal(input, false, true).unwrap(), + Expr::Literal(expected, None) + ); + } + } } diff --git a/datafusion/sql/src/planner.rs b/datafusion/sql/src/planner.rs index 89af194e1a4aa..a4f42653dd352 100644 --- a/datafusion/sql/src/planner.rs +++ b/datafusion/sql/src/planner.rs @@ -45,6 +45,8 @@ use sqlparser::ast::{DataType as SQLDataType, Ident, ObjectName, TableAlias}; pub struct ParserOptions { /// Whether to parse float as decimal. pub parse_float_as_decimal: bool, + /// Whether to remove insignificant trailing zeros from decimal literals. + pub trim_decimal_literal_trailing_zeros: bool, /// Whether to normalize identifiers. pub enable_ident_normalization: bool, /// Whether to support varchar with length. @@ -73,6 +75,7 @@ impl ParserOptions { pub fn new() -> Self { Self { parse_float_as_decimal: false, + trim_decimal_literal_trailing_zeros: false, enable_ident_normalization: true, support_varchar_with_length: true, map_string_types_to_utf8view: true, @@ -98,6 +101,12 @@ impl ParserOptions { self } + /// Sets the `trim_decimal_literal_trailing_zeros` option. + pub fn with_trim_decimal_literal_trailing_zeros(mut self, value: bool) -> Self { + self.trim_decimal_literal_trailing_zeros = value; + self + } + /// Sets the `enable_ident_normalization` option. /// /// # Examples @@ -153,6 +162,8 @@ impl From<&SqlParserOptions> for ParserOptions { fn from(options: &SqlParserOptions) -> Self { Self { parse_float_as_decimal: options.parse_float_as_decimal, + trim_decimal_literal_trailing_zeros: options + .trim_decimal_literal_trailing_zeros, enable_ident_normalization: options.enable_ident_normalization, support_varchar_with_length: options.support_varchar_with_length, map_string_types_to_utf8view: options.map_string_types_to_utf8view, diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index 7855e755e1a7d..d980a4de7eb21 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -194,6 +194,21 @@ fn parse_decimals_9() { ); } +#[test] +fn parse_decimals_trim_insignificant_trailing_zeros() { + let sql = "SELECT 10.00, 10.10, 0.00100, 100.0001, 1.2300e2"; + let options = + parse_decimals_parser_options().with_trim_decimal_literal_trailing_zeros(true); + let plan = logical_plan_with_options(sql, options).unwrap(); + assert_snapshot!( + plan, + @r" + Projection: Decimal128(10,2,0), Decimal128(10.1,3,1), Decimal128(0.001,4,3), Decimal128(100.0001,7,4), Decimal128(123,3,0) + EmptyRelation: rows=1 + " + ); +} + #[test] fn parse_ident_normalization_1() { let sql = "SELECT CHARACTER_LENGTH('str')"; @@ -4066,6 +4081,7 @@ impl ScalarUDFImpl for DummyUDF { fn parse_decimals_parser_options() -> ParserOptions { ParserOptions { parse_float_as_decimal: true, + trim_decimal_literal_trailing_zeros: false, enable_ident_normalization: false, support_varchar_with_length: false, map_string_types_to_utf8view: true, @@ -4078,6 +4094,7 @@ fn parse_decimals_parser_options() -> ParserOptions { fn ident_normalization_parser_options_no_ident_normalization() -> ParserOptions { ParserOptions { parse_float_as_decimal: true, + trim_decimal_literal_trailing_zeros: false, enable_ident_normalization: false, support_varchar_with_length: false, map_string_types_to_utf8view: true, @@ -4090,6 +4107,7 @@ fn ident_normalization_parser_options_no_ident_normalization() -> ParserOptions fn ident_normalization_parser_options_ident_normalization() -> ParserOptions { ParserOptions { parse_float_as_decimal: true, + trim_decimal_literal_trailing_zeros: false, enable_ident_normalization: true, support_varchar_with_length: false, map_string_types_to_utf8view: true, diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md index 088c94308699d..aeecf2772b869 100644 --- a/docs/source/user-guide/configs.md +++ b/docs/source/user-guide/configs.md @@ -197,6 +197,7 @@ The following configuration settings are available: | datafusion.explain.analyze_level | dev | Verbosity level for "EXPLAIN ANALYZE". Default is "dev" "summary" shows common metrics for high-level insights. "dev" provides deep operator-level introspection for developers. | | datafusion.explain.analyze_categories | all | Which metric categories to include in "EXPLAIN ANALYZE" output. Comma-separated list of: "rows", "bytes", "timing", "uncategorized". Use "none" to show plan structure only, or "all" (default) to show everything. Metrics without a declared category are treated as "uncategorized". | | datafusion.sql_parser.parse_float_as_decimal | false | When set to true, SQL parser will parse float as decimal type | +| datafusion.sql_parser.trim_decimal_literal_trailing_zeros | false | When set to true, insignificant trailing zeros are removed from decimal literals. For example, `10.00` is planned as `DECIMAL(2, 0)` instead of `DECIMAL(4, 2)`. | | datafusion.sql_parser.enable_ident_normalization | true | When set to true, SQL parser will normalize ident (convert ident to lowercase when not quoted) | | datafusion.sql_parser.enable_options_value_normalization | false | When set to true, SQL parser will normalize options value (convert value to lowercase). Note that this option is ignored and will be removed in the future. All case-insensitive values are normalized automatically. | | datafusion.sql_parser.dialect | generic | Configure the SQL dialect used by DataFusion's parser; supported values include: Generic, MySQL, PostgreSQL, Hive, SQLite, Snowflake, Redshift, MsSQL, ClickHouse, BigQuery, Ansi, DuckDB, Databricks, Spark. | From 57636801afb3da72e698b3b0bdeea090f12f334d Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 02:22:46 +0300 Subject: [PATCH 06/15] Format SQLParser workspace dependency --- Cargo.toml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index 13e26dafdc4ee..39735db249480 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -196,7 +196,10 @@ regex = "1.12" rstest = "0.26.1" serde_json = "1" sha2 = "^0.11.0" -sqlparser = { git = "https://github.com/Embucket/datafusion-sqlparser-rs.git", rev = "1a3f48f60802d0f135a968a27c89a3351e78b159", default-features = false, features = ["std", "visitor"] } +sqlparser = { git = "https://github.com/Embucket/datafusion-sqlparser-rs.git", rev = "1a3f48f60802d0f135a968a27c89a3351e78b159", default-features = false, features = [ + "std", + "visitor", +] } stacker = "0.1.24" strum = "0.28.0" strum_macros = "0.28.0" From 09477f0a7f0d4dbbc4ad137099e6f274fa9697ab Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 02:45:27 +0300 Subject: [PATCH 07/15] Update h2 to address RUSTSEC-2026-0258 --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 62c04d332b98e..f045e7697f576 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3284,9 +3284,9 @@ dependencies = [ [[package]] name = "h2" -version = "0.4.13" +version = "0.4.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f44da3a8150a6703ed5d34e164b875fd14c2cdab9af1252a9a1020bde2bdc54" +checksum = "a9f37a958b41b3b19ee2707c06439c0e9e547e847223eb791ecb0cb821c65e27" dependencies = [ "atomic-waker", "bytes", From 55182c3752ec8664bdbdaaf1ebaed68588481af8 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 04:18:18 +0300 Subject: [PATCH 08/15] Update settings SQL logic test expectations --- datafusion/sqllogictest/test_files/information_schema.slt | 2 ++ 1 file changed, 2 insertions(+) diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt index 00e48d04ebba3..399f81b907807 100644 --- a/datafusion/sqllogictest/test_files/information_schema.slt +++ b/datafusion/sqllogictest/test_files/information_schema.slt @@ -365,6 +365,7 @@ datafusion.sql_parser.map_string_types_to_utf8view true datafusion.sql_parser.parse_float_as_decimal false datafusion.sql_parser.recursion_limit 50 datafusion.sql_parser.support_varchar_with_length true +datafusion.sql_parser.trim_decimal_literal_trailing_zeros false # show all variables with verbose query TTT rowsort @@ -529,6 +530,7 @@ datafusion.sql_parser.map_string_types_to_utf8view true If true, string types (V datafusion.sql_parser.parse_float_as_decimal false When set to true, SQL parser will parse float as decimal type datafusion.sql_parser.recursion_limit 50 Specifies the recursion depth limit when parsing complex SQL Queries datafusion.sql_parser.support_varchar_with_length true If true, permit lengths for `VARCHAR` such as `VARCHAR(20)`, but ignore the length. If false, error if a `VARCHAR` with a length is specified. The Arrow type system does not have a notion of maximum string length and thus DataFusion can not enforce such limits. +datafusion.sql_parser.trim_decimal_literal_trailing_zeros false When set to true, insignificant trailing zeros are removed from decimal literals. For example, `10.00` is planned as `DECIMAL(2, 0)` instead of `DECIMAL(4, 2)`. # show_variable_in_config_options query TT From 30afc457c8c02890923a24b19a17f917dd7b4eb3 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 06:13:21 +0300 Subject: [PATCH 09/15] Fix wildcard keys for USING outer joins --- datafusion/core/tests/sql/joins.rs | 27 ++++++++++ datafusion/expr/src/utils.rs | 72 ++++++++++++++++++++++++- datafusion/sql/tests/sql_integration.rs | 24 +++++++++ 3 files changed, 121 insertions(+), 2 deletions(-) diff --git a/datafusion/core/tests/sql/joins.rs b/datafusion/core/tests/sql/joins.rs index 7c0e89ee96418..f1ffd6c5b12d0 100644 --- a/datafusion/core/tests/sql/joins.rs +++ b/datafusion/core/tests/sql/joins.rs @@ -25,6 +25,33 @@ use datafusion_sql::unparser::plan_to_sql; use super::*; +#[tokio::test] +async fn natural_full_join_wildcard_coalesces_join_key() -> Result<()> { + let ctx = SessionContext::new(); + let dataframe = ctx + .sql( + "WITH d1(id, name) AS (VALUES (1, 'a'), (2, 'b'), (4, 'c')), + d2(id, value) AS (VALUES (1, 'xx'), (2, 'yy'), (5, 'zz')) + SELECT * FROM d1 NATURAL FULL OUTER JOIN d2 ORDER BY id", + ) + .await?; + + assert_batches_eq!( + [ + "+----+------+-------+", + "| id | name | value |", + "+----+------+-------+", + "| 1 | a | xx |", + "| 2 | b | yy |", + "| 4 | c | |", + "| 5 | | zz |", + "+----+------+-------+", + ], + &dataframe.collect().await? + ); + Ok(()) +} + #[tokio::test] async fn join_change_in_planner() -> Result<()> { let config = SessionConfig::new().with_target_partitions(8); diff --git a/datafusion/expr/src/utils.rs b/datafusion/expr/src/utils.rs index 297fbce14c795..84185ffecff72 100644 --- a/datafusion/expr/src/utils.rs +++ b/datafusion/expr/src/utils.rs @@ -24,7 +24,8 @@ use std::sync::Arc; use crate::expr::{Alias, Sort, WildcardOptions, WindowFunctionParams}; use crate::expr_rewriter::strip_outer_reference; use crate::{ - BinaryExpr, Expr, ExprSchemable, Filter, GroupingSet, LogicalPlan, Operator, and, + BinaryExpr, Expr, ExprSchemable, Filter, GroupingSet, JoinConstraint, JoinType, + LogicalPlan, Operator, and, when, }; use datafusion_expr_common::signature::{Signature, TypeSignature}; @@ -442,6 +443,63 @@ fn exclude_using_columns(plan: &LogicalPlan) -> Result> { Ok(excluded) } +/// Adjusts an unqualified wildcard over a top-level `USING` join so the +/// retained join key has the value required by SQL outer-join semantics. +fn using_join_wildcard_replacements( + plan: &LogicalPlan, + columns_to_skip: &mut HashSet, +) -> Result> { + let LogicalPlan::Join(join) = plan else { + return Ok(HashMap::new()); + }; + if join.join_constraint != JoinConstraint::Using { + return Ok(HashMap::new()); + } + if !matches!( + join.join_type, + JoinType::Inner | JoinType::Left | JoinType::Right | JoinType::Full + ) { + return Ok(HashMap::new()); + } + + let mut replacements = HashMap::new(); + for (left_expr, right_expr) in &join.on { + let Some(left) = left_expr.get_as_join_column() else { + return internal_err!( + "Invalid USING join key. Expected column, found {left_expr:?}" + ); + }; + let Some(right) = right_expr.get_as_join_column() else { + return internal_err!( + "Invalid USING join key. Expected column, found {right_expr:?}" + ); + }; + + // Keep the left key in its original schema position and remove the + // duplicate right key. RIGHT and FULL joins replace its value below. + columns_to_skip.remove(left); + columns_to_skip.insert(right.clone()); + + let left_column = Expr::Column(left.clone()); + let right_column = Expr::Column(right.clone()); + let replacement = match join.join_type { + JoinType::Right => Some(right_column), + JoinType::Full => Some( + when(left_column.clone().is_not_null(), left_column) + .otherwise(right_column)?, + ), + _ => None, + }; + if let Some(replacement) = replacement { + replacements.insert( + left.clone(), + replacement.alias_qualified(left.relation.clone(), left.name.clone()), + ); + } + } + Ok(replacements) +} + /// Resolves an `Expr::Wildcard` to a collection of `Expr::Column`'s. pub fn expand_wildcard( schema: &DFSchema, @@ -449,12 +507,22 @@ pub fn expand_wildcard( wildcard_options: Option<&WildcardOptions>, ) -> Result> { let mut columns_to_skip = exclude_using_columns(plan)?; + let replacements = using_join_wildcard_replacements(plan, &mut columns_to_skip)?; columns_to_skip.extend(excluded_columns_from_schema( schema, wildcard_options, None, )?); - Ok(get_exprs_except_skipped(schema, &columns_to_skip)) + Ok(get_exprs_except_skipped(schema, &columns_to_skip) + .into_iter() + .map(|expr| match expr { + Expr::Column(column) => replacements + .get(&column) + .cloned() + .unwrap_or(Expr::Column(column)), + expr => expr, + }) + .collect()) } /// Resolves an unqualified wildcard using only the input schema. diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index d980a4de7eb21..8851b3212a6c3 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -5622,6 +5622,30 @@ fn test_using_join_wildcard_schema() { ] ); + // RIGHT and FULL joins must retain values from the non-null side of the + // merged USING column while preserving the wildcard schema. + let sql = "WITH t1 AS (SELECT 1 AS id, 'a' AS value1), + t2 AS (SELECT 2 AS id, 'x' AS value2) + SELECT * FROM t1 RIGHT JOIN t2 USING (id)"; + let plan = logical_plan(sql).unwrap(); + assert!( + plan.display_indent() + .to_string() + .contains("Projection: t2.id AS id, t1.value1, t2.value2"), + "{plan}" + ); + + let sql = "WITH t1 AS (SELECT 1 AS id, 'a' AS value1), + t2 AS (SELECT 2 AS id, 'x' AS value2) + SELECT * FROM t1 FULL OUTER JOIN t2 USING (id)"; + let plan = logical_plan(sql).unwrap(); + assert!( + plan.display_indent().to_string().contains( + "Projection: CASE WHEN t1.id IS NOT NULL THEN t1.id ELSE t2.id END AS id, t1.value1, t2.value2" + ), + "{plan}" + ); + // Multiple joins let sql = "WITH t1 AS (SELECT 1 AS a, 1 AS b), t2 AS (SELECT 1 AS a, 2 AS c), From 904df42381ed083e66626fd939d600d397640e84 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 08:01:02 +0300 Subject: [PATCH 10/15] Avoid target schema propagation into INSERT SELECT --- datafusion/sql/src/statement.rs | 9 ++++++--- datafusion/sql/tests/sql_integration.rs | 9 +++++++++ 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs index 93a8aaf186bc7..c25756a3fcff9 100644 --- a/datafusion/sql/src/statement.rs +++ b/datafusion/sql/src/statement.rs @@ -2828,6 +2828,7 @@ impl SqlToRel<'_, S> { let table_name = self.object_name_to_table_reference(table_name)?; let table_source = self.context_provider.get_table_source(table_name.clone())?; let table_schema = DFSchema::try_from(table_source.schema())?; + let source_is_values = matches!(source.body.as_ref(), SetExpr::Values(_)); let columns: Vec = columns .into_iter() @@ -2927,9 +2928,11 @@ impl SqlToRel<'_, S> { // Projection let mut planner_context = PlannerContext::new().with_prepare_param_data_types(prepare_param_data_types); - planner_context.set_table_schema(Some(DFSchemaRef::new( - DFSchema::from_unqualified_fields(fields.clone(), Default::default())?, - ))); + if source_is_values { + planner_context.set_table_schema(Some(DFSchemaRef::new( + DFSchema::from_unqualified_fields(fields.clone(), Default::default())?, + ))); + } let source = self.query_to_plan(*source, &mut planner_context)?; if fields.len() != source.schema().fields().len() { plan_err!("Column count doesn't match insert query!")?; diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index 8851b3212a6c3..390e8cc017921 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -737,6 +737,15 @@ fn plan_insert_no_target_columns() { ); } +#[test] +fn plan_insert_select_expression_from_values() { + let sql = "INSERT INTO array (\"left\") \ + SELECT make_array(column1) FROM (VALUES (1), (2))"; + let plan = logical_plan_with_dialect(sql, &GenericDialect {}).unwrap(); + + assert_contains!(plan.display_indent().to_string(), "make_array(column1)"); +} + #[rstest] #[case::duplicate_columns( "INSERT INTO test_decimal (id, price, price) VALUES (1, 2, 3), (4, 5, 6)", From 3048eab1a36e15c768772cf54d9028718319ce7d Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 14:07:53 +0300 Subject: [PATCH 11/15] Update USING join view snapshot --- datafusion/core/src/datasource/view_test.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/datafusion/core/src/datasource/view_test.rs b/datafusion/core/src/datasource/view_test.rs index 35418d6dea632..603209e2944c4 100644 --- a/datafusion/core/src/datasource/view_test.rs +++ b/datafusion/core/src/datasource/view_test.rs @@ -289,10 +289,10 @@ mod tests { insta::assert_snapshot!(batches_to_string(&results),@r" +---------+---------+---------+ - | column2 | column1 | column3 | + | column1 | column2 | column3 | +---------+---------+---------+ - | 2 | 1 | 3 | - | 5 | 4 | 6 | + | 1 | 2 | 3 | + | 4 | 5 | 6 | +---------+---------+---------+ "); From eaf0ce6afdb96fe312c106a8130f595b4f0a8044 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 14:28:41 +0300 Subject: [PATCH 12/15] Update USING join unparser expectation --- datafusion/core/tests/sql/unparser.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/datafusion/core/tests/sql/unparser.rs b/datafusion/core/tests/sql/unparser.rs index 355a58fd6f45b..99d663a0a9a6d 100644 --- a/datafusion/core/tests/sql/unparser.rs +++ b/datafusion/core/tests/sql/unparser.rs @@ -378,7 +378,7 @@ async fn optimized_duckdb_unparse_qualifies_nested_passthrough_column() -> Resul // `o` (which is only the base-table alias one level deeper). The bug emitted // `"o"."order_id"` inside that derived table; the fix emits a bare column. let expected = concat!( - r#"SELECT "o"."order_id", "o"."discount_pct_2" "#, + r#"SELECT "oi"."order_id", "o"."discount_pct_2" "#, r#"FROM "warehouse"."main"."order_items" AS "oi" "#, r#"INNER JOIN (SELECT "order_id", "#, r#"CASE WHEN "__common_expr_1" IS NOT NULL "#, From a95a514546badf08bd5bfbb88fc8d2d6cc4a95b7 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 20:36:38 +0300 Subject: [PATCH 13/15] Do not treat volatile expressions as constants --- .../user_defined_scalar_functions.rs | 18 ++++++++++ .../physical-expr/src/equivalence/class.rs | 36 +++++++++++++++++-- 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs b/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs index b758aeb5209e8..630e43de8690f 100644 --- a/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs +++ b/datafusion/core/tests/user_defined/user_defined_scalar_functions.rs @@ -609,6 +609,24 @@ async fn volatile_scalar_udf_with_params() -> Result<()> { ]; assert_batches_eq!(expected, &result); + let result = plan_and_collect( + &ctx, + "SELECT * FROM (SELECT add_index_to_string('test') AS str FROM t) \ + ORDER BY add_index_to_string('test') LIMIT 2 OFFSET 5", + ) + .await?; + assert_batches_eq!( + [ + "+---------+", + "| str |", + "+---------+", + "| 6) test |", + "| 7) test |", + "+---------+", + ], + &result + ); + let result = plan_and_collect(&ctx, "select add_index_to_string('test_value') as str") // with fixed function parameters .await?; diff --git a/datafusion/physical-expr/src/equivalence/class.rs b/datafusion/physical-expr/src/equivalence/class.rs index 1f9a6a583cc44..8783192384fc2 100644 --- a/datafusion/physical-expr/src/equivalence/class.rs +++ b/datafusion/physical-expr/src/equivalence/class.rs @@ -29,7 +29,9 @@ use crate::{PhysicalExpr, PhysicalExprRef, PhysicalSortExpr, PhysicalSortRequire use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode}; use datafusion_common::{JoinType, Result, ScalarValue}; -use datafusion_physical_expr_common::physical_expr::format_physical_expr_list; +use datafusion_physical_expr_common::physical_expr::{ + format_physical_expr_list, is_volatile, +}; use indexmap::{IndexMap, IndexSet}; @@ -747,6 +749,9 @@ impl EquivalenceGroup { &self, expr: &Arc, ) -> Option { + if is_volatile(expr) { + return None; + } if let Some(lit) = expr.downcast_ref::() { return Some(AcrossPartitions::Uniform(Some(lit.value().clone()))); } @@ -924,9 +929,36 @@ mod tests { use super::*; use crate::equivalence::tests::create_test_params; use crate::expressions::{BinaryExpr, Column, binary, col, lit}; + use crate::scalar_function::ScalarFunctionExpr; + use crate::utils::tests::TestScalarUDF; use arrow::datatypes::{DataType, Field, Schema}; - use datafusion_expr::Operator; + use datafusion_common::config::ConfigOptions; + use datafusion_expr::{Operator, ScalarUDF, Signature, Volatility}; + + #[test] + fn volatile_expression_with_constant_arguments_is_not_constant() -> Result<()> { + let function = Arc::new(ScalarUDF::new_from_impl(TestScalarUDF { + signature: Signature::uniform( + 1, + vec![DataType::Float64], + Volatility::Volatile, + ), + })); + let expression = Arc::new(ScalarFunctionExpr::try_new( + function, + vec![lit(1.0_f64)], + &Schema::empty(), + Arc::new(ConfigOptions::new()), + )?) as Arc; + + assert!( + EquivalenceGroup::default() + .is_expr_constant(&expression) + .is_none() + ); + Ok(()) + } #[test] fn test_bridge_groups() -> Result<()> { From db33ea3d228230c18eed18f3baed32516cb085a9 Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 20:50:58 +0300 Subject: [PATCH 14/15] Preserve rows when sorting zero-column batches --- datafusion/physical-plan/src/sorts/stream.rs | 41 ++++++++++++++++++-- 1 file changed, 38 insertions(+), 3 deletions(-) diff --git a/datafusion/physical-plan/src/sorts/stream.rs b/datafusion/physical-plan/src/sorts/stream.rs index 107631074ed3d..b1a6b4211d45f 100644 --- a/datafusion/physical-plan/src/sorts/stream.rs +++ b/datafusion/physical-plan/src/sorts/stream.rs @@ -21,7 +21,7 @@ use crate::{PhysicalExpr, PhysicalSortExpr}; use arrow::array::{Array, UInt32Array}; use arrow::compute::take_record_batch; use arrow::datatypes::Schema; -use arrow::record_batch::RecordBatch; +use arrow::record_batch::{RecordBatch, RecordBatchOptions}; use arrow::row::{RowConverter, Rows, SortField}; use arrow_ord::sort::lexsort_to_indices; use datafusion_common::{Result, internal_datafusion_err}; @@ -365,9 +365,12 @@ impl Iterator for IncrementalSortIterator { // Perform the take to produce the next batch let new_batch_indices = indices.slice(self.cursor, batch_size); - let new_batch = match take_record_batch(&self.batch, &new_batch_indices) { + let new_batch = match take_record_batch_preserving_row_count( + &self.batch, + &new_batch_indices, + ) { Ok(batch) => batch, - Err(e) => return Some(Err(e.into())), + Err(e) => return Some(Err(e)), }; self.cursor += batch_size; @@ -393,6 +396,22 @@ impl Iterator for IncrementalSortIterator { } } +fn take_record_batch_preserving_row_count( + batch: &RecordBatch, + indices: &UInt32Array, +) -> Result { + if batch.num_columns() == 0 { + let options = RecordBatchOptions::new().with_row_count(Some(indices.len())); + Ok(RecordBatch::try_new_with_options( + batch.schema(), + vec![], + &options, + )?) + } else { + Ok(take_record_batch(batch, indices)?) + } +} + impl FusedIterator for IncrementalSortIterator {} #[cfg(test)] @@ -407,6 +426,22 @@ mod tests { use futures::Stream; use std::pin::Pin; + #[test] + fn take_zero_column_batch_preserves_selected_row_count() -> Result<()> { + let schema = Arc::new(Schema::empty()); + let options = RecordBatchOptions::new().with_row_count(Some(3)); + let batch = RecordBatch::try_new_with_options(schema, vec![], &options)?; + + let taken = take_record_batch_preserving_row_count( + &batch, + &UInt32Array::from(vec![2, 0]), + )?; + + assert_eq!(taken.num_columns(), 0); + assert_eq!(taken.num_rows(), 2); + Ok(()) + } + /// Verifies that `take_record_batch` in `IncrementalSortIterator` actually /// copies the data into a new allocation rather than returning a zero-copy /// slice of the original batch. If the output arrays were slices, their From 7b12ae515309712ac014155ff0872321107c14df Mon Sep 17 00:00:00 2001 From: osipovartem Date: Wed, 2 Sep 2026 21:53:14 +0300 Subject: [PATCH 15/15] Fall back to floats for out-of-range decimal literals --- datafusion/sql/src/expr/value.rs | 24 ++++++++++++----- datafusion/sql/tests/sql_integration.rs | 34 ++++++++++++++++++++++++- 2 files changed, 50 insertions(+), 8 deletions(-) diff --git a/datafusion/sql/src/expr/value.rs b/datafusion/sql/src/expr/value.rs index e1a6965fca590..25c912eddd4fc 100644 --- a/datafusion/sql/src/expr/value.rs +++ b/datafusion/sql/src/expr/value.rs @@ -108,17 +108,21 @@ impl SqlToRel<'_, S> { } if self.options.parse_float_as_decimal { - parse_decimal( + let decimal = parse_decimal( unsigned_number, negative, self.options.trim_decimal_literal_trailing_zeros, - ) + ); + match decimal { + Err(DataFusionError::NotImplemented(_)) + if unsigned_number.contains(['.', 'e', 'E']) => + { + parse_float(&signed_number) + } + result => result, + } } else { - signed_number.parse::().map(lit).map_err(|_| { - DataFusionError::from(ParserError(format!( - "Cannot parse {signed_number} as f64" - ))) - }) + parse_float(&signed_number) } } @@ -379,6 +383,12 @@ fn bigint_to_i256(v: &BigInt) -> Option { } } +fn parse_float(signed_number: &str) -> Result { + signed_number.parse::().map(lit).map_err(|_| { + DataFusionError::from(ParserError(format!("Cannot parse {signed_number} as f64"))) + }) +} + fn parse_decimal( unsigned_number: &str, negative: bool, diff --git a/datafusion/sql/tests/sql_integration.rs b/datafusion/sql/tests/sql_integration.rs index 390e8cc017921..a83d070355e0b 100644 --- a/datafusion/sql/tests/sql_integration.rs +++ b/datafusion/sql/tests/sql_integration.rs @@ -26,7 +26,9 @@ use std::vec; use arrow::datatypes::{TimeUnit::Nanosecond, *}; use common::MockContextProvider; -use datafusion_common::{DFSchema, DataFusionError, Result, assert_contains}; +use datafusion_common::{ + DFSchema, DataFusionError, Result, ScalarValue, assert_contains, +}; use datafusion_expr::{ ColumnarValue, CreateIndex, DdlStatement, Expr, HigherOrderFunctionArgs, HigherOrderReturnFieldArgs, HigherOrderSignature, HigherOrderUDF, HigherOrderUDFImpl, @@ -194,6 +196,36 @@ fn parse_decimals_9() { ); } +#[test] +fn parse_decimal_out_of_range_float_falls_back_to_float64() { + let sql = "SELECT 1.7976931348623157e+308, 2.2250738585072014e-308"; + let options = parse_decimals_parser_options(); + let plan = logical_plan_with_options(sql, options).unwrap(); + let LogicalPlan::Projection(projection) = plan else { + panic!("expected a projection") + }; + let values = projection + .expr + .iter() + .map(|expr| match expr { + Expr::Literal(ScalarValue::Float64(Some(value)), _) => value.to_bits(), + _ => panic!("expected Float64 literals, got {expr}"), + }) + .collect::>(); + assert_eq!(values, [f64::MAX.to_bits(), f64::MIN_POSITIVE.to_bits()]); +} + +#[test] +fn parse_decimal_out_of_range_integer_remains_an_error() { + let sql = format!("SELECT {}", "1".repeat(77)); + let options = parse_decimals_parser_options(); + let error = logical_plan_with_options(&sql, options).unwrap_err(); + assert_contains!( + error.to_string(), + "Decimal precision 77 exceeds the maximum supported precision: 76" + ); +} + #[test] fn parse_decimals_trim_insignificant_trailing_zeros() { let sql = "SELECT 10.00, 10.10, 0.00100, 100.0001, 1.2300e2";