//! Logical optimizer rule that rewrites eligible
//! `lines + parse_json + EXPLODE+` plans onto the `json` scan.
//!
//! **The rule fires only when there is at least one `EXPLODE` in the
//! chain.** No-EXPLODE shapes (`SELECT a = parse_json(line).foo`) are
//! deliberately *not* fused — see "Why EXPLODE is required" below.
//!
//! ## How the rewrite works (bottom-up layered absorption)
//!
//! The rule walks the `LogicalPlan` post-order via `transform_up`. At each
//! visit it tries to *absorb* the current node into a fused json scan that
//! an earlier visit produced in the child position, or — in the case of an
//! Unnest sitting directly above the seed-shape Projection — *seed and
//! absorb in one step*. The absorption layers are:
//!
//! 1. **Projection over fused scan.** `Projection` with a carrier
//!    expression `from_variant?(variant_get*(<col>))` where `<col>`
//!    matches the fused scan's alias (or the carrier expression is
//!    just a plain `Column` rename). Extends the existing path with
//!    `Field` hops and re-emits with the new alias. Sibling
//!    expressions in the same projection are preserved verbatim if
//!    they are safely re-emitable (literals, immutable/stable UDFs,
//!    structures of those — see "Canonical fused-scan form" below);
//!    otherwise fusion refuses at this layer.
//!
//! 2. **Unnest absorption.** `Unnest` with a single `exec_columns`
//!    entry. Two cases:
//!    - Child is already a canonical fused scan (chained EXPLODE):
//!      append `Segment::ArrayElements`.
//!    - Child is a seed-shape Projection over either
//!      `TableScan(LinesFormat)` (innermost EXPLODE — fresh seed) or
//!      `TableScan(JsonFormat)` (a fused scan reached through enough
//!      view-projection layers that `match_fused_scan` couldn't pattern-
//!      match it directly): seed the state from it and append
//!      `Segment::ArrayElements`.
//!
//!    This is the only layer that introduces a fused scan — by design,
//!    so fusion never happens without an Unnest in the chain.
//!
//! Anything else passes through, and the climb stops naturally — whatever
//! sits above the topmost matched layer is left intact.
//!
//! ## Why EXPLODE is required
//!
//! Hamelin's standard view contract sets `event.original = line` on every
//! ingest connector — downstream queries can re-read the raw line as a
//! `Utf8View` for forensic purposes (regex, storage, round-tripping).
//!
//! - **No EXPLODE.** The lines reader's `Utf8View` output is the product:
//!   downstream code reads `event.original` as a string. Pushing
//!   `parse_json(line)` down to the json scan would force that downstream
//!   code to round-trip back through `to_json_string(event.original)` —
//!   costing more than what's saved by skipping the `Utf8View`
//!   intermediate.
//!
//! - **With EXPLODE.** The exploded element is inherently variant-shaped;
//!   there is no "raw string" to preserve for that row. The lines path
//!   here is doing wasted work — materializing a string column only for
//!   `parse_json` to immediately re-parse it into a Variant. The fused
//!   json scan does both in one pass with no string materialized. Pure
//!   win.
//!
//! The asymmetry tracks the actual semantic shape of the workload, not a
//! syntactic pattern that incidentally also matches.
//!
//! ### Canonical fused-scan form
//!
//! Throughout the climb every fused subtree has the same shape:
//!
//! ```text
//! Projection[ Column("value") AS <alias>, <safe-passthrough siblings>... ]
//! └─ TableScan(ListingTable(JsonFormat(segments, …)))
//! ```
//!
//! The first expression is always the carrier (`value AS <alias>`,
//! optionally wrapped in `from_variant(...)` to recover the user-visible
//! scalar type when a path-extending absorption captured one). Sibling
//! expressions are non-carrier projection columns the input projection
//! emitted alongside the carrier; we re-emit them verbatim when their
//! shape is safe.
//!
//! There are two sibling-shape validators — they diverge because the
//! kinds of inputs they can be fed differ:
//!
//! - **Chained-projection siblings** (Layer 1, [`is_safe_passthrough`]).
//!   These are siblings inside a Projection sitting on top of an
//!   already-fused scan. The fused scan emits `value` aliased to the
//!   prior carrier alias, so siblings can reference that alias freely.
//!   The validator allows: literals, `Column(<carrier_alias>)`,
//!   `ScalarFunction` with `Immutable`/`Stable` volatility over safe
//!   args, `Cast`, and `BinaryExpr` over safe subexpressions.
//!
//! - **Seed siblings** (Layer 2 case 2, [`is_safe_passthrough_seed`]).
//!   These are siblings on the seed-shape Projection that feeds the
//!   innermost Unnest. The fused scan we're about to introduce
//!   replaces the lines/json scan, so any column reference in a
//!   sibling is reading something that's about to disappear. The
//!   validator is correspondingly stricter: it only allows literals,
//!   `Column(<carrier_alias>)`, and `named_struct(...)` calls whose
//!   values recursively meet the same predicate or are
//!   `Column(<scan_column>)` (substituted to typed NULL by the
//!   rewriter — see "Scan-column substitution"). Top-level
//!   `Column(<scan_column>)` and arbitrary UDF calls / casts /
//!   binary ops are refused at this layer; running after
//!   `OptimizeProjections` means constant-folded scalars (`now()` →
//!   timestamp literal, `today()` → date literal, etc.) already
//!   arrive as `Expr::Literal`.
//!
//! Anything else (column references that aren't recognised, volatile
//! UDFs like `random`, unrecognized expression shapes) causes the rule
//! to **refuse fusion at that layer**. The plan is left intact above
//! the deepest layer the rule did succeed at.
//!
//! Volatility is the gate because the fused json scan produces the
//! same number of rows as the lines+unnest pipeline it replaced (one
//! per JSON array element), so re-emitting a `Stable` or `Immutable`
//! expression above the fused scan yields the same per-row value as
//! it would have in the original plan. `Volatile` expressions
//! (`random`, `uuid`) cannot be re-emitted safely: each exploded
//! child of a source row would get a fresh value instead of sharing
//! the source row's value, changing observable semantics.
//!
//! ### Scan-column substitution
//!
//! `Column(scan_col)` is allowed *only inside a `named_struct`* slot
//! at the seed layer, never as a top-level sibling expression. The
//! rewriter substitutes such struct-slot references with
//! `Cast(NULL AS <scan_col_type>)`. This is sound for two reasons:
//!
//! - If the user reads the slot through a compound-EXPLODE rebuild
//!   (`event.original`, where `event` was a struct sibling and the
//!   carrier writes back into `original`), the rebuild overwrites the
//!   NULL with the exploded value before any reader sees it.
//! - Otherwise the slot is read indirectly through some struct-field
//!   access. If the user actually consumes that field downstream,
//!   `OptimizeProjections` propagates the requirement back down and
//!   would have pulled the field-extraction up into the seed
//!   projection's siblings as a top-level expression — at which point
//!   the validator would see a top-level `Column(scan_col)` (or a
//!   field extraction reading the scan column directly) and refuse
//!   fusion.
//!
//! Top-level `Column(scan_col)` is *rejected* by the validator: a
//! sibling whose value IS the bare scan column means the user wants
//! to read the scan column's actual value, and the fused json scan
//! has dropped that column. Fusing here would silently NULL out a
//! read the user expected to succeed (forensic queries like
//! `SELECT raw_line = line, name = ...` after an EXPLODE).
//!
//! ### Why the optimizer stage (after `OptimizeProjections`)
//!
//! The rule runs as an `OptimizerRule` appended *after*
//! `OptimizeProjections`. This is load-bearing: a lowered EXPLODE
//! emits a seed projection that includes every upstream column as a
//! passthrough (`line = line`, `event = event`, …) so downstream
//! commands can still reference them. Most of those passthroughs are
//! immediately dropped by the user's outer SELECT/DROP. If the rule
//! ran early (analyzer stage), it would see the pre-pruning seed
//! projection — every `line = line` passthrough referencing the scan
//! column the fused json scan is about to delete forces fusion to
//! refuse. Letting `OptimizeProjections` propagate "actually-needed
//! columns" down through the seed projection first eliminates those
//! phantoms; the rule then sees only live siblings and can fuse.
//!
//! The seed-shape Projection itself is *not* eliminated by
//! `OptimizeProjections` — its carrier expression
//! (`from_variant(variant_get(json_to_variant(line), …))`) is non-trivial,
//! so [`is_projection_unnecessary`] keeps it intact.
//!
//! ### View pushdown
//!
//! The seed layer reaches the underlying `lines` scan through view-style
//! aliases — `event.original = line`, `event = { original: line, ... }`,
//! or a plain rename `original = line` — via
//! `resolve_innermost_through_views`.

use std::sync::Arc;

use datafusion::common::tree_node::{Transformed, TreeNode};
use datafusion::common::{Column, DFSchema, Result as DfResult, TableReference};
use datafusion::datasource::file_format::file_compression_type::FileCompressionType;
use datafusion::datasource::listing::{
    ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl,
};
use datafusion::datasource::{provider_as_source, source_as_provider};
use datafusion::logical_expr::{
    Expr, LogicalPlan, Projection, ScalarUDF, TableScan, Unnest, Volatility,
};
use datafusion_optimizer::{OptimizerConfig, OptimizerRule};

use super::path::Segment;
use super::reader::{json_schema, JsonFormat};
use crate::lines::LinesFormat;

/// DataFusion `OptimizerRule` that rewrites eligible
/// `lines + parse_json[ + EXPLODE]*` stacks onto a `json` scan.
/// See the module-level docstring for the layered absorption model.
#[derive(Debug, Default)]
pub struct JsonRewrite;

impl OptimizerRule for JsonRewrite {
    fn name(&self) -> &str {
        "json_rewrite"
    }

    fn rewrite(
        &self,
        plan: LogicalPlan,
        _config: &dyn OptimizerConfig,
    ) -> DfResult<Transformed<LogicalPlan>> {
        plan.transform_up(|p| match try_rewrite(&p)? {
            Some(new_plan) => Ok(Transformed::yes(new_plan)),
            None => Ok(Transformed::no(p)),
        })
    }
}

/// Try each absorption layer in order, returning the first match.
///
/// **Why no standalone seed layer.** Fusion only fires when there is at
/// least one `EXPLODE` in the chain. The lines reader's `Utf8View` output
/// is load-bearing for the standard `event.original = line` view contract:
/// downstream queries can re-read the raw line as a string for forensic
/// purposes (regex, storage as a string field, round-tripping). Pushing
/// down a no-EXPLODE `parse_json(line)` to the json scan would force any
/// such downstream reader to round-trip back through `to_json_string` to
/// recover the string — costing more than what's saved by skipping the
/// `Utf8View` intermediate.
///
/// With an EXPLODE in the path the calculus inverts: the exploded element
/// is inherently variant-shaped, so the lines path is just doing wasted
/// string materialization that the json scan eliminates. Pure win.
///
/// Therefore the seed-from-lines step lives *inside* `try_absorb_unnest`
/// rather than as its own layer — it only ever fires as part of an
/// Unnest absorption.
fn try_rewrite(plan: &LogicalPlan) -> DfResult<Option<LogicalPlan>> {
    if let Some(rewritten) = try_absorb_projection(plan)? {
        return Ok(Some(rewritten));
    }
    try_absorb_unnest(plan)
}

/// State extracted from a canonical fused-scan subtree: enough to extend
/// its path and re-emit a new canonical form.
struct FusedScan {
    /// Path segments built so far.
    segments: Vec<Segment>,
    /// Display path (e.g. `$.foo[*].bar`) built so far.
    path_display: String,
    /// File compression carried through from the original `lines` scan.
    compression: FileCompressionType,
    /// Table paths cloned off the original `ListingTable`. Re-used when
    /// re-emitting the scan with a `JsonFormat` instead of the original
    /// format.
    table_paths: Vec<ListingTableUrl>,
    /// File-extension hint cloned off the original `ListingTable`,
    /// re-used on the re-emitted scan.
    file_extension: String,
    /// Original TableScan's table name, preserved on the re-emitted scan.
    table_name: TableReference,
    /// The alias the renaming Projection assigns to `value`. Equals the
    /// column name a parent layer expects to consume.
    alias: String,
    /// Non-carrier sibling expressions from the input projection,
    /// validated by `is_safe_passthrough` and rewritten by
    /// `rename_carrier_refs` so any reference to the prior carrier
    /// alias points at the new TableScan's `value` input column —
    /// passthroughs are re-emitted verbatim *alongside* the carrier
    /// in `build_fused_scan`'s output Projection, so they reference
    /// input columns of the new TableScan rather than the carrier's
    /// output alias.
    passthroughs: Vec<PassthroughColumn>,
    /// Optional outer `hamelin_from_variant(...)` wrapper that the
    /// absorbed carrier expression had. Re-applied to the output
    /// projection's carrier in `build_fused_scan` so the rewritten
    /// plan keeps the same column data type as the original — the
    /// fused scan emits raw `Variant`, and `from_variant` is what
    /// unwraps it to the user-visible scalar type. Only set by
    /// `try_absorb_projection` when collapsing a path-extending
    /// `from_variant(variant_get*(...))` carrier; everywhere else
    /// the wrapper is dropped (the parent Unnest consumes Variant
    /// directly).
    carrier_wrapper: Option<Arc<ScalarUDF>>,
}

/// A non-carrier expression in an absorbed Projection that the rule
/// re-emits verbatim above the fused scan. The expression has been
/// validated by `is_safe_passthrough` (no scan-column references, no
/// volatile UDFs) and rewritten by `rename_carrier_refs` so any
/// reference to the prior carrier alias now reads from the new
/// TableScan's `value` input column.
struct PassthroughColumn {
    name: String,
    expr: Expr,
}

/// Recognize the canonical fused-scan form produced by an earlier visit:
///
/// ```text
/// Projection[ Column("value") AS <alias> (+ safe-passthrough siblings) ]
/// └─ TableScan(ListingTable(JsonFormat(...)))
/// ```
///
/// Returns `Ok(None)` when `plan` is not a fused-scan shape; returns
/// `Err` only if expression rewriting genuinely fails.
fn match_fused_scan(plan: &LogicalPlan) -> DfResult<Option<FusedScan>> {
    let LogicalPlan::Projection(proj) = plan else {
        return Ok(None);
    };
    let candidate = find_projection_entry(
        proj,
        |_name, body| matches!(body, Expr::Column(col) if col.name == "value" && col.relation.is_none()),
    );
    let Some((carrier_idx, alias, _)) = candidate else {
        return Ok(None);
    };
    let LogicalPlan::TableScan(ts) = proj.input.as_ref() else {
        return Ok(None);
    };
    let Ok(provider) = source_as_provider(&ts.source) else {
        return Ok(None);
    };
    let Some(listing) = provider.as_any().downcast_ref::<ListingTable>() else {
        return Ok(None);
    };
    let Some(je_format) = listing
        .options()
        .format
        .as_any()
        .downcast_ref::<JsonFormat>()
    else {
        return Ok(None);
    };
    let Some(passthroughs) = passthrough_columns_from_projection(proj, carrier_idx, &alias)? else {
        return Ok(None);
    };
    Ok(Some(FusedScan {
        segments: je_format.segments().to_vec(),
        path_display: je_format.path_display().to_string(),
        compression: je_format.compression(),
        table_paths: listing.table_paths().clone(),
        file_extension: listing.options().file_extension.clone(),
        table_name: ts.table_name.clone(),
        alias,
        passthroughs,
        carrier_wrapper: None,
    }))
}

/// Build the canonical fused-scan form from a (possibly extended) state.
fn build_fused_scan(state: FusedScan) -> DfResult<LogicalPlan> {
    let je_format = JsonFormat::new(state.segments, state.compression, None, state.path_display);
    let new_scan = build_json_scan(
        state.table_paths,
        state.file_extension,
        je_format,
        state.table_name,
    )?;
    let value_col = Expr::Column(Column::new_unqualified("value"));
    // Re-wrap with the original `hamelin_from_variant(...)` UDF when
    // a path-extending absorption captured one. Without this, the
    // carrier's output type would degrade from the user's scalar type
    // (Utf8View, Int64, …) back to raw `Variant`, breaking the
    // optimizer's schema-stability invariant.
    let carrier_expr = match state.carrier_wrapper {
        Some(udf) => {
            use datafusion::logical_expr::expr::ScalarFunction;
            Expr::ScalarFunction(ScalarFunction {
                func: udf,
                args: vec![value_col],
            })
        }
        None => value_col,
    };
    let mut projection_exprs = vec![carrier_expr.alias(&state.alias)];
    for passthrough in state.passthroughs {
        projection_exprs.push(passthrough.expr.alias(&passthrough.name));
    }
    Ok(LogicalPlan::Projection(Projection::try_new(
        projection_exprs,
        Arc::new(new_scan),
    )?))
}

/// Try to recognize a fusible EXPLODE seed shape: a `Projection` whose
/// exploded expression resolves, through transparent view-style aliases,
/// to either:
/// - `variant_get*(json_to_variant(<lines-col>))` over `TableScan(LinesFormat)`
/// - `variant_get*(value)` over `TableScan(JsonFormat)`
///
/// Returns the `FusedScan` state (without building the replacement plan),
/// so the caller — `try_absorb_unnest` — can extend it with
/// `ArrayElements` before materializing.
fn try_seed_from_scan_subtree(
    plan: &LogicalPlan,
    exploded_alias: &str,
) -> DfResult<Option<FusedScan>> {
    let LogicalPlan::Projection(outer) = plan else {
        return Ok(None);
    };
    let Some((carrier_idx, alias, expr)) = find_projection_expr(outer, exploded_alias) else {
        return Ok(None);
    };

    // Optionally peel an outer `from_variant(...)` wrapper. Innermost-EXPLODE
    // shapes wear it; the wrapper is only there to unwrap the Variant for
    // the Unnest parent and is dropped here (the seed-from-scan path
    // produces a Variant column that the parent Unnest then consumes).
    let (_from_variant_wrapper, core) = peel_optional_from_variant(expr);

    let Some((resolved, ts)) = resolve_innermost_through_views(core.clone(), outer.input.as_ref())
    else {
        return Ok(None);
    };

    let Ok(provider) = source_as_provider(&ts.source) else {
        return Ok(None);
    };
    let Some(listing) = provider.as_any().downcast_ref::<ListingTable>() else {
        return Ok(None);
    };
    let format = listing.options().format.as_any();

    let (segments, path_display, compression, scan_column_name, scan_column_type) =
        if let Some(lines_fmt) = format.downcast_ref::<LinesFormat>() {
            let Some((variant_arg, fields)) = decode_variant_get_chain_to_json(&resolved) else {
                return Ok(None);
            };
            let Expr::Column(root_col) = variant_arg else {
                return Ok(None);
            };
            let Some(field) = ts
                .projected_schema
                .fields()
                .iter()
                .find(|f| f.name() == &root_col.name)
            else {
                return Ok(None);
            };
            let segments: Vec<Segment> = fields.iter().map(|f| Segment::Field(f.clone())).collect();
            let mut path_display = String::from("$");
            for f in &fields {
                push_field(&mut path_display, f);
            }
            (
                segments,
                path_display,
                lines_fmt.compression(),
                root_col.name.clone(),
                field.data_type().clone(),
            )
        } else if let Some(json_fmt) = format.downcast_ref::<JsonFormat>() {
            let Some((bottom, fields)) = decode_variant_get_chain_to_column(&resolved) else {
                return Ok(None);
            };
            if bottom.name != "value" {
                return Ok(None);
            }
            let Some(field) = ts
                .projected_schema
                .fields()
                .iter()
                .find(|f| f.name() == "value")
            else {
                return Ok(None);
            };
            let mut segments = json_fmt.segments().to_vec();
            let mut path_display = json_fmt.path_display().to_string();
            for f in &fields {
                segments.push(Segment::Field(f.clone()));
                push_field(&mut path_display, f);
            }
            (
                segments,
                path_display,
                json_fmt.compression(),
                "value".to_string(),
                field.data_type().clone(),
            )
        } else {
            return Ok(None);
        };

    let Some(passthroughs) = seed_passthroughs_from_projection(
        outer,
        carrier_idx,
        exploded_alias,
        &alias,
        &scan_column_name,
        &scan_column_type,
    )?
    else {
        return Ok(None);
    };
    Ok(Some(FusedScan {
        segments,
        path_display,
        compression,
        table_paths: listing.table_paths().clone(),
        file_extension: listing.options().file_extension.clone(),
        table_name: ts.table_name.clone(),
        alias,
        passthroughs,
        carrier_wrapper: None,
    }))
}

/// Projection absorption over a fused scan.
///
/// Absorbs a single carrier-defining expression:
/// - `x = carrier`
/// - `x = carrier.foo.bar`
/// - optional outer `from_variant(...)`
///
/// Sibling expressions in the same Projection are preserved verbatim
/// when they meet `is_safe_passthrough`'s gate (literals, immutable
/// or stable UDFs, etc.). Any sibling that fails the gate causes
/// fusion to refuse at this layer.
fn try_absorb_projection(plan: &LogicalPlan) -> DfResult<Option<LogicalPlan>> {
    let LogicalPlan::Projection(proj) = plan else {
        return Ok(None);
    };
    let Some(state) = match_fused_scan(proj.input.as_ref())? else {
        return Ok(None);
    };
    let Some((carrier_idx, new_alias, expr)) = find_carrier_with_safe_siblings(proj, &state.alias)
    else {
        return Ok(None);
    };

    let (carrier_wrapper, core) = peel_optional_from_variant(expr);
    if let Expr::Column(col) = core {
        if col.name != state.alias {
            return Ok(None);
        }
        // Pure rename only — no path-extension. The carrier emits raw
        // Variant from the existing fused scan; if there's a from_variant
        // wrapper the user wanted the unwrapped scalar type, but we have
        // no path hops to fold so absorbing changes nothing useful.
        // Refuse so the wrapper stays put on the unfused plan layer.
        if carrier_wrapper.is_some() {
            return Ok(None);
        }
        let Some(passthroughs) =
            passthrough_columns_from_projection(proj, carrier_idx, &state.alias)?
        else {
            return Ok(None);
        };
        let mut state = state;
        state.alias = new_alias;
        state.passthroughs = passthroughs;
        return Ok(Some(build_fused_scan(state)?));
    }

    let Some((bottom, fields)) = decode_variant_get_chain_to_column(core) else {
        return Ok(None);
    };
    if bottom.name != state.alias || fields.is_empty() {
        return Ok(None);
    }

    // Path-extending absorption (the carrier dives into one or more
    // struct fields) changes what the fused scan emits. Any sibling
    // that reads the OLD carrier alias would, after rename, point at
    // the NEW (deeper) element — a different value entirely. Refuse
    // fusion when any sibling references the carrier alias.
    let any_sibling_reads_carrier = proj.expr.iter().enumerate().any(|(idx, expr)| {
        if idx == carrier_idx {
            return false;
        }
        let body = match expr {
            Expr::Alias(a) => a.expr.as_ref(),
            Expr::Column(_) => expr,
            _ => return false,
        };
        expr_references_column(body, &state.alias)
    });
    if any_sibling_reads_carrier {
        return Ok(None);
    }

    let Some(passthroughs) = passthrough_columns_from_projection(proj, carrier_idx, &state.alias)?
    else {
        return Ok(None);
    };
    let mut state = state;
    let mut path_display = state.path_display;
    for f in &fields {
        state.segments.push(Segment::Field(f.clone()));
        push_field(&mut path_display, f);
    }
    state.path_display = path_display;
    state.alias = new_alias;
    state.passthroughs = passthroughs;
    state.carrier_wrapper = carrier_wrapper;
    Ok(Some(build_fused_scan(state)?))
}

/// Layer 3 — Unnest absorption.
///
/// This is the only layer that *introduces* a fused scan. It matches in
/// two cases:
///
/// 1. **Chained EXPLODE.** The Unnest's child is already a canonical
///    fused scan (from a prior visit); just append `ArrayElements`.
///
/// 2. **Innermost EXPLODE.** The Unnest's child is the seed-shape
///    Projection over `TableScan(LinesFormat)` — see
///    `try_seed_from_lines_subtree`. Build the seed and append
///    `ArrayElements` in one step.
///
/// Both cases require a single `exec_columns` entry whose name matches
/// the (existing or seed) alias.
fn try_absorb_unnest(plan: &LogicalPlan) -> DfResult<Option<LogicalPlan>> {
    let LogicalPlan::Unnest(Unnest {
        input,
        exec_columns,
        ..
    }) = plan
    else {
        return Ok(None);
    };
    if exec_columns.len() != 1 {
        return Ok(None);
    }
    let exploded = &exec_columns[0];

    // Case 1: child is already a fused scan from an earlier visit.
    let mut state = if let Some(state) = match_fused_scan(input.as_ref())? {
        state
    } else if let Some(state) = try_seed_from_scan_subtree(input.as_ref(), &exploded.name)? {
        // Case 2: child resolves, through transparent wrappers, to either
        // a LinesFormat or JsonFormat scan.
        state
    } else {
        return Ok(None);
    };

    if exploded.name != state.alias {
        return Ok(None);
    }
    state.segments.push(Segment::ArrayElements);
    state.path_display.push_str("[*]");
    Ok(Some(build_fused_scan(state)?))
}

/// Peel an optional outer `hamelin_from_variant(<inner>)` wrapper.
/// Returns `(wrapper_udf, inner)` where `wrapper_udf` is `Some` when
/// `expr` had the wrapper. Path-extending absorption needs the wrapper
/// back at the top of the rewritten Projection so the column's output
/// type stays the unwrapped scalar type rather than the absorbed scan's
/// raw `Variant`.
fn peel_optional_from_variant(expr: &Expr) -> (Option<Arc<ScalarUDF>>, &Expr) {
    match expr {
        Expr::ScalarFunction(f)
            if udf_name(&f.func) == "hamelin_from_variant" && f.args.len() == 1 =>
        {
            (Some(Arc::clone(&f.func)), &f.args[0])
        }
        _ => (None, expr),
    }
}

/// Peel zero or more `hamelin_variant_get(prev, "field")` hops off
/// `expr`. Returns the deepest non-variant_get expression (whatever the
/// chain bottoms at — a Column, a `json_to_variant(...)` call, etc.)
/// plus the field names in document order. Callers match the bottom
/// shape themselves to enforce their layer's invariants.
fn peel_variant_get_chain(expr: &Expr) -> (&Expr, Vec<String>) {
    let mut fields: Vec<String> = Vec::new();
    let mut current = expr;
    loop {
        match current {
            Expr::ScalarFunction(f) if udf_name(&f.func) == "hamelin_variant_get" => {
                if f.args.len() != 2 {
                    break;
                }
                let Expr::Literal(lit, _) = &f.args[1] else {
                    break;
                };
                let Some(field) = lit.try_as_str().flatten() else {
                    break;
                };
                fields.push(field.to_string());
                current = &f.args[0];
            }
            _ => break,
        }
    }
    fields.reverse();
    (current, fields)
}

/// Decode `variant_get*(json_to_variant(<arg>))`: zero or more
/// `hamelin_variant_get(prev, "field")` hops bottoming at
/// `hamelin_json_to_variant(<arg>)`. Returns `(arg, fields_in_doc_order)`.
fn decode_variant_get_chain_to_json(expr: &Expr) -> Option<(&Expr, Vec<String>)> {
    let (bottom, fields) = peel_variant_get_chain(expr);
    let Expr::ScalarFunction(jf) = bottom else {
        return None;
    };
    if udf_name(&jf.func) != "hamelin_json_to_variant" || jf.args.len() != 1 {
        return None;
    }
    Some((&jf.args[0], fields))
}

/// Decode `variant_get*(<col>)`: zero or more
/// `hamelin_variant_get(prev, "field")` hops bottoming at a plain
/// `Column(<col>)`. Returns `(col, fields_in_doc_order)`.
fn decode_variant_get_chain_to_column(expr: &Expr) -> Option<(Column, Vec<String>)> {
    let (bottom, fields) = peel_variant_get_chain(expr);
    let Expr::Column(col) = bottom else {
        return None;
    };
    Some((col.clone(), fields))
}

/// Scan `proj.expr` for the first entry that satisfies `pred`, peeling
/// the projection-list alias to give callers `(idx, output_name, body)`.
/// `pred` runs against the body — either the inner expression of an
/// `Expr::Alias` or a bare `Expr::Column` (whose name *is* its output
/// name).
///
/// Used by `match_fused_scan`, `find_projection_expr`, and the candidate
/// finder inside `find_carrier_with_safe_siblings` — they all share this
/// shape.
fn find_projection_entry<'a>(
    proj: &'a Projection,
    pred: impl Fn(&str, &Expr) -> bool,
) -> Option<(usize, String, &'a Expr)> {
    proj.expr.iter().enumerate().find_map(|(idx, expr)| {
        let (name, body) = match expr {
            Expr::Alias(a) => (a.name.as_str(), a.expr.as_ref()),
            Expr::Column(c) => (c.name.as_str(), expr),
            _ => return None,
        };
        if pred(name, body) {
            Some((idx, name.to_string(), body))
        } else {
            None
        }
    })
}

fn find_projection_expr<'a>(
    proj: &'a Projection,
    alias: &str,
) -> Option<(usize, String, &'a Expr)> {
    find_projection_entry(proj, |name, _body| name == alias)
}

/// Find the carrier-defining expression in `proj` and verify every
/// other expression is a safe-passthrough sibling under the same
/// validator the collector uses. Returns `None` if no carrier
/// candidate exists OR any sibling fails the safety check.
///
/// Synchronizing this pre-check with the collector is load-bearing:
/// if the pre-check accepts a sibling shape that the collector
/// rejects, fusion would mis-fire; if the pre-check rejects a
/// shape the collector would accept, fusion would refuse cases it
/// could safely handle.
fn find_carrier_with_safe_siblings<'a>(
    proj: &'a Projection,
    carrier_alias: &str,
) -> Option<(usize, String, &'a Expr)> {
    let candidate = find_projection_entry(proj, |name, body| match body {
        Expr::Column(_) => name == carrier_alias,
        other => is_absorbable_carrier_expr(other, carrier_alias),
    })?;

    let any_unsafe_sibling = proj.expr.iter().enumerate().any(|(idx, expr)| {
        if idx == candidate.0 {
            return false;
        }
        let body = match expr {
            Expr::Alias(a) => a.expr.as_ref(),
            Expr::Column(_) => expr,
            _ => return true,
        };
        !is_safe_passthrough(body, carrier_alias)
    });
    if any_unsafe_sibling {
        return None;
    }
    Some(candidate)
}

fn is_absorbable_carrier_expr(expr: &Expr, carrier_alias: &str) -> bool {
    let (_wrapper, core) = peel_optional_from_variant(expr);
    match core {
        Expr::Column(col) => col.name == carrier_alias,
        _ => decode_variant_get_chain_to_column(core)
            .map(|(bottom, _fields)| bottom.name == carrier_alias)
            .unwrap_or(false),
    }
}

/// True if `expr` mentions `Column(name)` anywhere in its tree
/// (matching by column name only, ignoring relation).
fn expr_references_column(expr: &Expr, name: &str) -> bool {
    let mut found = false;
    expr.apply(|e| {
        if let Expr::Column(col) = e {
            if col.name == name {
                found = true;
                return Ok(datafusion::common::tree_node::TreeNodeRecursion::Stop);
            }
        }
        Ok(datafusion::common::tree_node::TreeNodeRecursion::Continue)
    })
    .ok();
    found
}

/// Returns true iff `expr` is safely re-emitable above the fused
/// scan: every leaf is a literal or a reference to the carrier
/// alias, and every interior node is an immutable/stable operation.
///
/// **Why volatility is the right gate.** An `Immutable` op
/// (`named_struct`, `make_array`, arithmetic, `date_trunc`) is
/// independent of evaluation context. A `Stable` op (`now`,
/// `current_timestamp`) returns the same value across all rows in
/// a query. The fused json scan produces the same number of rows
/// as the lines+unnest pipeline it replaced, so per-row
/// evaluation of either kind above the fused scan matches what
/// the original plan would have computed. A `Volatile` op
/// (`random`, `uuid`) does not satisfy that, so the rule refuses.
///
/// **Contract.** The caller passes an unwrapped expression body,
/// never an `Expr::Alias`. Aliases at the top level of a
/// projection expression carry the column name, which the caller
/// manages separately (stripping before this call, re-wrapping
/// after the renamer). Aliases nested deeper are not produced by
/// Hamelin or DataFusion translation, so they fall to the default
/// refuse arm and are out of contract.
fn is_safe_passthrough(expr: &Expr, carrier_alias: &str) -> bool {
    match expr {
        Expr::Literal(_, _) => true,
        // Match by name only — `relation` is ignored throughout the rule
        // (view-inlining in `inline_projection_aliases` looks up by name
        // too), so the validator must be consistent. The seed-side
        // validator (`is_safe_passthrough_seed`) does the same.
        Expr::Column(col) => col.name == carrier_alias,
        Expr::ScalarFunction(f) => match f.func.signature().volatility {
            Volatility::Immutable | Volatility::Stable => f
                .args
                .iter()
                .all(|arg| is_safe_passthrough(arg, carrier_alias)),
            Volatility::Volatile => false,
        },
        Expr::Cast(c) => is_safe_passthrough(&c.expr, carrier_alias),
        Expr::BinaryExpr(b) => {
            is_safe_passthrough(&b.left, carrier_alias)
                && is_safe_passthrough(&b.right, carrier_alias)
        }
        // All other Expr variants — including Expr::Alias (out of
        // contract), Expr::ScalarVariable, subqueries, window
        // functions, aggregates — refuse by default. This is the
        // safety boundary for any new variant DataFusion adds in
        // future versions.
        _ => false,
    }
}

/// Rewrite every `Column(old_carrier_alias)` reference in `expr` so
/// it reads from the new TableScan's `value` input column. Sibling
/// expressions in `build_fused_scan`'s rebuilt Projection evaluate
/// against the TableScan input (which only exposes `value`), not
/// against each other's output aliases — so the rename target is
/// always `"value"`, never the carrier's new output alias.
///
/// Called after `is_safe_passthrough` has validated the expression,
/// so the walker doesn't need to defend against unexpected shapes.
fn rename_carrier_refs(expr: Expr, old_carrier_alias: &str) -> DfResult<Expr> {
    expr.transform_up(|e| {
        if let Expr::Column(col) = &e {
            // Match by name only, mirroring `is_safe_passthrough`'s
            // gate. The seed-side `rewrite_seed_sibling` does the same.
            if col.name == old_carrier_alias {
                return Ok(Transformed::yes(Expr::Column(Column::new_unqualified(
                    "value",
                ))));
            }
        }
        Ok(Transformed::no(e))
    })
    .map(|t| t.data)
}

/// Walk every non-carrier expression in `proj.expr`, peel its
/// projection-list alias to a `(name, body)`, and hand the body to
/// `process` to validate-and-rewrite. `process` returns `Ok(Some(new))`
/// to keep the sibling, `Ok(None)` to refuse the whole fusion (one
/// poisoned sibling poisons the layer).
///
/// Both passthrough collectors share this iteration; only the body
/// processing differs (chained-projection siblings vs. seed-projection
/// siblings).
fn collect_passthroughs(
    proj: &Projection,
    carrier_idx: usize,
    mut process: impl FnMut(&Expr) -> DfResult<Option<Expr>>,
) -> DfResult<Option<Vec<PassthroughColumn>>> {
    let mut out = Vec::with_capacity(proj.expr.len().saturating_sub(1));
    for (idx, expr) in proj.expr.iter().enumerate() {
        if idx == carrier_idx {
            continue;
        }
        let (name, body) = match expr {
            Expr::Alias(a) => (a.name.clone(), a.expr.as_ref()),
            Expr::Column(c) => (c.name.clone(), expr),
            _ => return Ok(None),
        };
        let Some(rewritten) = process(body)? else {
            return Ok(None);
        };
        out.push(PassthroughColumn {
            name,
            expr: rewritten,
        });
    }
    Ok(Some(out))
}

/// Sibling-collection variant for the seed-from-scan layer.
///
/// Inlines each non-carrier sibling's body through the projection's
/// input chain (the connector view, plus any optimizer-introduced
/// extraction projections), then validates against a strict shape
/// predicate. After resolution, every leaf of the sibling's expression
/// tree must be one of:
///
/// - `Expr::Literal` (including constant-folded UDFs like `now()`),
/// - `Column(carrier_alias)` — renamed to `new_carrier_alias`,
/// - `Column(scan_column_name)` — re-emitted as a typed NULL (the scan
///   column does not survive fusion; the rebuild post-Unnest overwrites
///   any slot the user actually reads, and dead siblings are pruned by
///   later optimizer passes).
///
/// Internal nodes may only be `named_struct(name, value, …)` calls —
/// those are transparent (struct construction is just aliasing). Any
/// other shape (arbitrary UDF calls, casts, binary ops, non-alias
/// Column references) refuses fusion at this layer.
fn seed_passthroughs_from_projection(
    proj: &Projection,
    carrier_idx: usize,
    old_carrier_alias: &str,
    new_carrier_alias: &str,
    scan_column_name: &str,
    scan_column_type: &datafusion::arrow::datatypes::DataType,
) -> DfResult<Option<Vec<PassthroughColumn>>> {
    collect_passthroughs(proj, carrier_idx, |body| {
        let inlined = resolve_innermost_through_views(body.clone(), proj.input.as_ref())
            .map(|(expr, _)| expr)
            .unwrap_or_else(|| body.clone());
        if !is_safe_passthrough_seed(&inlined, old_carrier_alias, scan_column_name) {
            return Ok(None);
        }
        let rewritten = rewrite_seed_sibling(
            inlined,
            old_carrier_alias,
            new_carrier_alias,
            scan_column_name,
            scan_column_type,
        )?;
        Ok(Some(rewritten))
    })
}

/// Strict seed-sibling validator. A resolved sibling expression is
/// safe iff:
/// - it is `Expr::Literal`, or
/// - it is `Column(carrier_alias)` (re-emitted as a reference to the
///   fused scan's new alias), or
/// - it is `named_struct(name, val, …)` whose values are themselves
///   safe — note that *inside* a `named_struct` the validator further
///   accepts `Column(scan_column_name)` (the rewriter substitutes
///   typed NULL; the post-Unnest rebuild overwrites the slot if read
///   through the carrier, otherwise downstream readers were going to
///   discard the slot via a later projection-pruning pass).
///
/// Top-level `Column(scan_column_name)` is *not* safe: a sibling that
/// IS the bare scan column means the user wants the scan column's
/// value as that output column, and the fused json scan has already
/// dropped the scan column. Fusing here would silently NULL out a
/// read the user expected to succeed.
fn is_safe_passthrough_seed(expr: &Expr, carrier_alias: &str, scan_column_name: &str) -> bool {
    match expr {
        Expr::Literal(_, _) => true,
        Expr::Column(col) => col.name == carrier_alias,
        Expr::ScalarFunction(f) if udf_name(&f.func) == "named_struct" => {
            is_safe_named_struct_args(&f.args, carrier_alias, scan_column_name)
        }
        _ => false,
    }
}

/// Validate `named_struct(name1, val1, name2, val2, ...)` argument
/// pairs against the seed-sibling rule. Inside a `named_struct`, slot
/// values may additionally be `Column(scan_column_name)` — the
/// rewriter substitutes typed NULL there. Outside a `named_struct`
/// (i.e., as a top-level sibling expression),
/// `Column(scan_column_name)` is rejected by `is_safe_passthrough_seed`.
fn is_safe_named_struct_args(args: &[Expr], carrier_alias: &str, scan_column_name: &str) -> bool {
    if args.len() % 2 != 0 {
        return false;
    }
    let mut i = 0;
    while i + 1 < args.len() {
        let Expr::Literal(name_lit, _) = &args[i] else {
            return false;
        };
        if name_lit.try_as_str().flatten().is_none() {
            return false;
        }
        if !is_safe_named_struct_value(&args[i + 1], carrier_alias, scan_column_name) {
            return false;
        }
        i += 2;
    }
    true
}

/// A value inside a `named_struct` may additionally be
/// `Column(scan_column_name)` (substituted to typed NULL). Otherwise
/// follows the same rule as a top-level sibling.
fn is_safe_named_struct_value(expr: &Expr, carrier_alias: &str, scan_column_name: &str) -> bool {
    if let Expr::Column(col) = expr {
        if col.name == scan_column_name {
            return true;
        }
    }
    is_safe_passthrough_seed(expr, carrier_alias, scan_column_name)
}

/// Rewrite a resolved seed-projection sibling for re-emission above
/// the fused json scan:
/// - `Column(carrier_alias)` → `Column(new_carrier_alias)`
/// - `Column(scan_column_name)` → `Cast(Null AS scan_column_type)`
///   (the scan column is gone after fusion; any actually-read slot is
///   either overwritten by the rebuild or pruned downstream).
/// - Everything else is preserved verbatim. The validator already
///   rejected anything else.
fn rewrite_seed_sibling(
    expr: Expr,
    old_carrier_alias: &str,
    new_carrier_alias: &str,
    scan_column_name: &str,
    scan_column_type: &datafusion::arrow::datatypes::DataType,
) -> DfResult<Expr> {
    use datafusion::logical_expr::expr::Cast;
    expr.transform_up(|e| {
        if let Expr::Column(col) = &e {
            if col.name == old_carrier_alias {
                return Ok(Transformed::yes(Expr::Column(Column::new_unqualified(
                    new_carrier_alias,
                ))));
            }
            if col.name == scan_column_name {
                return Ok(Transformed::yes(Expr::Cast(Cast::new(
                    Box::new(Expr::Literal(datafusion::scalar::ScalarValue::Null, None)),
                    scan_column_type.clone(),
                ))));
            }
        }
        Ok(Transformed::no(e))
    })
    .map(|t| t.data)
}

/// Collect non-carrier sibling expressions from `proj` for re-emission
/// above the fused scan. Returns `None` if any sibling cannot be
/// safely preserved — one bad sibling poisons the whole fusion at
/// this layer.
///
/// `old_carrier_alias` is the alias the carrier expression reads
/// from in the input projection. Sibling expressions referencing
/// `old_carrier_alias` are rewritten to reference the canonical
/// `value` column — that's the column the new TableScan emits and
/// thus the only column the rebuilt Projection's expressions can
/// legitimately reference. (Previously this function took a
/// `new_carrier_alias` and renamed to that, which produced
/// self-referential Projections — referring to the carrier output's
/// alias rather than the input column the sibling actually needs.)
fn passthrough_columns_from_projection(
    proj: &Projection,
    carrier_idx: usize,
    old_carrier_alias: &str,
) -> DfResult<Option<Vec<PassthroughColumn>>> {
    collect_passthroughs(proj, carrier_idx, |body| {
        if !is_safe_passthrough(body, old_carrier_alias) {
            return Ok(None);
        }
        // Rewrite carrier references to `value` — the new TableScan's
        // input column. Any `Column(old_carrier_alias)` in the sibling
        // was reading the previously-fused scan's output; that output
        // is now `value` on the new scan.
        Ok(Some(rename_carrier_refs(body.clone(), old_carrier_alias)?))
    })
}

/// Append a field name to a path-display string. Uses dot-form
/// (`.fieldname`) when the JSONPath grammar accepts it directly;
/// bracket-quoted form (`["..."]`) otherwise. Bracket-quoted form
/// escapes `\`, `"`, and ASCII control characters per RFC 8259 so the
/// resulting JSONPath round-trips back through `path::parse` to the
/// same field name (a literal `\b` would otherwise be interpreted as
/// the JSON backspace escape on the way back in).
///
/// JSONPath dot-form is only legal when the name (a) is non-empty,
/// (b) starts with an ASCII letter or underscore, and (c) contains
/// only ASCII letters, digits, or underscores. Anything else — empty
/// strings, leading-digit names like `123`, names with spaces or
/// punctuation, multi-byte/non-ASCII characters — must take the
/// bracket-quoted form so the path round-trips through `path::parse`.
fn push_field(out: &mut String, f: &str) {
    let mut chars = f.chars();
    let dot_form_ok = match chars.next() {
        Some(c) if c.is_ascii_alphabetic() || c == '_' => {
            chars.all(|c| c.is_ascii_alphanumeric() || c == '_')
        }
        _ => false,
    };
    if dot_form_ok {
        out.push('.');
        out.push_str(f);
    } else {
        out.push_str("[\"");
        for c in f.chars() {
            match c {
                '\\' => out.push_str("\\\\"),
                '"' => out.push_str("\\\""),
                '\n' => out.push_str("\\n"),
                '\r' => out.push_str("\\r"),
                '\t' => out.push_str("\\t"),
                '\x08' => out.push_str("\\b"),
                '\x0c' => out.push_str("\\f"),
                c if (c as u32) < 0x20 => {
                    out.push_str(&format!("\\u{:04x}", c as u32));
                }
                _ => out.push(c),
            }
        }
        out.push_str("\"]");
    }
}

/// Resolve a Hamelin view-style expression chain down to an expression
/// that references only columns of the underlying `TableScan`.
///
/// Walks down through transparent wrappers:
/// - `SubqueryAlias(inner)` — pass through
/// - `Projection` whose expressions are all simple aliases (see below) —
///   inline the referenced aliases into `expr`, then recurse on the
///   Projection's input
///
/// Supported alias shapes inside those Projections:
/// - `named_struct("f1", e1, "f2", e2, ...) AS X` — when `expr` uses
///   `get_field(col("X"), "fi")`, substitute in `ei`
/// - `<expr> AS Y` where `Y` appears in `expr` as a plain `Column` —
///   substitute directly (covers simple renames, including `LET original = line`)
///
/// Stops at `TableScan`, returning the fully resolved expression and the
/// scan. Bails on unsupported shapes (Filter, Sort, etc.).
fn resolve_innermost_through_views<'a>(
    expr: Expr,
    plan: &'a LogicalPlan,
) -> Option<(Expr, &'a TableScan)> {
    let mut expr = expr;
    let mut current = plan;
    loop {
        match current {
            LogicalPlan::TableScan(ts) => return Some((expr, ts)),
            LogicalPlan::SubqueryAlias(sa) => {
                current = sa.input.as_ref();
            }
            LogicalPlan::Projection(p) => {
                expr = inline_projection_aliases(expr, &p.expr)?;
                current = p.input.as_ref();
            }
            _ => return None,
        }
    }
}

/// Substitute references to Projection-defined aliases within `expr`.
///
/// Iterates the column-substitution + `get_field(named_struct, …)`
/// simplification to a fixed point: substituting a column might inject
/// a `get_field(named_struct(…), …)` which then needs to be simplified,
/// and that simplification might expose new column references for the
/// next pass to substitute. A single `transform_up` pass would miss
/// these chained rewrites because the closure doesn't re-visit the
/// node it just replaced.
fn inline_projection_aliases(expr: Expr, proj_exprs: &[Expr]) -> Option<Expr> {
    let mut aliases: std::collections::HashMap<&str, &Expr> = std::collections::HashMap::new();
    for e in proj_exprs {
        match e {
            Expr::Alias(a) => {
                aliases.insert(a.name.as_str(), a.expr.as_ref());
            }
            Expr::Column(c) => {
                aliases.insert(c.name.as_str(), e);
            }
            _ => {}
        }
    }

    // Cap iterations so a pathological alias cycle can't loop forever.
    // Real chains produced by `OptimizeProjections` extraction (the
    // `__datafusion_extracted_N` shape) need at most two passes.
    let mut current = expr;
    for _ in 0..8 {
        let transformed = current
            .clone()
            .transform_up(|e| match e {
                Expr::Column(ref c) => {
                    if let Some(replacement) = aliases.get(c.name.as_str()) {
                        Ok(Transformed::yes((*replacement).clone()))
                    } else {
                        Ok(Transformed::no(e))
                    }
                }
                Expr::ScalarFunction(ref f) if udf_name(&f.func) == "get_field" => {
                    if f.args.len() != 2 {
                        return Ok(Transformed::no(e));
                    }
                    let Expr::Literal(lit, _) = &f.args[1] else {
                        return Ok(Transformed::no(e));
                    };
                    let Some(field_name) = lit.try_as_str().flatten() else {
                        return Ok(Transformed::no(e));
                    };
                    if let Expr::ScalarFunction(inner) = &f.args[0] {
                        if udf_name(&inner.func) == "named_struct" {
                            let mut i = 0;
                            while i + 1 < inner.args.len() {
                                if let Expr::Literal(name_lit, _) = &inner.args[i] {
                                    if name_lit.try_as_str().flatten() == Some(field_name) {
                                        return Ok(Transformed::yes(inner.args[i + 1].clone()));
                                    }
                                }
                                i += 2;
                            }
                        }
                    }
                    Ok(Transformed::no(e))
                }
                _ => Ok(Transformed::no(e)),
            })
            .ok()?;
        if !transformed.transformed {
            return Some(transformed.data);
        }
        current = transformed.data;
    }
    Some(current)
}

fn udf_name(udf: &Arc<ScalarUDF>) -> &str {
    udf.name()
}

/// Build a `json`-backed `TableScan` from a `JsonFormat`, table paths,
/// and file-extension hint. Schema is fixed (one Variant column), so no
/// async inference is needed.
fn build_json_scan(
    table_paths: Vec<ListingTableUrl>,
    file_extension: String,
    je_format: JsonFormat,
    table_name: TableReference,
) -> DfResult<LogicalPlan> {
    let listing_options =
        ListingOptions::new(Arc::new(je_format)).with_file_extension(file_extension);
    let schema = json_schema();
    let listing_config = ListingTableConfig::new_with_multi_paths(table_paths)
        .with_listing_options(listing_options)
        .with_schema(schema.clone());

    let listing = Arc::new(ListingTable::try_new(listing_config)?);

    let schema_dfs = DFSchema::try_from(schema.as_ref().clone())?;
    let ts = TableScan {
        table_name,
        source: provider_as_source(listing),
        projection: None,
        projected_schema: Arc::new(schema_dfs),
        filters: vec![],
        fetch: None,
    };
    Ok(LogicalPlan::TableScan(ts))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::json::path::{parse as parse_path, Segment};

    fn push(name: &str) -> String {
        let mut out = String::from("$");
        push_field(&mut out, name);
        out
    }

    /// Bracket-quoted field names that round-trip through
    /// `path::parse` must decode back to the original key. The encoder
    /// has one branch per JSON escape category, so this test covers at
    /// least one input from each branch — adding a new category to
    /// `push_field` should mean adding a case here.
    #[test]
    fn push_field_round_trips_through_path_parse() {
        for raw in [
            // ── dot-form eligible (non-empty, ALPHA-or-underscore start,
            // ── then alnum/underscore)
            "plain",
            "_underscore_leading",
            "a1b2",
            "foo_bar",
            "_",
            // ── must take bracket-quoted form ───────────────────────
            // empty key: vacuous `chars().all(...)` is true on empty,
            // so without the explicit `is_empty` guard this would emit
            // a bare `.` that doesn't round-trip.
            "",
            // digit-leading: JSONPath dot-form requires a leading ALPHA
            // or underscore. `is_ascii_alphanumeric` accepts digits, so
            // a naive predicate would emit `.123` which fails to
            // round-trip.
            "123",
            "0",
            "123_foo",
            // bracket-quoted, no escape needed inside
            "with space",
            "emoji 🦀",
            // ── special JSON characters ──────────────────────────────
            "has\"quote",
            "has\\backslash",
            // The original cursor[bot] case: a literal `\b` in the key
            // must NOT be emitted as the JSON backspace escape.
            "back\\b slash then b",
            // ── named JSON escapes ───────────────────────────────────
            "newline\nhere",
            "carriage\rreturn",
            "tab\there",
            "backspace\x08here",
            "formfeed\x0chere",
            // Sub-0x20 control char without a named escape (\u escape)
            "bell\x07here",
            // U+007F is not a control char (it's DEL) and is allowed
            // raw in JSON strings, so it falls through to the default
            // branch — exercise it to catch off-by-one in the < 0x20
            // boundary.
            "del\x7fhere",
            // ── punctuation and operators ────────────────────────────
            "has-hyphen",
            "has.dot.in.name",
            "has[bracket",
            "*",
            "$",
        ] {
            let path = push(raw);
            let segs = parse_path(&path)
                .unwrap_or_else(|e| panic!("parse failed for {raw:?} → {path:?}: {e:?}"));
            assert_eq!(
                segs,
                vec![Segment::Field(raw.to_string())],
                "round-trip mismatch for {raw:?} (path={path:?})"
            );
        }
    }
}
