knut-thund 0.1.4

//! The **SQL authoring surface** (feature `sql`): SDP-style DDL parsed into the
//! [`crate::ir::Pipeline`].
//!
//! This is the fourth front-end over the one IR, alongside the typed Rust
//! builder, the [RON DSL](super::dsl) and the facett graph editor. Like RON it
//! is *not* a second model — it parses to the same `Pipeline` value, so
//! `SQL -> IR -> RON -> IR -> SQL` round-trips.
//!
//! # Approach: extend, don't fork
//!
//! Built the way DataFusion's own `DFParser` extends `sqlparser-rs` — wrap a
//! [`Parser`], intercept the statements the dialect does not know, and delegate
//! everything else. We share sqlparser `0.62` with `datafusion 54`, so the
//! `native` backend and this parser link **one** AST, not two.
//!
//! The only genuinely new syntax in SDP is the `STREAM(t)` table factor, and it
//! needs *no* grammar fork: stock sqlparser already parses `FROM STREAM(t)` as
//! a table-valued function and `FROM STREAM t` as a table with an alias. We
//! parse with the stock grammar, then [rewrite those nodes](rewrite_stream) back
//! to a plain table reference and record the streaming-ness on the
//! [`Flow`] — which is where it belongs, since it is a property of the edge,
//! not of the `SELECT`.
//!
//! # Supported grammar
//!
//! Verified against Spark 4.1.3's SDP (`sdp-grammar-probe`); the two marked
//! **beyond SDP** are accepted here and rejected by Spark.
//!
//! ```sql
//! CREATE MATERIALIZED VIEW name
//!   [ ( col TYPE [NOT NULL], …
//!     [, CONSTRAINT c EXPECT (expr) [ON VIOLATION {DROP ROW | FAIL UPDATE}]] ) ]  -- beyond SDP
//!   [ PARTITIONED BY (col | months(col) | days(col) | years(col) | hours(col)
//!                    | bucket(n, col) | truncate(n, col)) ]                        -- transforms: beyond SDP
//!   [ COMMENT 'text' ]
//!   [ TBLPROPERTIES ('k' = 'v', …) ]
//!   AS SELECT …;
//!
//! CREATE [TEMPORARY] VIEW name AS SELECT …;
//! CREATE STREAMING TABLE name [ (…) ] [clauses] [ AS SELECT … FROM STREAM(src) ];
//! CREATE FLOW name AS INSERT INTO target [BY NAME] SELECT … FROM STREAM(src);
//! ```
//!
//! Anything else is handed to stock sqlparser and rejected as not-a-definition,
//! so a typo in a `SELECT` still reports sqlparser's own error.

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

use sqlparser::ast::{Query, SetExpr, TableFactor};
use sqlparser::dialect::GenericDialect;
use sqlparser::parser::Parser;
use sqlparser::tokenizer::Token;

use crate::error::{Result, ThundError};
use crate::ir::{
    Dataset, DatasetSchema, Expectation, Flow, OnViolation, OutputType, Pipeline, SourceSpec,
};

/// Parse an SDP-style SQL script into a named [`Pipeline`].
///
/// Statement order does not matter: the DAG is inferred from each flow's
/// `reads`, exactly as with the other authoring surfaces, so a `SELECT` may
/// reference a dataset defined further down the file.
pub fn from_sql(pipeline_name: impl Into<String>, text: &str) -> Result<Pipeline> {
    let name = pipeline_name.into();
    let result: Result<Pipeline> = (|| {
        let mut p = Pipeline::new(name.clone());
        for stmt in ThundParser::new(text)?.parse_all()? {
            match stmt {
                ThundStatement::Dataset { dataset, flow } => {
                    p = p.with_dataset(dataset);
                    if let Some(f) = flow {
                        p = p.with_flow(f);
                    }
                }
                ThundStatement::Flow(f) => p = p.with_flow(f),
            }
        }
        p.validate()?;
        Ok(p)
    })();
    crate::functional_status(
        "knut-thund/sql",
        "from_sql",
        result.is_ok(),
        &match &result {
            Ok(p) => format!("{} ({} datasets)", p.name, p.datasets.len()),
            Err(e) => e.to_string(),
        },
    );
    result
}

/// Render a [`Pipeline`] back as an SDP-style SQL script.
///
/// The inverse of [`from_sql`] for everything the grammar covers, so the
/// round-trip is checked by `tests/sql_frontend.rs`. Datasets are emitted in
/// [`Pipeline::topo_order`] where one exists, so the script reads
/// dependencies-first.
pub fn to_sql(pipeline: &Pipeline) -> Result<String> {
    let result: Result<String> = (|| {
        let order = pipeline.topo_order().ok_or(ThundError::Cyclic)?;
        let mut out = String::new();
        for name in &order {
            let Some(d) = pipeline.dataset(name) else {
                continue;
            };
            let flow = pipeline.flows.iter().find(|f| &f.target == name);
            out.push_str(&render_dataset(d, flow));
            out.push('\n');
        }
        // Extra flows (a second append into an existing streaming table) are
        // `CREATE FLOW` statements with no dataset of their own.
        for f in &pipeline.flows {
            if pipeline.dataset(&f.target).is_some()
                && pipeline.flows.iter().position(|g| g.target == f.target)
                    != pipeline.flows.iter().position(|g| g.name == f.name)
            {
                out.push_str(&render_standalone_flow(f));
                out.push('\n');
            }
        }
        Ok(out)
    })();
    crate::functional_status("knut-thund/sql", "to_sql", result.is_ok(), &pipeline.name);
    result
}

// ---------------------------------------------------------------------------
// Parser
// ---------------------------------------------------------------------------

/// What one parsed statement contributes to the graph.
enum ThundStatement {
    /// A dataset definition, with the flow that computes it (absent for a
    /// `CREATE STREAMING TABLE name;` whose flows arrive via `CREATE FLOW`).
    Dataset {
        dataset: Dataset,
        flow: Option<Flow>,
    },
    /// A standalone `CREATE FLOW` appending into an already-declared target.
    Flow(Flow),
}

/// An SDP-dialect parser wrapping stock `sqlparser`.
struct ThundParser<'a> {
    parser: Parser<'a>,
}

/// The post-name clauses of a dataset definition, which may appear in any order.
#[derive(Default)]
struct TableClauses {
    partition_cols: Vec<String>,
    comment: Option<String>,
    properties: BTreeMap<String, String>,
}

/// The dialect is shared by every parse; `GenericDialect` is the permissive one
/// DataFusion also starts from.
const DIALECT: GenericDialect = GenericDialect {};

impl<'a> ThundParser<'a> {
    fn new(sql: &str) -> Result<ThundParser<'_>> {
        let parser = Parser::new(&DIALECT)
            .try_with_sql(sql)
            .map_err(|e| ThundError::Sql(e.to_string()))?;
        Ok(ThundParser { parser })
    }

    /// Parse every `;`-separated statement in the script.
    fn parse_all(&mut self) -> Result<Vec<ThundStatement>> {
        let mut out = Vec::new();
        loop {
            while self.parser.consume_token(&Token::SemiColon) {}
            if self.parser.peek_token().token == Token::EOF {
                break;
            }
            out.push(self.parse_statement()?);
        }
        Ok(out)
    }

    /// Case-insensitive bare-word match that consumes on success.
    ///
    /// Used instead of `Keyword::*` for `MATERIALIZED` / `STREAMING` / `FLOW` /
    /// `EXPECT` and friends: which of those sqlparser promotes to a reserved
    /// keyword varies release to release, and a word match is stable across
    /// that churn.
    fn word(&mut self, w: &str) -> bool {
        let tok = self.parser.peek_token();
        if matches!(&tok.token, Token::Word(word) if word.value.eq_ignore_ascii_case(w)) {
            self.parser.next_token();
            return true;
        }
        false
    }

    /// [`Self::word`] for a whole phrase — all or nothing.
    ///
    /// Peeks the full phrase *before* consuming anything, so a partial match
    /// costs nothing and the caller can try the next alternative. (sqlparser
    /// exposes `index()` but no setter, so look-ahead is the way to backtrack.)
    fn words(&mut self, ws: &[&str]) -> bool {
        for (i, w) in ws.iter().enumerate() {
            match &self.parser.peek_nth_token(i).token {
                Token::Word(word) if word.value.eq_ignore_ascii_case(w) => {}
                _ => return false,
            }
        }
        for _ in ws {
            self.parser.next_token();
        }
        true
    }

    fn expect_word(&mut self, w: &str) -> Result<()> {
        if self.word(w) {
            Ok(())
        } else {
            Err(ThundError::Sql(format!(
                "expected `{w}`, found `{}`",
                self.parser.peek_token()
            )))
        }
    }

    fn parse_statement(&mut self) -> Result<ThundStatement> {
        if !self.word("CREATE") {
            return Err(ThundError::Sql(format!(
                "expected a CREATE definition, found `{}` — a Thund SQL script \
                 contains only dataset/flow definitions",
                self.parser.peek_token()
            )));
        }

        if self.words(&["MATERIALIZED", "VIEW"]) {
            self.parse_dataset(OutputType::MaterializedView, false)
        } else if self.words(&["STREAMING", "TABLE"]) {
            self.parse_dataset(OutputType::Table, true)
        } else if self.words(&["TEMPORARY", "VIEW"])
            || self.words(&["TEMP", "VIEW"])
            // A bare `CREATE VIEW` is a temporary view: SDP has no persisted
            // plain-view kind, and the alternative (silently promoting it to a
            // materialized view) would change what gets written.
            || self.word("VIEW")
        {
            self.parse_dataset(OutputType::TemporaryView, false)
        } else if self.word("FLOW") {
            self.parse_create_flow()
        } else {
            Err(ThundError::Sql(format!(
                "unsupported CREATE form at `{}` — expected MATERIALIZED VIEW, \
                 STREAMING TABLE, [TEMPORARY] VIEW or FLOW",
                self.parser.peek_token()
            )))
        }
    }

    /// `CREATE {MATERIALIZED VIEW | STREAMING TABLE | TEMPORARY VIEW} name …`
    fn parse_dataset(
        &mut self,
        output_type: OutputType,
        streaming: bool,
    ) -> Result<ThundStatement> {
        let name = self.object_name()?;

        let mut schema = DatasetSchema::new();
        let mut expectations = Vec::new();
        if self.parser.consume_token(&Token::LParen) {
            self.parse_column_list(&mut schema, &mut expectations)?;
        }

        let clauses = self.parse_table_clauses()?;

        let mut dataset = Dataset::new(&name, output_type);
        dataset.comment = clauses.comment;
        dataset.partition_cols = clauses.partition_cols;
        dataset.properties = clauses.properties;
        if !schema.is_empty() {
            dataset = dataset.with_schema(schema);
        }

        // `CREATE STREAMING TABLE t;` legally has no body — flows attach later.
        if !self.word("AS") {
            return Ok(ThundStatement::Dataset {
                dataset,
                flow: None,
            });
        }

        let flow = self.parse_flow_body(format!("f_{name}"), &name, streaming, expectations)?;
        Ok(ThundStatement::Dataset {
            dataset,
            flow: Some(flow),
        })
    }

    /// `CREATE FLOW name AS INSERT INTO target [BY NAME] SELECT …`
    fn parse_create_flow(&mut self) -> Result<ThundStatement> {
        let flow_name = self.object_name()?;
        self.expect_word("AS")?;
        self.expect_word("INSERT")?;
        self.expect_word("INTO")?;
        let target = self.object_name()?;
        // `BY NAME` is column-matching semantics, not graph shape — accepted
        // and not otherwise modelled.
        self.words(&["BY", "NAME"]);
        let flow = self.parse_flow_body(flow_name, &target, true, Vec::new())?;
        Ok(ThundStatement::Flow(flow))
    }

    /// Parse the `SELECT …` body shared by every definition and build the
    /// [`Flow`]: rewrite `STREAM(…)`, collect `reads`, pick batch vs streaming.
    fn parse_flow_body(
        &mut self,
        flow_name: impl Into<String>,
        target: &str,
        prefer_streaming: bool,
        expectations: Vec<Expectation>,
    ) -> Result<Flow> {
        let mut query = self
            .parser
            .parse_query()
            .map_err(|e| ThundError::Sql(e.to_string()))?;

        let streamed = rewrite_stream(&mut query);
        let reads = collect_reads(&query);
        let sql = query.to_string();

        // A `STREAM(…)` factor forces streaming; otherwise a streaming-table
        // target stays streaming only if it actually reads something unbounded.
        let mut flow = if let Some(source) = streamed.first() {
            let mut f = Flow::streaming(
                flow_name,
                target,
                SourceSpec::Cdc {
                    source: source.clone(),
                },
            );
            // `Flow::streaming` leaves `reads` empty; the DAG needs it.
            f.reads = reads;
            f
        } else if prefer_streaming && !reads.is_empty() {
            let mut f = Flow::streaming(
                flow_name,
                target,
                SourceSpec::Cdc {
                    source: reads[0].clone(),
                },
            );
            f.reads = reads;
            f
        } else {
            Flow::batch(flow_name, target, reads)
        };

        flow.query = Some(sql);
        flow.expectations = expectations;
        Ok(flow)
    }

    /// The parenthesised column list: `col TYPE [NOT NULL]` entries and
    /// `CONSTRAINT … EXPECT …` entries, comma-separated.
    fn parse_column_list(
        &mut self,
        schema: &mut DatasetSchema,
        expectations: &mut Vec<Expectation>,
    ) -> Result<()> {
        loop {
            if self.word("CONSTRAINT") {
                expectations.push(self.parse_expectation()?);
            } else {
                let col = self
                    .parser
                    .parse_identifier()
                    .map_err(|e| ThundError::Sql(e.to_string()))?;
                let ty = self
                    .parser
                    .parse_data_type()
                    .map_err(|e| ThundError::Sql(e.to_string()))?;
                let nullable = !self.words(&["NOT", "NULL"]);
                *schema = std::mem::take(schema).field(col.value, sql_type_to_arrow(&ty), nullable);
            }
            if self.parser.consume_token(&Token::Comma) {
                continue;
            }
            self.parser
                .expect_token(&Token::RParen)
                .map_err(|e| ThundError::Sql(e.to_string()))?;
            return Ok(());
        }
    }

    /// `CONSTRAINT name EXPECT (expr) [ON VIOLATION {DROP ROW | FAIL UPDATE}]`
    ///
    /// **Beyond SDP** — Spark 4.1.3 rejects this with `PARSE_SYNTAX_ERROR at or
    /// near 'EXPECT'`; it is a Databricks-DLT extension. The IR has modelled it
    /// all along ([`OnViolation`]), so the native backend can honour it.
    fn parse_expectation(&mut self) -> Result<Expectation> {
        let name = self
            .parser
            .parse_identifier()
            .map_err(|e| ThundError::Sql(e.to_string()))?;
        self.expect_word("EXPECT")?;
        self.parser
            .expect_token(&Token::LParen)
            .map_err(|e| ThundError::Sql(e.to_string()))?;
        let expr = self
            .parser
            .parse_expr()
            .map_err(|e| ThundError::Sql(e.to_string()))?;
        self.parser
            .expect_token(&Token::RParen)
            .map_err(|e| ThundError::Sql(e.to_string()))?;

        let policy = if self.words(&["ON", "VIOLATION"]) {
            if self.words(&["DROP", "ROW"]) {
                OnViolation::Drop
            } else if self.words(&["FAIL", "UPDATE"]) {
                OnViolation::Fail
            } else {
                return Err(ThundError::Sql(format!(
                    "expected DROP ROW or FAIL UPDATE after ON VIOLATION, found `{}`",
                    self.parser.peek_token()
                )));
            }
        } else {
            OnViolation::Warn
        };

        Ok(Expectation::new(name.value, expr.to_string()).on(policy))
    }

    /// The post-name clauses, in any order: `PARTITIONED BY`, `COMMENT`,
    /// `TBLPROPERTIES`.
    fn parse_table_clauses(&mut self) -> Result<TableClauses> {
        let mut c = TableClauses::default();
        loop {
            if self.words(&["PARTITIONED", "BY"]) {
                c.partition_cols = self.parse_partition_list()?;
            } else if self.word("COMMENT") {
                c.comment = Some(
                    self.parser
                        .parse_literal_string()
                        .map_err(|e| ThundError::Sql(e.to_string()))?,
                );
            } else if self.word("TBLPROPERTIES") {
                c.properties = self.parse_property_list()?;
            } else {
                return Ok(c);
            }
        }
    }

    /// `TBLPROPERTIES ('k' = 'v', 'k2' = 'v2')`
    ///
    /// Keys and values may be quoted strings or bare words; `=` is optional, as
    /// in Spark. Collected into [`Dataset::properties`] and lowered onto
    /// `DefineOutput.TableDetails.table_properties`, so they reach Spark.
    fn parse_property_list(&mut self) -> Result<BTreeMap<String, String>> {
        self.parser
            .expect_token(&Token::LParen)
            .map_err(|e| ThundError::Sql(e.to_string()))?;
        let mut props = BTreeMap::new();
        // `TBLPROPERTIES ()` is legal and empty.
        if self.parser.consume_token(&Token::RParen) {
            return Ok(props);
        }
        loop {
            let key = self.property_token()?;
            // Spark accepts `'k' 'v'` as well as `'k' = 'v'`.
            let _ = self.parser.consume_token(&Token::Eq);
            let value = self.property_token()?;
            props.insert(key, value);
            if self.parser.consume_token(&Token::Comma) {
                continue;
            }
            self.parser
                .expect_token(&Token::RParen)
                .map_err(|e| ThundError::Sql(e.to_string()))?;
            return Ok(props);
        }
    }

    /// One `TBLPROPERTIES` key or value: a quoted string, a bare word, or a
    /// number. Dotted keys (`delta.appendOnly`) tokenize as several words, so
    /// they are re-joined here rather than truncated at the first `.`.
    fn property_token(&mut self) -> Result<String> {
        let mut s = match self.parser.next_token().token {
            Token::SingleQuotedString(v) | Token::DoubleQuotedString(v) => return Ok(v),
            Token::Word(w) => w.value,
            Token::Number(n, _) => n,
            other => {
                return Err(ThundError::Sql(format!(
                    "expected a TBLPROPERTIES key or value, found `{other}`"
                )));
            }
        };
        while self.parser.consume_token(&Token::Period) {
            match self.parser.next_token().token {
                Token::Word(w) => {
                    s.push('.');
                    s.push_str(&w.value);
                }
                other => {
                    return Err(ThundError::Sql(format!(
                        "expected an identifier after `.` in a TBLPROPERTIES key, found `{other}`"
                    )));
                }
            }
        }
        Ok(s)
    }

    /// `PARTITIONED BY (col, months(ts), bucket(16, id), …)`
    ///
    /// Transforms are kept **verbatim** as `partition_cols` strings, e.g.
    /// `"months(event_ts)"`. This is the L2-additive choice: the field stays
    /// `Vec<String>`, so every existing RON spec deserializes unchanged, and a
    /// backend that cannot honour a transform can reject it by inspection.
    /// Spark's SDP is exactly such a backend — its `PartitionHelper` accepts
    /// only identity — so the Spark lowering must screen these out.
    fn parse_partition_list(&mut self) -> Result<Vec<String>> {
        self.parser
            .expect_token(&Token::LParen)
            .map_err(|e| ThundError::Sql(e.to_string()))?;
        let mut cols = Vec::new();
        loop {
            let expr = self
                .parser
                .parse_expr()
                .map_err(|e| ThundError::Sql(e.to_string()))?;
            cols.push(expr.to_string());
            if self.parser.consume_token(&Token::Comma) {
                continue;
            }
            self.parser
                .expect_token(&Token::RParen)
                .map_err(|e| ThundError::Sql(e.to_string()))?;
            return Ok(cols);
        }
    }

    /// A possibly-qualified name (`catalog.db.name`), rendered back to a string
    /// because the IR names datasets by string.
    fn object_name(&mut self) -> Result<String> {
        self.parser
            .parse_object_name(false)
            .map(|n| n.to_string())
            .map_err(|e| ThundError::Sql(e.to_string()))
    }
}

// ---------------------------------------------------------------------------
// AST helpers — the STREAM rewrite and `reads` extraction
// ---------------------------------------------------------------------------

/// Rewrite every `STREAM(t)` / `STREAM t` table factor to a plain reference to
/// `t`, returning the sources found in visit order.
///
/// This is the whole of the "new syntax" problem. `FROM STREAM(t)` already
/// parses as a table-valued function and `FROM STREAM t` as a table with an
/// alias, so both arrive as [`TableFactor::Table`] named `STREAM` — we only
/// have to *reinterpret* them. Mutating `name`/`args`/`alias` in place (rather
/// than rebuilding the variant) keeps this immune to sqlparser adding fields.
fn rewrite_stream(query: &mut Query) -> Vec<String> {
    let mut found = Vec::new();
    walk_factors_mut(query, &mut |tf: &mut TableFactor| {
        if let TableFactor::Table {
            name, args, alias, ..
        } = tf
        {
            let is_stream = name
                .0
                .last()
                .map(|p| {
                    p.to_string()
                        .trim_matches('"')
                        .eq_ignore_ascii_case("STREAM")
                })
                .unwrap_or(false);
            if !is_stream {
                return;
            }
            // `STREAM(t)` — the source is the single argument.
            if let Some(a) = args.take() {
                if let Some(first) = a.args.first() {
                    let src = first.to_string();
                    if let Ok(parsed) = Parser::new(&DIALECT)
                        .try_with_sql(&src)
                        .and_then(|mut p| p.parse_object_name(false))
                    {
                        *name = parsed;
                        found.push(src);
                        return;
                    }
                }
                return;
            }
            // `STREAM t` — sqlparser read `t` as the alias.
            if let Some(al) = alias.take() {
                let src = al.name.value.clone();
                if let Ok(parsed) = Parser::new(&DIALECT)
                    .try_with_sql(&src)
                    .and_then(|mut p| p.parse_object_name(false))
                {
                    *name = parsed;
                    found.push(src);
                }
            }
        }
    });
    found
}

/// Every table referenced by the query, deduplicated and sorted — the flow's
/// `reads`, and hence the DAG edges. CTE names are excluded: they are local to
/// the query, not datasets in the graph.
fn collect_reads(query: &Query) -> Vec<String> {
    let mut ctes = BTreeSet::new();
    if let Some(with) = &query.with {
        for cte in &with.cte_tables {
            ctes.insert(cte.alias.name.value.to_ascii_lowercase());
        }
    }
    let mut names = BTreeSet::new();
    walk_factors(query, &mut |tf: &TableFactor| {
        if let TableFactor::Table { name, .. } = tf {
            let n = name.to_string();
            if !ctes.contains(&n.to_ascii_lowercase()) {
                names.insert(n);
            }
        }
    });
    names.into_iter().collect()
}

/// Visit every [`TableFactor`] in a query, including joins, subqueries and CTEs.
fn walk_factors(query: &Query, f: &mut impl FnMut(&TableFactor)) {
    if let Some(with) = &query.with {
        for cte in &with.cte_tables {
            walk_factors(&cte.query, f);
        }
    }
    walk_set_expr(&query.body, f);
}

fn walk_set_expr(body: &SetExpr, f: &mut impl FnMut(&TableFactor)) {
    match body {
        SetExpr::Select(select) => {
            for twj in &select.from {
                walk_factor(&twj.relation, f);
                for join in &twj.joins {
                    walk_factor(&join.relation, f);
                }
            }
        }
        SetExpr::Query(q) => walk_factors(q, f),
        SetExpr::SetOperation { left, right, .. } => {
            walk_set_expr(left, f);
            walk_set_expr(right, f);
        }
        _ => {}
    }
}

fn walk_factor(tf: &TableFactor, f: &mut impl FnMut(&TableFactor)) {
    f(tf);
    match tf {
        TableFactor::Derived { subquery, .. } => walk_factors(subquery, f),
        TableFactor::NestedJoin {
            table_with_joins, ..
        } => {
            walk_factor(&table_with_joins.relation, f);
            for join in &table_with_joins.joins {
                walk_factor(&join.relation, f);
            }
        }
        _ => {}
    }
}

/// `walk_factors`, mutably.
fn walk_factors_mut(query: &mut Query, f: &mut impl FnMut(&mut TableFactor)) {
    if let Some(with) = &mut query.with {
        for cte in &mut with.cte_tables {
            walk_factors_mut(&mut cte.query, f);
        }
    }
    walk_set_expr_mut(&mut query.body, f);
}

fn walk_set_expr_mut(body: &mut SetExpr, f: &mut impl FnMut(&mut TableFactor)) {
    match body {
        SetExpr::Select(select) => {
            for twj in &mut select.from {
                walk_factor_mut(&mut twj.relation, f);
                for join in &mut twj.joins {
                    walk_factor_mut(&mut join.relation, f);
                }
            }
        }
        SetExpr::Query(q) => walk_factors_mut(q, f),
        SetExpr::SetOperation { left, right, .. } => {
            walk_set_expr_mut(left, f);
            walk_set_expr_mut(right, f);
        }
        _ => {}
    }
}

fn walk_factor_mut(tf: &mut TableFactor, f: &mut impl FnMut(&mut TableFactor)) {
    f(tf);
    match tf {
        TableFactor::Derived { subquery, .. } => walk_factors_mut(subquery, f),
        TableFactor::NestedJoin {
            table_with_joins, ..
        } => {
            walk_factor_mut(&mut table_with_joins.relation, f);
            for join in &mut table_with_joins.joins {
                walk_factor_mut(&mut join.relation, f);
            }
        }
        _ => {}
    }
}

/// Map a SQL type to the IR's Arrow **display string** (`Utf8`, `Int64`,
/// `Timestamp(Microsecond, None)`) — the form [`DatasetSchema`] stores so it
/// round-trips through RON and back into `arrow_schema::DataType`.
fn sql_type_to_arrow(ty: &sqlparser::ast::DataType) -> String {
    use sqlparser::ast::DataType as D;
    match ty {
        D::Boolean | D::Bool => "Boolean".into(),
        D::TinyInt(_) => "Int8".into(),
        D::SmallInt(_) => "Int16".into(),
        D::Int(_) | D::Integer(_) => "Int32".into(),
        D::BigInt(_) => "Int64".into(),
        D::TinyIntUnsigned(_) | D::UTinyInt => "UInt8".into(),
        D::SmallIntUnsigned(_) => "UInt16".into(),
        D::IntUnsigned(_) | D::IntegerUnsigned(_) => "UInt32".into(),
        D::BigIntUnsigned(_) | D::UBigInt => "UInt64".into(),
        D::Real | D::Float(_) | D::Float4 => "Float32".into(),
        D::Double(_) | D::DoublePrecision | D::Float8 => "Float64".into(),
        D::Date => "Date32".into(),
        D::Timestamp(_, _) | D::Datetime(_) => "Timestamp(Microsecond, None)".into(),
        D::Time(_, _) => "Time64(Microsecond)".into(),
        D::Decimal(info) | D::Numeric(info) | D::Dec(info) => match info {
            sqlparser::ast::ExactNumberInfo::PrecisionAndScale(p, s) => {
                format!("Decimal128({p}, {s})")
            }
            sqlparser::ast::ExactNumberInfo::Precision(p) => format!("Decimal128({p}, 0)"),
            sqlparser::ast::ExactNumberInfo::None => "Decimal128(38, 10)".into(),
        },
        D::Bytea | D::Blob(_) | D::Binary(_) | D::Varbinary(_) => "Binary".into(),
        // Utf8 is the right default: every remaining SQL type the IR can carry
        // is textual, and an unknown type is better stored as a string than
        // rejected outright.
        _ => "Utf8".into(),
    }
}

// ---------------------------------------------------------------------------
// Rendering (IR -> SQL)
// ---------------------------------------------------------------------------

fn render_dataset(d: &Dataset, flow: Option<&Flow>) -> String {
    let kind = match d.output_type {
        OutputType::MaterializedView => "CREATE MATERIALIZED VIEW",
        OutputType::Table => "CREATE STREAMING TABLE",
        OutputType::TemporaryView => "CREATE TEMPORARY VIEW",
        OutputType::Sink => "CREATE STREAMING TABLE",
    };
    let mut s = format!("{kind} {}", d.name);

    let expectations = flow.map(|f| f.expectations.as_slice()).unwrap_or(&[]);
    if !d.schema.is_empty() || !expectations.is_empty() {
        let mut parts: Vec<String> = d
            .schema
            .fields
            .iter()
            .map(|f| {
                format!(
                    "{} {}{}",
                    f.name,
                    arrow_to_sql_type(&f.arrow_type),
                    if f.nullable { "" } else { " NOT NULL" }
                )
            })
            .collect();
        for e in expectations {
            let policy = match e.on_violation {
                OnViolation::Warn => "",
                OnViolation::Drop => " ON VIOLATION DROP ROW",
                OnViolation::Fail => " ON VIOLATION FAIL UPDATE",
            };
            parts.push(format!(
                "CONSTRAINT {} EXPECT ({}){}",
                e.name, e.constraint, policy
            ));
        }
        s.push_str(&format!(" (\n  {}\n)", parts.join(",\n  ")));
    }

    if !d.partition_cols.is_empty() {
        s.push_str(&format!(
            "\nPARTITIONED BY ({})",
            d.partition_cols.join(", ")
        ));
    }
    if let Some(c) = &d.comment {
        s.push_str(&format!("\nCOMMENT '{}'", c.replace('\'', "''")));
    }
    if !d.properties.is_empty() {
        // BTreeMap iteration is sorted, so the rendered script is stable across
        // runs — this text gets diffed and historized.
        let props = d
            .properties
            .iter()
            .map(|(k, v)| format!("'{}' = '{}'", k.replace('\'', "''"), v.replace('\'', "''")))
            .collect::<Vec<_>>()
            .join(", ");
        s.push_str(&format!("\nTBLPROPERTIES ({props})"));
    }
    match flow.and_then(|f| f.query.clone()) {
        Some(q) => s.push_str(&format!("\nAS {q};\n")),
        None => s.push_str(";\n"),
    }
    s
}

fn render_standalone_flow(f: &Flow) -> String {
    let body = f.query.clone().unwrap_or_else(|| "SELECT 1".into());
    format!(
        "CREATE FLOW {} AS INSERT INTO {} BY NAME {body};\n",
        f.name, f.target
    )
}

/// The inverse of [`sql_type_to_arrow`] for the types the renderer emits.
fn arrow_to_sql_type(arrow: &str) -> &str {
    match arrow {
        "Boolean" => "BOOLEAN",
        "Int8" => "TINYINT",
        "Int16" => "SMALLINT",
        "Int32" => "INT",
        "Int64" => "BIGINT",
        "Float32" => "REAL",
        "Float64" => "DOUBLE",
        "Date32" => "DATE",
        "Timestamp(Microsecond, None)" => "TIMESTAMP",
        "Binary" => "BINARY",
        "Utf8" => "STRING",
        // Decimal128(p, s) and anything else the IR carries verbatim.
        other => other,
    }
}

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

    #[test]
    fn stream_function_and_bare_forms_both_resolve() {
        for sql in [
            "CREATE MATERIALIZED VIEW src AS SELECT 1 AS x;
             CREATE STREAMING TABLE st AS SELECT * FROM STREAM(src);",
            "CREATE MATERIALIZED VIEW src AS SELECT 1 AS x;
             CREATE STREAMING TABLE st AS SELECT * FROM STREAM src;",
        ] {
            let p = from_sql("p", sql).unwrap();
            let f = p.flows.iter().find(|f| f.target == "st").unwrap();
            assert_eq!(f.reads, vec!["src".to_string()]);
            assert!(
                f.kind.is_unbounded(),
                "STREAM(...) must produce a streaming flow"
            );
            // The rendered query must no longer mention STREAM.
            assert!(!f.query.as_ref().unwrap().to_uppercase().contains("STREAM"));
        }
    }

    #[test]
    fn partition_transform_survives_verbatim() {
        let p = from_sql(
            "p",
            "CREATE MATERIALIZED VIEW e PARTITIONED BY (months(event_ts), bucket(16, id))
             AS SELECT 1 AS id, CAST(1 AS TIMESTAMP) AS event_ts;",
        )
        .unwrap();
        assert_eq!(
            p.dataset("e").unwrap().partition_cols,
            vec!["months(event_ts)".to_string(), "bucket(16, id)".to_string()]
        );
    }

    #[test]
    fn cte_names_are_not_graph_edges() {
        let p = from_sql(
            "p",
            "CREATE MATERIALIZED VIEW base AS SELECT 1 AS x;
             CREATE MATERIALIZED VIEW roll AS
               WITH tmp AS (SELECT x FROM base) SELECT x FROM tmp;",
        )
        .unwrap();
        let f = p.flows.iter().find(|f| f.target == "roll").unwrap();
        assert_eq!(
            f.reads,
            vec!["base".to_string()],
            "`tmp` is a CTE, not a dataset"
        );
    }
}