mz_expr_parser/

parser.rs

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// As of the Change Date specified in that file, in accordance with
// the Business Source License, use of this software will be governed
// by the Apache License, Version 2.0.

use mz_ore::collections::CollectionExt;
use proc_macro2::LineColumn;
use syn::Error;
use syn::parse::{Parse, ParseStream, Parser};
use syn::spanned::Spanned;

use super::TestCatalog;

use self::util::*;

/// Builds a [mz_expr::MirRelationExpr] from a string.
pub fn try_parse_mir(catalog: &TestCatalog, s: &str) -> Result<mz_expr::MirRelationExpr, String> {
    // Define a Parser that constructs a (read-only) parsing context `ctx` and
    // delegates to `relation::parse_expr` by passing a `ctx` as a shared ref.
    let parser = move |input: ParseStream| {
        let ctx = Ctx { catalog };
        relation::parse_expr(&ctx, input)
    };
    // Since the syn lexer doesn't parse comments, we replace all `// {`
    // occurrences in the input string with `:: {`.
    let s = s.replace("// {", ":: {");
    // Call the parser with the given input string.
    let mut expr = parser.parse_str(&s).map_err(|err| {
        let (line, column) = (err.span().start().line, err.span().start().column);
        format!("parse error at {line}:{column}:\n{err}\n")
    })?;
    // Fix the types of the local let bindings of the parsed expression in a
    // post-processing pass.
    relation::fix_types(&mut expr, &mut relation::FixTypesCtx::default())?;
    // Return the parsed, post-processed expression.
    Ok(expr)
}

/// Builds a [mz_expr::MirScalarExpr] from a string.
pub fn try_parse_scalar(s: &str) -> Result<mz_expr::MirScalarExpr, String> {
    let parser = |input: syn::parse::ParseStream| {
        let expr = scalar::parse_expr(input)?;
        if !input.is_empty() {
            Err(Error::new(
                input.span(),
                "unexpected input after expression",
            ))?
        }
        Ok(expr)
    };
    parser.parse_str(s).map_err(|err| {
        let (line, column) = (err.span().start().line, err.span().start().column);
        format!("parse error at {line}:{column}:\n{err}\n")
    })
}

/// Builds a comma-separated list of [mz_expr::MirScalarExpr]s from a string.
pub fn try_parse_scalars(s: &str) -> Result<Vec<mz_expr::MirScalarExpr>, String> {
    let parser = |input: syn::parse::ParseStream| {
        let exprs = scalar::parse_exprs(input)?;
        if !input.is_empty() {
            Err(Error::new(
                input.span(),
                "unexpected input after expressions",
            ))?
        }
        Ok(exprs)
    };
    parser.parse_str(s).map_err(|err| {
        let (line, column) = (err.span().start().line, err.span().start().column);
        format!("parse error at {line}:{column}:\n{err}\n")
    })
}

/// Parses a parenthesized, comma-separated column type list, for example
/// `(bigint, text?)`. A trailing `?` marks a column as nullable.
pub fn try_parse_column_types(s: &str) -> Result<Vec<mz_repr::SqlColumnType>, String> {
    let parser = |input: syn::parse::ParseStream| {
        let inner;
        syn::parenthesized!(inner in input);
        let types = inner.parse_comma_sep(analyses::parse_column_type)?;
        if !input.is_empty() {
            Err(Error::new(input.span(), "unexpected input after type list"))?
        }
        Ok(types
            .iter()
            .map(mz_repr::SqlColumnType::from_repr)
            .collect())
    };
    parser.parse_str(s).map_err(|err| {
        let (line, column) = (err.span().start().line, err.span().start().column);
        format!("parse error at {line}:{column}:\n{err}\n")
    })
}

/// Builds a source definition from a string.
pub fn try_parse_def(catalog: &TestCatalog, s: &str) -> Result<Def, String> {
    // Define a Parser that constructs a (read-only) parsing context `ctx` and
    // delegates to `relation::parse_expr` by passing a `ctx` as a shared ref.
    let parser = move |input: ParseStream| {
        let ctx = Ctx { catalog };
        def::parse_def(&ctx, input)
    };
    // Call the parser with the given input string.
    let def = parser.parse_str(s).map_err(|err| {
        let (line, column) = (err.span().start().line, err.span().start().column);
        format!("parse error at {line}:{column}:\n{err}\n")
    })?;
    // Return the parsed, post-processed expression.
    Ok(def)
}

/// Support for parsing [mz_expr::MirRelationExpr].
mod relation {
    use std::collections::BTreeMap;

    use mz_expr::{AccessStrategy, Id, JoinImplementation, LocalId, MirRelationExpr};
    use mz_repr::{Diff, ReprRelationType, Row, SqlScalarType};

    use crate::parser::analyses::Analyses;

    use super::*;

    type Result = syn::Result<MirRelationExpr>;

    pub fn parse_expr(ctx: CtxRef, input: ParseStream) -> Result {
        let lookahead = input.lookahead1();
        if lookahead.peek(kw::Constant) {
            parse_constant(ctx, input)
        } else if lookahead.peek(kw::Get) {
            parse_get(ctx, input)
        } else if lookahead.peek(kw::Return) {
            parse_let_or_letrec_old(ctx, input)
        } else if lookahead.peek(kw::With) {
            parse_let_or_letrec(ctx, input)
        } else if lookahead.peek(kw::Project) {
            parse_project(ctx, input)
        } else if lookahead.peek(kw::Map) {
            parse_map(ctx, input)
        } else if lookahead.peek(kw::FlatMap) {
            parse_flat_map(ctx, input)
        } else if lookahead.peek(kw::Filter) {
            parse_filter(ctx, input)
        } else if lookahead.peek(kw::CrossJoin) {
            parse_cross_join(ctx, input)
        } else if lookahead.peek(kw::Join) {
            parse_join(ctx, input)
        } else if lookahead.peek(kw::Distinct) {
            parse_distinct(ctx, input)
        } else if lookahead.peek(kw::Reduce) {
            parse_reduce(ctx, input)
        } else if lookahead.peek(kw::TopK) {
            parse_top_k(ctx, input)
        } else if lookahead.peek(kw::Negate) {
            parse_negate(ctx, input)
        } else if lookahead.peek(kw::Threshold) {
            parse_threshold(ctx, input)
        } else if lookahead.peek(kw::Union) {
            parse_union(ctx, input)
        } else if lookahead.peek(kw::ArrangeBy) {
            parse_arrange_by(ctx, input)
        } else {
            Err(lookahead.error())
        }
    }

    fn parse_constant(_ctx: CtxRef, input: ParseStream) -> Result {
        let constant = input.parse::<kw::Constant>()?;

        let parse_typ = |input: ParseStream| -> syn::Result<ReprRelationType> {
            let analyses = analyses::parse_analyses(input)?;
            let Some(column_types) = analyses.types else {
                let msg = "Missing expected `types` analyses for Constant line";
                Err(Error::new(input.span(), msg))?
            };
            let keys = analyses.keys.unwrap_or_default();
            Ok(ReprRelationType::new(column_types).with_keys(keys))
        };
        if input.eat3(syn::Token![<], kw::empty, syn::Token![>]) {
            let typ = parse_typ(input)?;
            Ok(MirRelationExpr::Constant {
                rows: Ok(vec![]),
                typ,
            })
        } else {
            let typ = parse_typ(input)?;
            let parse_children = ParseChildren::new(input, constant.span().start());
            let rows = Ok(parse_children.parse_many(&typ, parse_constant_entry)?);
            Ok(MirRelationExpr::Constant { rows, typ })
        }
    }

    fn parse_constant_entry(
        typ: &ReprRelationType,
        input: ParseStream,
    ) -> syn::Result<(Row, Diff)> {
        input.parse::<syn::Token![-]>()?;

        let (row, diff);

        let inner1;
        syn::parenthesized!(inner1 in input);

        if inner1.peek(syn::token::Paren) {
            let inner2;
            syn::parenthesized!(inner2 in inner1);
            row = row::parse_row(&inner2, &typ.column_types)?;
            inner1.parse::<kw::x>()?;
            diff = match inner1.parse::<syn::Lit>()? {
                syn::Lit::Int(l) => Ok(l.base10_parse::<Diff>()?),
                _ => Err(Error::new(inner1.span(), "expected Diff literal")),
            }?;
        } else {
            row = row::parse_row(&inner1, &typ.column_types)?;
            diff = Diff::ONE;
        }

        Ok((row, diff))
    }

    fn parse_get(ctx: CtxRef, input: ParseStream) -> Result {
        input.parse::<kw::Get>()?;

        let ident = input.parse::<syn::Ident>()?;
        match ctx.catalog.get(&ident.to_string()) {
            Some((id, _cols, typ)) => Ok(MirRelationExpr::Get {
                id: Id::Global(*id),
                typ: ReprRelationType::from(typ),
                access_strategy: AccessStrategy::UnknownOrLocal,
            }),
            None => Ok(MirRelationExpr::Get {
                id: Id::Local(parse_local_id(ident)?),
                typ: ReprRelationType::empty(),
                access_strategy: AccessStrategy::UnknownOrLocal,
            }),
        }
    }

    /// Parses a Let or a LetRec with the old order: Return first, and then CTEs in descending order.
    fn parse_let_or_letrec_old(ctx: CtxRef, input: ParseStream) -> Result {
        let return_ = input.parse::<kw::Return>()?;
        let parse_body = ParseChildren::new(input, return_.span().start());
        let body = parse_body.parse_one(ctx, parse_expr)?;

        let with = input.parse::<kw::With>()?;
        let recursive = input.eat2(kw::Mutually, kw::Recursive);
        let parse_ctes = ParseChildren::new(input, with.span().start());
        let mut ctes = parse_ctes.parse_many(ctx, parse_cte)?;

        if ctes.is_empty() {
            let msg = "At least one Let/LetRec cte binding expected";
            Err(Error::new(input.span(), msg))?
        }

        ctes.reverse();
        let cte_ids = ctes.iter().map(|(id, _, _)| id);
        if !cte_ids.clone().is_sorted() {
            let msg = format!(
                "Error parsing Let/LetRec: seen Return before With, but cte ids are not ordered descending: {:?}",
                cte_ids.collect::<Vec<_>>()
            );
            Err(Error::new(input.span(), msg))?
        }
        build_let_or_let_rec(ctes, body, recursive, with)
    }

    /// Parses a Let or a LetRec with the new order: CTEs first in ascending order, and then Return.
    fn parse_let_or_letrec(ctx: CtxRef, input: ParseStream) -> Result {
        let with = input.parse::<kw::With>()?;
        let recursive = input.eat2(kw::Mutually, kw::Recursive);
        let parse_ctes = ParseChildren::new(input, with.span().start());
        let ctes = parse_ctes.parse_many(ctx, parse_cte)?;

        let return_ = input.parse::<kw::Return>()?;
        let parse_body = ParseChildren::new(input, return_.span().start());
        let body = parse_body.parse_one(ctx, parse_expr)?;

        if ctes.is_empty() {
            let msg = "At least one `let cte` binding expected";
            Err(Error::new(input.span(), msg))?
        }

        let cte_ids = ctes.iter().map(|(id, _, _)| id);
        if !cte_ids.clone().is_sorted() {
            let msg = format!(
                "Error parsing Let/LetRec: seen With before Return, but cte ids are not ordered ascending: {:?}",
                cte_ids.collect::<Vec<_>>()
            );
            Err(Error::new(input.span(), msg))?
        }
        build_let_or_let_rec(ctes, body, recursive, with)
    }

    fn build_let_or_let_rec(
        ctes: Vec<(LocalId, Analyses, MirRelationExpr)>,
        body: MirRelationExpr,
        recursive: bool,
        with: kw::With,
    ) -> Result {
        if recursive {
            let (mut ids, mut values, mut limits) = (vec![], vec![], vec![]);
            for (id, analyses, value) in ctes.into_iter() {
                let typ = {
                    let Some(column_types) = analyses.types else {
                        let msg = format!("`let {}` needs a `types` analyses", id);
                        Err(Error::new(with.span(), msg))?
                    };
                    let keys = analyses.keys.unwrap_or_default();
                    ReprRelationType::new(column_types).with_keys(keys)
                };
                // An ugly-ugly hack to pass the type information of the WMR CTE
                // to the `fix_types` pass.
                let value = {
                    let get_cte = MirRelationExpr::Get {
                        id: Id::Local(id),
                        typ,
                        access_strategy: AccessStrategy::UnknownOrLocal,
                    };
                    // Do not use the `union` smart constructor here!
                    MirRelationExpr::Union {
                        base: Box::new(get_cte),
                        inputs: vec![value],
                    }
                };

                ids.push(id);
                values.push(value);
                limits.push(None); // TODO: support limits
            }

            Ok(MirRelationExpr::LetRec {
                ids,
                values,
                limits,
                body: Box::new(body),
            })
        } else {
            let mut body = body;
            for (id, _, value) in ctes.into_iter().rev() {
                body = MirRelationExpr::Let {
                    id,
                    value: Box::new(value),
                    body: Box::new(body),
                };
            }
            Ok(body)
        }
    }

    fn parse_cte(
        ctx: CtxRef,
        input: ParseStream,
    ) -> syn::Result<(LocalId, analyses::Analyses, MirRelationExpr)> {
        let cte = input.parse::<kw::cte>()?;

        let ident = input.parse::<syn::Ident>()?;
        let id = parse_local_id(ident)?;

        input.parse::<syn::Token![=]>()?;

        let analyses = analyses::parse_analyses(input)?;

        let parse_value = ParseChildren::new(input, cte.span().start());
        let value = parse_value.parse_one(ctx, parse_expr)?;

        Ok((id, analyses, value))
    }

    fn parse_project(ctx: CtxRef, input: ParseStream) -> Result {
        let project = input.parse::<kw::Project>()?;

        let content;
        syn::parenthesized!(content in input);
        let outputs = content.parse_comma_sep(scalar::parse_column_index)?;
        let parse_input = ParseChildren::new(input, project.span().start());
        let input = Box::new(parse_input.parse_one(ctx, parse_expr)?);

        Ok(MirRelationExpr::Project { input, outputs })
    }

    fn parse_map(ctx: CtxRef, input: ParseStream) -> Result {
        let map = input.parse::<kw::Map>()?;

        let scalars = {
            let inner;
            syn::parenthesized!(inner in input);
            scalar::parse_exprs(&inner)?
        };

        let parse_input = ParseChildren::new(input, map.span().start());
        let input = Box::new(parse_input.parse_one(ctx, parse_expr)?);

        Ok(MirRelationExpr::Map { input, scalars })
    }

    fn parse_flat_map(ctx: CtxRef, input: ParseStream) -> Result {
        use mz_expr::TableFunc::*;

        let flat_map = input.parse::<kw::FlatMap>()?;

        let ident = input.parse::<syn::Ident>()?;
        let func = match ident.to_string().to_lowercase().as_str() {
            "unnest_list" => UnnestList {
                el_typ: SqlScalarType::Int64, // FIXME
            },
            "unnest_array" => UnnestArray {
                el_typ: SqlScalarType::Int64, // FIXME
            },
            "wrap1" => Wrap {
                types: vec![
                    SqlScalarType::Int64.nullable(true), // FIXME
                ],
                width: 1,
            },
            "wrap2" => Wrap {
                types: vec![
                    SqlScalarType::Int64.nullable(true), // FIXME
                    SqlScalarType::Int64.nullable(true), // FIXME
                ],
                width: 2,
            },
            "wrap3" => Wrap {
                types: vec![
                    SqlScalarType::Int64.nullable(true), // FIXME
                    SqlScalarType::Int64.nullable(true), // FIXME
                    SqlScalarType::Int64.nullable(true), // FIXME
                ],
                width: 3,
            },
            "generate_series" => GenerateSeriesInt64,
            // Both int widths display as `generate_series`; this spelling lets a
            // test pin the 32-bit variant (e.g. its widening casts) on input.
            "generate_series_i32" => GenerateSeriesInt32,
            "jsonb_object_keys" => JsonbObjectKeys,
            _ => Err(Error::new(ident.span(), "unsupported function name"))?,
        };

        let exprs = {
            let inner;
            syn::parenthesized!(inner in input);
            scalar::parse_exprs(&inner)?
        };

        let parse_input = ParseChildren::new(input, flat_map.span().start());
        let input = Box::new(parse_input.parse_one(ctx, parse_expr)?);

        Ok(MirRelationExpr::FlatMap { input, func, exprs })
    }

    fn parse_filter(ctx: CtxRef, input: ParseStream) -> Result {
        use mz_expr::MirScalarExpr::CallVariadic;
        use mz_expr::VariadicFunc::And;

        let filter = input.parse::<kw::Filter>()?;

        let predicates = match scalar::parse_expr(input)? {
            CallVariadic {
                func: And(_),
                exprs,
            } => exprs,
            expr => vec![expr],
        };

        let parse_input = ParseChildren::new(input, filter.span().start());
        let input = Box::new(parse_input.parse_one(ctx, parse_expr)?);

        Ok(MirRelationExpr::Filter { input, predicates })
    }

    fn parse_cross_join(ctx: CtxRef, input: ParseStream) -> Result {
        let join = input.parse::<kw::CrossJoin>()?;

        let parse_inputs = ParseChildren::new(input, join.span().start());
        let inputs = parse_inputs.parse_many(ctx, parse_expr)?;

        Ok(MirRelationExpr::Join {
            inputs,
            equivalences: vec![],
            implementation: JoinImplementation::Unimplemented,
        })
    }

    fn parse_join(ctx: CtxRef, input: ParseStream) -> Result {
        let join = input.parse::<kw::Join>()?;

        input.parse::<kw::on>()?;
        input.parse::<syn::Token![=]>()?;
        let inner;
        syn::parenthesized!(inner in input);
        let equivalences = scalar::parse_join_equivalences(&inner)?;

        let parse_inputs = ParseChildren::new(input, join.span().start());
        let inputs = parse_inputs.parse_many(ctx, parse_expr)?;

        Ok(MirRelationExpr::Join {
            inputs,
            equivalences,
            implementation: JoinImplementation::Unimplemented,
        })
    }

    fn parse_distinct(ctx: CtxRef, input: ParseStream) -> Result {
        let reduce = input.parse::<kw::Distinct>()?;

        let group_key = if input.eat(kw::project) {
            input.parse::<syn::Token![=]>()?;
            let inner;
            syn::bracketed!(inner in input);
            inner.parse_comma_sep(scalar::parse_expr)?
        } else {
            vec![]
        };

        let monotonic = input.eat(kw::monotonic);

        let expected_group_size = if input.eat(kw::exp_group_size) {
            input.parse::<syn::Token![=]>()?;
            Some(input.parse::<syn::LitInt>()?.base10_parse::<u64>()?)
        } else {
            None
        };

        let parse_inputs = ParseChildren::new(input, reduce.span().start());
        let input = Box::new(parse_inputs.parse_one(ctx, parse_expr)?);

        Ok(MirRelationExpr::Reduce {
            input,
            group_key,
            aggregates: vec![],
            monotonic,
            expected_group_size,
        })
    }

    fn parse_reduce(ctx: CtxRef, input: ParseStream) -> Result {
        let reduce = input.parse::<kw::Reduce>()?;

        let group_key = if input.eat(kw::group_by) {
            input.parse::<syn::Token![=]>()?;
            let inner;
            syn::bracketed!(inner in input);
            inner.parse_comma_sep(scalar::parse_expr)?
        } else {
            vec![]
        };

        let aggregates = {
            input.parse::<kw::aggregates>()?;
            input.parse::<syn::Token![=]>()?;
            let inner;
            syn::bracketed!(inner in input);
            inner.parse_comma_sep(aggregate::parse_expr)?
        };

        let monotonic = input.eat(kw::monotonic);

        let expected_group_size = if input.eat(kw::exp_group_size) {
            input.parse::<syn::Token![=]>()?;
            Some(input.parse::<syn::LitInt>()?.base10_parse::<u64>()?)
        } else {
            None
        };

        let parse_inputs = ParseChildren::new(input, reduce.span().start());
        let input = Box::new(parse_inputs.parse_one(ctx, parse_expr)?);

        Ok(MirRelationExpr::Reduce {
            input,
            group_key,
            aggregates,
            monotonic,
            expected_group_size,
        })
    }

    fn parse_top_k(ctx: CtxRef, input: ParseStream) -> Result {
        let top_k = input.parse::<kw::TopK>()?;

        let group_key = if input.eat(kw::group_by) {
            input.parse::<syn::Token![=]>()?;
            let inner;
            syn::bracketed!(inner in input);
            inner.parse_comma_sep(scalar::parse_column_index)?
        } else {
            vec![]
        };

        let order_key = if input.eat(kw::order_by) {
            input.parse::<syn::Token![=]>()?;
            let inner;
            syn::bracketed!(inner in input);
            inner.parse_comma_sep(scalar::parse_column_order)?
        } else {
            vec![]
        };

        let limit = if input.eat(kw::limit) {
            input.parse::<syn::Token![=]>()?;
            Some(scalar::parse_expr(input)?)
        } else {
            None
        };

        let offset = if input.eat(kw::offset) {
            input.parse::<syn::Token![=]>()?;
            input.parse::<syn::LitInt>()?.base10_parse::<usize>()?
        } else {
            0
        };

        let monotonic = input.eat(kw::monotonic);

        let expected_group_size = if input.eat(kw::exp_group_size) {
            input.parse::<syn::Token![=]>()?;
            Some(input.parse::<syn::LitInt>()?.base10_parse::<u64>()?)
        } else {
            None
        };

        let parse_inputs = ParseChildren::new(input, top_k.span().start());
        let input = Box::new(parse_inputs.parse_one(ctx, parse_expr)?);

        Ok(MirRelationExpr::TopK {
            input,
            group_key,
            order_key,
            limit,
            offset,
            monotonic,
            expected_group_size,
        })
    }

    fn parse_negate(ctx: CtxRef, input: ParseStream) -> Result {
        let negate = input.parse::<kw::Negate>()?;

        let parse_input = ParseChildren::new(input, negate.span().start());
        let input = Box::new(parse_input.parse_one(ctx, parse_expr)?);

        Ok(MirRelationExpr::Negate { input })
    }

    fn parse_threshold(ctx: CtxRef, input: ParseStream) -> Result {
        let threshold = input.parse::<kw::Threshold>()?;

        let parse_input = ParseChildren::new(input, threshold.span().start());
        let input = Box::new(parse_input.parse_one(ctx, parse_expr)?);

        Ok(MirRelationExpr::Threshold { input })
    }

    fn parse_union(ctx: CtxRef, input: ParseStream) -> Result {
        let union = input.parse::<kw::Union>()?;

        let parse_inputs = ParseChildren::new(input, union.span().start());
        let mut children = parse_inputs.parse_many(ctx, parse_expr)?;
        let inputs = children.split_off(1);
        let base = Box::new(children.into_element());

        Ok(MirRelationExpr::Union { base, inputs })
    }

    fn parse_arrange_by(ctx: CtxRef, input: ParseStream) -> Result {
        let arrange_by = input.parse::<kw::ArrangeBy>()?;

        let keys = {
            input.parse::<kw::keys>()?;
            input.parse::<syn::Token![=]>()?;
            let inner;
            syn::bracketed!(inner in input);
            inner.parse_comma_sep(|input| {
                let inner;
                syn::bracketed!(inner in input);
                scalar::parse_exprs(&inner)
            })?
        };

        let parse_input = ParseChildren::new(input, arrange_by.span().start());
        let input = Box::new(parse_input.parse_one(ctx, parse_expr)?);

        Ok(MirRelationExpr::ArrangeBy { input, keys })
    }

    fn parse_local_id(ident: syn::Ident) -> syn::Result<LocalId> {
        if ident.to_string().starts_with('l') {
            let n = ident.to_string()[1..]
                .parse::<u64>()
                .map_err(|err| Error::new(ident.span(), err.to_string()))?;
            Ok(mz_expr::LocalId::new(n))
        } else {
            Err(Error::new(ident.span(), "invalid LocalId"))
        }
    }

    #[derive(Default)]
    pub struct FixTypesCtx {
        env: BTreeMap<LocalId, ReprRelationType>,
        typ: Vec<ReprRelationType>,
    }

    pub fn fix_types(
        expr: &mut MirRelationExpr,
        ctx: &mut FixTypesCtx,
    ) -> std::result::Result<(), String> {
        match expr {
            MirRelationExpr::Let { id, value, body } => {
                fix_types(value, ctx)?;
                let value_typ = ctx.typ.pop().expect("value type");
                let prior_typ = ctx.env.insert(id.clone(), value_typ);
                fix_types(body, ctx)?;
                ctx.env.remove(id);
                if let Some(prior_typ) = prior_typ {
                    ctx.env.insert(id.clone(), prior_typ);
                }
            }
            MirRelationExpr::LetRec {
                ids,
                values,
                body,
                limits: _,
            } => {
                // An ugly-ugly hack to pass the type information of the WMR CTE
                // to the `fix_types` pass.
                let mut prior_typs = BTreeMap::default();
                for (id, value) in std::iter::zip(ids.iter_mut(), values.iter_mut()) {
                    let MirRelationExpr::Union { base, mut inputs } = value.take_dangerous() else {
                        unreachable!("ensured by construction");
                    };
                    let MirRelationExpr::Get { id: _, typ, .. } = *base else {
                        unreachable!("ensured by construction");
                    };
                    if let Some(prior_typ) = ctx.env.insert(id.clone(), typ) {
                        prior_typs.insert(id.clone(), prior_typ);
                    }
                    *value = inputs.pop().expect("ensured by construction");
                }
                for value in values.iter_mut() {
                    fix_types(value, ctx)?;
                }
                fix_types(body, ctx)?;
                for id in ids.iter() {
                    ctx.env.remove(id);
                    if let Some(prior_typ) = prior_typs.remove(id) {
                        ctx.env.insert(id.clone(), prior_typ);
                    }
                }
            }
            MirRelationExpr::Get {
                id: Id::Local(id),
                typ,
                ..
            } => {
                let env_typ = match ctx.env.get(&*id) {
                    Some(env_typ) => env_typ,
                    None => Err(format!("Cannot fix type of unbound CTE {}", id))?,
                };
                *typ = env_typ.clone();
                ctx.typ.push(env_typ.clone());
            }
            _ => {
                for input in expr.children_mut() {
                    fix_types(input, ctx)?;
                }
                let input_types = ctx.typ.split_off(ctx.typ.len() - expr.num_inputs());
                ctx.typ.push(expr.typ_with_input_types(&input_types));
            }
        };

        Ok(())
    }
}

/// Support for parsing [mz_expr::MirScalarExpr].
mod scalar {
    use mz_expr::{BinaryFunc, ColumnOrder, MirScalarExpr, UnaryFunc, VariadicFunc, func};
    use mz_ore::collections::CollectionExt;
    use mz_repr::adt::jsonb::JsonbPacker;
    use mz_repr::adt::numeric::NumericMaxScale;
    use mz_repr::{
        AsColumnType, ColumnName, Datum, ReprColumnType, ReprScalarType, Row, RowArena,
        SqlScalarType, strconv,
    };

    use super::*;

    type Result = syn::Result<MirScalarExpr>;

    pub fn parse_exprs(input: ParseStream) -> syn::Result<Vec<MirScalarExpr>> {
        input.parse_comma_sep(parse_expr)
    }

    /// Parses a single expression.
    ///
    /// Because in EXPLAIN contexts parentheses might be optional, we need to
    /// correctly handle operator precedence of infix operators.
    ///
    /// Currently, this works in two steps:
    ///
    /// 1. Convert the original infix expression to a postfix expression using
    ///    an adapted variant of this [algorithm] with precedence taken from the
    ///    Postgres [precedence] docs. Parenthesized operands are parsed in one
    ///    step, so steps (3-4) from the [algorithm] are not needed here.
    /// 2. Convert the postfix vector into a single [MirScalarExpr].
    ///
    /// [algorithm]: <https://www.prepbytes.com/blog/stacks/infix-to-postfix-conversion-using-stack/>
    /// [precedence]: <https://www.postgresql.org/docs/7.2/sql-precedence.html>
    pub fn parse_expr(input: ParseStream) -> Result {
        let line = input.span().start().line;

        /// Helper struct to keep track of the parsing state.
        #[derive(Debug)]
        enum Op {
            Unr(mz_expr::UnaryFunc), // unary
            Neg(mz_expr::UnaryFunc), // negated unary (append -.not() on fold)
            Bin(mz_expr::BinaryFunc),
            Var(mz_expr::VariadicFunc),
        }

        impl Op {
            fn precedence(&self) -> Option<usize> {
                match self {
                    // 01: logical disjunction
                    Op::Var(mz_expr::VariadicFunc::Or(_)) => Some(1),
                    // 02: logical conjunction
                    Op::Var(mz_expr::VariadicFunc::And(_)) => Some(2),
                    // 04: equality, assignment
                    Op::Bin(mz_expr::BinaryFunc::Eq(_)) => Some(4),
                    Op::Bin(mz_expr::BinaryFunc::NotEq(_)) => Some(4),
                    // 05: less than, greater than
                    Op::Bin(mz_expr::BinaryFunc::Gt(_)) => Some(5),
                    Op::Bin(mz_expr::BinaryFunc::Gte(_)) => Some(5),
                    Op::Bin(mz_expr::BinaryFunc::Lt(_)) => Some(5),
                    Op::Bin(mz_expr::BinaryFunc::Lte(_)) => Some(5),
                    // 13: test for TRUE, FALSE, UNKNOWN, NULL
                    Op::Unr(mz_expr::UnaryFunc::IsNull(_)) => Some(13),
                    Op::Neg(mz_expr::UnaryFunc::IsNull(_)) => Some(13),
                    Op::Unr(mz_expr::UnaryFunc::IsTrue(_)) => Some(13),
                    Op::Neg(mz_expr::UnaryFunc::IsTrue(_)) => Some(13),
                    Op::Unr(mz_expr::UnaryFunc::IsFalse(_)) => Some(13),
                    Op::Neg(mz_expr::UnaryFunc::IsFalse(_)) => Some(13),
                    // 14: addition, subtraction
                    Op::Bin(mz_expr::BinaryFunc::AddInt64(_)) => Some(14),
                    // 14: multiplication, division, modulo
                    Op::Bin(mz_expr::BinaryFunc::MulInt64(_)) => Some(15),
                    Op::Bin(mz_expr::BinaryFunc::DivInt64(_)) => Some(15),
                    Op::Bin(mz_expr::BinaryFunc::ModInt64(_)) => Some(15),
                    // unsupported
                    _ => None,
                }
            }
        }

        /// Helper struct for entries in the postfix vector.
        #[derive(Debug)]
        enum Entry {
            Operand(MirScalarExpr),
            Operator(Op),
        }

        let mut opstack = vec![];
        let mut postfix = vec![];
        let mut exp_opd = true; // expects an argument of an operator

        // Scan the given infix expression from left to right.
        while !input.is_empty() && input.span().start().line == line {
            // Operands and operators alternate.
            if exp_opd {
                postfix.push(Entry::Operand(parse_operand(input)?));
                exp_opd = false;
            } else {
                // If the current symbol is an operator, then bind it to op.
                // Else it is an operand - append it to postfix and continue.
                let op = if input.eat(syn::Token![=]) {
                    exp_opd = true;
                    Op::Bin(func::Eq.into())
                } else if input.eat(syn::Token![!=]) {
                    exp_opd = true;
                    Op::Bin(func::NotEq.into())
                } else if input.eat(syn::Token![>=]) {
                    exp_opd = true;
                    Op::Bin(func::Gte.into())
                } else if input.eat(syn::Token![>]) {
                    exp_opd = true;
                    Op::Bin(func::Gt.into())
                } else if input.eat(syn::Token![<=]) {
                    exp_opd = true;
                    Op::Bin(func::Lte.into())
                } else if input.eat(syn::Token![<]) {
                    exp_opd = true;
                    Op::Bin(func::Lt.into())
                } else if input.eat(syn::Token![+]) {
                    exp_opd = true;
                    Op::Bin(func::AddInt64.into()) // TODO: fix placeholder
                } else if input.eat(syn::Token![*]) {
                    exp_opd = true;
                    Op::Bin(func::MulInt64.into()) // TODO: fix placeholder
                } else if input.eat(syn::Token![/]) {
                    exp_opd = true;
                    Op::Bin(func::DivInt64.into()) // TODO: fix placeholder
                } else if input.eat(syn::Token![%]) {
                    exp_opd = true;
                    Op::Bin(func::ModInt64.into()) // TODO: fix placeholder
                } else if input.eat(kw::AND) {
                    exp_opd = true;
                    Op::Var(VariadicFunc::And(func::variadic::And))
                } else if input.eat(kw::OR) {
                    exp_opd = true;
                    Op::Var(VariadicFunc::Or(func::variadic::Or))
                } else if input.eat(kw::coalesce) {
                    exp_opd = true;
                    Op::Var(VariadicFunc::Coalesce(func::variadic::Coalesce))
                } else if input.eat(kw::IS) {
                    let negate = input.eat(kw::NOT);

                    let lookahead = input.lookahead1();
                    let func = if input.look_and_eat(kw::NULL, &lookahead) {
                        mz_expr::func::IsNull.into()
                    } else if input.look_and_eat(kw::TRUE, &lookahead) {
                        mz_expr::func::IsTrue.into()
                    } else if input.look_and_eat(kw::FALSE, &lookahead) {
                        mz_expr::func::IsFalse.into()
                    } else {
                        Err(lookahead.error())?
                    };

                    if negate { Op::Neg(func) } else { Op::Unr(func) }
                } else {
                    // We were expecting an optional operator but didn't find
                    // anything. Exit the parsing loop and process the postfix
                    // vector.
                    break;
                };

                // First, pop the operators which are already on the opstack that
                // have higher or equal precedence than the current operator and
                // append them to the postfix.
                while opstack
                    .last()
                    .map(|op1: &Op| op1.precedence() >= op.precedence())
                    .unwrap_or(false)
                {
                    let op1 = opstack.pop().expect("non-empty opstack");
                    postfix.push(Entry::Operator(op1));
                }

                // Then push the op from this iteration onto the stack.
                opstack.push(op);
            }
        }

        // Pop all remaining symbols from opstack and append them to postfix.
        postfix.extend(opstack.into_iter().rev().map(Entry::Operator));

        if postfix.is_empty() {
            let msg = "Cannot parse an empty expression";
            Err(Error::new(input.span(), msg))?
        }

        // Flatten the postfix vector into a single MirScalarExpr.
        let mut stack = vec![];
        postfix.reverse();
        while let Some(entry) = postfix.pop() {
            match entry {
                Entry::Operand(expr) => {
                    stack.push(expr);
                }
                Entry::Operator(Op::Unr(func)) => {
                    let expr = Box::new(stack.pop().expect("non-empty stack"));
                    stack.push(MirScalarExpr::CallUnary { func, expr });
                }
                Entry::Operator(Op::Neg(func)) => {
                    let expr = Box::new(stack.pop().expect("non-empty stack"));
                    stack.push(MirScalarExpr::CallUnary { func, expr }.not());
                }
                Entry::Operator(Op::Bin(func)) => {
                    let expr2 = Box::new(stack.pop().expect("non-empty stack"));
                    let expr1 = Box::new(stack.pop().expect("non-empty stack"));
                    stack.push(MirScalarExpr::CallBinary { func, expr1, expr2 });
                }
                Entry::Operator(Op::Var(func)) => {
                    let expr2 = stack.pop().expect("non-empty stack");
                    let expr1 = stack.pop().expect("non-empty stack");
                    let mut exprs = vec![];
                    for expr in [expr1, expr2] {
                        match expr {
                            MirScalarExpr::CallVariadic { func: f, exprs: es } if f == func => {
                                exprs.extend(es);
                            }
                            expr => {
                                exprs.push(expr);
                            }
                        }
                    }
                    stack.push(MirScalarExpr::CallVariadic { func, exprs });
                }
            }
        }

        if stack.len() != 1 {
            let msg = "Cannot fold postfix vector into a single MirScalarExpr";
            Err(Error::new(input.span(), msg))?
        }

        Ok(stack.pop().unwrap())
    }

    pub fn parse_operand(input: ParseStream) -> Result {
        let lookahead = input.lookahead1();
        if lookahead.peek(syn::Token![#]) {
            parse_column(input)
        } else if lookahead.peek(syn::Lit) || lookahead.peek(kw::null) {
            parse_literal_ok(input)
        } else if lookahead.peek(kw::error) {
            parse_literal_err(input)
        } else if lookahead.peek(kw::array) {
            parse_array(input)
        } else if lookahead.peek(kw::list) {
            parse_list(input)
        } else if lookahead.peek(kw::case) {
            parse_case(input)
        } else if lookahead.peek(syn::Ident) {
            parse_apply(input)
        } else if lookahead.peek(syn::token::Brace) {
            let inner;
            syn::braced!(inner in input);
            parse_literal_array(&inner)
        } else if lookahead.peek(syn::token::Bracket) {
            let inner;
            syn::bracketed!(inner in input);
            parse_literal_list(&inner)
        } else if lookahead.peek(syn::token::Paren) {
            let inner;
            syn::parenthesized!(inner in input);
            parse_expr(&inner)
        } else {
            Err(lookahead.error())
        }
    }

    /// Parses `case when {cond} then {then} else {els} end`.
    fn parse_case(input: ParseStream) -> Result {
        input.parse::<kw::case>()?;
        if input.peek(kw::when) {
            input.parse::<kw::when>()?;
            let cond = parse_expr(input)?;
            input.parse::<kw::then>()?;
            let then = parse_expr(input)?;
            input.parse::<syn::Token![else]>()?;
            let els = parse_expr(input)?;
            input.parse::<kw::end>()?;
            Ok(MirScalarExpr::If {
                cond: Box::new(cond),
                then: Box::new(then),
                els: Box::new(els),
            })
        } else {
            Err(Error::new(input.span(), "expected 'when' after 'case'"))
        }
    }

    pub fn parse_column(input: ParseStream) -> Result {
        Ok(MirScalarExpr::column(parse_column_index(input)?))
    }

    pub fn parse_column_index(input: ParseStream) -> syn::Result<usize> {
        input.parse::<syn::Token![#]>()?;
        input.parse::<syn::LitInt>()?.base10_parse::<usize>()
    }

    pub fn parse_column_order(input: ParseStream) -> syn::Result<ColumnOrder> {
        input.parse::<syn::Token![#]>()?;
        let column = input.parse::<syn::LitInt>()?.base10_parse::<usize>()?;
        let desc = input.eat(kw::desc) || !input.eat(kw::asc);
        let nulls_last = input.eat(kw::nulls_last) || !input.eat(kw::nulls_first);
        Ok(ColumnOrder {
            column,
            desc,
            nulls_last,
        })
    }

    fn parse_literal_ok(input: ParseStream) -> Result {
        let mut row = Row::default();
        let mut packer = row.packer();

        let typ = if input.eat(kw::null) {
            packer.push(Datum::Null);
            input.parse::<syn::Token![::]>()?;
            ReprColumnType {
                scalar_type: analyses::parse_scalar_type(input)?,
                nullable: true,
            }
        } else {
            let lit = input.parse::<syn::Lit>()?;
            if input.peek(syn::Token![::]) {
                // A literal with an explicit type, e.g. `2000.0::numeric` or
                // `"{}"::jsonb`. Coerce the literal token to the target type.
                input.parse::<syn::Token![::]>()?;
                let scalar_type = analyses::parse_scalar_type(input)?;
                parse_typed_literal(&mut packer, &lit, &scalar_type)?;
                ReprColumnType {
                    scalar_type,
                    nullable: false,
                }
            } else {
                match lit {
                    syn::Lit::Str(l) => {
                        packer.push(Datum::from(l.value().as_str()));
                        Ok(ReprColumnType::from(&String::as_column_type()))
                    }
                    syn::Lit::Int(l) => {
                        packer.push(Datum::from(l.base10_parse::<i64>()?));
                        Ok(ReprColumnType::from(&i64::as_column_type()))
                    }
                    syn::Lit::Float(l) => {
                        packer.push(Datum::from(l.base10_parse::<f64>()?));
                        Ok(ReprColumnType::from(&f64::as_column_type()))
                    }
                    syn::Lit::Bool(l) => {
                        packer.push(Datum::from(l.value));
                        Ok(ReprColumnType::from(&bool::as_column_type()))
                    }
                    _ => Err(Error::new(input.span(), "cannot parse literal")),
                }?
            }
        };

        Ok(MirScalarExpr::Literal(Ok(row), typ))
    }

    /// Packs a literal token coerced to the given target type.
    fn parse_typed_literal(
        packer: &mut mz_repr::RowPacker,
        lit: &syn::Lit,
        scalar_type: &ReprScalarType,
    ) -> syn::Result<()> {
        use syn::Lit::*;
        let err = |msg: String| Error::new(lit.span(), msg);
        match (lit, scalar_type) {
            (Int(l), ReprScalarType::Int16) => packer.push(Datum::from(l.base10_parse::<i16>()?)),
            (Int(l), ReprScalarType::Int32) => packer.push(Datum::from(l.base10_parse::<i32>()?)),
            (Int(l), ReprScalarType::Int64) => packer.push(Datum::from(l.base10_parse::<i64>()?)),
            (Int(l), ReprScalarType::Float64) => packer.push(Datum::from(l.base10_parse::<f64>()?)),
            (Float(l), ReprScalarType::Float64) => {
                packer.push(Datum::from(l.base10_parse::<f64>()?))
            }
            (Int(l), ReprScalarType::Numeric { .. }) => {
                let n = strconv::parse_numeric(l.base10_digits())
                    .map_err(|e| err(format!("invalid numeric literal: {e}")))?;
                packer.push(Datum::Numeric(n));
            }
            (Float(l), ReprScalarType::Numeric { .. }) => {
                let n = strconv::parse_numeric(l.base10_digits())
                    .map_err(|e| err(format!("invalid numeric literal: {e}")))?;
                packer.push(Datum::Numeric(n));
            }
            (Str(l), ReprScalarType::String) => packer.push(Datum::from(l.value().as_str())),
            (Str(l), ReprScalarType::Jsonb) => {
                JsonbPacker::new(packer)
                    .pack_str(&l.value())
                    .map_err(|e| err(format!("invalid jsonb literal: {e}")))?;
            }
            _ => Err(err("unsupported literal type annotation".to_string()))?,
        }
        Ok(())
    }
    fn parse_literal_err(input: ParseStream) -> Result {
        input.parse::<kw::error>()?;
        let mut msg = {
            let content;
            syn::parenthesized!(content in input);
            content.parse::<syn::LitStr>()?.value()
        };
        let err = if msg.starts_with("internal error: ") {
            Ok(mz_expr::EvalError::Internal(msg.split_off(16).into()))
        } else {
            Err(Error::new(msg.span(), "expected `internal error: $msg`"))
        }?;
        Ok(MirScalarExpr::literal(Err(err), ReprScalarType::Bool))
    }

    fn parse_literal_array(input: ParseStream) -> Result {
        use mz_expr::func::variadic::ArrayCreate;

        let elem_type = SqlScalarType::Int64; // FIXME
        let func = VariadicFunc::ArrayCreate(ArrayCreate { elem_type });
        let exprs = input.parse_comma_sep(parse_literal_ok)?;

        // Evaluate into a datum
        let temp_storage = RowArena::default();
        let datum = func.eval(&[], &temp_storage, &exprs).expect("datum");
        let typ = ReprScalarType::from(&SqlScalarType::Array(Box::new(SqlScalarType::Int64))); // FIXME
        Ok(MirScalarExpr::literal_ok(datum, typ))
    }
    fn parse_literal_list(input: ParseStream) -> Result {
        use mz_expr::func::variadic::ListCreate;

        let elem_type = SqlScalarType::Int64; // FIXME
        let func = VariadicFunc::ListCreate(ListCreate { elem_type });
        let exprs = input.parse_comma_sep(parse_literal_ok)?;

        // Evaluate into a datum
        let temp_storage = RowArena::default();
        let datum = func.eval(&[], &temp_storage, &exprs).expect("datum");
        let typ = ReprScalarType::from(&SqlScalarType::Array(Box::new(SqlScalarType::Int64))); // FIXME
        Ok(MirScalarExpr::literal_ok(datum, typ))
    }
    fn parse_array(input: ParseStream) -> Result {
        use mz_expr::func::variadic::ArrayCreate;

        input.parse::<kw::array>()?;

        // parse brackets
        let inner;
        syn::bracketed!(inner in input);

        let elem_type = SqlScalarType::Int64; // FIXME
        let func = ArrayCreate { elem_type };
        let exprs = inner.parse_comma_sep(parse_expr)?;

        Ok(MirScalarExpr::call_variadic(func, exprs))
    }

    fn parse_list(input: ParseStream) -> Result {
        use mz_expr::func::variadic::ListCreate;

        input.parse::<kw::list>()?;

        // parse brackets
        let inner;
        syn::bracketed!(inner in input);

        let elem_type = SqlScalarType::Int64; // FIXME
        let func = ListCreate { elem_type };
        let exprs = inner.parse_comma_sep(parse_expr)?;

        Ok(MirScalarExpr::call_variadic(func, exprs))
    }

    fn parse_apply(input: ParseStream) -> Result {
        let ident = input.parse::<syn::Ident>()?;

        // Function variants with parameters take them in brackets before the
        // argument list, e.g. `cast_int32_to_numeric[127](#0)`.
        if input.peek(syn::token::Bracket) {
            return parse_apply_parameterized(&ident, input);
        }

        // parse parentheses
        let inner;
        syn::parenthesized!(inner in input);

        let parse_unary = |func: UnaryFunc| -> Result {
            let expr = Box::new(parse_expr(&inner)?);
            Ok(MirScalarExpr::CallUnary { func, expr })
        };
        let parse_binary = |func: BinaryFunc| -> Result {
            let expr1 = Box::new(parse_expr(&inner)?);
            inner.parse::<syn::Token![,]>()?;
            let expr2 = Box::new(parse_expr(&inner)?);
            Ok(MirScalarExpr::CallBinary { func, expr1, expr2 })
        };
        let parse_variadic = |func: VariadicFunc| -> Result {
            let exprs = inner.parse_comma_sep(parse_expr)?;
            Ok(MirScalarExpr::call_variadic(func, exprs))
        };

        // Infix binary and variadic function calls are handled in `parse_scalar_expr`.
        //
        // Some restrictions apply with the current state of the code,
        // most notably one cannot handle overloaded function names because we don't want to do
        // name resolution in the parser.
        match ident.to_string().to_lowercase().as_str() {
            // Supported unary functions:
            "abs" => parse_unary(func::AbsInt64.into()),
            "not" => parse_unary(func::Not.into()),
            // Supported binary functions:
            "ltrim" => parse_binary(func::TrimLeading.into()),
            // Supported variadic functions:
            "greatest" => parse_variadic(VariadicFunc::Greatest(func::variadic::Greatest)),
            "coalesce" => parse_variadic(VariadicFunc::Coalesce(func::variadic::Coalesce)),
            // Supported unmaterializable functions:
            "mz_now" => Ok(MirScalarExpr::CallUnmaterializable(
                mz_expr::UnmaterializableFunc::MzNow,
            )),
            // Exact function variants by their canonical name, dispatched on
            // the argument count (unary and binary win over variadic).
            name => {
                let exprs = inner.parse_comma_sep(parse_expr)?;
                parse_apply_variant(&ident, name, exprs)
            }
        }
    }

    /// Applies a function variant named by its canonical (snake_case) name,
    /// e.g. `add_int32(#0, #1)`. See `FuncName` in `mz_expr`.
    fn parse_apply_variant(
        ident: &syn::Ident,
        name: &str,
        mut exprs: Vec<MirScalarExpr>,
    ) -> Result {
        if exprs.len() == 1 {
            if let Some(func) = UnaryFunc::from_variant_name(name) {
                let expr = Box::new(exprs.into_element());
                return Ok(MirScalarExpr::CallUnary { func, expr });
            }
        }
        if exprs.len() == 2 {
            if let Some(func) = BinaryFunc::from_variant_name(name) {
                let expr2 = Box::new(exprs.pop().expect("two exprs"));
                let expr1 = Box::new(exprs.pop().expect("two exprs"));
                return Ok(MirScalarExpr::CallBinary { func, expr1, expr2 });
            }
        }
        if let Some(func) = VariadicFunc::from_variant_name(name) {
            return Ok(MirScalarExpr::CallVariadic { func, exprs });
        }
        Err(Error::new(ident.span(), "unsupported function name"))
    }

    /// Applies a parameterized function variant. The variant's parameters
    /// appear in brackets between the name and the argument list.
    fn parse_apply_parameterized(ident: &syn::Ident, input: ParseStream) -> Result {
        let params;
        syn::bracketed!(params in input);
        let inner;
        syn::parenthesized!(inner in input);

        match ident.to_string().to_lowercase().as_str() {
            "cast_int32_to_numeric" => {
                let max_scale = if params.is_empty() {
                    None
                } else {
                    let scale = params.parse::<syn::LitInt>()?.base10_parse::<i64>()?;
                    let scale = NumericMaxScale::try_from(scale)
                        .map_err(|e| Error::new(params.span(), e.to_string()))?;
                    Some(scale)
                };
                let expr = Box::new(parse_expr(&inner)?);
                Ok(MirScalarExpr::CallUnary {
                    func: func::CastInt32ToNumeric(max_scale).into(),
                    expr,
                })
            }
            "record_get" => {
                let index = params.parse::<syn::LitInt>()?.base10_parse::<usize>()?;
                let expr = Box::new(parse_expr(&inner)?);
                Ok(MirScalarExpr::CallUnary {
                    func: func::RecordGet(index).into(),
                    expr,
                })
            }
            "record_create" => {
                let field_names = params
                    .parse_comma_sep(|p| Ok(ColumnName::from(p.parse::<syn::LitStr>()?.value())))?;
                let exprs = inner.parse_comma_sep(parse_expr)?;
                Ok(MirScalarExpr::call_variadic(
                    func::variadic::RecordCreate { field_names },
                    exprs,
                ))
            }
            "list_create" => {
                let elem_type = SqlScalarType::from_repr(&analyses::parse_scalar_type(&params)?);
                let exprs = inner.parse_comma_sep(parse_expr)?;
                Ok(MirScalarExpr::call_variadic(
                    func::variadic::ListCreate { elem_type },
                    exprs,
                ))
            }
            _ => Err(Error::new(
                ident.span(),
                "unsupported parameterized function",
            )),
        }
    }

    pub fn parse_join_equivalences(input: ParseStream) -> syn::Result<Vec<Vec<MirScalarExpr>>> {
        let mut equivalences = vec![];
        while !input.is_empty() {
            let mut equivalence = vec![];
            loop {
                let mut worklist = vec![parse_operand(input)?];
                while let Some(operand) = worklist.pop() {
                    // Be more lenient and support parenthesized equivalences,
                    // e.g. `... AND (x = u + v = z + 1) AND ...`.
                    if let MirScalarExpr::CallBinary {
                        func: BinaryFunc::Eq(_),
                        expr1,
                        expr2,
                    } = operand
                    {
                        // We reverse the order in the worklist in order to get
                        // the correct order in the equivalence class.
                        worklist.push(*expr2);
                        worklist.push(*expr1);
                    } else {
                        equivalence.push(operand);
                    }
                }
                if !input.eat(syn::Token![=]) {
                    break;
                }
            }
            equivalences.push(equivalence);
            input.eat(kw::AND);
        }
        Ok(equivalences)
    }
}

/// Support for parsing [mz_expr::AggregateExpr].
mod aggregate {
    use mz_expr::{AggregateExpr, MirScalarExpr};

    use super::*;

    type Result = syn::Result<AggregateExpr>;

    pub fn parse_expr(input: ParseStream) -> Result {
        use mz_expr::AggregateFunc::*;

        // Some restrictions apply with the current state of the code,
        // most notably one cannot handle overloaded function names because we don't want to do
        // name resolution in the parser.
        let ident = input.parse::<syn::Ident>()?;
        let func = match ident.to_string().to_lowercase().as_str() {
            "count" => Count,
            "any" => Any,
            "all" => All,
            "max" => MaxInt64,
            "min" => MinInt64,
            "sum" => SumInt64,
            // Exact variants, for tests that pin a specific width.
            "max_int32" => MaxInt32,
            "max_int64" => MaxInt64,
            "min_int32" => MinInt32,
            "min_int64" => MinInt64,
            "sum_int16" => SumInt16,
            "sum_int32" => SumInt32,
            "sum_int64" => SumInt64,
            _ => Err(Error::new(ident.span(), "unsupported function name"))?,
        };

        // parse parentheses
        let inner;
        syn::parenthesized!(inner in input);

        if func == Count && inner.eat(syn::Token![*]) {
            Ok(AggregateExpr {
                func,
                expr: MirScalarExpr::literal_true(),
                distinct: false, // TODO: fix explain output
            })
        } else {
            let distinct = inner.eat(kw::distinct);
            let expr = scalar::parse_expr(&inner)?;
            Ok(AggregateExpr {
                func,
                expr,
                distinct,
            })
        }
    }
}

/// Support for parsing [mz_repr::Row].
mod row {
    use mz_repr::{Datum, ReprColumnType, ReprScalarType, Row};

    use super::*;

    /// Parses a comma-separated datum list into a [`Row`], coercing each datum
    /// to the corresponding column type where the literal admits it.
    ///
    /// Literals that do not match their column type (and datums beyond the
    /// column count) fall back to type inference from the literal alone. The
    /// parser deliberately does not reject such rows, ill-typed constants are
    /// valid parser output that tests feed to the typechecker.
    pub fn parse_row(input: ParseStream, types: &[ReprColumnType]) -> syn::Result<Row> {
        let mut row = Row::default();
        let mut packer = row.packer();

        let mut types = types.iter();
        loop {
            if input.is_empty() {
                break;
            }
            parse_datum(input, &mut packer, types.next())?;
            if input.is_empty() {
                break;
            }
            input.parse::<syn::Token![,]>()?;
        }

        Ok(row)
    }

    fn parse_datum(
        input: ParseStream,
        packer: &mut mz_repr::RowPacker,
        typ: Option<&ReprColumnType>,
    ) -> syn::Result<()> {
        use ReprScalarType::*;
        if input.eat(kw::null) {
            packer.push(Datum::Null);
            return Ok(());
        }
        let lit = input.parse::<syn::Lit>()?;
        match (&lit, typ.map(|t| &t.scalar_type)) {
            (syn::Lit::Int(l), Some(Int16)) => packer.push(Datum::from(l.base10_parse::<i16>()?)),
            (syn::Lit::Int(l), Some(Int32)) => packer.push(Datum::from(l.base10_parse::<i32>()?)),
            (syn::Lit::Int(l), Some(Float64)) => packer.push(Datum::from(l.base10_parse::<f64>()?)),
            // Literal-only inference, also the fallback for literals that do
            // not match their column type.
            (syn::Lit::Str(l), _) => packer.push(Datum::from(l.value().as_str())),
            (syn::Lit::Int(l), _) => packer.push(Datum::from(l.base10_parse::<i64>()?)),
            (syn::Lit::Float(l), _) => packer.push(Datum::from(l.base10_parse::<f64>()?)),
            (syn::Lit::Bool(l), _) => packer.push(Datum::from(l.value)),
            (lit, _) => Err(Error::new(lit.span(), "cannot parse literal"))?,
        }
        Ok(())
    }
}

mod analyses {
    use mz_repr::{ReprColumnType, ReprScalarType};

    use super::*;

    #[derive(Default)]
    pub struct Analyses {
        pub types: Option<Vec<ReprColumnType>>,
        pub keys: Option<Vec<Vec<usize>>>,
    }

    pub fn parse_analyses(input: ParseStream) -> syn::Result<Analyses> {
        let mut analyses = Analyses::default();

        // Analyses are optional, appearing after a `//` at the end of the
        // line. However, since the syn lexer eats comments, we assume that `//`
        // was replaced with `::` upfront.
        if input.eat(syn::Token![::]) {
            let inner;
            syn::braced!(inner in input);

            let (start, end) = (inner.span().start(), inner.span().end());
            if start.line != end.line {
                let msg = "analyses should not span more than one line".to_string();
                Err(Error::new(inner.span(), msg))?
            }

            while inner.peek(syn::Ident) {
                let ident = inner.parse::<syn::Ident>()?.to_string();
                match ident.as_str() {
                    "types" => {
                        inner.parse::<syn::Token![:]>()?;
                        let value = inner.parse::<syn::LitStr>()?.value();
                        analyses.types = Some(parse_types.parse_str(&value)?);
                    }
                    "keys" => {
                        inner.parse::<syn::Token![:]>()?;
                        let value = inner.parse::<syn::LitStr>()?.value();
                        analyses.keys = Some(parse_keys.parse_str(&value)?);
                    }
                    key => {
                        let msg = format!("unexpected analysis type `{}`", key);
                        Err(Error::new(inner.span(), msg))?;
                    }
                }
                inner.eat(syn::Token![,]);
            }
        }
        Ok(analyses)
    }

    fn parse_types(input: ParseStream) -> syn::Result<Vec<ReprColumnType>> {
        let inner;
        syn::parenthesized!(inner in input);
        inner.parse_comma_sep(parse_column_type)
    }

    /// Parses a unique keys annotation in the format printed by EXPLAIN,
    /// e.g. `([0], [1, 2])`.
    fn parse_keys(input: ParseStream) -> syn::Result<Vec<Vec<usize>>> {
        let inner;
        syn::parenthesized!(inner in input);
        inner.parse_comma_sep(|input| {
            let inner;
            syn::bracketed!(inner in input);
            inner.parse_comma_sep(|input| input.parse::<syn::LitInt>()?.base10_parse::<usize>())
        })
    }

    pub fn parse_column_type(input: ParseStream) -> syn::Result<ReprColumnType> {
        let scalar_type = parse_scalar_type(input)?;
        Ok(ReprColumnType {
            scalar_type,
            nullable: input.eat(syn::Token![?]),
        })
    }

    pub fn parse_scalar_type(input: ParseStream) -> syn::Result<ReprScalarType> {
        let lookahead = input.lookahead1();

        let scalar_type = if input.look_and_eat(bigint, &lookahead) {
            ReprScalarType::Int64
        } else if input.look_and_eat(double, &lookahead) {
            input.parse::<precision>()?;
            ReprScalarType::Float64
        } else if input.look_and_eat(boolean, &lookahead) {
            ReprScalarType::Bool
        } else if input.look_and_eat(character, &lookahead) {
            input.parse::<varying>()?;
            ReprScalarType::String
        } else if input.look_and_eat(integer, &lookahead) {
            ReprScalarType::Int32
        } else if input.look_and_eat(smallint, &lookahead) {
            ReprScalarType::Int16
        } else if input.look_and_eat(text, &lookahead) {
            ReprScalarType::String
        } else if input.look_and_eat(jsonb, &lookahead) {
            ReprScalarType::Jsonb
        } else if input.look_and_eat(numeric, &lookahead) {
            ReprScalarType::Numeric
        } else {
            Err(lookahead.error())?
        };

        Ok(scalar_type)
    }

    syn::custom_keyword!(bigint);
    syn::custom_keyword!(boolean);
    syn::custom_keyword!(character);
    syn::custom_keyword!(double);
    syn::custom_keyword!(integer);
    syn::custom_keyword!(jsonb);
    syn::custom_keyword!(numeric);
    syn::custom_keyword!(precision);
    syn::custom_keyword!(smallint);
    syn::custom_keyword!(text);
    syn::custom_keyword!(varying);
}

pub enum Def {
    Source {
        name: String,
        cols: Vec<String>,
        typ: mz_repr::SqlRelationType,
    },
}

mod def {
    use mz_repr::{SqlColumnType, SqlRelationType};

    use super::*;

    pub fn parse_def(ctx: CtxRef, input: ParseStream) -> syn::Result<Def> {
        parse_def_source(ctx, input) // only one variant for now
    }

    fn parse_def_source(ctx: CtxRef, input: ParseStream) -> syn::Result<Def> {
        let reduce = input.parse::<def::DefSource>()?;

        let name = {
            input.parse::<def::name>()?;
            input.parse::<syn::Token![=]>()?;
            input.parse::<syn::Ident>()?.to_string()
        };

        let keys = if input.eat(kw::keys) {
            input.parse::<syn::Token![=]>()?;
            let inner;
            syn::bracketed!(inner in input);
            inner.parse_comma_sep(|input| {
                let inner;
                syn::bracketed!(inner in input);
                inner.parse_comma_sep(scalar::parse_column_index)
            })?
        } else {
            vec![]
        };

        let parse_inputs = ParseChildren::new(input, reduce.span().start());
        let (cols, column_types) = {
            let source_columns = parse_inputs.parse_many(ctx, parse_def_source_column)?;
            let mut column_names = vec![];
            let mut column_types = vec![];
            for (column_name, column_type) in source_columns {
                column_names.push(column_name);
                column_types.push(column_type);
            }
            (column_names, column_types)
        };

        let typ = SqlRelationType { column_types, keys };

        Ok(Def::Source { name, cols, typ })
    }

    fn parse_def_source_column(
        _ctx: CtxRef,
        input: ParseStream,
    ) -> syn::Result<(String, SqlColumnType)> {
        input.parse::<syn::Token![-]>()?;
        let column_name = input.parse::<syn::Ident>()?.to_string();
        input.parse::<syn::Token![:]>()?;
        let column_type = SqlColumnType::from_repr(&analyses::parse_column_type(input)?);
        Ok((column_name, column_type))
    }

    syn::custom_keyword!(DefSource);
    syn::custom_keyword!(name);
}

/// Help utilities used by sibling modules.
mod util {
    use syn::parse::{Lookahead1, ParseBuffer, Peek};

    use super::*;

    /// Extension methods for [`syn::parse::ParseBuffer`].
    pub trait ParseBufferExt<'a> {
        fn look_and_eat<T: Eat>(&self, token: T, lookahead: &Lookahead1<'a>) -> bool;

        /// Consumes a token `T` if present.
        fn eat<T: Eat>(&self, t: T) -> bool;

        /// Consumes two tokens `T1 T2` if present in that order.
        fn eat2<T1: Eat, T2: Eat>(&self, t1: T1, t2: T2) -> bool;

        /// Consumes three tokens `T1 T2 T3` if present in that order.
        fn eat3<T1: Eat, T2: Eat, T3: Eat>(&self, t1: T1, t2: T2, t3: T3) -> bool;

        // Parse a comma-separated list of items into a vector.
        fn parse_comma_sep<T>(&self, p: fn(ParseStream) -> syn::Result<T>) -> syn::Result<Vec<T>>;
    }

    impl<'a> ParseBufferExt<'a> for ParseBuffer<'a> {
        /// Consumes a token `T` if present, looking it up using the provided
        /// [`Lookahead1`] instance.
        fn look_and_eat<T: Eat>(&self, token: T, lookahead: &Lookahead1<'a>) -> bool {
            if lookahead.peek(token) {
                self.parse::<T::Token>().unwrap();
                true
            } else {
                false
            }
        }

        fn eat<T: Eat>(&self, t: T) -> bool {
            if self.peek(t) {
                self.parse::<T::Token>().unwrap();
                true
            } else {
                false
            }
        }

        fn eat2<T1: Eat, T2: Eat>(&self, t1: T1, t2: T2) -> bool {
            if self.peek(t1) && self.peek2(t2) {
                self.parse::<T1::Token>().unwrap();
                self.parse::<T2::Token>().unwrap();
                true
            } else {
                false
            }
        }

        fn eat3<T1: Eat, T2: Eat, T3: Eat>(&self, t1: T1, t2: T2, t3: T3) -> bool {
            if self.peek(t1) && self.peek2(t2) && self.peek3(t3) {
                self.parse::<T1::Token>().unwrap();
                self.parse::<T2::Token>().unwrap();
                self.parse::<T3::Token>().unwrap();
                true
            } else {
                false
            }
        }

        fn parse_comma_sep<T>(&self, p: fn(ParseStream) -> syn::Result<T>) -> syn::Result<Vec<T>> {
            Ok(self
                .parse_terminated(p, syn::Token![,])?
                .into_iter()
                .collect::<Vec<_>>())
        }
    }

    // Helper trait for types that can be eaten.
    //
    // Implementing types must also implement [`Peek`], and the associated
    // [`Peek::Token`] type should implement [`Parse`]). For some reason the
    // latter bound is not present in [`Peek`] even if it makes a lot of sense,
    // which is why we need this helper.
    pub trait Eat: Peek<Token = Self::_Token> {
        type _Token: Parse;
    }

    impl<T> Eat for T
    where
        T: Peek,
        T::Token: Parse,
    {
        type _Token = T::Token;
    }

    pub struct Ctx<'a> {
        pub catalog: &'a TestCatalog,
    }

    pub type CtxRef<'a> = &'a Ctx<'a>;

    /// Provides facilities for parsing
    pub struct ParseChildren<'a> {
        stream: ParseStream<'a>,
        parent: LineColumn,
    }

    impl<'a> ParseChildren<'a> {
        pub fn new(stream: ParseStream<'a>, parent: LineColumn) -> Self {
            Self { stream, parent }
        }

        pub fn parse_one<C, T>(
            &self,
            ctx: C,
            function: fn(C, ParseStream) -> syn::Result<T>,
        ) -> syn::Result<T> {
            match self.maybe_child() {
                Ok(_) => function(ctx, self.stream),
                Err(e) => Err(e),
            }
        }

        pub fn parse_many<C: Copy, T>(
            &self,
            ctx: C,
            function: fn(C, ParseStream) -> syn::Result<T>,
        ) -> syn::Result<Vec<T>> {
            let mut inputs = vec![self.parse_one(ctx, function)?];
            while self.maybe_child().is_ok() {
                inputs.push(function(ctx, self.stream)?);
            }
            Ok(inputs)
        }

        fn maybe_child(&self) -> syn::Result<()> {
            let start = self.stream.span().start();
            if start.line <= self.parent.line {
                let msg = format!("child expected at line > {}", self.parent.line);
                Err(Error::new(self.stream.span(), msg))?
            }
            if start.column != self.parent.column + 2 {
                let msg = format!("child expected at column {}", self.parent.column + 2);
                Err(Error::new(self.stream.span(), msg))?
            }
            Ok(())
        }
    }
}

/// Custom keywords used while parsing.
mod kw {
    syn::custom_keyword!(aggregates);
    syn::custom_keyword!(AND);
    syn::custom_keyword!(ArrangeBy);
    syn::custom_keyword!(array);
    syn::custom_keyword!(asc);
    // case when ... then ... else ... end
    syn::custom_keyword!(case);
    syn::custom_keyword!(coalesce);
    syn::custom_keyword!(Constant);
    syn::custom_keyword!(CrossJoin);
    syn::custom_keyword!(cte);
    syn::custom_keyword!(desc);
    syn::custom_keyword!(distinct);
    syn::custom_keyword!(Distinct);
    syn::custom_keyword!(empty);
    syn::custom_keyword!(end);
    syn::custom_keyword!(eq);
    syn::custom_keyword!(error);
    syn::custom_keyword!(exp_group_size);
    syn::custom_keyword!(FALSE);
    syn::custom_keyword!(Filter);
    syn::custom_keyword!(FlatMap);
    syn::custom_keyword!(Get);
    syn::custom_keyword!(group_by);
    syn::custom_keyword!(IS);
    syn::custom_keyword!(Join);
    syn::custom_keyword!(keys);
    syn::custom_keyword!(limit);
    syn::custom_keyword!(list);
    syn::custom_keyword!(Map);
    syn::custom_keyword!(monotonic);
    syn::custom_keyword!(Mutually);
    syn::custom_keyword!(Negate);
    syn::custom_keyword!(NOT);
    syn::custom_keyword!(null);
    syn::custom_keyword!(NULL);
    syn::custom_keyword!(nulls_first);
    syn::custom_keyword!(nulls_last);
    syn::custom_keyword!(offset);
    syn::custom_keyword!(on);
    syn::custom_keyword!(OR);
    syn::custom_keyword!(order_by);
    syn::custom_keyword!(project);
    syn::custom_keyword!(Project);
    syn::custom_keyword!(Recursive);
    syn::custom_keyword!(Reduce);
    syn::custom_keyword!(Return);
    syn::custom_keyword!(then);
    syn::custom_keyword!(Threshold);
    syn::custom_keyword!(TopK);
    syn::custom_keyword!(TRUE);
    syn::custom_keyword!(Union);
    syn::custom_keyword!(when);
    syn::custom_keyword!(With);
    syn::custom_keyword!(x);
}

#[cfg(test)]
mod tests {
    use mz_expr::{MirScalarExpr, func};
    use mz_repr::{ReprScalarType, SqlScalarType};

    use super::*;

    #[mz_ore::test]
    fn parse_scalar_variant_names() {
        let actual = try_parse_scalar("is_null(add_int32(#1, #0))").unwrap();
        let expected = MirScalarExpr::CallUnary {
            func: func::IsNull.into(),
            expr: Box::new(MirScalarExpr::CallBinary {
                func: func::AddInt32.into(),
                expr1: Box::new(MirScalarExpr::column(1)),
                expr2: Box::new(MirScalarExpr::column(0)),
            }),
        };
        assert_eq!(actual, expected);
    }

    #[mz_ore::test]
    fn parse_scalar_parameterized() {
        let actual = try_parse_scalar("record_get[1](#0)").unwrap();
        let expected = MirScalarExpr::CallUnary {
            func: func::RecordGet(1).into(),
            expr: Box::new(MirScalarExpr::column(0)),
        };
        assert_eq!(actual, expected);

        let actual = try_parse_scalar("list_create[integer](#0)").unwrap();
        let expected = MirScalarExpr::call_variadic(
            func::variadic::ListCreate {
                elem_type: SqlScalarType::Int32,
            },
            vec![MirScalarExpr::column(0)],
        );
        assert_eq!(actual, expected);
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn parse_scalar_typed_literals() {
        let actual = try_parse_scalar(r#""{\"a\": 1}"::jsonb"#).unwrap();
        let MirScalarExpr::Literal(Ok(_), typ) = &actual else {
            panic!("expected literal, got {actual:?}");
        };
        assert_eq!(typ.scalar_type, ReprScalarType::Jsonb);

        let actual = try_parse_scalar("2000.5::numeric").unwrap();
        let MirScalarExpr::Literal(Ok(row), typ) = &actual else {
            panic!("expected literal, got {actual:?}");
        };
        assert_eq!(typ.scalar_type, ReprScalarType::Numeric);
        assert_eq!(row.unpack_first().to_string(), "2000.5");
    }

    #[mz_ore::test]
    fn parse_column_types() {
        let actual = try_parse_column_types("(bigint, text?)").unwrap();
        assert_eq!(actual.len(), 2);
        assert_eq!(actual[0].scalar_type, SqlScalarType::Int64);
        assert!(!actual[0].nullable);
        assert_eq!(actual[1].scalar_type, SqlScalarType::String);
        assert!(actual[1].nullable);
    }
}