arcform 0.2.2

//! SQL introspection: what a SQL step reads and produces.
//!
//! Two readers take a step apart. By default DuckDB's own parse is read, through
//! [`crate::duckdb_lineage`], from the DuckDB arc runs, which has to be a 2.x build; a
//! statement that reading cannot read is parsed by sqlparser-rs with its DuckDB dialect.
//! `ARC_SQL_READER=sqlparser` has sqlparser-rs read every statement. Either reader's
//! statements pass through the same rules here: which table function reads a
//! file, which reads no table, which one's arguments arc does not read, and whether a
//! `COPY … TO` writes a file or a directory.
//!
//! Extracted from each statement:
//! - **Outputs**: tables/views created or written to (CREATE TABLE, CREATE VIEW, CTAS, INSERT INTO, COPY TO)
//! - **Inputs**: tables read from (FROM, JOIN clauses)
//!
//! **File-path lineage.** A DuckDB file-reader in a FROM clause — `read_parquet('x.parquet')`,
//! `read_csv(...)`, `read_json(['a.json','b.json'])` — is a table-valued function whose *first
//! argument is a filesystem path*. Rather than record the opaque function name (`read_parquet`)
//! as the input, we lift the path literal(s) it reads and a `COPY … TO 'file'` writes: those
//! path-shaped names become filesystem-backed assets downstream (see [`crate::contract`]) —
//! one file, a directory of files, or a glob, per [`SqlAssets::kinds`]. Lineage into and out of
//! files is thus *discovered from the SQL*, never hand-declared via `depends_on:`.

use std::collections::{BTreeMap, BTreeSet};
use std::ops::ControlFlow;

use sqlparser::ast::{
    CopyOption, CopySource, CopyTarget, Expr, FunctionArg, FunctionArgExpr, Insert, ObjectName,
    Statement, TableFactor, TableFunctionArgs, TableObject, Value, visit_expressions,
};
use sqlparser::dialect::DuckDbDialect;
use sqlparser::parser::Parser;

use crate::asset_kind::AssetKind;
use crate::duckdb_lineage::{self, ArgValue, Relation, TableCall};

/// Assets discovered from parsing a SQL file — four-set model.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SqlAssets {
    /// Tables/views this SQL creates, writes to, or modifies (ALTER).
    pub outputs: BTreeSet<String>,
    /// External tables this SQL reads data from (CTEs excluded).
    pub inputs: BTreeSet<String>,
    /// CTE names — step-internal assets visible in lineage but not cross-step dependencies.
    pub internal: BTreeSet<String>,
    /// Tables/views this SQL drops — destructive operations tracked separately.
    pub destroys: BTreeSet<String>,
    /// What each name in `outputs`/`inputs` actually is — set here, at the one place
    /// that already knows. A bare identifier from CREATE/FROM/JOIN is a `Table`. A
    /// path literal lifted from a file-reader argument is a `File`, or a `Pattern`
    /// when [`is_glob`] holds. A `COPY … TO` target is classified by
    /// [`copy_to_target_kind`] from the statement's own options and is never tested
    /// for glob metacharacters — DuckDB's COPY target is a literal path, not a
    /// pattern. Never reconstructed later from the string.
    pub kinds: BTreeMap<String, AssetKind>,
    /// Table-valued functions this SQL calls with a string or a quoted identifier among
    /// their arguments, other than the file readers and row-generators this layer knows —
    /// `mlpack_random_forest_train("X", "Y", "params", "model")`, `query('SELECT …')`. What
    /// such an argument names depends on the function's signature, which arc does not
    /// carry, so the call records no input at all, not even one under the function's own
    /// name: the function is listed here for `AssetGraph::build` to ask the step for a
    /// `depends_on:` and a `produces:`. Lowercased; a function called twice is listed once.
    pub unread_table_functions: BTreeSet<String>,
}

impl SqlAssets {
    fn record_output(&mut self, name: String, kind: AssetKind) {
        self.kinds.insert(name.clone(), kind);
        self.outputs.insert(name);
    }

    fn record_input(&mut self, name: String, kind: AssetKind) {
        self.kinds.insert(name.clone(), kind);
        self.inputs.insert(name);
    }
}

/// Whether a lifted path literal is a glob pattern rather than one literal path.
fn is_glob(path: &str) -> bool {
    path.contains(['*', '?', '['])
}

/// What a path a statement reads names: a glob pattern, or one file.
fn read_path_kind(path: &str) -> AssetKind {
    if is_glob(path) {
        AssetKind::Pattern
    } else {
        AssetKind::File
    }
}

/// The `COPY … TO 'target'` option names under which DuckDB writes a *directory* of
/// files at `target` instead of one file at `target`.
///
/// Read out of DuckDB's own source rather than assembled from the cases someone
/// happened to hit. `PhysicalCopyToFile::GetGlobalSinkState`
/// (`src/execution/operator/persistent/physical_copy_to_file.cpp`) creates `target`
/// as a directory on `partition_output || per_thread_output || rotate`;
/// `Binder::BindCopyTo` (`src/planner/binder/statement/bind_copy.cpp`) sets the
/// first from a non-empty `PARTITION_BY` column list, the second from
/// `PER_THREAD_OUTPUT`, and the third from `CopyFunction::rotate_files`. The two
/// `rotate_files` implementations that exist in the tree — `WriteCSVRotateFiles`
/// (`src/function/table/copy_csv.cpp`) and `ParquetWriteRotateFiles`
/// (`extension/parquet/parquet_extension.cpp`) — return true for `FILE_SIZE_BYTES`,
/// and the parquet one additionally for `ROW_GROUPS_PER_FILE`. That branch and its
/// three inputs read the same in v1.5.2 and v1.5.6, the version this crate pins and CI
/// links, compared in the DuckDB sources `libduckdb-sys` 1.10502.0 and 1.10506.0 carry:
/// `physical_copy_to_file.cpp` and `copy_csv.cpp` are byte-identical, `bind_copy.cpp`
/// differs only in how three error-message arguments are built, and
/// `parquet_extension.cpp` only in registering the variant functions.
const DIRECTORY_WRITING_COPY_OPTIONS: [&str; 4] = [
    "PARTITION_BY",
    "PER_THREAD_OUTPUT",
    "FILE_SIZE_BYTES",
    "ROW_GROUPS_PER_FILE",
];

/// A `COPY … TO` option's value, as either reader gives it: what
/// [`copy_option_is_on`] reads of it.
#[derive(Debug, Clone, PartialEq)]
enum CopyOptionValue {
    /// The option is given without a value: `PER_THREAD_OUTPUT`.
    Absent,
    Boolean(bool),
    Number(String),
    String(String),
    /// `()`.
    EmptyList,
    /// Anything else: a bare word, a list with something in it, an expression.
    Other,
}

/// What a `COPY … TO 'filename'` writes, decided from the statement's own options
/// against [`DIRECTORY_WRITING_COPY_OPTIONS`].
fn copy_to_target_kind(options: &[(String, CopyOptionValue)]) -> AssetKind {
    let writes_a_directory = options.iter().any(|(name, value)| {
        DIRECTORY_WRITING_COPY_OPTIONS
            .iter()
            .any(|known| name.eq_ignore_ascii_case(known))
            && copy_option_is_on(name, value)
    });
    if writes_a_directory {
        AssetKind::Directory
    } else {
        AssetKind::File
    }
}

/// sqlparser's `COPY` options, as [`copy_to_target_kind`] reads them.
fn sqlparser_copy_options(options: &[CopyOption]) -> Vec<(String, CopyOptionValue)> {
    let value = |value: &Option<Expr>| match value {
        None => CopyOptionValue::Absent,
        Some(Expr::Tuple(items)) if items.is_empty() => CopyOptionValue::EmptyList,
        Some(Expr::Value(v)) => match &v.value {
            Value::Boolean(b) => CopyOptionValue::Boolean(*b),
            Value::Number(n, _) => CopyOptionValue::Number(n.clone()),
            Value::SingleQuotedString(s)
            | Value::DoubleQuotedString(s)
            | Value::TripleSingleQuotedString(s)
            | Value::TripleDoubleQuotedString(s) => CopyOptionValue::String(s.clone()),
            _ => CopyOptionValue::Other,
        },
        Some(_) => CopyOptionValue::Other,
    };
    options
        .iter()
        .filter_map(|opt| match opt {
            CopyOption::DuckDbOption { name, value: v } => Some((name.value.clone(), value(v))),
            _ => None,
        })
        .collect()
}

/// Whether an option carrying one of those names is actually switched on. DuckDB's
/// binder applies three different rules to these four tokens, so this does too:
///
/// * `PARTITION_BY` — `partition_output = !partition_cols.empty()`, so an empty
///   column list leaves it off.
/// * `PER_THREAD_OUTPUT` — `GetBooleanArg`, which is
///   `arg.empty() || arg[0].CastAs(BOOLEAN).GetValue<bool>()`. It **casts**, so the
///   argument does not have to be the `false` keyword: see [`boolean_arg`].
/// * `FILE_SIZE_BYTES` and `ROW_GROUPS_PER_FILE` — neither is read as a boolean at
///   all. `rotate` is `file_size_bytes.IsValid() || row_groups_per_file.IsValid()`,
///   set from the option carrying any value, so presence is the whole test. Measured
///   on DuckDB v1.5.4 and v1.5.5: `FILE_SIZE_BYTES 0` writes a directory.
fn copy_option_is_on(name: &str, value: &CopyOptionValue) -> bool {
    if name.eq_ignore_ascii_case("PARTITION_BY") {
        *value != CopyOptionValue::EmptyList
    } else if name.eq_ignore_ascii_case("PER_THREAD_OUTPUT") {
        boolean_arg(value)
    } else {
        true
    }
}

/// DuckDB's `GetBooleanArg` for a `COPY` option: no argument is true, and otherwise
/// the argument is **cast** to BOOLEAN rather than compared against a keyword.
///
/// Recognising only the `false` keyword is what this replaced, and it was wrong in
/// the direction that never settles: `PER_THREAD_OUTPUT 0` and
/// `PER_THREAD_OUTPUT 'false'` each wrote a single file on DuckDB v1.5.4 and v1.5.5,
/// while a `Directory` classification would `read_dir` that file, get `None`, and
/// re-run the step on every run while warning that it produced nothing.
///
/// The string arm is `TryCastStringBool` with `strict = false`, which is what
/// `Value::CastAs` defaults to: `t`/`y`/`1`/`yes`/`true` and `f`/`n`/`0`/`no`/`false`,
/// case-insensitively. A string outside that set is a conversion error in DuckDB and
/// the statement writes nothing at all, so what this returns for it cannot be
/// observed on disk; it stays `true`, the answer that forces staleness rather than
/// certifying an artifact. A value that is not a literal is not something this can
/// evaluate, and stays `true` for the same reason.
fn boolean_arg(value: &CopyOptionValue) -> bool {
    match value {
        CopyOptionValue::Boolean(b) => *b,
        CopyOptionValue::Number(n) => n.parse::<f64>().map(|x| x != 0.0).unwrap_or(true),
        CopyOptionValue::String(s) => cast_string_to_bool(s).unwrap_or(true),
        CopyOptionValue::Absent | CopyOptionValue::EmptyList | CopyOptionValue::Other => true,
    }
}

/// `TryCastStringBool` with `strict = false`, from DuckDB's `cast_operators.hpp`.
/// `None` where DuckDB raises a conversion error.
fn cast_string_to_bool(s: &str) -> Option<bool> {
    match s.to_ascii_lowercase().as_str() {
        "t" | "y" | "1" | "yes" | "true" => Some(true),
        "f" | "n" | "0" | "no" | "false" => Some(false),
        _ => None,
    }
}

/// Names the reader that takes a SQL step's reads and produces. Unset or `duckdb`, DuckDB's
/// own parse is read, with sqlparser-rs reading each statement DuckDB's reading cannot;
/// `sqlparser` has sqlparser-rs read every statement, a way round a mapping bug in DuckDB's
/// reading while it is fixed.
pub(crate) const SQL_READER_ENV: &str = "ARC_SQL_READER";

/// The value of [`SQL_READER_ENV`] that chooses DuckDB's parse, as unset does.
const DUCKDB_READER: &str = "duckdb";

/// The value of [`SQL_READER_ENV`] that chooses sqlparser-rs for every statement.
const SQLPARSER_READER: &str = "sqlparser";

/// Which reader [`SQL_READER_ENV`] chooses.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Reader {
    DuckDb,
    Sqlparser,
}

fn reader() -> Result<Reader, String> {
    match std::env::var_os(SQL_READER_ENV) {
        None => Ok(Reader::DuckDb),
        Some(value) if value == DUCKDB_READER => Ok(Reader::DuckDb),
        Some(value) if value == SQLPARSER_READER => Ok(Reader::Sqlparser),
        Some(value) => Err(format!(
            "{SQL_READER_ENV} is '{}', and the values it takes are `{DUCKDB_READER}`, which \
             reads a SQL step with DuckDB's own parser as arc does with it unset, and \
             `{SQLPARSER_READER}`, which reads it with arc's own",
            value.to_string_lossy()
        )),
    }
}

/// Refuses a run, before a step runs, when [`SQL_READER_ENV`] names no reader. Whether the
/// DuckDB arc runs gives a parse is asked in the asset graph's own questions, which `arc run`
/// asks before its first step: a DuckDB 2 that gives no tree refuses each SQL step that holds
/// a statement as unreadable (see [`StepSql::unreadable`]), and starts no process of its own.
pub(crate) fn check_reader() -> Result<(), String> {
    reader().map(|_| ())
}

/// What each statement of a SQL step reads and produces, in source order, and the
/// statements no reader could read.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct StepSql {
    pub statements: Vec<SqlAssets>,
    /// Which reader read each statement, in the order of `statements`.
    pub read_by: Vec<ReadBy>,
    /// Each statement's `[start, end)` bytes in the step's SQL, in the order of
    /// `statements`: where DuckDB's tokens place it when DuckDB read the step, else where
    /// [`statement_byte_ranges`] cuts it. `None` for a statement that cannot be placed.
    pub byte_ranges: Vec<Option<(usize, usize)>>,
    /// Each statement neither reader could read, as it reads nothing; see [`unread_warning`].
    pub unread: Vec<UnreadStatement>,
    /// Why DuckDB's answer for the step could not be read, when a DuckDB 2 answered and
    /// sqlparser-rs read the whole step instead. `arc run` refuses such a step.
    pub unreadable: Option<String>,
}

/// Which reader read a statement, as the run contract records it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReadBy {
    /// DuckDB's reading.
    DuckDb,
    /// sqlparser-rs's: the reader chosen, or the one that read a statement DuckDB's reading
    /// could not take.
    Sqlparser,
    /// Neither: the statement reads nothing, and draws [`unread_warning`].
    Nothing,
}

impl ReadBy {
    /// The name the run contract gives it.
    pub fn as_str(self) -> &'static str {
        match self {
            ReadBy::DuckDb => "duckdb",
            ReadBy::Sqlparser => "sqlparser",
            ReadBy::Nothing => "nothing",
        }
    }
}

/// A statement of a SQL step that neither reader could read.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnreadStatement {
    /// Its place among the step's statements, counting from 1.
    pub statement: usize,
    /// The line of the file it starts on, counting from 1.
    pub line: usize,
}

impl StepSql {
    /// The step's statements as one set: what any of them produces, reads, holds as a
    /// `WITH` name or drops. A `WITH` name is left out of the reads of the statement it is
    /// declared in, and only of that one.
    pub fn merged(&self) -> SqlAssets {
        let mut assets = SqlAssets::default();
        for statement in &self.statements {
            merge(&mut assets, statement.clone());
        }
        assets
    }
}

fn merge(into: &mut SqlAssets, from: SqlAssets) {
    into.outputs.extend(from.outputs);
    into.inputs.extend(from.inputs);
    into.internal.extend(from.internal);
    into.destroys.extend(from.destroys);
    into.kinds.extend(from.kinds);
    into.unread_table_functions
        .extend(from.unread_table_functions);
}

/// The warning for a statement of `step`, in `file`, that no reader could read. Both
/// readers' paths give it in these words.
pub fn unread_warning(step: &str, file: &str, unread: &UnreadStatement) -> String {
    format!(
        "could not read statement {} of {step} ({file}:{}) \u{2014} treating it as reading \
         nothing",
        unread.statement, unread.line
    )
}

/// What each statement of `sql` reads and produces, by the reader [`SQL_READER_ENV`]
/// chooses. `Err` only when it names no reader.
///
/// With DuckDB's reader, a statement it cannot read is read by sqlparser-rs. When the
/// DuckDB arc runs does not answer, or is not a 2.x build, the whole step is read by
/// sqlparser-rs, and arc says so once in the process: `arc run` refuses before reading
/// (see [`check_reader`]), so this is a command that reads the graph without running it.
/// When a DuckDB 2 answered and its answer for this step could not be read, sqlparser-rs
/// reads the step and [`StepSql::unreadable`] says why; see [`from_duckdb`].
pub fn read_sql(sql: &str) -> Result<StepSql, String> {
    match reader()? {
        Reader::Sqlparser => Ok(sqlparser_step(sql)),
        #[cfg(test)]
        Reader::DuckDb if !test_reading::is_set() => Ok(sqlparser_step(sql)),
        Reader::DuckDb => Ok(from_duckdb(sql, duckdb_reading(sql))),
    }
}

/// DuckDB's reading of `sql`, from the DuckDB arc runs.
#[cfg(not(test))]
fn duckdb_reading(sql: &str) -> Result<duckdb_lineage::StepReading, duckdb_lineage::Refusal> {
    let program =
        crate::engine::duckdb_program().map_err(|e| duckdb_lineage::Refusal::NotStarted {
            program: std::env::var_os(crate::engine::DUCKDB_BIN_ENV).map_or_else(
                || "duckdb".to_string(),
                |bin| bin.to_string_lossy().into_owned(),
            ),
            reason: e.to_string(),
        })?;
    duckdb_lineage::read_step(&program, sql)
}

/// DuckDB's reading of `sql`, as the test running gives it with [`test_reading::with`].
#[cfg(test)]
fn duckdb_reading(sql: &str) -> Result<duckdb_lineage::StepReading, duckdb_lineage::Refusal> {
    test_reading::read(sql)
}

/// The reading the lib's tests give the default reader. A test that sets none reads with
/// sqlparser-rs, so no lib test starts the `duckdb` on PATH and each reads the same way on
/// every machine; a test that wants DuckDB's reading gives it from recorded answers.
#[cfg(test)]
pub(crate) mod test_reading {
    use std::cell::RefCell;

    use crate::duckdb_lineage::{Refusal, StepReading};

    type Read = Box<dyn Fn(&str) -> Result<StepReading, Refusal>>;

    thread_local! {
        static READING: RefCell<Option<Read>> = const { RefCell::new(None) };
    }

    pub(crate) fn is_set() -> bool {
        READING.with(|reading| reading.borrow().is_some())
    }

    pub(super) fn read(sql: &str) -> Result<StepReading, Refusal> {
        READING.with(|reading| (reading.borrow().as_ref().expect("a reading is set"))(sql))
    }

    /// Runs `body` with the default reader reading through `read`, on this thread.
    pub(crate) fn with<T>(
        read: impl Fn(&str) -> Result<StepReading, Refusal> + 'static,
        body: impl FnOnce() -> T,
    ) -> T {
        READING.with(|reading| *reading.borrow_mut() = Some(Box::new(read)));
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(body));
        READING.with(|reading| *reading.borrow_mut() = None);
        result.unwrap_or_else(|panic| std::panic::resume_unwind(panic))
    }

    /// DuckDB v2.0.0's reading of `sql`, from the recorded answers.
    pub(crate) fn recorded(sql: &str) -> Result<StepReading, Refusal> {
        Ok(crate::duckdb_lineage::recorded_steps::reading(sql))
    }
}

/// The step's reading from DuckDB's answer. A DuckDB that could not be asked, or is not a
/// 2.x build, leaves the step to sqlparser-rs, and arc says so once in the process; a DuckDB 2
/// whose answer for this step could not be read leaves it to sqlparser-rs too, with the
/// reason in [`StepSql::unreadable`], for the caller to name the step.
fn from_duckdb(
    sql: &str,
    reading: Result<duckdb_lineage::StepReading, duckdb_lineage::Refusal>,
) -> StepSql {
    use duckdb_lineage::Refusal;
    match reading {
        Ok(reading) => duckdb_step(&reading),
        Err(Refusal::Unreadable(reason)) => StepSql {
            unreadable: Some(reason),
            ..sqlparser_step(sql)
        },
        Err(
            no_duckdb @ (Refusal::NotStarted { .. }
            | Refusal::NotRun { .. }
            | Refusal::NotDuckDb2 { .. }),
        ) => {
            warn_once_no_duckdb(&no_duckdb);
            sqlparser_step(sql)
        }
    }
}

/// The warning for a step DuckDB's answer could not be read for, which sqlparser-rs read.
pub fn unreadable_warning(step: &str, file: &str, reason: &str) -> String {
    format!(
        "arc read {file} of step '{step}' with its own parser, because DuckDB's answer for it \
         could not be read: {reason}"
    )
}

/// The refusal of a run for a step DuckDB's answer could not be read for: arc run reads
/// each step with DuckDB's parse, and does not read one with its own in its place.
pub fn unreadable_refusal(step: &str, file: &str, reason: &str) -> String {
    format!(
        "step '{step}' ({file}): arc reads what each SQL step reads and produces from DuckDB \
         2's own parse, and DuckDB's answer for it could not be read: {reason}; \
         {SQL_READER_ENV}={SQLPARSER_READER} reads the steps with arc's own parser instead"
    )
}

/// Prints that the asset graph is sqlparser-rs's reading: once for each cause, and again
/// when the same cause outlasts [`crate::duckdb_tokens::NOT_STARTED_LIFETIME`], so a command
/// reading many files says it once and a long-lived `arc mcp` says it again when the cause
/// changes or persists. When DuckDB ran and failed, it is asked whether it tokenises at
/// all, and a DuckDB that does not, as one built without its autocomplete extension, is
/// named with the fix.
fn warn_once_no_duckdb(refusal: &duckdb_lineage::Refusal) {
    use owo_colors::OwoColorize;
    static SAID: std::sync::Mutex<Option<(String, std::time::Instant)>> =
        std::sync::Mutex::new(None);
    let cause = no_duckdb_cause(refusal);
    let now = std::time::Instant::now();
    {
        let mut said = SAID
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if !no_duckdb_warning_due(
            said.as_ref(),
            &cause,
            now,
            crate::duckdb_tokens::NOT_STARTED_LIFETIME,
        ) {
            return;
        }
        *said = Some((cause, now));
    }
    let warning = match refusal {
        duckdb_lineage::Refusal::NotRun { .. } => no_tokenizer_named(),
        _ => None,
    }
    .unwrap_or_else(|| no_duckdb_warning(&refusal.to_string()));
    eprintln!("{} {warning}", "warning:".yellow());
}

/// What a no-DuckDB warning is said once for: the kind of failure and the program, or the
/// version that answered. Not the reason, which can differ from file to file.
fn no_duckdb_cause(refusal: &duckdb_lineage::Refusal) -> String {
    use duckdb_lineage::Refusal;
    match refusal {
        Refusal::NotStarted { program, .. } => format!("not started: {program}"),
        Refusal::NotRun { program, .. } => format!("not run: {program}"),
        Refusal::NotDuckDb2 { version } => format!("version: {version}"),
        Refusal::Unreadable(_) => "unreadable".to_string(),
    }
}

/// Whether the no-DuckDB warning for `cause` is due at `now`, given the last one `said`
/// and when: when none was said, the cause changed, or `lifetime` has passed since.
fn no_duckdb_warning_due(
    said: Option<&(String, std::time::Instant)>,
    cause: &str,
    now: std::time::Instant,
    lifetime: std::time::Duration,
) -> bool {
    match said {
        None => true,
        Some((last, at)) => last != cause || now.saturating_duration_since(*at) >= lifetime,
    }
}

/// [`no_tokenizer_warning`] for the DuckDB arc runs, when its tokenizer does not answer.
#[cfg(not(test))]
fn no_tokenizer_named() -> Option<String> {
    let program = crate::engine::duckdb_program().ok()?;
    let error = crate::duckdb_tokens::tokenizer_failure(&program)?;
    let printed = crate::engine::printed_version(&program);
    Some(no_tokenizer_warning(
        &crate::engine::which_duckdb(&program, &printed),
        &error,
    ))
}

/// The lib's tests start no DuckDB, so none is found to lack a tokenizer.
#[cfg(test)]
fn no_tokenizer_named() -> Option<String> {
    None
}

/// The words of the warning for a DuckDB, named by `which`, whose tokenizer does not answer,
/// saying `error`.
pub(crate) fn no_tokenizer_warning(which: &str, error: &str) -> String {
    format!(
        "{} The asset graph is arc's own reading of the SQL instead. {}",
        crate::engine::no_tokenizer_cause(which, error),
        crate::engine::release_install_fix()
    )
}

/// The words of [`warn_once_no_duckdb`]'s warning.
pub(crate) fn no_duckdb_warning(reason: &str) -> String {
    format!(
        "the asset graph is arc's own reading of the SQL, because no DuckDB 2 answered \
         ({reason})"
    )
}

/// DuckDB's reading of a step, with sqlparser-rs reading each statement DuckDB's reading
/// cannot: one of a form the reader does not frame, one holding a query DuckDB gives no
/// tree for, and one DuckDB cannot parse. What DuckDB's words made certain is kept beside
/// sqlparser-rs's reading; a statement sqlparser-rs cannot parse either reads nothing else,
/// and is listed in [`StepSql::unread`].
fn duckdb_step(reading: &duckdb_lineage::StepReading) -> StepSql {
    let mut step = StepSql::default();
    for (n, statement) in reading.statements.iter().enumerate() {
        let mut assets = duckdb_statement(statement);
        let mut read_by = ReadBy::DuckDb;
        if matches!(
            statement.form,
            duckdb_lineage::Form::Unread(_) | duckdb_lineage::Form::NotParsed(_)
        ) {
            match sqlparser_per_statement(&statement.text) {
                Ok(read) => {
                    read.into_iter().for_each(|other| merge(&mut assets, other));
                    read_by = ReadBy::Sqlparser;
                }
                Err(_) => {
                    read_by = ReadBy::Nothing;
                    step.unread.push(UnreadStatement {
                        statement: n + 1,
                        line: statement.line,
                    });
                }
            }
        }
        step.statements.push(assets);
        step.read_by.push(read_by);
        step.byte_ranges.push(statement.byte_range);
    }
    step
}

/// [`statement_byte_ranges`] on each batch the DuckDB command line cuts from `sql`, mapped
/// back to `sql`'s bytes, so a dot command, a `#` line or any other line the command line
/// drops is in no range. `None` when a batch or a statement cannot be placed in `sql`.
fn batched_byte_ranges(sql: &str) -> Option<Vec<(usize, usize)>> {
    let mut ranges = Vec::new();
    for batch in crate::engine::cli_batches_mapped(sql.as_bytes()) {
        // A NUL can cut a character in two, leaving a batch that is not UTF-8.
        let text = String::from_utf8(batch.text).ok()?;
        for (start, end) in statement_byte_ranges(&text) {
            let placed = batch
                .origin
                .at(start)
                .zip(batch.origin.at(end - 1).map(|last| last + 1));
            ranges.push(placed?);
        }
    }
    Some(ranges)
}

/// The ranges [`batched_byte_ranges`] cuts in `sql`, one per statement of a step that has
/// `count`. A cut that counts statements differently from the reader gives none, rather than
/// ranges misaligned with the statements.
fn lexical_ranges(sql: &str, count: usize) -> Vec<Option<(usize, usize)>> {
    let ranges = batched_byte_ranges(sql).unwrap_or_default();
    if ranges.len() == count {
        ranges.into_iter().map(Some).collect()
    } else {
        vec![None; count]
    }
}

/// sqlparser-rs's reading of a step. A step it cannot parse whole is cut where
/// [`statement_byte_ranges`] cuts it and each statement is read alone, so a statement it
/// cannot parse reads nothing and the others are still read.
fn sqlparser_step(sql: &str) -> StepSql {
    if let Ok(statements) = sqlparser_per_statement(sql) {
        return StepSql {
            read_by: vec![ReadBy::Sqlparser; statements.len()],
            byte_ranges: lexical_ranges(sql, statements.len()),
            statements,
            ..StepSql::default()
        };
    }
    let mut step = StepSql::default();
    // A step the command line's batches cannot place is still cut where the plain lexical cut
    // cuts it, so no statement goes unread; its statements have no range.
    let (cut, placed) = match batched_byte_ranges(sql) {
        Some(ranges) => (ranges, true),
        None => (statement_byte_ranges(sql), false),
    };
    for (n, (start, end)) in cut.into_iter().enumerate() {
        step.byte_ranges.push(placed.then_some((start, end)));
        match sql
            .get(start..end)
            .ok_or_else(Vec::new)
            .and_then(sqlparser_per_statement)
        {
            Ok(read) => {
                let mut assets = SqlAssets::default();
                read.into_iter().for_each(|other| merge(&mut assets, other));
                step.statements.push(assets);
                step.read_by.push(ReadBy::Sqlparser);
            }
            Err(_) => {
                step.statements.push(SqlAssets::default());
                step.read_by.push(ReadBy::Nothing);
                step.unread.push(UnreadStatement {
                    statement: n + 1,
                    line: sql[..start].matches('\n').count() + 1,
                });
            }
        }
    }
    step
}

/// sqlparser-rs's reading of each statement of `sql`, or its error when it cannot parse
/// the whole of it. CTE names are filtered out of each statement's `inputs`.
fn sqlparser_per_statement(sql: &str) -> Result<Vec<SqlAssets>, Vec<String>> {
    let dialect = DuckDbDialect {};
    let statements = Parser::parse_sql(&dialect, sql).map_err(|e| vec![e.to_string()])?;

    let mut per_statement = Vec::with_capacity(statements.len());
    for stmt in &statements {
        let mut assets = SqlAssets::default();
        extract_from_statement(stmt, &mut assets);
        // Per-statement CTE filtering: a CTE reference is step-internal, not an input.
        let internal: Vec<String> = assets.internal.iter().cloned().collect();
        for cte_name in internal {
            assets.inputs.remove(&cte_name);
        }
        per_statement.push(assets);
    }
    Ok(per_statement)
}

/// One statement of DuckDB's reading, through the rules sqlparser's statements pass
/// through. A table is named by its last part, lowercased, as [`object_name_to_string`]
/// names one. A statement the reader could not read keeps what its words made certain.
fn duckdb_statement(statement: &duckdb_lineage::Statement) -> SqlAssets {
    let mut assets = SqlAssets::default();
    let options: Vec<(String, CopyOptionValue)> = statement
        .options
        .iter()
        .map(|option| (option.name.clone(), duckdb_copy_option(&option.value)))
        .collect();
    for produced in &statement.produces {
        match produced {
            Relation::Table(table) => {
                assets.record_output(table.name.to_lowercase(), AssetKind::Table)
            }
            // The one file a statement produces is a `COPY … TO` target.
            Relation::File(path) => {
                assets.record_output(path.clone(), copy_to_target_kind(&options))
            }
        }
    }
    for read in &statement.reads {
        match read {
            Relation::Table(table) => {
                assets.record_input(table.name.to_lowercase(), AssetKind::Table)
            }
            Relation::File(path) => assets.record_input(path.clone(), read_path_kind(path)),
        }
    }
    for call in &statement.calls {
        record_table_call(&duckdb_call(call), &mut assets);
    }
    // DuckDB's reading leaves a `WITH` name out of the reads of the statement it is in scope
    // in; it is the step's own, as sqlparser-rs's reading records it.
    assets.internal.extend(statement.with.iter().cloned());
    assets
}

/// A DuckDB `COPY` option's value, as [`copy_to_target_kind`] reads it.
fn duckdb_copy_option(value: &Option<ArgValue>) -> CopyOptionValue {
    match value {
        None => CopyOptionValue::Absent,
        Some(ArgValue::Identifier(word)) if word.eq_ignore_ascii_case("true") => {
            CopyOptionValue::Boolean(true)
        }
        Some(ArgValue::Identifier(word)) if word.eq_ignore_ascii_case("false") => {
            CopyOptionValue::Boolean(false)
        }
        Some(ArgValue::Number(n)) => CopyOptionValue::Number(n.clone()),
        Some(ArgValue::String(s)) => CopyOptionValue::String(s.clone()),
        Some(ArgValue::List(items)) if items.is_empty() => CopyOptionValue::EmptyList,
        Some(_) => CopyOptionValue::Other,
    }
}

/// A DuckDB table function call, as [`record_table_call`] reads it.
fn duckdb_call(call: &TableCall) -> FunctionCall {
    fn paths(value: &ArgValue, out: &mut Vec<String>) {
        match value {
            ArgValue::String(path) => out.push(path.clone()),
            ArgValue::List(items) => items.iter().for_each(|item| paths(item, out)),
            _ => {}
        }
    }
    fn names_something(value: &ArgValue) -> bool {
        match value {
            ArgValue::String(_) | ArgValue::QuotedIdentifier(_) => true,
            ArgValue::List(items) => items.iter().any(names_something),
            ArgValue::Expression { quotes, .. } => *quotes,
            // A query's strings are its own: they name nothing the call reads.
            ArgValue::Query | ArgValue::Number(_) | ArgValue::Identifier(_) => false,
        }
    }
    let mut unnamed_paths = Vec::new();
    for arg in call.args.iter().filter(|arg| arg.name.is_none()) {
        paths(&arg.value, &mut unnamed_paths);
    }
    FunctionCall {
        function: call.function.to_lowercase(),
        paths: unnamed_paths,
        names_something: call.args.iter().any(|arg| names_something(&arg.value)),
        reads_a_query: call
            .args
            .iter()
            .any(|arg| matches!(arg.value, ArgValue::Query)),
    }
}

/// A table function call, as either reader gives it: what [`record_table_call`] reads.
struct FunctionCall {
    /// Lowercased, without a schema.
    function: String,
    /// The strings among its unnamed arguments, and in the lists among them: the paths a
    /// file reader reads.
    paths: Vec<String>,
    /// Whether any argument, named or not, holds a string or a quoted identifier at any
    /// depth — the two forms in which a DuckDB call names a table, a file or a query
    /// (`"X"`, `'x.parquet'`, `'SELECT …'`). A call with neither (`recent()`,
    /// `range(10)`, `f(days := 7)`) names nothing arc could be missing.
    names_something: bool,
    /// Whether any argument, named or not, is a parenthesised query:
    /// `summary((SELECT src, dst FROM edges))`. The reader has read that query's tables
    /// as the statement's own, so the call's name is not one.
    reads_a_query: bool,
}

/// What a table function call reads, by the same rules whichever reader found it.
fn record_table_call(call: &FunctionCall, assets: &mut SqlAssets) {
    if is_file_reader(&call.function) || call.function == "glob" {
        // `read_parquet('x.parquet')` / `read_csv([...])` / `glob('data/*.csv')` etc: the
        // function reads a path — file contents, or, for `glob`, the filenames it lists —
        // so its path literal(s) are file/pattern inputs, not the opaque fn name.
        for path in &call.paths {
            assets.record_input(path.clone(), read_path_kind(path));
        }
    } else if reads_no_table(&call.function) {
        // DuckDB row-generators (`range(…)`, `generate_series(…)`) and catalog
        // introspection functions (`duckdb_functions()`, `duckdb_tables()`,
        // `duckdb_secrets()`) produce rows with no backing table at all — there is
        // nothing to record as an input, and none as the fn name either.
    } else if call.names_something {
        // Any other table-valued function whose arguments hold a string or a quoted
        // identifier (an extension's `mlpack_…_train("X", "Y", …)`, `query('…')`): those
        // arguments name tables, files or queries, and arc does not carry the signature
        // that says which. The call records no input — the function's own name is not a
        // table — and is reported as unread instead.
        assets.unread_table_functions.insert(call.function.clone());
    } else if call.reads_a_query {
        // A function called on a query (`onager_pth_dijkstra((SELECT src, dst, w FROM
        // edges))`): the query's tables are what the call reads, and the reader has
        // recorded them. The function's own name is not a table.
    } else {
        // A table macro (`recent()`), or a function called with numbers and bare names
        // only: record the name itself as the input, as before.
        assets.record_input(call.function.clone(), AssetKind::Table);
    }
}

/// The `[start, end)` byte offset of each top-level statement in `sql`, in source order.
///
/// A lexical splitter (not the AST) that walks the raw bytes so a renderer can slice
/// `sql` and show the exact source of each statement. It splits on top-level `;`,
/// skipping over single-quoted strings, double-quoted identifiers, `--` line comments,
/// `/* … */` block comments, and `$tag$ … $tag$` dollar-quoted bodies so a `;` inside
/// any of those never splits. Comment-only / whitespace-only segments are dropped, so
/// on well-formed SQL the count matches [`read_sql`]'s; the caller zips the
/// two and falls back to no ranges if they ever disagree. Ranges are trimmed of leading
/// whitespace/comments and trailing whitespace.
///
/// This is the fallback cut: [`read_sql`] places a statement by DuckDB's own tokens when
/// DuckDB answers, and otherwise cuts each batch of the command line with this and maps the
/// ranges back to the file ([`batched_byte_ranges`]).
pub fn statement_byte_ranges(sql: &str) -> Vec<(usize, usize)> {
    let bytes = sql.as_bytes();
    let n = bytes.len();
    let mut ranges = Vec::new();
    let mut seg_start = 0usize;
    let mut i = 0usize;

    while i < n {
        match bytes[i] {
            b'\'' => i = skip_string(bytes, i, b'\''),
            b'"' => i = skip_string(bytes, i, b'"'),
            b'-' if i + 1 < n && bytes[i + 1] == b'-' => i = skip_line_comment(bytes, i),
            b'/' if i + 1 < n && bytes[i + 1] == b'*' => i = skip_block_comment(bytes, i),
            b'$' => match skip_dollar_quote(bytes, i) {
                Some(j) => i = j,
                None => i += 1,
            },
            b';' => {
                // Include the terminating `;` in the range so a rendered slice reads as a
                // complete statement.
                if let Some(r) = trim_code_span(sql, seg_start, i + 1) {
                    ranges.push(r);
                }
                i += 1;
                seg_start = i;
            }
            _ => i += 1,
        }
    }
    // The tail after the last `;` (a file need not terminate its final statement).
    if let Some(r) = trim_code_span(sql, seg_start, n) {
        ranges.push(r);
    }
    ranges
}

/// Advance past a quoted string/identifier opened by `quote` at `open`. Handles the
/// doubled-delimiter escape (`''` / `""`). Returns the index just past the closing quote
/// (or the input end if unterminated).
fn skip_string(bytes: &[u8], open: usize, quote: u8) -> usize {
    let n = bytes.len();
    let mut i = open + 1;
    while i < n {
        if bytes[i] == quote {
            if i + 1 < n && bytes[i + 1] == quote {
                i += 2; // Escaped delimiter — stay inside the string.
            } else {
                return i + 1;
            }
        } else {
            i += 1;
        }
    }
    n
}

/// Advance past a `-- …` line comment. Returns the index just past the newline (or end).
fn skip_line_comment(bytes: &[u8], open: usize) -> usize {
    let n = bytes.len();
    let mut i = open + 2;
    while i < n && bytes[i] != b'\n' {
        i += 1;
    }
    if i < n { i + 1 } else { n }
}

/// Advance past a `/* … */` block comment. Returns the index just past `*/` (or end).
fn skip_block_comment(bytes: &[u8], open: usize) -> usize {
    let n = bytes.len();
    let mut i = open + 2;
    while i + 1 < n {
        if bytes[i] == b'*' && bytes[i + 1] == b'/' {
            return i + 2;
        }
        i += 1;
    }
    n
}

/// If a `$tag$` dollar-quote opens at `open`, return the index just past the matching
/// `$tag$` close (or the input end if unterminated). Returns `None` if `open` is not a
/// valid dollar-quote opener, so the caller treats `$` as an ordinary byte.
fn skip_dollar_quote(bytes: &[u8], open: usize) -> Option<usize> {
    let n = bytes.len();
    // Tag runs from just after the opening `$` to the next `$`; tags are [A-Za-z0-9_]*.
    let mut j = open + 1;
    while j < n && (bytes[j].is_ascii_alphanumeric() || bytes[j] == b'_') {
        j += 1;
    }
    if j >= n || bytes[j] != b'$' {
        return None; // Not a `$…$`-delimited opener.
    }
    let tag = &bytes[open..=j]; // The full `$tag$` delimiter, reused to find the close.
    let mut i = j + 1;
    while i < n {
        if bytes[i] == b'$' && bytes[i..].starts_with(tag) {
            return Some(i + tag.len());
        }
        i += 1;
    }
    Some(n)
}

/// Trim `[start, end)` to the code it contains: skip leading whitespace and comments,
/// then drop trailing ASCII whitespace. Returns `None` if the span is empty or made up
/// entirely of whitespace/comments (so blank or comment-only segments are not counted).
fn trim_code_span(sql: &str, start: usize, end: usize) -> Option<(usize, usize)> {
    let bytes = sql.as_bytes();
    let mut i = start;
    loop {
        while i < end && bytes[i].is_ascii_whitespace() {
            i += 1;
        }
        if i + 1 < end && bytes[i] == b'-' && bytes[i + 1] == b'-' {
            i = skip_line_comment(bytes, i).min(end);
            continue;
        }
        if i + 1 < end && bytes[i] == b'/' && bytes[i + 1] == b'*' {
            i = skip_block_comment(bytes, i).min(end);
            continue;
        }
        break;
    }
    if i >= end {
        return None; // Nothing but whitespace/comments.
    }
    let mut j = end;
    while j > i && bytes[j - 1].is_ascii_whitespace() {
        j -= 1;
    }
    Some((i, j))
}

/// Extract table names from a single SQL statement.
fn extract_from_statement(stmt: &Statement, assets: &mut SqlAssets) {
    match stmt {
        // CREATE TABLE foo (...)
        // CREATE TABLE foo AS SELECT ...
        Statement::CreateTable(create) => {
            let name = object_name_to_string(&create.name);
            assets.record_output(name, AssetKind::Table);

            // If it's a CTAS, the query's FROM tables are inputs.
            if let Some(ref query) = create.query {
                extract_inputs_from_query(query, assets);
            }
        }

        // CREATE VIEW foo AS SELECT ...
        Statement::CreateView { name, query, .. } => {
            assets.record_output(object_name_to_string(name), AssetKind::Table);
            extract_inputs_from_query(query, assets);
        }

        // INSERT INTO foo SELECT ...
        Statement::Insert(Insert { table, source, .. }) => {
            if let TableObject::TableName(name) = table {
                assets.record_output(object_name_to_string(name), AssetKind::Table);
            }
            if let Some(src) = source {
                extract_inputs_from_query(src.as_ref(), assets);
            }
        }

        // COPY foo TO 'file.csv'
        // COPY foo FROM 'file.csv'
        Statement::Copy {
            source,
            target,
            options,
            ..
        } => {
            match source {
                CopySource::Table { table_name, .. } => {
                    // COPY <table> ... — table is the source being read/written
                    match target {
                        CopyTarget::File { filename } => {
                            // COPY table TO 'file' — reading the table, producing the file.
                            // The file path is a first-class produced asset (file-path
                            // lineage). Whether that name is one file or a directory of
                            // files is decided by the COPY's own options — see
                            // `copy_to_target_kind` for the enumeration and where it came
                            // from — so it is known here rather than guessed later.
                            assets
                                .record_input(object_name_to_string(table_name), AssetKind::Table);
                            assets.record_output(
                                filename.clone(),
                                copy_to_target_kind(&sqlparser_copy_options(options)),
                            );
                        }
                        CopyTarget::Stdout => {
                            // COPY table TO STDOUT — reading from the table
                            assets
                                .record_input(object_name_to_string(table_name), AssetKind::Table);
                        }
                        CopyTarget::Stdin => {
                            // COPY table FROM STDIN — writing to the table
                            assets
                                .record_output(object_name_to_string(table_name), AssetKind::Table);
                        }
                        _ => {}
                    }
                }
                CopySource::Query(query) => {
                    // COPY (SELECT …) TO 'file' — the query reads its tables and the
                    // COPY produces the file, the same two facts `COPY <table> TO
                    // 'file'` records above, with the target classified the same way.
                    // The parser refuses `COPY (query) FROM`, so a query source is
                    // always a `TO`.
                    extract_inputs_from_query(query, assets);
                    if let CopyTarget::File { filename } = target {
                        assets.record_output(
                            filename.clone(),
                            copy_to_target_kind(&sqlparser_copy_options(options)),
                        );
                    }
                }
            }
        }

        // DROP TABLE/VIEW — destructive operation
        Statement::Drop { names, .. } => {
            for name in names {
                assets.destroys.insert(object_name_to_string(name));
            }
        }

        // ALTER TABLE — modifies the asset (output), does not read data from it
        Statement::AlterTable { name, .. } => {
            assets.record_output(object_name_to_string(name), AssetKind::Table);
        }

        // ALTER VIEW — modifies the view (output), new query reads from tables (inputs)
        Statement::AlterView { name, query, .. } => {
            assets.record_output(object_name_to_string(name), AssetKind::Table);
            extract_inputs_from_query(query, assets);
        }

        // MERGE INTO target USING source — target is written, source is read
        Statement::Merge { table, source, .. } => {
            // Target table → outputs
            if let TableFactor::Table { name, .. } = table {
                assets.record_output(object_name_to_string(name), AssetKind::Table);
            }
            // Source table → inputs
            extract_inputs_from_table_factor(source, assets);
        }

        // SELECT ... FROM — standalone select, extract inputs
        Statement::Query(query) => {
            extract_inputs_from_query(query, assets);
        }

        // All other statements — no asset extraction
        _ => {}
    }
}

/// Extract input table names from a query (SELECT ... FROM ... JOIN ...).
/// Also collects CTE names into `assets.internal`.
fn extract_inputs_from_query(query: &sqlparser::ast::Query, assets: &mut SqlAssets) {
    extract_inputs_from_set_expr(&query.body, assets);

    // Handle CTEs — they define local names, and their queries read from tables.
    // CTE names are captured in `internal` (step-internal assets).
    if let Some(with) = &query.with {
        for cte in &with.cte_tables {
            // Record the CTE name as an internal asset.
            assets.internal.insert(cte.alias.name.value.to_lowercase());
            // The CTE's body reads from tables — those are real inputs.
            extract_inputs_from_query(&cte.query, assets);
        }
    }
}

/// Recursively extract input table names from a set expression.
/// Handles SELECT, UNION/EXCEPT/INTERSECT, and nested queries.
fn extract_inputs_from_set_expr(set_expr: &sqlparser::ast::SetExpr, assets: &mut SqlAssets) {
    match set_expr {
        sqlparser::ast::SetExpr::Select(select) => {
            for table in &select.from {
                extract_inputs_from_table_factor(&table.relation, assets);
                for join in &table.joins {
                    extract_inputs_from_table_factor(&join.relation, assets);
                }
            }
        }
        sqlparser::ast::SetExpr::SetOperation { left, right, .. } => {
            extract_inputs_from_set_expr(left, assets);
            extract_inputs_from_set_expr(right, assets);
        }
        sqlparser::ast::SetExpr::Query(query) => {
            extract_inputs_from_query(query, assets);
        }
        // DuckDB statement-form PIVOT/UNPIVOT (`PIVOT t ON … USING …`): the source
        // relation being (un)pivoted is a real input, exactly as a FROM table is.
        // (Needs the vendored sqlparser fork; the SQL-standard `FROM t PIVOT (…)`
        // table-factor form is handled in `extract_inputs_from_table_factor`.)
        sqlparser::ast::SetExpr::Pivot(pivot) => {
            extract_inputs_from_table_factor(&pivot.source, assets);
        }
        sqlparser::ast::SetExpr::Unpivot(unpivot) => {
            extract_inputs_from_table_factor(&unpivot.source, assets);
        }
        // Values, Insert, Update, Table — no table references to extract.
        _ => {}
    }
}

/// Extract a table name from a table factor (FROM clause item).
fn extract_inputs_from_table_factor(factor: &TableFactor, assets: &mut SqlAssets) {
    match factor {
        // A bare table reference, or a table-valued function call (`args: Some`).
        TableFactor::Table { name, args, .. } => {
            let fn_name = object_name_to_string(name);
            match args {
                Some(table_args) => {
                    let mut paths = Vec::new();
                    let mut reads_a_query = false;
                    for arg in &table_args.args {
                        if let Some(query) = query_argument(arg) {
                            extract_inputs_from_query(query, assets);
                            reads_a_query = true;
                        } else if let FunctionArg::Unnamed(FunctionArgExpr::Expr(expr)) = arg {
                            path_literals(expr, &mut paths);
                        }
                    }
                    let call = FunctionCall {
                        function: fn_name,
                        paths,
                        names_something: args_name_something(table_args),
                        reads_a_query,
                    };
                    record_table_call(&call, assets);
                }
                None => assets.record_input(fn_name, AssetKind::Table),
            }
        }
        TableFactor::Derived { subquery, .. } => {
            extract_inputs_from_query(subquery, assets);
        }
        TableFactor::NestedJoin {
            table_with_joins, ..
        } => {
            extract_inputs_from_table_factor(&table_with_joins.relation, assets);
            for join in &table_with_joins.joins {
                extract_inputs_from_table_factor(&join.relation, assets);
            }
        }
        // PIVOT wraps a source table — extract the inner table as an input.
        TableFactor::Pivot { table, .. } => {
            extract_inputs_from_table_factor(table, assets);
        }
        // UNPIVOT wraps a source table — extract the inner table as an input.
        TableFactor::Unpivot { table, .. } => {
            extract_inputs_from_table_factor(table, assets);
        }
        // TableFunction, MatchRecognize, etc. — skip
        _ => {}
    }
}

/// Convert an ObjectName (potentially qualified: schema.table) to a simple string.
/// Uses the last identifier (the table name itself), lowercased for consistency.
fn object_name_to_string(name: &ObjectName) -> String {
    // ObjectName contains Vec<ObjectNamePart>; take the last part (table name).
    name.0
        .last()
        .and_then(|part| part.as_ident())
        .map(|ident| ident.value.to_lowercase())
        .unwrap_or_default()
}

/// Whether a table-valued function name is a DuckDB file reader whose first argument
/// is a path (or list of paths). Matched case-insensitively against the fn name that
/// [`object_name_to_string`] already lowercased.
fn is_file_reader(fn_name: &str) -> bool {
    const READERS: [&str; 17] = [
        "read_parquet",
        "parquet_scan",
        "read_csv",
        "read_csv_auto",
        "read_json",
        "read_json_auto",
        "read_json_objects",
        "read_ndjson",
        "read_ndjson_auto",
        "read_ndjson_objects",
        "read_text",
        "read_blob",
        "read_xlsx",
        "read_xml",
        "read_dta",
        "read_stat",
        "st_read",
    ];
    READERS.contains(&fn_name)
}

/// Whether a table-valued function name produces rows with no backing table to
/// record — a row-generator (`range`, `generate_series`) or a catalog/introspection
/// function (`duckdb_functions`, `duckdb_tables`, `duckdb_secrets`). Matched
/// case-insensitively against the fn name that [`object_name_to_string`] already
/// lowercased. `glob` is not here: it reads a path pattern, handled alongside
/// [`is_file_reader`] instead.
fn reads_no_table(fn_name: &str) -> bool {
    const NO_TABLE: [&str; 5] = [
        "range",
        "generate_series",
        "duckdb_functions",
        "duckdb_tables",
        "duckdb_secrets",
    ];
    NO_TABLE.contains(&fn_name)
}

/// Whether a table-valued function's arguments hold a string literal or a quoted
/// identifier, at any depth — the two forms in which a DuckDB call names a table, a file
/// or a query (`"X"`, `'x.parquet'`, `'SELECT …'`). A call with neither
/// (`recent()`, `range(10)`, `f(days := 7)`) names nothing arc could be missing.
///
/// A query among the arguments (see [`query_argument`]) is not walked: its strings are
/// its own (`WHERE kind = 'road'`), and its tables are read, not named.
fn args_name_something(table_args: &TableFunctionArgs) -> bool {
    table_args
        .args
        .iter()
        .filter(|arg| query_argument(arg).is_none())
        .any(arg_names_something)
}

fn arg_names_something(arg: &FunctionArg) -> bool {
    visit_expressions(arg, |expr| {
        let names = match expr {
            Expr::Value(v) => matches!(
                v.value,
                Value::SingleQuotedString(_)
                    | Value::DoubleQuotedString(_)
                    | Value::DollarQuotedString(_)
                    | Value::EscapedStringLiteral(_)
                    | Value::NationalStringLiteral(_)
            ),
            Expr::Identifier(ident) => ident.quote_style.is_some(),
            Expr::CompoundIdentifier(parts) => parts.iter().any(|p| p.quote_style.is_some()),
            _ => false,
        };
        if names {
            ControlFlow::Break(())
        } else {
            ControlFlow::Continue(())
        }
    })
    .is_break()
}

/// The query an argument is, when it is one: `(SELECT src, dst FROM edges)`, named or
/// not, and with any parentheses around it. A query nested deeper in an argument's
/// expression is not one: arc's own reader reads a select's `FROM` and its joins, and
/// not the subqueries an expression holds.
fn query_argument(arg: &FunctionArg) -> Option<&sqlparser::ast::Query> {
    fn query(expr: &Expr) -> Option<&sqlparser::ast::Query> {
        match expr {
            Expr::Subquery(query) => Some(query),
            Expr::Nested(inner) => query(inner),
            _ => None,
        }
    }
    match arg {
        FunctionArg::Named {
            arg: FunctionArgExpr::Expr(expr),
            ..
        }
        | FunctionArg::ExprNamed {
            arg: FunctionArgExpr::Expr(expr),
            ..
        }
        | FunctionArg::Unnamed(FunctionArgExpr::Expr(expr)) => query(expr),
        _ => None,
    }
}

/// The filesystem-path string literals of a file-reader argument expression.
///
/// Handles a single quoted path (`'x.parquet'`) and a bracketed/`ARRAY` list of them
/// (`['a.json', 'b.json']`) — DuckDB's multi-file glob form. Paths keep their original
/// case (filesystems are case-sensitive); everything else is ignored, so reader options
/// like `format => 'array'` never masquerade as inputs (they arrive as named args, which
/// the caller already skips, but a stray literal is harmless).
fn path_literals(expr: &Expr, paths: &mut Vec<String>) {
    match expr {
        Expr::Value(v) => {
            if let Value::SingleQuotedString(path) = &v.value {
                paths.push(path.clone());
            }
        }
        Expr::Array(array) => {
            for elem in &array.elem {
                path_literals(elem, paths);
            }
        }
        _ => {}
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::duckdb_lineage::recorded_steps;

    /// The no-DuckDB warning is said once for a cause, again when the cause changes, and
    /// again for the same cause once the lifetime has passed.
    #[test]
    fn the_no_duckdb_warning_repeats_on_a_new_cause_or_after_its_lifetime() {
        let lifetime = std::time::Duration::from_millis(30);
        let start = std::time::Instant::now();
        let said = ("not started: duckdb".to_string(), start);
        assert!(no_duckdb_warning_due(
            None,
            "not started: duckdb",
            start,
            lifetime
        ));
        assert!(!no_duckdb_warning_due(
            Some(&said),
            "not started: duckdb",
            start + lifetime / 2,
            lifetime
        ));
        assert!(no_duckdb_warning_due(
            Some(&said),
            "not run: duckdb",
            start + lifetime / 2,
            lifetime
        ));
        assert!(no_duckdb_warning_due(
            Some(&said),
            "not started: duckdb",
            start + lifetime,
            lifetime
        ));
        let not_started = |reason: &str| duckdb_lineage::Refusal::NotStarted {
            program: "duckdb".to_string(),
            reason: reason.to_string(),
        };
        assert_eq!(
            no_duckdb_cause(&not_started("No such file")),
            no_duckdb_cause(&not_started("Permission denied")),
            "the cause is the failure and the program, not the reason"
        );
    }

    /// What the rules make of DuckDB's reading of `sql`, from the answers DuckDB v2.0.0
    /// gave for it: what [`read_sql`] reads by default, before the fallback.
    fn read_by_duckdb(sql: &str) -> Vec<SqlAssets> {
        recorded_steps::read(sql)
            .iter()
            .map(duckdb_statement)
            .collect()
    }

    /// What sqlparser-rs reads in `sql`, as one set. The tests of the mapping read with it
    /// directly, so none of them starts a DuckDB.
    fn sqlparser_assets(sql: &str) -> Result<SqlAssets, Vec<String>> {
        let mut assets = SqlAssets::default();
        for statement in sqlparser_per_statement(sql)? {
            merge(&mut assets, statement);
        }
        Ok(assets)
    }

    fn read_by_sqlparser(sql: &str) -> Vec<SqlAssets> {
        sqlparser_per_statement(sql).unwrap_or_else(|e| panic!("sqlparser refused {sql:?}: {e:?}"))
    }

    fn set(names: &[&str]) -> BTreeSet<String> {
        names.iter().map(ToString::to_string).collect()
    }

    #[test]
    fn duckdb_reads_an_asof_join_sqlparser_cannot_parse() {
        let sql = recorded_steps::ASOF;
        assert!(
            sqlparser_per_statement(sql).is_err(),
            "sqlparser parses {sql:?} now; the step no longer shows what the switch is for"
        );
        let read = read_by_duckdb(sql);
        assert_eq!(read.len(), 1);
        assert_eq!(read[0].outputs, set(&["r"]), "{sql}");
        assert_eq!(read[0].inputs, set(&["t", "u"]), "{sql}");
        assert_eq!(read[0].kinds.get("r"), Some(&AssetKind::Table));
    }

    #[test]
    fn table_functions_are_read_by_the_same_rules_under_either_reader() {
        for sql in [
            recorded_steps::READ_XLSX,
            recorded_steps::RANGE,
            recorded_steps::MLPACK,
        ] {
            assert_eq!(read_by_duckdb(sql), read_by_sqlparser(sql), "{sql}");
        }
        let xlsx = &read_by_duckdb(recorded_steps::READ_XLSX)[0];
        assert_eq!(
            xlsx.inputs,
            set(&["build/budget.xlsx"]),
            "read_xlsx reads its file"
        );
        assert_eq!(xlsx.kinds.get("build/budget.xlsx"), Some(&AssetKind::File));
        let range = &read_by_duckdb(recorded_steps::RANGE)[0];
        assert_eq!(range.inputs, set(&[]), "range reads no table");
        assert_eq!(range.unread_table_functions, set(&[]));
        let mlpack = &read_by_duckdb(recorded_steps::MLPACK)[0];
        assert_eq!(
            mlpack.inputs,
            set(&[]),
            "the call names no table arc can read"
        );
        assert_eq!(
            mlpack.unread_table_functions,
            set(&["mlpack_random_forest_train"]),
            "the call is listed for the warning that asks for depends_on: and produces:"
        );
    }

    #[test]
    fn a_table_function_called_on_a_subquery_is_read_by_the_same_rules_under_either_reader() {
        let cases: [(&str, &[&str], &[&str]); 3] = [
            (recorded_steps::SUBQUERY_CALL, &["edges"], &[]),
            // A string inside the subquery is the subquery's own.
            (recorded_steps::SUBQUERY_WITH_STRING, &["edges"], &[]),
            // A string beside it still names something arc does not read.
            (
                recorded_steps::SUBQUERY_BESIDE_STRING,
                &["train"],
                &["my_ext_fit"],
            ),
        ];
        for (sql, reads, unread) in cases {
            let by_duckdb = read_by_duckdb(sql);
            assert_eq!(by_duckdb, read_by_sqlparser(sql), "{sql}");
            assert_eq!(by_duckdb[0].inputs, set(reads), "reads of {sql}");
            assert_eq!(
                by_duckdb[0].unread_table_functions,
                set(unread),
                "unread functions of {sql}"
            );
        }
    }

    #[test]
    fn a_copy_is_read_by_the_same_rules_under_either_reader() {
        for (sql, file, kind) in [
            (recorded_steps::COPY_QUERY, "out.csv", AssetKind::File),
            (
                recorded_steps::COPY_PARTITIONED,
                "parts",
                AssetKind::Directory,
            ),
            (
                recorded_steps::COPY_ONE_FILE_PER_THREAD_OFF,
                "one.parquet",
                AssetKind::File,
            ),
        ] {
            let read = read_by_duckdb(sql);
            assert_eq!(read, read_by_sqlparser(sql), "{sql}");
            assert_eq!(read[0].outputs, set(&[file]), "{sql}");
            assert_eq!(read[0].kinds.get(file), Some(&kind), "{sql}");
            assert_eq!(read[0].inputs, set(&["a"]), "{sql}");
        }
    }

    #[test]
    fn the_reader_mapping_cases_are_read_by_the_same_rules_under_either_reader() {
        use AssetKind::{File, Pattern, Table};
        // Each step, what it reads with the kind of each, and the table it makes.
        type Case = (
            &'static str,
            &'static [(&'static str, AssetKind)],
            &'static str,
        );
        let cases: [Case; 6] = [
            // Each path in a list is a file the reader reads.
            (
                recorded_steps::READ_JSON_LIST,
                &[("a.json", File), ("b.json", File)],
                "j",
            ),
            // A named argument's string is an option, not a path.
            (recorded_steps::READ_CSV_NAMED, &[("x.csv", File)], "c"),
            (
                recorded_steps::STRING_ARGUMENT,
                &[("data/in.parquet", File)],
                "p",
            ),
            // A call with no string or quoted name is a table macro, read under its name.
            (
                recorded_steps::EXPRESSION_ARGUMENT,
                &[("recent", Table)],
                "w",
            ),
            // A table is named in lower case, quoted or not.
            (
                recorded_steps::MIXED_CASE,
                &[("mytable", Table), ("other", Table)],
                "out",
            ),
            (recorded_steps::GLOB, &[("data/*.csv", Pattern)], "d"),
        ];
        for (sql, inputs, output) in cases {
            let read = read_by_duckdb(sql);
            assert_eq!(read, read_by_sqlparser(sql), "{sql}");
            assert_eq!(read.len(), 1, "{sql}");
            let names: Vec<&str> = inputs.iter().map(|(name, _)| *name).collect();
            assert_eq!(read[0].inputs, set(&names), "reads of {sql}");
            for (name, kind) in inputs {
                assert_eq!(read[0].kinds.get(*name), Some(kind), "{name} in {sql}");
            }
            assert_eq!(read[0].outputs, set(&[output]), "produces of {sql}");
            assert_eq!(read[0].unread_table_functions, set(&[]), "{sql}");
        }
    }

    /// DuckDB 2.0 reads a statement of a bare name as `FROM` that name; sqlparser-rs
    /// cannot parse it, so here DuckDB's reading is the one that holds.
    #[test]
    fn duckdb_reads_a_bare_name_as_a_read_of_that_table() {
        let sql = recorded_steps::BARE_NAME;
        assert!(
            sqlparser_per_statement(sql).is_err(),
            "sqlparser parses {sql:?} now; compare the two readings instead"
        );
        let read = read_by_duckdb(sql);
        assert_eq!(read.len(), 1);
        assert_eq!(read[0].inputs, set(&["tbl"]), "{sql}");
        assert_eq!(read[0].kinds.get("tbl"), Some(&AssetKind::Table));
        assert_eq!(read[0].outputs, set(&[]), "{sql}");
    }

    /// What the default reader makes of `sql`, from the answers DuckDB v2.0.0 gave for it:
    /// DuckDB's reading, with sqlparser-rs reading each statement it cannot.
    fn read_by_default(sql: &str) -> StepSql {
        duckdb_step(&recorded_steps::reading(sql))
    }

    #[test]
    fn a_step_duckdb_did_not_answer_keeps_the_lexical_ranges() {
        let sql = "SELECT 1;\nCREATE TABLE t AS SELECT 2;\n";
        let step = test_reading::with(
            |_| Err(duckdb_lineage::Refusal::Unreadable("no answer".to_string())),
            || read_sql(sql).unwrap(),
        );
        assert_eq!(step.unreadable.as_deref(), Some("no answer"));
        assert_eq!(
            step.byte_ranges,
            [Some((0, 9)), Some((10, 37))],
            "sqlparser-rs read the step, and the ranges are the lexical cut's"
        );
    }

    /// `ARC_SQL_READER=sqlparser` reads through `sqlparser_step`, as a lib test with no
    /// recorded reading does.
    #[test]
    fn the_sqlparser_reader_places_ranges_past_the_lines_the_command_line_drops() {
        let sql = ".mode list\n# note\nSELECT 1;\r\n.print x\nSELECT 2;\n";
        let step = read_sql(sql).unwrap();
        let slices: Vec<_> = step
            .byte_ranges
            .iter()
            .map(|r| r.map(|(start, end)| &sql[start..end]))
            .collect();
        assert_eq!(slices, [Some("SELECT 1;"), Some("SELECT 2;")]);
    }

    #[test]
    fn a_step_the_batches_cannot_place_is_still_read_and_has_no_ranges() {
        // A NUL ends what the command line keeps of its 99-byte chunk, which here ends
        // inside the `\u{e9}`: the second batch is not UTF-8 and cannot be placed.
        let sql = format!(
            "SELECT * FROM t;\nSELECT * FROM u;\n{}\0{}\u{e9};\n",
            "x".repeat(50),
            "y".repeat(47)
        );
        assert_eq!(batched_byte_ranges(&sql), None);
        let step = sqlparser_step(&sql);
        assert_eq!(step.statements.len(), 3, "{step:?}");
        assert!(step.statements[0].inputs.contains("t"));
        assert!(step.statements[1].inputs.contains("u"));
        assert_eq!(step.byte_ranges, [None, None, None]);
    }

    #[test]
    fn lexical_ranges_are_dropped_when_the_cut_counts_statements_differently() {
        assert_eq!(lexical_ranges("SELECT 1; SELECT 2;", 3), [None; 3]);
        assert_eq!(
            lexical_ranges("SELECT 1; SELECT 2;", 2),
            [Some((0, 9)), Some((10, 19))]
        );
    }

    #[test]
    fn duckdbs_range_for_a_statement_follows_its_tokens() {
        let sql = ".mode list\nSELECT * FROM a;\n";
        let step = read_by_default(sql);
        assert_eq!(
            step.byte_ranges,
            [Some((11, 27))],
            "DuckDB's tokens start the range at SELECT"
        );
        assert_eq!(
            statement_byte_ranges(sql),
            [(0, 27)],
            "the lexical cut of the whole file starts at the dot command"
        );
        assert_eq!(
            sqlparser_step(sql).byte_ranges,
            [Some((11, 27))],
            "the fallback cuts the command line's batches, so it does not"
        );
    }

    #[test]
    fn a_statement_duckdbs_reading_cannot_take_is_read_by_sqlparser() {
        let sql = "SET VARIABLE m = (SELECT max(d) FROM events);\n\
                   SET threads TO 4;\n\
                   FORCE INSTALL spatial;\n\
                   DELETE FROM t USING u WHERE t.k = u.k;\n\
                   CREATE TABLE r AS PIVOT t ON c USING sum(v);";
        let step = read_by_default(sql);
        assert_eq!(
            step.unread,
            [],
            "sqlparser-rs reads each statement DuckDB's cannot"
        );
        let by_sqlparser = read_by_sqlparser(sql);
        assert_eq!(
            step.statements[3], by_sqlparser[3],
            "the DELETE, which DuckDB's reading does not frame, is sqlparser-rs's reading"
        );
        assert_eq!(step.statements[4].outputs, set(&["r"]));
        assert_eq!(
            step.statements[4].inputs,
            set(&["t"]),
            "the PIVOT DuckDB gives no tree for reads its table, as sqlparser-rs reads it"
        );
    }

    #[test]
    fn a_statement_neither_reader_can_read_is_listed_with_its_line() {
        let sql = "SELECT 1;\nCREATE TABL r AS SELECT 1;";
        let step = read_by_default(sql);
        assert_eq!(
            step.unread,
            [UnreadStatement {
                statement: 2,
                line: 2
            }],
            "the statement DuckDB and sqlparser-rs both refuse"
        );
        assert_eq!(step.statements.len(), 2);
        assert_eq!(step.statements[1], SqlAssets::default(), "it reads nothing");
    }

    #[test]
    fn sqlparser_reads_each_statement_of_a_step_it_cannot_parse_whole() {
        let sql = "SELECT * FROM a;\n\n\
                   CREATE TABLE r AS SELECT * EXCLUDE (a) FROM t ASOF LEFT JOIN u USING (k);\n\
                   CREATE TABLE s AS SELECT * FROM b;";
        let step = sqlparser_step(sql);
        let reads: Vec<BTreeSet<String>> =
            step.statements.iter().map(|s| s.inputs.clone()).collect();
        assert_eq!(reads, [set(&["a"]), set(&[]), set(&["b"])]);
        assert_eq!(step.statements[2].outputs, set(&["s"]));
        assert_eq!(
            step.unread,
            [UnreadStatement {
                statement: 2,
                line: 3
            }],
            "the ASOF join sqlparser-rs cannot parse reads nothing and is listed"
        );
    }

    #[test]
    fn a_duckdb_that_cannot_be_asked_leaves_the_step_to_sqlparser() {
        use crate::duckdb_lineage::Refusal;
        let sql = "CREATE TABLE r AS SELECT * FROM t;";
        for refusal in [
            Refusal::NotStarted {
                program: "duckdb".to_string(),
                reason: "No such file".to_string(),
            },
            Refusal::NotRun {
                program: "duckdb".to_string(),
                reason: "exit 1".to_string(),
            },
            Refusal::NotDuckDb2 {
                version: "v1.5.6".to_string(),
            },
        ] {
            let step = from_duckdb(sql, Err(refusal.clone()));
            assert_eq!(step, sqlparser_step(sql), "{refusal:?}");
            assert_eq!(step.unreadable, None, "{refusal:?}: no DuckDB 2 answered");
        }
    }

    #[test]
    fn an_answer_that_cannot_be_read_is_named_for_the_step() {
        let sql = "CREATE TABLE r AS SELECT * FROM t;";
        let step = from_duckdb(
            sql,
            Err(crate::duckdb_lineage::Refusal::Unreadable(
                "binder: m".to_string(),
            )),
        );
        assert_eq!(step.unreadable.as_deref(), Some("binder: m"));
        assert_eq!(step.statements, sqlparser_step(sql).statements);
        assert_eq!(step.read_by, [ReadBy::Sqlparser]);
    }

    #[test]
    fn each_statement_says_which_reader_read_it() {
        let step = read_by_default(
            "SET VARIABLE m = (SELECT max(d) FROM events);\n\
             SET threads TO 4;\n\
             FORCE INSTALL spatial;\n\
             DELETE FROM t USING u WHERE t.k = u.k;\n\
             CREATE TABLE r AS PIVOT t ON c USING sum(v);",
        );
        use ReadBy::{DuckDb, Sqlparser};
        assert_eq!(step.read_by, [DuckDb, DuckDb, DuckDb, Sqlparser, Sqlparser]);
        let step = read_by_default("SELECT 1;\nCREATE TABL r AS SELECT 1;");
        assert_eq!(step.read_by, [DuckDb, ReadBy::Nothing]);
    }

    #[test]
    fn the_unread_statement_warning_names_the_step_the_file_and_the_line() {
        let unread = UnreadStatement {
            statement: 2,
            line: 7,
        };
        assert_eq!(
            unread_warning("load", "models/load.sql", &unread),
            "could not read statement 2 of load (models/load.sql:7) \u{2014} treating it as \
             reading nothing"
        );
    }

    #[test]
    fn the_no_tokenizer_warning_names_the_cause_and_the_fix() {
        assert_eq!(
            no_tokenizer_warning(
                "the duckdb on PATH, `/usr/bin/duckdb`, which prints `v2.0.0`,",
                "Catalog Error: Table Function with name sql_tokenize does not exist!"
            ),
            "the duckdb on PATH, `/usr/bin/duckdb`, which prints `v2.0.0`, cannot read SQL into tokens, because \
             it was built without DuckDB's autocomplete extension. arc needs those tokens to \
             check what each SQL step installs. The asset graph is arc's own reading of the SQL \
             instead. Install DuckDB's release build with DuckDB's own installer, pinned to the \
             version arc is tested on: `curl https://install.duckdb.org | DUCKDB_VERSION=2.0.0 \
             sh`, which installs it to ~/.duckdb/cli/2.0.0. Then put ~/.duckdb/cli/2.0.0 first \
             on PATH, or set ARC_DUCKDB_BIN=~/.duckdb/cli/2.0.0/duckdb."
        );
    }

    #[test]
    fn the_no_duckdb_warning_says_whose_reading_the_graph_is() {
        assert_eq!(
            no_duckdb_warning("no such file"),
            "the asset graph is arc's own reading of the SQL, because no DuckDB 2 answered \
             (no such file)"
        );
    }

    // DuckDB reads `1_000` as 1000, which casts to true: measured on v1.5.5,
    // `PER_THREAD_OUTPUT 1_000` wrote a directory, and DuckDB 2.0's tokenizer gives
    // `1_000` as one number. Rust's `f64` parse refuses the separator, so the number is
    // one the rule cannot read, and a value it cannot read leaves the option on.
    #[test]
    fn a_number_the_rule_cannot_read_leaves_per_thread_output_on() {
        let value = duckdb_copy_option(&Some(ArgValue::Number("1_000".to_string())));
        assert_eq!(value, CopyOptionValue::Number("1_000".to_string()));
        assert_eq!(
            copy_to_target_kind(&[("PER_THREAD_OUTPUT".to_string(), value)]),
            AssetKind::Directory,
            "PER_THREAD_OUTPUT 1_000 writes a directory"
        );
    }

    #[test]
    fn a_pivot_joined_to_another_table_is_not_read_as_its_first() {
        let read = read_by_duckdb(recorded_steps::PIVOT_JOIN);
        assert_eq!(read.len(), 1);
        assert_ne!(
            read[0].inputs,
            set(&["t"]),
            "the pivot reads the rows of t and u, and is read as reading t alone"
        );
        assert_eq!(read[0].inputs, set(&[]), "the statement is unread");
    }

    // CREATE TABLE is discovered as an output.
    #[test]
    fn test_create_table_output() {
        let sql = "CREATE TABLE customers (id INT, name TEXT);";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.outputs.contains("customers"));
        assert!(assets.inputs.is_empty());
    }

    // CREATE VIEW is discovered as an output.
    #[test]
    fn test_create_view_output() {
        let sql = "CREATE VIEW active_customers AS SELECT * FROM customers WHERE active = true;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.outputs.contains("active_customers"));
        assert!(assets.inputs.contains("customers"));
    }

    // CREATE TABLE AS SELECT (CTAS) discovers both output and inputs.
    #[test]
    fn test_ctas_output_and_inputs() {
        let sql = "CREATE TABLE summary AS SELECT count(*) AS total FROM orders;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.outputs.contains("summary"));
        assert!(assets.inputs.contains("orders"));
    }

    // Multiple DDL statements in one file.
    #[test]
    fn test_multiple_creates() {
        let sql = "CREATE TABLE foo (id INT);\nCREATE TABLE bar (id INT);\nCREATE VIEW baz AS SELECT * FROM foo;";
        let assets = sqlparser_assets(sql).unwrap();
        assert_eq!(
            assets.outputs,
            BTreeSet::from(["foo".into(), "bar".into(), "baz".into()])
        );
        assert!(assets.inputs.contains("foo"));
    }

    // FROM clause tables are discovered as inputs.
    #[test]
    fn test_from_clause_inputs() {
        let sql = "SELECT * FROM customers;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.inputs.contains("customers"));
        assert!(assets.outputs.is_empty());
    }

    // JOIN tables are discovered as inputs.
    #[test]
    fn test_join_inputs() {
        let sql = "SELECT c.name, o.total FROM customers c JOIN orders o ON c.id = o.customer_id;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.inputs.contains("customers"));
        assert!(assets.inputs.contains("orders"));
    }

    // Subqueries in FROM clause.
    #[test]
    fn test_subquery_inputs() {
        let sql = "SELECT * FROM (SELECT * FROM raw_data) sub;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.inputs.contains("raw_data"));
    }

    // INSERT INTO is discovered as an output.
    #[test]
    fn test_insert_into_output() {
        let sql = "INSERT INTO summary SELECT count(*) FROM customers;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.outputs.contains("summary"));
        assert!(assets.inputs.contains("customers"));
    }

    // COPY TO reads from a table (input).
    #[test]
    fn test_copy_to_file() {
        let sql = "COPY customers TO 'customers.csv';";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.inputs.contains("customers"));
    }

    // Unparseable SQL returns an error (caller treats as opaque).
    #[test]
    fn test_unparseable_sql() {
        let sql = "THIS IS NOT VALID SQL AT ALL %%%";
        let result = sqlparser_assets(sql);
        assert!(result.is_err());
    }

    // UNION ALL discovers inputs from both branches.
    #[test]
    fn test_union_all_inputs() {
        let sql = "SELECT * FROM customers UNION ALL SELECT * FROM archived_customers;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.inputs.contains("customers"));
        assert!(assets.inputs.contains("archived_customers"));
    }

    // CTAS with UNION discovers output and all inputs.
    #[test]
    fn test_ctas_union_inputs() {
        let sql = "CREATE TABLE all_customers AS SELECT * FROM customers UNION ALL SELECT * FROM archived_customers;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.outputs.contains("all_customers"));
        assert!(assets.inputs.contains("customers"));
        assert!(assets.inputs.contains("archived_customers"));
    }

    // EXCEPT discovers inputs from both sides.
    #[test]
    fn test_except_inputs() {
        let sql = "SELECT id FROM customers EXCEPT SELECT id FROM blocklist;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.inputs.contains("customers"));
        assert!(assets.inputs.contains("blocklist"));
    }

    // CTE names go to internal, not inputs.
    #[test]
    fn test_cte_internal_not_inputs() {
        let sql =
            "WITH recent AS (SELECT * FROM orders WHERE date > '2026-01-01') SELECT * FROM recent;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.inputs.contains("orders"),
            "real table should be in inputs"
        );
        assert!(
            !assets.inputs.contains("recent"),
            "CTE name should NOT be in inputs"
        );
        assert!(
            assets.internal.contains("recent"),
            "CTE name should be in internal"
        );
    }

    // Edge case: Qualified table names use the last component.
    #[test]
    fn test_qualified_name() {
        let sql = "CREATE TABLE main.customers (id INT);";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.outputs.contains("customers"));
    }

    // Edge case: Empty SQL produces empty assets.
    #[test]
    fn test_empty_sql() {
        // sqlparser may reject empty input, so use a comment-only file
        let sql = "-- just a comment";
        // This may either parse as empty or error — both are acceptable
        let result = sqlparser_assets(sql);
        // An Err is also acceptable — comment-only input is treated as opaque.
        if let Ok(assets) = result {
            assert!(assets.outputs.is_empty());
            assert!(assets.inputs.is_empty());
        }
    }

    // Nested CTEs — both captured in internal.
    #[test]
    fn test_nested_ctes_in_internal() {
        let sql = "WITH a AS (SELECT * FROM raw_data), b AS (SELECT * FROM a) SELECT * FROM b;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.internal.contains("a"),
            "CTE 'a' should be in internal"
        );
        assert!(
            assets.internal.contains("b"),
            "CTE 'b' should be in internal"
        );
        assert!(
            assets.inputs.contains("raw_data"),
            "real table should be in inputs"
        );
        assert!(
            !assets.inputs.contains("a"),
            "CTE 'a' should NOT be in inputs"
        );
        assert!(
            !assets.inputs.contains("b"),
            "CTE 'b' should NOT be in inputs"
        );
    }

    // CTE name shadowing a real table — CTE goes to internal, real table stays in inputs.
    #[test]
    fn test_cte_shadows_real_table() {
        let sql = "WITH customers AS (SELECT * FROM raw_customers) SELECT * FROM customers;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.internal.contains("customers"),
            "CTE 'customers' should be in internal"
        );
        assert!(
            assets.inputs.contains("raw_customers"),
            "real table should be in inputs"
        );
        assert!(
            !assets.inputs.contains("customers"),
            "CTE 'customers' should NOT be in inputs"
        );
    }

    // DROP TABLE populates destroys.
    #[test]
    fn test_drop_table_destroys() {
        let sql = "DROP TABLE foo;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.destroys.contains("foo"),
            "dropped table should be in destroys"
        );
        assert!(assets.outputs.is_empty(), "drop should not add to outputs");
        assert!(assets.inputs.is_empty(), "drop should not add to inputs");
    }

    // DROP VIEW also populates destroys.
    #[test]
    fn test_drop_view_destroys() {
        let sql = "DROP VIEW IF EXISTS my_view;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.destroys.contains("my_view"),
            "dropped view should be in destroys"
        );
    }

    // DROP + CREATE in same file — both destroys and outputs populated.
    #[test]
    fn test_drop_then_create() {
        let sql = "DROP TABLE IF EXISTS foo; CREATE TABLE foo AS SELECT * FROM bar;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.destroys.contains("foo"),
            "dropped table should be in destroys"
        );
        assert!(
            assets.outputs.contains("foo"),
            "created table should be in outputs"
        );
        assert!(
            assets.inputs.contains("bar"),
            "source table should be in inputs"
        );
    }

    // ALTER TABLE populates outputs only.
    #[test]
    fn test_alter_table_outputs_only() {
        let sql = "ALTER TABLE customers ADD COLUMN email TEXT;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.outputs.contains("customers"),
            "altered table should be in outputs"
        );
        assert!(assets.inputs.is_empty(), "alter should not add to inputs");
    }

    // MERGE INTO — target in outputs, source in inputs.
    #[test]
    fn test_merge_into() {
        let sql = "MERGE INTO target USING source ON target.id = source.id WHEN MATCHED THEN UPDATE SET target.name = source.name;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.outputs.contains("target"),
            "merge target should be in outputs"
        );
        assert!(
            assets.inputs.contains("source"),
            "merge source should be in inputs"
        );
    }

    // CREATE OR REPLACE TABLE is handled as output.
    #[test]
    fn test_create_or_replace() {
        let sql = "CREATE OR REPLACE TABLE foo AS SELECT * FROM bar;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.outputs.contains("foo"),
            "replaced table should be in outputs"
        );
        assert!(
            assets.inputs.contains("bar"),
            "source table should be in inputs"
        );
    }

    // PIVOT source table is extracted as input.
    #[test]
    fn test_pivot_source_table() {
        // sqlparser-rs 0.55 supports PIVOT syntax
        let sql =
            "SELECT * FROM monthly_sales PIVOT (SUM(amount) FOR month IN ('Jan', 'Feb', 'Mar'));";
        let result = sqlparser_assets(sql);
        match result {
            Ok(assets) => {
                assert!(
                    assets.inputs.contains("monthly_sales"),
                    "pivot source should be in inputs"
                );
            }
            Err(_) => {
                // If sqlparser doesn't support this syntax, graceful degradation is acceptable
            }
        }
    }

    // UNPIVOT source table is extracted as input.
    #[test]
    fn test_unpivot_source_table() {
        let sql = "SELECT * FROM quarterly_report UNPIVOT (value FOR quarter IN (q1, q2, q3, q4));";
        let result = sqlparser_assets(sql);
        match result {
            Ok(assets) => {
                assert!(
                    assets.inputs.contains("quarterly_report"),
                    "unpivot source should be in inputs"
                );
            }
            Err(_) => {
                // If sqlparser doesn't support this syntax, graceful degradation is acceptable
            }
        }
    }

    // Edge case: Recursive CTE — self-reference within CTE body.
    #[test]
    fn test_recursive_cte() {
        let sql = "WITH RECURSIVE tree AS (SELECT id, parent_id FROM nodes WHERE parent_id IS NULL UNION ALL SELECT n.id, n.parent_id FROM nodes n JOIN tree t ON n.parent_id = t.id) SELECT * FROM tree;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.internal.contains("tree"),
            "recursive CTE should be in internal"
        );
        assert!(
            assets.inputs.contains("nodes"),
            "real table should be in inputs"
        );
        assert!(
            !assets.inputs.contains("tree"),
            "CTE should NOT be in inputs"
        );
    }

    // Edge case: CTE with subquery — inner subquery tables discovered.
    #[test]
    fn test_cte_with_subquery() {
        let sql = "WITH a AS (SELECT * FROM (SELECT * FROM raw) sub) SELECT * FROM a;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.internal.contains("a"), "CTE should be in internal");
        assert!(
            assets.inputs.contains("raw"),
            "subquery source should be in inputs"
        );
        assert!(!assets.inputs.contains("a"), "CTE should NOT be in inputs");
    }

    // Edge case: ALTER VIEW — modifies view (output), reads from tables (inputs).
    #[test]
    fn test_alter_view_outputs_and_inputs() {
        let sql = "ALTER VIEW active_customers AS SELECT * FROM customers WHERE active = true;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.outputs.contains("active_customers"),
            "altered view should be in outputs"
        );
        assert!(
            assets.inputs.contains("customers"),
            "source table should be in inputs"
        );
    }

    // Edge case: DROP multiple tables in one statement.
    #[test]
    fn test_drop_multiple_tables() {
        let sql = "DROP TABLE foo, bar, baz;";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.destroys.contains("foo"));
        assert!(assets.destroys.contains("bar"));
        assert!(assets.destroys.contains("baz"));
    }

    // read_parquet('path') contributes the *file path* as input, not the fn name.
    #[test]
    fn test_read_parquet_lifts_path() {
        let sql = "CREATE TABLE t AS SELECT * FROM read_parquet('build/edgar.parquet');";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.inputs.contains("build/edgar.parquet"),
            "path is the input"
        );
        assert!(
            !assets.inputs.contains("read_parquet"),
            "fn name is not an input"
        );
        assert!(assets.outputs.contains("t"));
    }

    // read_csv keeps original case in the path (filesystems are case-sensitive).
    #[test]
    fn test_read_csv_preserves_case() {
        let sql = "SELECT * FROM read_csv('Data/Raw/GLEIF.csv');";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.inputs.contains("Data/Raw/GLEIF.csv"),
            "path case preserved"
        );
        assert!(!assets.inputs.contains("read_csv"));
    }

    // read_json over a list of files lifts every path; named options are ignored.
    #[test]
    fn test_read_json_list_and_options() {
        let sql = "CREATE TABLE brew AS SELECT * FROM read_json(['a/30d.json', 'a/90d.json'], format = 'array');";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.inputs.contains("a/30d.json"), "first path lifted");
        assert!(assets.inputs.contains("a/90d.json"), "second path lifted");
        assert!(
            !assets.inputs.contains("read_json"),
            "fn name is not an input"
        );
        assert!(
            !assets.inputs.contains("array"),
            "option value is not an input"
        );
        assert!(assets.outputs.contains("brew"));
    }

    // read_xlsx, read_xml, read_dta, read_stat and ST_Read lift the path they read,
    // exactly like read_parquet/read_csv/read_json — not the opaque fn name.
    #[test]
    fn test_five_new_readers_lift_path_not_function_name() {
        for (reader, path) in [
            ("read_xlsx", "build/budget.xlsx"),
            ("read_xml", "build/catalog.xml"),
            ("read_dta", "build/survey.dta"),
            ("read_stat", "build/survey.sas7bdat"),
            ("ST_Read", "data/shapes.shp"),
        ] {
            let sql = format!("CREATE TABLE r AS SELECT * FROM {reader}('{path}');");
            let assets = sqlparser_assets(&sql).unwrap();
            assert!(
                assets.inputs.contains(path),
                "{reader}: path {path} should be the input, got {:?}",
                assets.inputs
            );
            assert!(
                !assets.inputs.contains(&reader.to_lowercase()),
                "{reader}: fn name must not be recorded as a table input"
            );
            assert_eq!(assets.kinds.get(path), Some(&AssetKind::File));
        }
    }

    // A glob-shaped path to one of the five new readers is a Pattern, same as it
    // already is for the twelve existing readers.
    #[test]
    fn test_five_new_readers_glob_path_is_a_pattern() {
        for (reader, pattern) in [
            ("read_xlsx", "build/*.xlsx"),
            ("read_xml", "build/report-?.xml"),
            ("ST_Read", "data/[a-z].shp"),
        ] {
            let sql = format!("CREATE TABLE r AS SELECT * FROM {reader}('{pattern}');");
            let assets = sqlparser_assets(&sql).unwrap();
            assert_eq!(
                assets.kinds.get(pattern),
                Some(&AssetKind::Pattern),
                "{reader}: {pattern} should be a Pattern"
            );
        }
    }

    // range, generate_series and DuckDB's catalog-introspection table functions
    // produce rows from no backing table at all: nothing is recorded as an input,
    // and the fn name is not recorded as a table either.
    #[test]
    fn test_row_generators_and_catalog_functions_read_no_table() {
        for call in [
            "range(3)",
            "generate_series(1, 5)",
            "duckdb_functions()",
            "duckdb_tables()",
            "duckdb_secrets()",
        ] {
            let sql = format!("CREATE TABLE c AS SELECT * FROM {call};");
            let assets = sqlparser_assets(&sql).unwrap();
            assert!(
                assets.inputs.is_empty(),
                "{call}: no table should be read, got {:?}",
                assets.inputs
            );
        }
    }

    // glob('pattern') lifts the pattern itself as an input, not a table named `glob`.
    #[test]
    fn test_glob_function_lifts_pattern_not_table_name() {
        let sql = "CREATE TABLE c AS SELECT * FROM glob('data/*.csv');";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.inputs.contains("data/*.csv"), "pattern is the input");
        assert!(!assets.inputs.contains("glob"), "fn name is not an input");
        assert_eq!(assets.kinds.get("data/*.csv"), Some(&AssetKind::Pattern));
    }

    // An extension-supplied table function that reads tables named in its own
    // arguments (`"X"` is a quoted identifier) records no input — not even one under its
    // own name, which is not a table — and is listed as unread, so the step can be
    // asked for what it reads and writes.
    #[test]
    fn test_table_function_with_quoted_identifier_args_is_unread_not_recorded() {
        let sql = r#"CREATE TABLE fitted AS SELECT * FROM mlpack_random_forest_train("X", "Y", "params", "model");"#;
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            !assets.inputs.contains("mlpack_random_forest_train"),
            "the function's name is not a table, got {:?}",
            assets.inputs
        );
        assert!(
            assets.inputs.is_empty(),
            "no input at all: {:?}",
            assets.inputs
        );
        assert_eq!(
            assets.unread_table_functions,
            BTreeSet::from(["mlpack_random_forest_train".to_string()])
        );
        assert!(
            assets.outputs.contains("fitted"),
            "the CTAS target is still produced"
        );
    }

    // The other form a call names something in: a string. Each of these is a table
    // function arc has no signature for, called with a string at some depth — a plain
    // argument, a list element, a named argument's value, a schema-qualified call.
    #[test]
    fn test_table_function_with_a_string_arg_is_unread() {
        for sql in [
            "SELECT * FROM query('SELECT 1');",
            "SELECT * FROM my_ext_scan(['a', 'b']);",
            "SELECT * FROM my_ext_scan(3, source := 'orders');",
            "SELECT * FROM my_ext.my_ext_scan('orders');",
        ] {
            let assets = sqlparser_assets(sql).unwrap();
            assert!(
                assets.inputs.is_empty(),
                "{sql}: records no input, got {:?}",
                assets.inputs
            );
            assert_eq!(
                assets.unread_table_functions.len(),
                1,
                "{sql}: one function is unread, got {:?}",
                assets.unread_table_functions
            );
        }
    }

    // A function called twice is listed once; two functions are listed twice.
    #[test]
    fn test_unread_table_functions_are_a_set_of_names() {
        let sql = "SELECT * FROM my_scan('a'); SELECT * FROM my_scan('b'); SELECT * FROM other_scan('c');";
        let assets = sqlparser_assets(sql).unwrap();
        assert_eq!(
            assets.unread_table_functions,
            BTreeSet::from(["my_scan".to_string(), "other_scan".to_string()])
        );
    }

    // A call with no string and no quoted identifier names nothing arc could be missing:
    // it is not listed as unread. `recent()` and `recent(days := 7)` (a table macro)
    // keep recording their own name; `range(10)` records nothing, as before. A call on a
    // subquery records what the subquery reads instead; see the tests for it below.
    #[test]
    fn test_table_function_with_no_string_or_quoted_identifier_is_not_unread() {
        for (sql, records) in [
            ("SELECT * FROM range(10);", None),
            ("SELECT * FROM recent();", Some("recent")),
            (
                "SELECT * FROM recent(days := 7, strict := true);",
                Some("recent"),
            ),
            (
                "SELECT * FROM my_ext_scan(10, 2.5, x);",
                Some("my_ext_scan"),
            ),
            // An argument that is not a leaf but holds no string: a list of numbers, a
            // sum, a negation. Only a string or a quoted identifier names anything.
            ("SELECT * FROM my_ext_scan([1, 2]);", Some("my_ext_scan")),
            ("SELECT * FROM my_ext_scan(1 + 2, -3);", Some("my_ext_scan")),
        ] {
            let assets = sqlparser_assets(sql).unwrap();
            assert!(
                assets.unread_table_functions.is_empty(),
                "{sql}: not unread, got {:?}",
                assets.unread_table_functions
            );
            assert_eq!(
                assets.inputs,
                records
                    .map(str::to_string)
                    .into_iter()
                    .collect::<BTreeSet<_>>(),
                "{sql}: inputs"
            );
        }
    }

    // A table function called on a subquery reads the tables the subquery reads, and its
    // own name is not one — whether the function is an extension's, DuckDB's own, or a
    // row generator. The subquery may be named, parenthesised again, or nest another in
    // its `FROM`.
    #[test]
    fn test_table_function_called_on_a_subquery_reads_the_subquerys_tables() {
        for (sql, reads) in [
            (
                "CREATE TABLE paths AS SELECT * FROM onager_pth_dijkstra((SELECT src, dst, w FROM edges));",
                &["edges"][..],
            ),
            (
                "CREATE TABLE r AS SELECT * FROM onager_pth_dijkstra((SELECT src, dst, w FROM (SELECT * FROM edges) e));",
                &["edges"],
            ),
            (
                "CREATE TABLE r AS SELECT * FROM summary((SELECT e.src FROM edges e JOIN nodes n ON n.id = e.src));",
                &["edges", "nodes"],
            ),
            (
                "CREATE TABLE r AS SELECT * FROM range((SELECT count(*) FROM edges));",
                &["edges"],
            ),
            (
                "CREATE TABLE r AS SELECT * FROM my_ext_scan(source := (SELECT src FROM edges));",
                &["edges"],
            ),
            (
                "CREATE TABLE r AS SELECT * FROM my_ext_scan(((SELECT src FROM edges)));",
                &["edges"],
            ),
        ] {
            let assets = sqlparser_assets(sql).unwrap();
            assert_eq!(assets.inputs, set(reads), "{sql}: the subquery's tables");
            assert!(
                assets.unread_table_functions.is_empty(),
                "{sql}: the call is read, got {:?}",
                assets.unread_table_functions
            );
            assert_eq!(assets.outputs.len(), 1, "{sql}: the target is produced");
        }
    }

    // The statement that showed the defect: two extension functions called on subqueries
    // and joined.
    #[test]
    fn test_two_table_functions_called_on_subqueries_and_joined_read_the_table() {
        let sql = "CREATE TEMP TABLE result AS \
            SELECT d.node_id, CASE WHEN isinf(d.distance) THEN NULL ELSE d.distance END AS distance, c.component \
            FROM onager_pth_dijkstra((SELECT src, dst, w FROM edges), directed := false, source := 0) d \
            JOIN onager_cmm_components((SELECT src, dst FROM edges)) c USING (node_id);";
        let assets = sqlparser_assets(sql).unwrap();
        assert_eq!(assets.inputs, set(&["edges"]));
        assert_eq!(assets.outputs, set(&["result"]));
        assert!(assets.unread_table_functions.is_empty());
    }

    // A string, or a quoted identifier, inside the subquery is the subquery's own: it
    // names nothing the call reads, so it does not send the call to the unread arm.
    #[test]
    fn test_a_string_inside_a_subquery_argument_does_not_make_the_call_unread() {
        for sql in [
            "CREATE TABLE r AS SELECT * FROM summary((SELECT src, dst FROM edges WHERE kind = 'road'));",
            r#"CREATE TABLE r AS SELECT * FROM summary((SELECT "src" FROM edges));"#,
        ] {
            let assets = sqlparser_assets(sql).unwrap();
            assert_eq!(assets.inputs, set(&["edges"]), "{sql}");
            assert!(
                assets.unread_table_functions.is_empty(),
                "{sql}: not unread, got {:?}",
                assets.unread_table_functions
            );
        }
    }

    // A string or a quoted identifier beside the subquery still names something arc does
    // not read: the call keeps the warning, and the subquery's tables are still read.
    #[test]
    fn test_a_string_beside_a_subquery_argument_keeps_the_call_unread() {
        for sql in [
            "CREATE TABLE m AS SELECT * FROM my_ext_fit((SELECT x, y FROM train), 'model');",
            r#"CREATE TABLE m AS SELECT * FROM my_ext_fit((SELECT x, y FROM train), "model");"#,
            "CREATE TABLE m AS SELECT * FROM my_ext_fit((SELECT x, y FROM train), target := 'model');",
        ] {
            let assets = sqlparser_assets(sql).unwrap();
            assert_eq!(
                assets.inputs,
                set(&["train"]),
                "{sql}: no read named for the call"
            );
            assert_eq!(
                assets.unread_table_functions,
                set(&["my_ext_fit"]),
                "{sql}: the call is listed for the warning"
            );
        }
    }

    // COPY <table> TO 'file' produces the file path as an output (file-path lineage).
    #[test]
    fn test_copy_to_produces_file() {
        let sql = "COPY ranking TO 'data/ranking.parquet';";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(assets.inputs.contains("ranking"), "table is read");
        assert!(
            assets.outputs.contains("data/ranking.parquet"),
            "file is produced"
        );
    }

    // COPY (SELECT …) TO 'file' reads what its query reads and produces the file, as
    // COPY <table> TO 'file' does.
    #[test]
    fn test_copy_from_query_produces_file() {
        let sql = "COPY (SELECT * FROM a WHERE x > 1) TO 'out.csv' (HEADER);";
        let assets = sqlparser_assets(sql).unwrap();
        assert_eq!(
            assets.inputs,
            BTreeSet::from(["a".to_string()]),
            "the query's table is read"
        );
        assert_eq!(
            assets.outputs,
            BTreeSet::from(["out.csv".to_string()]),
            "the file is produced, and nothing else is"
        );
        assert_eq!(assets.kinds.get("out.csv"), Some(&AssetKind::File));
    }

    // A query that reads several tables (a CTE and a join) has each of them recorded as
    // read, and the CTE's own name is not one of them.
    #[test]
    fn test_copy_from_query_reads_every_table_its_query_reads() {
        let sql = "COPY (WITH c AS (SELECT * FROM a) \
                   SELECT * FROM c JOIN b ON c.x = b.x) \
                   TO 'out.csv';";
        let assets = sqlparser_per_statement(sql).unwrap().remove(0);
        assert_eq!(
            assets.inputs,
            ["a", "b"].map(str::to_string).into_iter().collect(),
            "tables the query reads"
        );
        assert_eq!(assets.outputs, BTreeSet::from(["out.csv".to_string()]));
    }

    // The target of COPY (query) is classified from the statement's own options, and
    // classified as COPY <table> TO classifies it: the same options give the same kind
    // whichever the source. The expected kinds are listed too, so two arms that were
    // both wrong the same way would not pass.
    #[test]
    fn test_copy_from_query_target_kind_follows_the_options_as_copy_from_table_does() {
        for (options, expected) in [
            ("(HEADER)", AssetKind::File),
            ("(FORMAT parquet)", AssetKind::File),
            (
                "(FORMAT parquet, PARTITION_BY (region))",
                AssetKind::Directory,
            ),
            ("(PER_THREAD_OUTPUT)", AssetKind::Directory),
            ("(PER_THREAD_OUTPUT false)", AssetKind::File),
            ("(FORMAT csv, FILE_SIZE_BYTES '1GB')", AssetKind::Directory),
            (
                "(FORMAT parquet, ROW_GROUPS_PER_FILE 2)",
                AssetKind::Directory,
            ),
        ] {
            let from_table = sqlparser_assets(&format!("COPY a TO 'out' {options};")).unwrap();
            let from_query =
                sqlparser_assets(&format!("COPY (SELECT * FROM a) TO 'out' {options};")).unwrap();
            assert_eq!(
                from_table.kinds.get("out"),
                Some(&expected),
                "COPY a TO 'out' {options}"
            );
            assert_eq!(
                from_query.kinds.get("out"),
                Some(&expected),
                "COPY (SELECT * FROM a) TO 'out' {options}"
            );
        }
    }

    // Only a file target produces an asset: a COPY (query) to STDOUT reads its tables
    // and writes nothing arc can track.
    #[test]
    fn test_copy_from_query_to_stdout_produces_nothing() {
        let assets = sqlparser_assets("COPY (SELECT * FROM a) TO STDOUT;").unwrap();
        assert_eq!(assets.inputs, BTreeSet::from(["a".to_string()]));
        assert!(
            assets.outputs.is_empty(),
            "no output for STDOUT, got {:?}",
            assets.outputs
        );
    }

    // The two COPY <table> forms record what they recorded before COPY (query) did.
    // `COPY a FROM 'in.csv'` records the file as *produced* — the direction is not
    // read from the statement — and this change leaves that as it is.
    #[test]
    fn test_copy_table_forms_record_what_they_recorded_before() {
        let to = sqlparser_assets("COPY a TO 'out.csv' (HEADER);").unwrap();
        assert_eq!(to.inputs, BTreeSet::from(["a".to_string()]), "TO: reads");
        assert_eq!(
            to.outputs,
            BTreeSet::from(["out.csv".to_string()]),
            "TO: produces"
        );
        assert_eq!(to.kinds.get("out.csv"), Some(&AssetKind::File));

        let from = sqlparser_assets("COPY a FROM 'in.csv';").unwrap();
        assert_eq!(
            from.inputs,
            BTreeSet::from(["a".to_string()]),
            "FROM: reads"
        );
        assert_eq!(
            from.outputs,
            BTreeSet::from(["in.csv".to_string()]),
            "FROM: produces"
        );
        assert_eq!(from.kinds.get("in.csv"), Some(&AssetKind::File));
    }

    // A table function called with no string, no quoted identifier and no subquery — a
    // table macro such as `recent()` — keeps recording its name. One called with a string
    // or a quoted identifier records nothing under its name; see the `unread` tests
    // above. One called on a subquery records the tables the subquery reads.
    #[test]
    fn test_non_file_table_function_unchanged() {
        // A table macro (an extension-supplied table function that is neither a file
        // reader nor a row-generator, called with no string, no quoted identifier and no
        // subquery) still records its own name — it may read tables named in its own definition,
        // which this layer cannot see, so the name is the only handle lineage has on it.
        let sql = "SELECT * FROM recent();";
        let assets = sqlparser_assets(sql).unwrap();
        assert!(
            assets.inputs.contains("recent"),
            "a table macro called with no string or quoted identifier still records its name"
        );
    }

    // Byte ranges: one range per top-level statement, each slicing its own source.
    #[test]
    fn byte_ranges_split_top_level_statements() {
        let sql = "CREATE TABLE foo (id INT);\nSELECT * FROM foo;";
        let ranges = statement_byte_ranges(sql);
        assert_eq!(ranges.len(), 2);
        assert_eq!(&sql[ranges[0].0..ranges[0].1], "CREATE TABLE foo (id INT);");
        assert_eq!(&sql[ranges[1].0..ranges[1].1], "SELECT * FROM foo;");
        // Ranges align 1:1 with the parsed statements.
        assert_eq!(ranges.len(), sqlparser_per_statement(sql).unwrap().len());
    }

    // Byte ranges: a `;` inside a string literal must not split the statement.
    #[test]
    fn byte_ranges_ignore_semicolons_in_strings() {
        let sql = "INSERT INTO t VALUES ('a;b;c');";
        let ranges = statement_byte_ranges(sql);
        assert_eq!(ranges.len(), 1);
        assert_eq!(
            &sql[ranges[0].0..ranges[0].1],
            "INSERT INTO t VALUES ('a;b;c');"
        );
    }

    // Byte ranges: comment-only and blank segments are dropped, not counted.
    #[test]
    fn byte_ranges_drop_comment_and_blank_segments() {
        let sql = "-- header comment\nSELECT 1; /* trailing */ \n\n";
        let ranges = statement_byte_ranges(sql);
        assert_eq!(ranges.len(), 1);
        // Range starts at the code, past the leading comment, and trims trailing space.
        assert_eq!(&sql[ranges[0].0..ranges[0].1], "SELECT 1;");
    }

    // Byte ranges: a `;` inside a `--` line comment does not split.
    #[test]
    fn byte_ranges_ignore_semicolons_in_line_comments() {
        let sql = "SELECT 1 -- a; b; c\nFROM t;";
        let ranges = statement_byte_ranges(sql);
        assert_eq!(ranges.len(), 1);
        assert_eq!(&sql[ranges[0].0..ranges[0].1], sql.trim_end_matches('\n'));
    }

    // Byte ranges: a `;` inside a dollar-quoted body does not split.
    #[test]
    fn byte_ranges_ignore_semicolons_in_dollar_quotes() {
        let sql = "SELECT $$a; b; c$$ AS s;";
        let ranges = statement_byte_ranges(sql);
        assert_eq!(ranges.len(), 1);
        assert_eq!(&sql[ranges[0].0..ranges[0].1], "SELECT $$a; b; c$$ AS s;");
    }

    // ---- DuckDB statement-form PIVOT/UNPIVOT + multi-option COPY. ----
    // These forms need the vendored sqlparser fork; without it the parse fails and the
    // whole step degrades to an opaque node (see AssetGraph::build), defeating the
    // structural-transparency principle.

    // Statement-form PIVOT over a real table lifts the source as an input, not an
    // opaque step.
    #[test]
    fn test_pivot_statement_lifts_source() {
        let sql = "PIVOT monthly_sales ON month USING SUM(amount) GROUP BY country;";
        let assets = sqlparser_assets(sql).expect("statement-form PIVOT must parse");
        assert!(
            assets.inputs.contains("monthly_sales"),
            "pivot source should be an input, got {:?}",
            assets.inputs
        );
    }

    // PIVOT as a CTAS body — output AND source input are both discovered.
    #[test]
    fn test_pivot_statement_as_ctas_body() {
        let sql =
            "CREATE OR REPLACE TABLE installs AS PIVOT wide_sales ON days USING SUM(installs);";
        let assets = sqlparser_assets(sql).expect("CTAS over a PIVOT must parse");
        assert!(
            assets.outputs.contains("installs"),
            "CTAS output discovered"
        );
        assert!(
            assets.inputs.contains("wide_sales"),
            "pivot source lifted as input, got {:?}",
            assets.inputs
        );
    }

    // The brewtrend shape — CTAS + WITH + PIVOT over the CTE. The CTE name is filtered
    // out of inputs; the CTE's own source table is the real input.
    #[test]
    fn test_pivot_ctas_with_cte_shape() {
        let sql = "CREATE OR REPLACE TABLE installs AS \
                   WITH install_counts AS (SELECT category, name, days, installs FROM categories) \
                   PIVOT install_counts ON days USING SUM(installs);";
        let assets = sqlparser_assets(sql).expect("brewtrend-shape PIVOT must parse");
        assert!(assets.outputs.contains("installs"), "output discovered");
        assert!(
            assets.inputs.contains("categories"),
            "real underlying table is the input, got {:?}",
            assets.inputs
        );
        assert!(
            !assets.inputs.contains("install_counts"),
            "the pivoted CTE name is internal, not an external input"
        );
        assert!(
            assets.internal.contains("install_counts"),
            "CTE tracked as internal"
        );
    }

    // Statement-form UNPIVOT lifts the source as an input.
    #[test]
    fn test_unpivot_statement_lifts_source() {
        let sql = "UNPIVOT quarterly_report ON q1, q2, q3, q4 INTO NAME quarter VALUE amount;";
        let assets = sqlparser_assets(sql).expect("statement-form UNPIVOT must parse");
        assert!(
            assets.inputs.contains("quarterly_report"),
            "unpivot source should be an input, got {:?}",
            assets.inputs
        );
    }

    // UNPIVOT with the shorthand (no INTO clause) still parses + lifts source.
    #[test]
    fn test_unpivot_statement_shorthand() {
        let sql = "UNPIVOT sensor_readings ON temp, humidity, pressure;";
        let assets = sqlparser_assets(sql).expect("shorthand UNPIVOT must parse");
        assert!(assets.inputs.contains("sensor_readings"), "source lifted");
    }

    // Multi-option COPY parses; the table is read and the file is produced, and the
    // extra DuckDB options (COMPRESSION) do not break introspection.
    #[test]
    fn test_multi_option_copy() {
        let sql = "COPY ranking TO 'data/ranking.parquet' (FORMAT parquet, COMPRESSION zstd);";
        let assets = sqlparser_assets(sql).expect("multi-option COPY must parse");
        assert!(assets.inputs.contains("ranking"), "table is read");
        assert!(
            assets.outputs.contains("data/ranking.parquet"),
            "file is produced (file-path lineage), got {:?}",
            assets.outputs
        );
    }

    // COPY with a parenthesized PARTITION_BY value list also parses.
    #[test]
    fn test_copy_partition_by() {
        let sql = "COPY orders TO 'out/orders' (FORMAT parquet, PARTITION_BY (year, month), OVERWRITE_OR_IGNORE);";
        let assets = sqlparser_assets(sql).expect("COPY with PARTITION_BY must parse");
        assert!(assets.inputs.contains("orders"), "table is read");
        assert!(
            assets.outputs.contains("out/orders"),
            "output path produced"
        );
        // PARTITION_BY makes DuckDB write a directory of Hive-partitioned files under
        // this name, not one file — the COPY's own options say so, so this is known
        // here rather than guessed later from the string or the filesystem.
        assert_eq!(
            assets.kinds.get("out/orders"),
            Some(&AssetKind::Directory),
            "PARTITION_BY target must be classified as a directory, not a file"
        );
    }

    // A COPY … TO carrying none of `copy_to_target_kind`'s directory-writing options
    // writes one file — it must not be classified a directory just for sharing the
    // COPY statement shape.
    #[test]
    fn test_copy_without_a_directory_writing_option_is_a_file() {
        let sql = "COPY orders TO 'out/orders.parquet' (FORMAT parquet);";
        let assets = sqlparser_assets(sql).expect("plain COPY must parse");
        assert_eq!(
            assets.kinds.get("out/orders.parquet"),
            Some(&AssetKind::File)
        );
    }

    // The other three directory-writing options, each on its own. Until this round
    // only PARTITION_BY was tested for, and the comment above this test asserted that
    // a COPY without it "writes exactly one file" — false on the DuckDB this crate
    // links: PER_THREAD_OUTPUT and FILE_SIZE_BYTES each wrote a directory, the
    // File-kind classification then made `fs::read` fail on it, and the step re-ran
    // forever while warning that nothing had been produced.
    #[test]
    fn test_per_thread_output_target_is_a_directory() {
        let sql = "COPY orders TO 'out/pto' (FORMAT parquet, PER_THREAD_OUTPUT true);";
        let assets = sqlparser_assets(sql).expect("PER_THREAD_OUTPUT COPY must parse");
        assert_eq!(assets.kinds.get("out/pto"), Some(&AssetKind::Directory));
    }

    #[test]
    fn test_per_thread_output_bare_flag_is_a_directory() {
        let sql = "COPY orders TO 'out/pto' (FORMAT parquet, PER_THREAD_OUTPUT);";
        let assets = sqlparser_assets(sql).expect("bare-flag COPY must parse");
        assert_eq!(assets.kinds.get("out/pto"), Some(&AssetKind::Directory));
    }

    // DuckDB's own `GetBooleanArg` reads an explicit `false` as off, so this one
    // really does write a single file and classifying it a directory would send the
    // step into the same perpetual re-run from the other side.
    #[test]
    fn test_per_thread_output_false_is_a_file() {
        let sql = "COPY orders TO 'out/one.parquet' (FORMAT parquet, PER_THREAD_OUTPUT false);";
        let assets = sqlparser_assets(sql).expect("PER_THREAD_OUTPUT false COPY must parse");
        assert_eq!(assets.kinds.get("out/one.parquet"), Some(&AssetKind::File));
    }

    // `GetBooleanArg` CASTS its argument to BOOLEAN; it does not compare it against
    // the `false` keyword. Each of these spellings wrote a single 198-byte parquet
    // file when driven on the DuckDB CLI at v1.5.4 and at v1.5.5, and each was
    // classified `Directory` here until this round — `read_dir` on a regular file
    // returns `None`, so the step re-ran on every run while warning that it had
    // produced nothing.
    #[test]
    fn test_per_thread_output_cast_to_false_is_a_file() {
        for arg in ["0", "'false'", "'FALSE'", "'no'", "'f'"] {
            let sql = format!(
                "COPY orders TO 'out/one.parquet' (FORMAT parquet, PER_THREAD_OUTPUT {arg});"
            );
            let assets = sqlparser_assets(&sql).expect("COPY must parse");
            assert_eq!(
                assets.kinds.get("out/one.parquet"),
                Some(&AssetKind::File),
                "PER_THREAD_OUTPUT {arg} casts to false and writes one file"
            );
        }
    }

    // The same cast in the other direction, so the arm above cannot be satisfied by
    // reading every PER_THREAD_OUTPUT argument as off. Each of these wrote a
    // directory on both engines.
    #[test]
    fn test_per_thread_output_cast_to_true_is_a_directory() {
        for arg in ["1", "'true'", "'yes'", "'t'", "'Y'"] {
            let sql =
                format!("COPY orders TO 'out/pto' (FORMAT parquet, PER_THREAD_OUTPUT {arg});");
            let assets = sqlparser_assets(&sql).expect("COPY must parse");
            assert_eq!(
                assets.kinds.get("out/pto"),
                Some(&AssetKind::Directory),
                "PER_THREAD_OUTPUT {arg} casts to true and writes a directory"
            );
        }
    }

    // The cast belongs to PER_THREAD_OUTPUT alone. `rotate` is set from
    // `file_size_bytes.IsValid()`, not from a boolean, so a zero here is still on —
    // `FILE_SIZE_BYTES 0` wrote a directory on both engines. Applying the boolean
    // cast to all four names uniformly would get this one wrong.
    #[test]
    fn test_file_size_bytes_zero_is_still_a_directory() {
        let sql = "COPY orders TO 'out/sized' (FORMAT parquet, FILE_SIZE_BYTES 0);";
        let assets = sqlparser_assets(sql).expect("FILE_SIZE_BYTES 0 COPY must parse");
        assert_eq!(assets.kinds.get("out/sized"), Some(&AssetKind::Directory));
    }

    #[test]
    fn test_file_size_bytes_target_is_a_directory() {
        let sql = "COPY orders TO 'out/sized' (FORMAT parquet, FILE_SIZE_BYTES '1MB');";
        let assets = sqlparser_assets(sql).expect("FILE_SIZE_BYTES COPY must parse");
        assert_eq!(assets.kinds.get("out/sized"), Some(&AssetKind::Directory));
    }

    #[test]
    fn test_row_groups_per_file_target_is_a_directory() {
        let sql = "COPY orders TO 'out/rgpf' (FORMAT parquet, ROW_GROUPS_PER_FILE 1);";
        let assets = sqlparser_assets(sql).expect("ROW_GROUPS_PER_FILE COPY must parse");
        assert_eq!(assets.kinds.get("out/rgpf"), Some(&AssetKind::Directory));
    }

    // An option NOT in the directory-writing set must not flip the classification,
    // however directory-ish it reads: FILENAME_PATTERN and FILE_EXTENSION only shape
    // the names DuckDB uses once something else has already made the target a
    // directory, and OVERWRITE only decides what happens to what is already there.
    #[test]
    fn test_neighbouring_copy_options_do_not_make_a_directory() {
        for sql in [
            "COPY orders TO 'out/o.parquet' (FORMAT parquet, FILENAME_PATTERN 'part_{i}');",
            "COPY orders TO 'out/o.parquet' (FORMAT parquet, FILE_EXTENSION 'pq');",
            "COPY orders TO 'out/o.parquet' (FORMAT parquet, OVERWRITE_OR_IGNORE);",
            "COPY orders TO 'out/o.parquet' (FORMAT parquet, ROW_GROUP_SIZE 100000);",
        ] {
            let assets = sqlparser_assets(sql).expect("COPY must parse");
            assert_eq!(
                assets.kinds.get("out/o.parquet"),
                Some(&AssetKind::File),
                "not a directory-writing option: {sql}"
            );
        }
    }

    // ---- DuckDB's Python-style `lambda x: expr` lambda syntax. ----
    // DuckDB is retiring the single-arrow lambda; without this the fork, a model
    // using the new form fails to parse and the whole step degrades to an opaque
    // node (see AssetGraph::build), contributing no assets to the lineage graph.

    // A CTAS whose SELECT list uses `lambda c: ...` still parses and still
    // discovers both the output and the FROM-clause input — the lambda sits in
    // an expression that asset extraction never inspects, so the only way it
    // can affect `assets` at all is by breaking the parse.
    #[test]
    fn test_lambda_colon_single_param_does_not_block_introspection() {
        let sql = "CREATE TABLE bumped AS \
                   SELECT list_transform([x], lambda c: c + 1) AS y FROM source_table;";
        let assets = sqlparser_assets(sql).expect("lambda colon syntax must parse");
        assert!(assets.outputs.contains("bumped"), "CTAS output discovered");
        assert!(
            assets.inputs.contains("source_table"),
            "FROM table still discovered as input, got {:?}",
            assets.inputs
        );
    }

    // The multi-param colon form (`lambda acc, v: ...`, no parens) parses too.
    #[test]
    fn test_lambda_colon_multi_param_does_not_block_introspection() {
        let sql = "CREATE TABLE totals AS \
                   SELECT list_reduce(xs, lambda acc, v: acc + v) AS total FROM source_table;";
        let assets = sqlparser_assets(sql).expect("multi-param lambda colon syntax must parse");
        assert!(assets.outputs.contains("totals"));
        assert!(assets.inputs.contains("source_table"));
    }
}