SQL planner calls LogicalPlan::using_columns() once per SELECT item, making wide nested SELECTs quadratic #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 (#26314), 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 (2c568b4), 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, L1041). LogicalPlan::using_columns 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 2c568b4 vs the same commit with the call hoisted. statement_to_plan only, median of 3 runs; reproduction below.

The fix pays off more once the linear scans in DFSchema lookups (#26314) are also gone, because those otherwise dominate. With the name index applied to both sides:

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 perf: reuse column normalization context across expressions #25010 and perf: avoid quadratic planning for SELECTs with many aggregates #25739 backported
• 5.8 s with this change added
• 2.4 s with the DFSchema name index added as well

[dependencies]
datafusion = { path = "<datafusion checkout>/datafusion/core", default-features = false, features = ["sql"] }
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

//! 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(())
}

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: perf: reuse column normalization context across expressions #25010 (normalization context) and perf: avoid quadratic planning for SELECTs with many aggregates #25739 (Columnizer).
• The umbrella issue for wide-SELECT planning cost is SqlToRel::statement_to_plan is superlinear for a wide Aggregate (exponent ≈ 1.8, DataFusion 54.1.0) #25248.