feat: LogicalPlanningPipeline

[notfilippo]

POC for design validation only — not ready for review or merge. Will open a proper issue when the design is more stable.

Why

DataFusion's planning architecture has two implicit constraints that block a class of real use cases:

1. All rules are synchronous. No async I/O is possible during planning. Systems that need to resolve remote metadata — catalog fetches, schema-on-read stores like Elasticsearch, Delta Lake, or Iceberg — must either block a thread or pre-resolve everything before planning begins. Neither is acceptable at scale.

2. The pipeline is a closed two-slot system. Frameworks and engines built on DataFusion have no way to inject custom planning stages without replacing the entire Analyzer or Optimizer. There is no named hook, no ordering guarantee, no enable/disable knob.

Concrete scenarios that motivate this:

• Schema-on-read catalogs (Elasticsearch, Delta Lake, Iceberg): field schemas are resolved by async metadata fetches. An AsyncAnalyzerRule can do those fetches and rewrite table scans in-place, without blocking or pre-warming a cache.
• Join-driven metadata enrichment: you may want to fetch metadata based on the shape of a join in the plan — that requires seeing the plan first, then doing async I/O, then transforming it. Exactly the async rule model.
• Custom pipeline stages: an engine embedding DataFusion may need a pre-analysis phase that rewrites vendor-specific SQL constructs before TypeCoercion runs, or a post-optimization phase that applies cost-based rewrites using stats fetched asynchronously.

What

1. Add async rule variants

New traits alongside the existing sync ones:

// analysis
pub trait AsyncAnalyzerRule { async fn analyze(...) }

// optimization
pub trait AsyncOptimizerRule { async fn rewrite(...) }

2. Unify everything into an extensible pipeline

Replaces the two hardcoded slots with a single LogicalPlanningPipeline — an ordered Vec<Phase> where each phase is named, can be enabled/disabled, carries a Strategy (Once or FixedPoint), and holds rules of one type:

Phase variant	Rule type
SyncAnalysis	AnalyzerRule
SyncOptimization	OptimizerRule
AsyncAnalysis	AsyncAnalyzerRule
AsyncOptimization	AsyncOptimizerRule

The design is directly inspired by Spark SQL's Batch(name, strategy, rules*) and Apache Calcite's HepPlanner batches: named, ordered phases with explicit iteration strategy, strict analysis/optimization separation.

The default pipeline has two phases that replicate existing behavior exactly:

"analysis"      SyncAnalysis     FixedPoint  ← Analyzer::new().rules
"optimization"  SyncOptimization FixedPoint  ← Optimizer::new().rules

SessionState replaces analyzer: Analyzer and optimizer: Optimizer with logical_pipeline: LogicalPlanningPipeline. Users can inject custom phases (including async ones) at any position:

let mut phase = AsyncAnalysisPhase::new("my-async-phase", Strategy::Once);
phase.rules.push(Arc::new(MyAsyncRule));

let state = SessionStateBuilder::new()
    .with_phase_before(DEFAULT_ANALYSIS_PHASE, Phase::AsyncAnalysis(phase))
    .build();

The sync optimize() path (DataFrame API, PREPARE) fails fast with a clear error if an async phase is registered. The async create_physical_plan() path handles all four variants, including EXPLAIN output that captures per-rule intermediate plans across both sync and async phases.

Known gaps

• SessionState::analyzer() / optimizer() now return owned values instead of references.