JoeryH opened a new issue, #26313: URL: https://github.com/apache/datafusion/issues/26313
**TL;DR:** `SqlToRel::sql_select_to_rex` calls `LogicalPlan::using_columns()` once for every SELECT item, and each call walks the whole input plan, subqueries and expressions included. Planning a SELECT therefore costs O(items × input plan size), and nesting such SELECTs makes it roughly O(width² × depth²). Computing `using_columns()` once per SELECT gives the same result and makes SQL planning **1.7–1.9× faster** on wide, deeply nested queries. Combined with a name index on `DFSchema` (#ISSUE_B), it is **11.8×** faster: 8.9 s → 0.76 s at 4000 columns × 12 levels. ### Is your feature request related to a problem or challenge? On `main` (2c568b478), `prepare_select_exprs` calls `sql_select_to_rex` once per projection item. For `UnnamedExpr` and `ExprWithAlias`, `sql_select_to_rex` evaluates `&plan.using_columns()?` inline ([select.rs#L1030](https://github.com/apache/datafusion/blob/2c568b478/datafusion/sql/src/select.rs#L1030), [L1041](https://github.com/apache/datafusion/blob/2c568b478/datafusion/sql/src/select.rs#L1041)). [`LogicalPlan::using_columns`](https://github.com/apache/datafusion/blob/2c568b478/datafusion/expr/src/logical_plan/plan.rs#L506) is `apply_with_subqueries` over the entire input tree, so every call: - visits every plan node below the SELECT, - visits every expression of every node, looking for subqueries, - allocates a fresh `Vec<HashSet<Column>>`. Its result depends only on `plan`, which is the same for every item of one SELECT. A SELECT naming W columns over an input whose plan holds about W × D expressions (D nested layers below it) therefore does W × (W × D) work. Planning all D layers is about W² × D². This is separate from #14118, which is about the *correctness* of `using_columns()`. This issue is only about how often it is called. ### Numbers Release build, Apple M1 Max, `main` at 2c568b478 vs the same commit with the call hoisted. `statement_to_plan` only, median of 3 runs; reproduction below. | width | depth | main | hoisted | speedup | |---:|---:|---:|---:|---:| | 1000 | 3 | 119 ms | 98 ms | 1.2× | | 4000 | 3 | 1594 ms | 1308 ms | 1.2× | | 1000 | 12 | 558 ms | 323 ms | 1.7× | | 2000 | 12 | 2010 ms | 1085 ms | 1.9× | | 4000 | 12 | 8046 ms | 4388 ms | 1.8× | The fix pays off more once the linear scans in `DFSchema` lookups (#ISSUE_B) are also gone, because those otherwise dominate. With the name index applied to both sides: | width | depth | name index only | + hoisted | speedup | |---:|---:|---:|---:|---:| | 1000 | 12 | 307 ms | 75 ms | 4.1× | | 2000 | 12 | 1104 ms | 195 ms | 5.7× | | 4000 | 12 | 4345 ms | 622 ms | 7.0× | On a real ~4,500-column view (about 20 SELECT blocks, 1.4 MB of SQL), backported to 54.1.0, cold logical planning went: - 7.8 s on stock 54.1.0 - 7.3 s with #25010 and #25739 backported - 5.8 s with this change added - 2.4 s with the `DFSchema` name index added as well <details><summary>Reproduction (standalone binary, release build)</summary> `Cargo.toml`: ```toml [dependencies] datafusion = { path = "<datafusion checkout>/datafusion/core", default-features = false, features = ["sql"] } tokio = { version = "1", features = ["rt-multi-thread", "macros"] } ``` `src/main.rs`: ```rust //! Times logical planning of a wide, nested SELECT: `depth` subquery layers, each naming all //! `width` columns of the layer below, over a `width`-column table joined to a small one. use std::sync::Arc; use std::time::{Duration, Instant}; use datafusion::arrow::datatypes::{DataType, Field, Schema}; use datafusion::datasource::MemTable; use datafusion::prelude::*; fn query(width: usize, depth: usize) -> String { let cols = |q: &str| { (0..width) .map(|i| format!("{q}c{i} AS c{i}")) .collect::<Vec<_>>() .join(", ") }; let mut sql = format!("SELECT {} FROM t JOIN u ON t.c0 = u.k", cols("t.")); for level in 0..depth { sql = format!("SELECT {} FROM ({sql}) l{level}", cols(&format!("l{level}."))); } sql } #[tokio::main] async fn main() -> datafusion::error::Result<()> { let depth: usize = std::env::var("DEPTH").map(|d| d.parse().unwrap()).unwrap_or(3); let widths: Vec<usize> = std::env::args().skip(1).map(|w| w.parse().unwrap()).collect(); println!("| width | parse | statement_to_plan | analyzer | total |"); println!("|---:|---:|---:|---:|---:|"); for width in widths { let ctx = SessionContext::new(); let fields = (0..width).map(|i| Field::new(format!("c{i}"), DataType::Int64, true)); let t = Arc::new(Schema::new(fields.collect::<Vec<_>>())); ctx.register_table("t", Arc::new(MemTable::try_new(t, vec![vec![]])?))?; let u = Arc::new(Schema::new(vec![Field::new("k", DataType::Int64, true)])); ctx.register_table("u", Arc::new(MemTable::try_new(u, vec![vec![]])?))?; let sql = query(width, depth); let state = ctx.state(); let mut runs = vec![]; for _ in 0..3 { let started = Instant::now(); let statement = state.sql_to_statement(&sql, &datafusion::config::Dialect::Generic)?; let parsed = started.elapsed(); let started = Instant::now(); let plan = state.statement_to_plan(statement).await?; let planned = started.elapsed(); let started = Instant::now(); state .analyzer() .execute_and_check(plan, state.config_options(), |_, _| {})?; let analyzed = started.elapsed(); runs.push((parsed, planned, analyzed)); } runs.sort_by_key(|(p, s, a)| *p + *s + *a); let (p, s, a) = runs[1]; let ms = |d: Duration| format!("{:.0} ms", d.as_secs_f64() * 1000.0); println!("| {width} | {} | {} | {} | {} |", ms(p), ms(s), ms(a), ms(p + s + a)); } Ok(()) } ``` Run with `DEPTH=3 cargo run --release -- 1000 2000 4000`. The query is `depth` nested subqueries. Each one names every column of the layer below as `lN.cI AS cI`, over a `width`-column table joined to a one-column table. Each cell is the median of 3 runs. </details> ### Describe the solution you'd like Compute `plan.using_columns()?` once in `prepare_select_exprs` and pass `&[HashSet<Column>]` into `sql_select_to_rex`. Do the same at the other caller, the `REPLACE` items in `plan_wildcard_options`. The planned output is unchanged. One subtlety: an error from `using_columns()` would surface once for the whole SELECT, instead of once per item through `DataFusionErrorBuilder`. A lazily filled `Option` keeps the per-item behaviour exactly, if that matters. I have a patch (about 15 lines) and am happy to open a PR. ### Describe alternatives you've considered Caching `using_columns` on the plan. That is more invasive, and #14118 suggests the API itself may change. ### Additional context - Similar per-item recomputation was already fixed in `LogicalPlanBuilder::project`: #25010 (normalization context) and #25739 (`Columnizer`). - The umbrella issue for wide-SELECT planning cost is #25248. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
