mz_catalog_decode/

create_sql.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.

//! Decoders for catalog `create_sql` strings.
//!
//! Each function parses a persisted `create_sql` and returns a JSON value with
//! the fields a builtin catalog view projects. Errors are human-readable
//! strings.

use std::collections::BTreeMap;

use mz_repr::adt::interval::Interval;
use mz_repr::strconv;
use mz_sql_parser::ast::display::AstDisplay;
use mz_sql_parser::ast::item_refs::collect_item_references;
use mz_sql_parser::ast::{
    AstInfo, AvroSchema, ConnectionOption, ConnectionOptionName, CreateConnectionType,
    CreateSinkConnection, CreateSourceOptionName, CreateSubsourceOptionName, Format,
    FormatSpecifier, IcebergSinkConfigOptionName, IcebergSinkMode, IndexOptionName,
    KafkaSinkConfigOptionName, KafkaSourceConfigOptionName, MaterializedViewOptionName,
    PgConfigOptionName, ProtobufSchema, Raw, RawClusterName, RawItemName, RefreshAtOptionValue,
    RefreshEveryOptionValue, RefreshOptionValue, SinkEnvelope, SourceEnvelope, SourceErrorPolicy,
    Statement, TableFromSourceOptionName, TableOptionName, UnresolvedItemName, Value,
    WithOptionValue,
};
use prost::Message as _;
use serde_json::json;

use crate::durable;

/// Parses `sql`, which must contain exactly one statement.
fn parse_single_statement(sql: &str) -> Result<Statement<Raw>, String> {
    let mut stmts = mz_sql_parser::parser::parse_statements(sql)
        .map_err(|e| format!("failed to parse create_sql: {e}"))?;
    match stmts.len() {
        1 => Ok(stmts.remove(0).ast),
        n => Err(format!("expected a single statement, found {n}")),
    }
}

fn item_id(item: RawItemName) -> Result<String, &'static str> {
    match item {
        RawItemName::Id(id, _, _) => Ok(id),
        RawItemName::Name(_) => Err("unresolved item name"),
    }
}

fn cluster_id(in_cluster: RawClusterName) -> Result<String, &'static str> {
    match in_cluster {
        RawClusterName::Resolved(s) => Ok(s),
        RawClusterName::Unresolved(_) => Err("unresolved cluster name"),
    }
}

fn format_name<T: AstInfo>(fmt: &Format<T>) -> &'static str {
    match fmt {
        Format::Bytes => "bytes",
        Format::Avro(_) => "avro",
        Format::Protobuf(_) => "protobuf",
        Format::Regex(_) => "regex",
        Format::Csv { .. } => "csv",
        Format::Json { .. } => "json",
        Format::Text => "text",
    }
}

/// Extracts the string form of a `WithOptionValue`, matching how the planner
/// coerces option values. The blanket `TryFromValue<WithOptionValue<T>>` impl in
/// `src/sql/src/plan/with_options.rs` accepts a quoted string, a bare identifier,
/// and a 1-part unresolved item name, all yielding a string. Any other variant
/// yields None.
///
/// `AstDisplay` does not re-quote bare identifiers or 1-part names, so those
/// forms persist unquoted in `create_sql`. Matching only `Value::String` here
/// would miss them and silently fall back to a default, so the catalog-raw
/// parsers below route their string options through this helper.
fn option_string<T: AstInfo>(value: &WithOptionValue<T>) -> Option<String> {
    match value {
        WithOptionValue::Value(Value::String(s)) => Some(s.clone()),
        WithOptionValue::Ident(ident) => Some(ident.clone().into_string()),
        WithOptionValue::UnresolvedItemName(UnresolvedItemName(parts)) if parts.len() == 1 => {
            Some(parts[0].clone().into_string())
        }
        _ => None,
    }
}

/// The compaction window a `RETAIN HISTORY` option plans to, in milliseconds.
///
/// Mirrors `plan_retain_history_option` and `OptionalDuration` in `mz_sql`: `NULL` and a zero
/// duration disable compaction, which `CompactionWindow::comparable_timestamp` reports as
/// `u64::MAX`.
fn retain_history_millis<T: AstInfo>(value: Option<&WithOptionValue<T>>) -> Result<u64, String> {
    let value = match value {
        Some(WithOptionValue::RetainHistoryFor(value) | WithOptionValue::Value(value)) => value,
        _ => return Err("invalid RETAIN HISTORY value".into()),
    };
    let interval = match value {
        Value::Null => None,
        Value::Interval(literal) => Some(durable::interval_literal(literal)?),
        Value::Number(s) | Value::String(s) => {
            Some(strconv::parse_interval(s).map_err(|e| e.to_string())?)
        }
        Value::HexString(_) | Value::Boolean(_) => {
            return Err("invalid RETAIN HISTORY value".into());
        }
    };
    let duration = interval
        .map(|interval| interval.duration().map_err(|e| e.to_string()))
        .transpose()?
        .filter(|duration| !duration.is_zero());
    match duration {
        None => Ok(u64::MAX),
        Some(duration) => u64::try_from(duration.as_millis())
            .map_err(|_| "RETAIN HISTORY duration out of range".to_string()),
    }
}

/// Records the window of the last `RETAIN HISTORY` option among `options`, the one
/// `CatalogItem::update_retain_history` keeps, as `retain_history_millis` in `info`.
///
/// Each statement type has its own option type, so `options` yields each option's name and
/// value, and `retain_history` is that type's `RetainHistory` name.
fn insert_retain_history<'a, N, T>(
    info: &mut BTreeMap<&str, serde_json::Value>,
    options: impl DoubleEndedIterator<Item = (&'a N, &'a Option<WithOptionValue<T>>)>,
    retain_history: N,
) -> Result<(), String>
where
    N: PartialEq + 'a,
    T: AstInfo + 'a,
{
    if let Some((_, value)) = options.rev().find(|(name, _)| **name == retain_history) {
        let millis = retain_history_millis(value.as_ref())?;
        info.insert("retain_history_millis", json!(millis));
    }
    Ok(())
}

/// Records the `REFRESH` options among `options` as `refresh` in `info`, one entry per option in
/// statement order.
fn insert_refresh_strategies<'a, T: AstInfo + 'a>(
    info: &mut BTreeMap<&str, serde_json::Value>,
    options: impl Iterator<Item = &'a Option<WithOptionValue<T>>>,
) -> Result<(), String> {
    let mut refresh = Vec::new();
    for value in options {
        let Some(WithOptionValue::Refresh(value)) = value else {
            continue;
        };
        refresh.push(match value {
            RefreshOptionValue::OnCommit => json!({"type": "on-commit"}),
            RefreshOptionValue::AtCreation => json!({"type": "at", "at": null}),
            RefreshOptionValue::At(RefreshAtOptionValue { time }) => {
                json!({"type": "at", "at": time.to_ast_string_stable()})
            }
            RefreshOptionValue::Every(RefreshEveryOptionValue {
                interval,
                aligned_to,
            }) => {
                // The same duration round trip as planning, so that `1 day`
                // renders as `24:00:00` rather than as a day.
                let interval = durable::interval_literal(interval)
                    .and_then(|interval| {
                        interval
                            .duration()
                            .and_then(|duration| Interval::from_duration(&duration))
                            .map_err(|e| e.to_string())
                    })
                    .map_err(|e| format!("invalid REFRESH EVERY interval: {e}"))?;
                json!({
                    "type": "every",
                    "interval": interval.to_string(),
                    "aligned_to": aligned_to.as_ref().map(|time| time.to_ast_string_stable()),
                })
            }
        });
    }
    // Purification leaves at least one REFRESH option, but a statement stored
    // before it did has none and means ON COMMIT.
    if refresh.is_empty() {
        refresh.push(json!({"type": "on-commit"}));
    }
    info.insert("refresh", json!(refresh));
    Ok(())
}

/// Parses a catalog `create_sql` string into a JSON object.
///
/// The returned JSON does not fully reflect the parsed SQL and instead contains only fields
/// required by current callers.
pub fn item_details(a: &str) -> Result<serde_json::Value, String> {
    let stmt = parse_single_statement(a)?;

    let mut info = BTreeMap::<&str, serde_json::Value>::new();

    use mz_sql_parser::ast::Statement::*;
    let item_type = match stmt {
        CreateSecret(_) => "secret",
        CreateConnection(stmt) => {
            let connection_type = stmt.connection_type.as_str();
            info.insert("connection_type", json!(connection_type));

            "connection"
        }
        CreateView(stmt) => {
            let mut definition = stmt.definition.query.to_ast_string_stable();
            // PostgreSQL appends a semicolon in `pg_views.definition`, we
            // do the same for compatibility's sake.
            definition.push(';');
            info.insert("definition", json!(definition));

            "view"
        }
        CreateMaterializedView(stmt) => {
            let Some(in_cluster) = stmt.in_cluster else {
                return Err("missing IN CLUSTER".into());
            };
            let cluster_id = match in_cluster {
                RawClusterName::Unresolved(ident) => ident.into_string(),
                RawClusterName::Resolved(s) => s,
            };
            info.insert("cluster_id", json!(cluster_id));

            let mut definition = stmt.query.to_ast_string_stable();
            definition.push(';');
            info.insert("definition", json!(definition));

            insert_retain_history(
                &mut info,
                stmt.with_options.iter().map(|o| (&o.name, &o.value)),
                MaterializedViewOptionName::RetainHistory,
            )?;

            insert_refresh_strategies(&mut info, stmt.with_options.iter().map(|o| &o.value))?;

            if let Some(target) = stmt.replacement_for {
                info.insert("replacement_target", json!(item_id(target)?));
            }

            "materialized-view"
        }
        CreateTable(stmt) => {
            insert_retain_history(
                &mut info,
                stmt.with_options.iter().map(|o| (&o.name, &o.value)),
                TableOptionName::RetainHistory,
            )?;

            "table"
        }
        CreateTableFromSource(stmt) => {
            let source_id = item_id(stmt.source)?;
            info.insert("source_id", json!(source_id));

            insert_retain_history(
                &mut info,
                stmt.with_options.iter().map(|o| (&o.name, &o.value)),
                TableFromSourceOptionName::RetainHistory,
            )?;

            "table"
        }
        CreateSource(stmt) => {
            insert_retain_history(
                &mut info,
                stmt.with_options.iter().map(|o| (&o.name, &o.value)),
                CreateSourceOptionName::RetainHistory,
            )?;

            let Some(in_cluster) = stmt.in_cluster else {
                return Err("missing IN CLUSTER".into());
            };
            let cluster_id = cluster_id(in_cluster)?;
            info.insert("cluster_id", json!(cluster_id));

            use mz_sql_parser::ast::CreateSourceConnection::*;
            let (source_type, connection) = match stmt.connection {
                Kafka { connection, .. } => ("kafka", Some(connection)),
                Postgres { connection, .. } => ("postgres", Some(connection)),
                MySql { connection, .. } => ("mysql", Some(connection)),
                SqlServer { connection, .. } => ("sql-server", Some(connection)),
                LoadGenerator { .. } => ("load-generator", None),
            };
            info.insert("source_type", json!(source_type));
            if let Some(conn) = connection {
                let conn_id = item_id(conn)?;
                info.insert("connection_id", json!(conn_id));
            }

            let is_debezium = matches!(
                stmt.envelope,
                Some(mz_sql_parser::ast::SourceEnvelope::Debezium)
            );

            // An old-syntax kafka source ingests into its own relation, so an
            // omitted ENVELOPE means the default ENVELOPE NONE and the pre-MV
            // packer reported 'none'. A new-syntax source (no progress
            // subsource, hence no EXPOSE PROGRESS AS in create_sql) ingests
            // nothing itself. Its envelopes live on the per-table exports, so
            // its own envelope_type stays absent (SQL NULL), matching released
            // behavior. `progress_subsource.is_some()` is the planner's own
            // old-vs-new discriminator (see `OldSyntaxIngestion` in
            // plan_create_source). Non-kafka sources carry no envelope either.
            // See the `mz_sources.envelope_type` column.
            let envelope_type = match &stmt.envelope {
                Some(envelope) => {
                    use mz_sql_parser::ast::SourceEnvelope::*;
                    Some(match envelope {
                        None => "none",
                        Debezium => "debezium",
                        Upsert { .. } => "upsert",
                        CdcV2 => "materialize",
                    })
                }
                None if source_type == "kafka" && stmt.progress_subsource.is_some() => Some("none"),
                None => None,
            };
            if let Some(envelope_type) = envelope_type {
                info.insert("envelope_type", json!(envelope_type));
            }

            if let Some(format_spec) = stmt.format {
                match &format_spec {
                    FormatSpecifier::Bare(fmt) => {
                        // Debezium sources with a single format spec implicitly use
                        // the same format for both key and value.
                        if is_debezium {
                            info.insert("key_format", json!(format_name(fmt)));
                        }
                        info.insert("value_format", json!(format_name(fmt)));
                    }
                    FormatSpecifier::KeyValue { key, value } => {
                        info.insert("key_format", json!(format_name(key)));
                        info.insert("value_format", json!(format_name(value)));
                    }
                }
            }

            "source"
        }
        CreateWebhookSource(stmt) => {
            if stmt.is_table {
                "table"
            } else {
                info.insert("source_type", json!("webhook"));
                if let Some(in_cluster) = stmt.in_cluster {
                    let cluster_id = cluster_id(in_cluster)?;
                    info.insert("cluster_id", json!(cluster_id));
                }
                "source"
            }
        }
        CreateSubsource(stmt) => {
            use mz_sql_parser::ast::CreateSubsourceOptionName;
            let is_progress = stmt
                .with_options
                .iter()
                .any(|o| matches!(o.name, CreateSubsourceOptionName::Progress));
            let source_type = if is_progress { "progress" } else { "subsource" };
            info.insert("source_type", json!(source_type));

            insert_retain_history(
                &mut info,
                stmt.with_options.iter().map(|o| (&o.name, &o.value)),
                CreateSubsourceOptionName::RetainHistory,
            )?;

            if let Some(of_source) = stmt.of_source {
                let of_source_id = item_id(of_source)?;
                info.insert("of_source_id", json!(of_source_id));
            }

            "subsource"
        }
        // Everything the mz_sinks, mz_kafka_sinks and mz_iceberg_sinks
        // views read. The Rust side of each value lives in `Sink` and
        // `StorageSinkConnection`, so those and this have to move together.
        //
        // NOTE: we bail below if a sink has no resolved `IN CLUSTER`, no
        // `TOPIC` on kafka, or no `NAMESPACE`/`TABLE` on iceberg. Planning
        // guarantees all four. But if one ever slipped through it would
        // take down every view built on this function, not just the sink
        // ones.
        CreateSink(stmt) => {
            let Some(in_cluster) = stmt.in_cluster else {
                return Err("missing IN CLUSTER".into());
            };
            info.insert("cluster_id", json!(cluster_id(in_cluster)?));

            match stmt.connection {
                CreateSinkConnection::Kafka {
                    connection,
                    options,
                    key: sink_key,
                    ..
                } => {
                    info.insert("sink_type", json!("kafka"));
                    info.insert("connection_id", json!(item_id(connection)?));

                    let topic = options
                        .into_iter()
                        .find(|o| o.name == KafkaSinkConfigOptionName::Topic)
                        .and_then(|o| o.value.as_ref().and_then(option_string))
                        .ok_or("kafka sink missing TOPIC")?;
                    info.insert("topic", json!(topic));

                    if let Some(envelope) = stmt.envelope {
                        let envelope_type = match envelope {
                            SinkEnvelope::Upsert => "upsert",
                            SinkEnvelope::Debezium => "debezium",
                        };
                        info.insert("envelope_type", json!(envelope_type));
                    }

                    if let Some(format_spec) = stmt.format {
                        // A key format only survives if the sink has a
                        // `KEY`. Without one `kafka_sink_builder` throws
                        // away the key half of a key/value spec, and does
                        // not copy a bare spec over to the key either.
                        let (key_format, value_format) = match (&format_spec, sink_key.is_some()) {
                            (FormatSpecifier::Bare(fmt), false) => (None, format_name(fmt)),
                            (FormatSpecifier::Bare(fmt), true) => {
                                (Some(format_name(fmt)), format_name(fmt))
                            }
                            (FormatSpecifier::KeyValue { value, .. }, false) => {
                                (None, format_name(value))
                            }
                            (FormatSpecifier::KeyValue { key, value }, true) => {
                                (Some(format_name(key)), format_name(value))
                            }
                        };
                        if let Some(key_format) = key_format {
                            info.insert("key_format", json!(key_format));
                        }
                        info.insert("value_format", json!(value_format));

                        // The deprecated combined `format`. Only avro/avro
                        // and json/json collapse to a single name.
                        // Everything else, text/text and bytes/bytes
                        // included, gets the composite form.
                        let combined = match key_format {
                            None => value_format.to_string(),
                            Some(key_format)
                                if key_format == value_format
                                    && matches!(value_format, "avro" | "json") =>
                            {
                                value_format.to_string()
                            }
                            Some(key_format) => {
                                format!("key-{key_format}-value-{value_format}")
                            }
                        };
                        info.insert("format", json!(combined));
                    }
                }
                CreateSinkConnection::Iceberg {
                    catalog_connection,
                    options,
                    ..
                } => {
                    info.insert("sink_type", json!("iceberg"));
                    // The catalog connection, not the optional AWS one.
                    info.insert("connection_id", json!(item_id(catalog_connection)?));

                    let mut namespace = None;
                    let mut table = None;
                    for option in options {
                        match option.name {
                            IcebergSinkConfigOptionName::Namespace => {
                                namespace = option.value.as_ref().and_then(option_string)
                            }
                            IcebergSinkConfigOptionName::Table => {
                                table = option.value.as_ref().and_then(option_string)
                            }
                        }
                    }
                    info.insert(
                        "namespace",
                        json!(namespace.ok_or("iceberg sink missing NAMESPACE")?),
                    );
                    info.insert("table", json!(table.ok_or("iceberg sink missing TABLE")?));

                    // Iceberg spells the envelope `MODE`, and has no format
                    // columns at all.
                    if let Some(mode) = stmt.mode {
                        let envelope_type = match mode {
                            IcebergSinkMode::Upsert => "upsert",
                            IcebergSinkMode::Append => "append",
                        };
                        info.insert("envelope_type", json!(envelope_type));
                    }
                }
            }

            "sink"
        }
        CreateMetricSink(stmt) => {
            let Some(in_cluster) = stmt.in_cluster else {
                return Err("missing IN CLUSTER".into());
            };
            let cluster_id = cluster_id(in_cluster)?;
            info.insert("cluster_id", json!(cluster_id));
            let from_id = item_id(stmt.from)?;
            info.insert("from_id", json!(from_id));
            "metric-sink"
        }
        CreateIndex(stmt) => {
            let Some(in_cluster) = stmt.in_cluster else {
                return Err("missing IN CLUSTER".into());
            };
            let cluster_id = cluster_id(in_cluster)?;
            info.insert("cluster_id", json!(cluster_id));
            let on_id = item_id(stmt.on_name)?;
            info.insert("on_id", json!(on_id));
            insert_retain_history(
                &mut info,
                stmt.with_options.iter().map(|o| (&o.name, &o.value)),
                IndexOptionName::RetainHistory,
            )?;
            "index"
        }
        CreateType(_) => "type",
        // NOTE: every statement that creates a catalog item needs an arm above. These
        // catalog views run this over every item row before their type filter drops the
        // unwanted rows, so one unclassified `create_sql` takes out `mz_objects`,
        // `mz_indexes`, and every sibling view at once. The match is exhaustive to make
        // that a compile error here, not a runtime failure.
        Select(_)
        | Insert(_)
        | Copy(_)
        | Update(_)
        | Delete(_)
        | CreateDatabase(_)
        | CreateSchema(_)
        | CreateRole(_)
        | CreateCluster(_)
        | CreateClusterReplica(_)
        | CreateNetworkPolicy(_)
        | AlterCluster(_)
        | AlterOwner(_)
        | AlterObjectRename(_)
        | AlterObjectSwap(_)
        | AlterRetainHistory(_)
        | AlterIndex(_)
        | AlterSecret(_)
        | AlterSetCluster(_)
        | AlterSink(_)
        | AlterSource(_)
        | AlterSystemSet(_)
        | AlterSystemReset(_)
        | AlterSystemResetAll(_)
        | AlterConnection(_)
        | AlterNetworkPolicy(_)
        | AlterRole(_)
        | AlterTableAddColumn(_)
        | AlterMaterializedViewApplyReplacement(_)
        | Discard(_)
        | DropObjects(_)
        | DropOwned(_)
        | SetVariable(_)
        | ResetVariable(_)
        | Show(_)
        | StartTransaction(_)
        | SetTransaction(_)
        | Commit(_)
        | Rollback(_)
        | Subscribe(_)
        | ExplainPlan(_)
        | ExplainPushdown(_)
        | ExplainTimestamp(_)
        | ExplainSinkSchema(_)
        | ExplainAnalyzeObject(_)
        | ExplainAnalyzeCluster(_)
        | Declare(_)
        | Fetch(_)
        | Close(_)
        | Prepare(_)
        | Execute(_)
        | ExecuteUnitTest(_)
        | Deallocate(_)
        | Raise(_)
        | GrantRole(_)
        | RevokeRole(_)
        | GrantPrivileges(_)
        | RevokePrivileges(_)
        | AlterDefaultPrivileges(_)
        | ReassignOwned(_)
        | ValidateConnection(_)
        | Comment(_) => return Err("not a CREATE item statement".into()),
    };
    info.insert("type", json!(item_type));

    let info = info.into_iter().map(|(k, v)| (k.to_string(), v)).collect();
    Ok(info)
}

/// Extracts the catalog item references from a catalog `create_sql` string as a JSONB object.
///
/// - `ids`: array of catalog id strings from `[<id> AS <name>]` references.
/// - `named_funcs`: array of function references such as "pg_catalog"."max".
/// - `named_types`: array of type references such as "pg_catalog"."int4", exclusive from the ones
///   in `ids`.
/// - `named_relations`: array of relation references, exclusive from the ones in `ids`.
pub fn item_references(a: &str) -> Result<serde_json::Value, String> {
    fn qualified(name: &UnresolvedItemName) -> serde_json::Value {
        match &name.0[..] {
            [.., schema, item] => json!({"schema": schema.as_str(), "name": item.as_str()}),
            [item] => json!({"schema": serde_json::Value::Null, "name": item.as_str()}),
            [] => json!({"schema": serde_json::Value::Null, "name": ""}),
        }
    }

    let stmt = parse_single_statement(a)?;

    let refs = collect_item_references(&stmt);
    mz_ore::soft_assert_or_log!(
        refs.named_array_elements.is_empty(),
        "persisted create_sql should never carry an array type T[] \
        and resolve its array type in `ids`: {:?}",
        refs.named_array_elements
    );
    Ok(json!({
        "ids": refs.ids.iter().collect::<Vec<_>>(),
        "named_funcs": refs.named_funcs.iter().map(qualified).collect::<Vec<_>>(),
        "named_types": refs.named_types.iter().map(qualified).collect::<Vec<_>>(),
        "named_relations": refs.named_relations.iter().map(qualified).collect::<Vec<_>>(),
    }))
}

/// Minimal decoder for `ProtoPostgresSourcePublicationDetails`. The
/// canonical proto lives in `mz-storage-types`, which depends on this
/// crate via `mz-expr`, so we redeclare the two tags we read here. Upstream tag
/// renumbers slip past silently. The `mz_postgres_sources` lockdown
/// SLTs catch them.
#[derive(Clone, PartialEq, ::prost::Message)]
struct PostgresPublicationDetailsSubset {
    #[prost(string, tag = "2")]
    slot: String,
    #[prost(uint64, optional, tag = "3")]
    timeline_id: Option<u64>,
}

/// Extracts postgres source publication details (slot, timeline_id) from a
/// catalog `create_sql`. Returns:
///
/// - jsonb `{"slot": <text>, "timeline_id": <u64 | null>}` for `CREATE SOURCE ... FROM POSTGRES
///   CONNECTION ... (DETAILS = ...)` statements.
/// - jsonb `null` for any other statement.
///
/// Errors if the statement fails to parse, is a postgres source without
/// a `DETAILS` option, or if the `DETAILS` value can't be hex- and
/// proto-decoded.
pub fn postgres_source_details(a: &str) -> Result<serde_json::Value, String> {
    let stmt = parse_single_statement(a)?;

    use mz_sql_parser::ast::CreateSourceConnection;
    use mz_sql_parser::ast::Statement::CreateSource;
    let options = match stmt {
        CreateSource(stmt) => match stmt.connection {
            CreateSourceConnection::Postgres { options, .. } => options,
            _ => return Ok(serde_json::Value::Null),
        },
        _ => return Ok(serde_json::Value::Null),
    };

    let details_hex = options
        .into_iter()
        .find(|opt| opt.name == PgConfigOptionName::Details)
        .and_then(|opt| match opt.value {
            Some(WithOptionValue::Value(Value::String(s))) => Some(s),
            _ => None,
        })
        .ok_or("missing DETAILS option on postgres source")?;

    let details_bytes =
        hex::decode(&details_hex).map_err(|e| format!("DETAILS is not valid hex: {e}"))?;

    let details = PostgresPublicationDetailsSubset::decode(&*details_bytes)
        .map_err(|e| format!("DETAILS is not a valid publication-details proto: {e}"))?;

    Ok(json!({
        "slot": details.slot,
        "timeline_id": details.timeline_id,
    }))
}

/// Extracts kafka source configuration (topic, group id prefix, connection
/// id) from a catalog `create_sql`. Returns:
///
/// - jsonb `{"topic": <text>, "group_id_prefix": <text | null>, "connection_id": <text>}` for
///   `CREATE SOURCE ... FROM KAFKA CONNECTION ... (TOPIC = ..., [GROUP ID PREFIX = ...])`
///   statements.
/// - jsonb `null` for any other statement.
///
/// Errors if the statement fails to parse, is a kafka source without a
/// `TOPIC` option, or references an unresolved connection name (i.e. one
/// that hasn't been through purification).
pub fn kafka_source_details(a: &str) -> Result<serde_json::Value, String> {
    let stmt = parse_single_statement(a)?;

    use mz_sql_parser::ast::CreateSourceConnection;
    use mz_sql_parser::ast::Statement::CreateSource;
    let (connection, options) = match stmt {
        CreateSource(stmt) => match stmt.connection {
            CreateSourceConnection::Kafka {
                connection,
                options,
            } => (connection, options),
            _ => return Ok(serde_json::Value::Null),
        },
        _ => return Ok(serde_json::Value::Null),
    };

    let connection_id = item_id(connection)?;

    let mut topic: Option<String> = None;
    let mut group_id_prefix: Option<String> = None;
    for opt in options {
        let string_value = opt.value.as_ref().and_then(option_string);
        match opt.name {
            KafkaSourceConfigOptionName::Topic => topic = string_value,
            KafkaSourceConfigOptionName::GroupIdPrefix => group_id_prefix = string_value,
            _ => {}
        }
    }

    let topic = topic.ok_or("missing TOPIC option on kafka source")?;

    Ok(json!({
        "topic": topic,
        "group_id_prefix": group_id_prefix,
        "connection_id": connection_id,
    }))
}

/// Extracts source-export (source table) metadata from a catalog `create_sql`.
///
/// Returns, for a `CREATE TABLE ... FROM SOURCE` or a non-progress
/// `CREATE SUBSOURCE ... OF SOURCE ...` statement:
///
/// ```json
/// {
///   "source_id": "<parent source item id>",
///   "external_reference": ["part1", "part2", ...],
///   "envelope_type": <text | null>,
///   "key_format": <text | null>,
///   "value_format": <text | null>
/// }
/// ```
///
/// `envelope_type`, `key_format`, and `value_format` are always null for a
/// `CREATE SUBSOURCE` (the postgres/mysql/sql-server exports that use the old
/// subsource syntax carry neither format nor envelope). They may also be null
/// for a `CREATE TABLE ... FROM SOURCE` that omits FORMAT/ENVELOPE.
///
/// Returns jsonb `null` for progress subsources and for any statement that is
/// not a source export. The caller distinguishes the four source-table views
/// by joining `source_id` against `mz_sources` and filtering on the parent's
/// type, so this helper stays connection-type agnostic.
///
/// Errors if the statement fails to parse, references an unresolved item name,
/// or is a non-progress subsource missing its OF SOURCE or EXTERNAL REFERENCE.
///
/// The `key_format`/`value_format` derivation mirrors the runtime
/// `DataSourceDesc::formats()` that the removed `pack_kafka_source_tables_update`
/// packer read. A bare FORMAT only carries a key when it resolves to an
/// encoding that has one, which among bare formats is only Avro or Protobuf
/// read from a Confluent Schema Registry whose purified seed carries a key
/// schema. A KEY FORMAT ... VALUE FORMAT ... spec always carries both.
pub fn source_export_details(a: &str) -> Result<serde_json::Value, String> {
    // A bare FORMAT resolves to an encoding with a key only for Avro or
    // Protobuf read from a schema registry whose purified seed carries a key
    // schema. Every other bare format is value-only.
    fn bare_format_has_key<T: AstInfo>(fmt: &Format<T>) -> bool {
        match fmt {
            Format::Avro(AvroSchema::Csr { csr_connection }) => csr_connection
                .seed
                .as_ref()
                .is_some_and(|seed| seed.key_schema.is_some()),
            Format::Protobuf(ProtobufSchema::Csr { csr_connection }) => csr_connection
                .seed
                .as_ref()
                .is_some_and(|seed| seed.key.is_some()),
            _ => false,
        }
    }

    fn key_value_formats<T: AstInfo>(
        spec: &FormatSpecifier<T>,
    ) -> (Option<&'static str>, Option<&'static str>) {
        match spec {
            FormatSpecifier::KeyValue { key, value } => {
                (Some(format_name(key)), Some(format_name(value)))
            }
            FormatSpecifier::Bare(fmt) => {
                let value = Some(format_name(fmt));
                let key = bare_format_has_key(fmt).then(|| format_name(fmt));
                (key, value)
            }
        }
    }

    fn envelope_name(envelope: &SourceEnvelope) -> &'static str {
        match envelope {
            SourceEnvelope::None => "none",
            SourceEnvelope::Debezium => "debezium",
            SourceEnvelope::Upsert {
                value_decode_err_policy,
            } => {
                if value_decode_err_policy
                    .iter()
                    .any(|p| matches!(p, SourceErrorPolicy::Inline { .. }))
                {
                    "upsert-value-err-inline"
                } else {
                    "upsert"
                }
            }
            SourceEnvelope::CdcV2 => "materialize",
        }
    }

    let stmt = parse_single_statement(a)?;

    use mz_sql_parser::ast::Statement::{CreateSubsource, CreateTableFromSource};
    match stmt {
        CreateTableFromSource(stmt) => {
            let source_id = item_id(stmt.source)?;
            let external_reference = stmt
                .external_reference
                .ok_or("missing external reference on CREATE TABLE FROM SOURCE")?
                .0
                .into_iter()
                .map(|ident| ident.into_string())
                .collect::<Vec<_>>();

            let envelope_type = stmt.envelope.as_ref().map(envelope_name);
            let (key_format, value_format) = match &stmt.format {
                Some(spec) => key_value_formats(spec),
                None => (None, None),
            };

            Ok(json!({
                "source_id": source_id,
                "external_reference": external_reference,
                "envelope_type": envelope_type,
                "key_format": key_format,
                "value_format": value_format,
            }))
        }
        CreateSubsource(stmt) => {
            // Progress subsources track ingestion progress and are not
            // source tables. They have no external reference.
            let is_progress = stmt
                .with_options
                .iter()
                .any(|o| matches!(o.name, CreateSubsourceOptionName::Progress));
            if is_progress {
                return Ok(serde_json::Value::Null);
            }

            let source_id = stmt
                .of_source
                .ok_or("non-progress CREATE SUBSOURCE without OF SOURCE")
                .and_then(item_id)?;

            let external_reference = stmt
                .with_options
                .into_iter()
                .find(|o| matches!(o.name, CreateSubsourceOptionName::ExternalReference))
                .and_then(|o| match o.value {
                    Some(WithOptionValue::UnresolvedItemName(name)) => Some(name),
                    _ => None,
                })
                .ok_or("CREATE SUBSOURCE missing EXTERNAL REFERENCE option")?
                .0
                .into_iter()
                .map(|ident| ident.into_string())
                .collect::<Vec<_>>();

            Ok(json!({
                "source_id": source_id,
                "external_reference": external_reference,
                "envelope_type": serde_json::Value::Null,
                "key_format": serde_json::Value::Null,
                "value_format": serde_json::Value::Null,
            }))
        }
        _ => Ok(serde_json::Value::Null),
    }
}

/// Extracts connection-detail metadata from a catalog `create_sql`.
///
/// Returns a per-connection-type object with the fields that the
/// `mz_kafka_connections`, `mz_ssh_tunnel_connections`, and `mz_aws_connections`
/// builtin views need. For everything else (other connection types, including
/// aws-privatelink whose only detail is context-derived, and non-connection
/// statements) it returns jsonb `null`, so callers filter on `IS NOT NULL` and
/// gate on the connection type separately (via
/// `parse_catalog_create_sql(...)->>'connection_type'`, the way `mz_connections`
/// already does).
///
/// The shape per type:
///
/// ```json
/// // kafka
/// { "brokers": ["host:port", ...], "progress_topic": <text | null> }
/// // ssh-tunnel
/// { "public_key_1": "<text>", "public_key_2": "<text>" }
/// // aws
/// {
///   "auth_kind": "credentials" | "assume-role",
///   "endpoint": <text | null>, "region": <text | null>,
///   "access_key_id": <text | null>, "access_key_id_secret_id": <text | null>,
///   "secret_access_key_secret_id": <text | null>,
///   "session_token": <text | null>, "session_token_secret_id": <text | null>,
///   "assume_role_arn": <text | null>, "assume_role_session_name": <text | null>
/// }
/// ```
///
/// `progress_topic` is null when the connection does not set an explicit
/// `PROGRESS TOPIC`. The default (`_materialize-progress-<env>-<conn_id>`) is
/// reconstructed by the view, not here, because it needs the environment id and
/// the connection's own id. Values derived only from environment context
/// (AWS principal, external id, trust policy, privatelink principal) are also
/// left to the view. This keeps the helper a pure function of the `create_sql`.
///
/// For aws, an option is either an inline value or a secret reference. Inline
/// values land in `access_key_id`/`session_token`; a secret reference lands in
/// the matching `*_secret_id` as the referenced secret's catalog item id (the
/// persisted `create_sql` stores resolved references as `[uNNN AS name]`).
/// `auth_kind` is `assume-role` when `ASSUME ROLE ARN` is present, else
/// `credentials`, matching the `AwsAuth` variant the removed packer read.
///
/// Errors if the statement fails to parse.
pub fn connection_details(a: &str) -> Result<serde_json::Value, String> {
    // The persisted `create_sql` stores an inline broker as a single `BROKER`
    // option and a broker list as a `BROKERS (...)` sequence. Either way we
    // only need the addresses, which are present regardless of the tunnel
    // (direct, SSH, or PrivateLink).
    fn broker_addresses<T: AstInfo>(values: &[ConnectionOption<T>]) -> Vec<String> {
        let mut brokers = Vec::new();
        for opt in values {
            match (&opt.name, &opt.value) {
                (ConnectionOptionName::Broker, Some(WithOptionValue::ConnectionKafkaBroker(b))) => {
                    brokers.push(b.address.clone());
                }
                (ConnectionOptionName::Brokers, Some(WithOptionValue::Sequence(seq))) => {
                    for v in seq {
                        if let WithOptionValue::ConnectionKafkaBroker(b) = v {
                            brokers.push(b.address.clone());
                        }
                    }
                }
                _ => {}
            }
        }
        brokers
    }

    fn string_option<T: AstInfo>(
        values: &[ConnectionOption<T>],
        name: ConnectionOptionName,
    ) -> Option<String> {
        values
            .iter()
            .find(|o| o.name == name)
            .and_then(|o| o.value.as_ref())
            .and_then(option_string)
    }

    // The catalog id of the secret a `SECRET ...` option references. Resolved
    // references persist as `RawItemName::Id`, so an unresolved name yields
    // None (the same treatment [`item_id`] gives item names).
    fn secret_id_option<T: AstInfo<ItemName = RawItemName>>(
        values: &[ConnectionOption<T>],
        name: ConnectionOptionName,
    ) -> Option<String> {
        values.iter().find_map(|o| match &o.value {
            Some(WithOptionValue::Secret(RawItemName::Id(id, _, _))) if o.name == name => {
                Some(id.clone())
            }
            _ => None,
        })
    }

    let stmt = parse_single_statement(a)?;

    use mz_sql_parser::ast::Statement::CreateConnection;
    let stmt = match stmt {
        CreateConnection(stmt) => stmt,
        _ => return Ok(serde_json::Value::Null),
    };

    match stmt.connection_type {
        CreateConnectionType::Kafka => Ok(json!({
            "brokers": broker_addresses(&stmt.values),
            "progress_topic": string_option(&stmt.values, ConnectionOptionName::ProgressTopic),
        })),
        CreateConnectionType::Ssh => Ok(json!({
            "public_key_1": string_option(&stmt.values, ConnectionOptionName::PublicKey1),
            "public_key_2": string_option(&stmt.values, ConnectionOptionName::PublicKey2),
        })),
        CreateConnectionType::Aws => {
            let assume_role_arn = string_option(&stmt.values, ConnectionOptionName::AssumeRoleArn);
            let auth_kind = if assume_role_arn.is_some() {
                "assume-role"
            } else {
                "credentials"
            };
            Ok(json!({
                "auth_kind": auth_kind,
                // Planning coerces an empty ENDPOINT to None (see
                // `src/sql/src/plan/statement/ddl/connection.rs`), so the
                // removed packer wrote NULL for `ENDPOINT = ''`. Match that.
                "endpoint": string_option(&stmt.values, ConnectionOptionName::Endpoint)
                    .filter(|s| !s.is_empty()),
                "region": string_option(&stmt.values, ConnectionOptionName::Region),
                "access_key_id": string_option(&stmt.values, ConnectionOptionName::AccessKeyId),
                "access_key_id_secret_id":
                    secret_id_option(&stmt.values, ConnectionOptionName::AccessKeyId),
                "secret_access_key_secret_id":
                    secret_id_option(&stmt.values, ConnectionOptionName::SecretAccessKey),
                "session_token": string_option(&stmt.values, ConnectionOptionName::SessionToken),
                "session_token_secret_id":
                    secret_id_option(&stmt.values, ConnectionOptionName::SessionToken),
                "assume_role_arn": assume_role_arn,
                "assume_role_session_name":
                    string_option(&stmt.values, ConnectionOptionName::AssumeRoleSessionName),
            }))
        }
        _ => Ok(serde_json::Value::Null),
    }
}

#[cfg(test)]
mod tests {
    use prost::Message as _;
    use serde_json::json;

    /// Encode the two proto fields our decoder cares about, using the same
    /// tag numbering as the canonical proto.
    fn encode_pg_details(slot: &str, timeline_id: Option<u64>) -> String {
        let details = super::PostgresPublicationDetailsSubset {
            slot: slot.to_string(),
            timeline_id,
        };
        hex::encode(details.encode_to_vec())
    }

    fn pg_source_sql(details_hex: &str) -> String {
        format!(
            "CREATE SOURCE \"materialize\".\"public\".\"pg_src\" \
             IN CLUSTER [u42] \
             FROM POSTGRES CONNECTION [u10 AS \"materialize\".\"public\".\"pg_conn\"] \
             (DETAILS = '{details_hex}', PUBLICATION = 'mz_source') \
             FOR ALL TABLES"
        )
    }

    fn kafka_source_sql(with_prefix: bool) -> String {
        let prefix_opt = if with_prefix {
            ", GROUP ID PREFIX 'my-prefix-'"
        } else {
            ""
        };
        format!(
            "CREATE SOURCE \"materialize\".\"public\".\"k_src\" \
             IN CLUSTER [u42] \
             FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"] \
             (TOPIC 'test'{prefix_opt}) FORMAT TEXT"
        )
    }

    // --- parse_postgres_source_details ---------------------------------------

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `decContextDefault` on OS `linux`
    fn pg_happy_path_with_timeline() {
        let hex = encode_pg_details("materialize_abc", Some(42));
        let sql = pg_source_sql(&hex);
        let out = super::postgres_source_details(&sql).expect("ok");
        assert_eq!(out, json!({ "slot": "materialize_abc", "timeline_id": 42 }));
    }

    #[mz_ore::test]
    fn pg_happy_path_null_timeline() {
        // Pre-2024 sources have no timeline_id field. The decoder must
        // surface that as JSON null, not error.
        let hex = encode_pg_details("materialize_legacy", None);
        let sql = pg_source_sql(&hex);
        let out = super::postgres_source_details(&sql).expect("ok");
        assert_eq!(
            out,
            json!({ "slot": "materialize_legacy", "timeline_id": null }),
        );
    }

    #[mz_ore::test]
    fn pg_non_postgres_source_returns_null_jsonb() {
        let sql = "CREATE SOURCE \"materialize\".\"public\".\"lg\" \
             IN CLUSTER [u42] FROM LOAD GENERATOR COUNTER";
        let out = super::postgres_source_details(sql).expect("ok");
        assert_eq!(out, serde_json::Value::Null);
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn pg_non_create_source_returns_null_jsonb() {
        let sql = "CREATE VIEW v AS SELECT 1";
        let out = super::postgres_source_details(sql).expect("ok");
        assert_eq!(out, serde_json::Value::Null);
    }

    #[mz_ore::test]
    fn pg_missing_details_option_errors() {
        let sql = "CREATE SOURCE \"materialize\".\"public\".\"pg_src\" \
             IN CLUSTER [u42] \
             FROM POSTGRES CONNECTION [u10 AS \"materialize\".\"public\".\"pg_conn\"] \
             (PUBLICATION = 'mz_source') FOR ALL TABLES";
        let err = super::postgres_source_details(sql).unwrap_err();
        assert!(
            err.contains("missing DETAILS"),
            "wrong error message: {err}"
        );
    }

    #[mz_ore::test]
    fn pg_malformed_hex_errors() {
        let sql = pg_source_sql("not-hex!!");
        let err = super::postgres_source_details(&sql).unwrap_err();
        assert!(err.contains("valid hex"), "wrong error message: {err}");
    }

    #[mz_ore::test]
    fn pg_malformed_proto_errors() {
        // Valid hex, garbage bytes. Prost decoding fails on unexpected wire
        // format.
        let sql = pg_source_sql("ffff");
        let err = super::postgres_source_details(&sql).unwrap_err();
        assert!(
            err.contains("publication-details proto"),
            "wrong error message: {err}"
        );
    }

    // --- parse_kafka_source_details ------------------------------------------

    #[mz_ore::test]
    fn kafka_happy_path_with_prefix() {
        let sql = kafka_source_sql(true);
        let out = super::kafka_source_details(&sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "topic": "test",
                "group_id_prefix": "my-prefix-",
                "connection_id": "u11",
            }),
        );
    }

    #[mz_ore::test]
    fn kafka_happy_path_without_prefix() {
        let sql = kafka_source_sql(false);
        let out = super::kafka_source_details(&sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "topic": "test",
                "group_id_prefix": null,
                "connection_id": "u11",
            }),
        );
    }

    #[mz_ore::test]
    fn kafka_non_kafka_source_returns_null_jsonb() {
        let sql = "CREATE SOURCE \"materialize\".\"public\".\"lg\" \
             IN CLUSTER [u42] FROM LOAD GENERATOR COUNTER";
        let out = super::kafka_source_details(sql).expect("ok");
        assert_eq!(out, serde_json::Value::Null);
    }

    #[mz_ore::test]
    fn kafka_missing_topic_errors() {
        let sql = "CREATE SOURCE \"materialize\".\"public\".\"k_src\" \
             IN CLUSTER [u42] \
             FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"] \
             FORMAT TEXT";
        let err = super::kafka_source_details(sql).unwrap_err();
        assert!(err.contains("missing TOPIC"), "wrong error message: {err}");
    }

    #[mz_ore::test]
    fn kafka_unquoted_topic_and_prefix() {
        // Planning accepts a bare identifier for TOPIC / GROUP ID PREFIX, and it
        // persists unquoted in create_sql. Matching only quoted strings would
        // drop TOPIC and error the whole mz_kafka_source_tables view.
        let sql = "CREATE SOURCE \"materialize\".\"public\".\"k_src\" \
             IN CLUSTER [u42] \
             FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"] \
             (TOPIC = my_topic, GROUP ID PREFIX = my_prefix) FORMAT TEXT";
        let out = super::kafka_source_details(sql).expect("ok");
        assert_eq!(out["topic"], json!("my_topic"));
        assert_eq!(out["group_id_prefix"], json!("my_prefix"));
    }

    #[mz_ore::test]
    fn kafka_unresolved_connection_errors() {
        // A bare-name connection reference never happens after purification,
        // but the decoder must reject it explicitly rather than silently
        // dropping the connection_id.
        let sql = "CREATE SOURCE \"materialize\".\"public\".\"k_src\" \
             IN CLUSTER [u42] \
             FROM KAFKA CONNECTION k_conn (TOPIC 'test') FORMAT TEXT";
        let err = super::kafka_source_details(sql).unwrap_err();
        assert!(
            err.contains("unresolved item name"),
            "wrong error message: {err}"
        );
    }

    // --- parse_source_export_details -----------------------------------------

    fn table_from_source_sql(reference: &str, suffix: &str) -> String {
        format!(
            "CREATE TABLE \"materialize\".\"public\".\"tbl\" \
             FROM SOURCE [u1 AS \"materialize\".\"public\".\"src\"] \
             (REFERENCE = {reference}){suffix}"
        )
    }

    #[mz_ore::test]
    fn export_table_postgres_style_no_format() {
        // Postgres/mysql/sql-server tables carry a multi-part external
        // reference and no format or envelope.
        let sql = table_from_source_sql("\"db\".\"public\".\"t\"", "");
        let out = super::source_export_details(&sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "source_id": "u1",
                "external_reference": ["db", "public", "t"],
                "envelope_type": null,
                "key_format": null,
                "value_format": null,
            }),
        );
    }

    #[mz_ore::test]
    fn export_table_kafka_bare_value_only() {
        // A bare non-registry FORMAT is value-only: no key format.
        let sql = table_from_source_sql("\"topic\"", " FORMAT TEXT ENVELOPE NONE");
        let out = super::source_export_details(&sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "source_id": "u1",
                "external_reference": ["topic"],
                "envelope_type": "none",
                "key_format": null,
                "value_format": "text",
            }),
        );
    }

    #[mz_ore::test]
    fn export_table_kafka_omitted_envelope_is_null() {
        // Omitting ENVELOPE persists as absent in create_sql, so this
        // source-type-agnostic helper reports null. The mz_kafka_source_tables
        // view is responsible for defaulting kafka's null envelope to 'none'.
        let sql = table_from_source_sql("\"topic\"", " FORMAT TEXT");
        let out = super::source_export_details(&sql).expect("ok");
        assert_eq!(out["envelope_type"], serde_json::Value::Null);
    }

    #[mz_ore::test]
    fn export_table_kafka_key_value_format() {
        let sql = table_from_source_sql(
            "\"topic\"",
            " KEY FORMAT TEXT VALUE FORMAT TEXT ENVELOPE NONE",
        );
        let out = super::source_export_details(&sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "source_id": "u1",
                "external_reference": ["topic"],
                "envelope_type": "none",
                "key_format": "text",
                "value_format": "text",
            }),
        );
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn export_table_kafka_bare_avro_seed_with_key() {
        // A bare Avro CSR format whose seed carries a key schema resolves to
        // an encoding with a key, so key_format mirrors value_format. This is
        // the upsert/debezium path.
        let sql = table_from_source_sql(
            "\"topic\"",
            " FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY \
             CONNECTION [u5 AS \"materialize\".\"public\".\"csr\"] \
             SEED KEY SCHEMA 'k' VALUE SCHEMA 'v' ENVELOPE UPSERT",
        );
        let out = super::source_export_details(&sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "source_id": "u1",
                "external_reference": ["topic"],
                "envelope_type": "upsert",
                "key_format": "avro",
                "value_format": "avro",
            }),
        );
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn export_table_kafka_bare_avro_seed_without_key() {
        // A bare Avro CSR seed with only a value schema is value-only.
        let sql = table_from_source_sql(
            "\"topic\"",
            " FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY \
             CONNECTION [u5 AS \"materialize\".\"public\".\"csr\"] \
             SEED VALUE SCHEMA 'v' ENVELOPE NONE",
        );
        let out = super::source_export_details(&sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "source_id": "u1",
                "external_reference": ["topic"],
                "envelope_type": "none",
                "key_format": null,
                "value_format": "avro",
            }),
        );
    }

    #[mz_ore::test]
    fn export_subsource_non_progress() {
        // Old-syntax subsource: external reference lives in a WITH option, and
        // there is never a format or envelope.
        let sql = "CREATE SUBSOURCE \"materialize\".\"public\".\"sub\" (id int4) \
             OF SOURCE [u1 AS \"materialize\".\"public\".\"src\"] \
             WITH (EXTERNAL REFERENCE = \"db\".\"public\".\"t\")";
        let out = super::source_export_details(sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "source_id": "u1",
                "external_reference": ["db", "public", "t"],
                "envelope_type": null,
                "key_format": null,
                "value_format": null,
            }),
        );
    }

    #[mz_ore::test]
    fn export_progress_subsource_returns_null_jsonb() {
        let sql = "CREATE SUBSOURCE \"materialize\".\"public\".\"progress\" (id int4) \
             WITH (PROGRESS)";
        let out = super::source_export_details(sql).expect("ok");
        assert_eq!(out, serde_json::Value::Null);
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn export_non_source_export_returns_null_jsonb() {
        let sql = "CREATE VIEW v AS SELECT 1";
        let out = super::source_export_details(sql).expect("ok");
        assert_eq!(out, serde_json::Value::Null);
    }

    #[mz_ore::test]
    fn export_unresolved_source_name_errors() {
        let sql = "CREATE TABLE \"materialize\".\"public\".\"tbl\" \
             FROM SOURCE src (REFERENCE = \"topic\")";
        let err = super::source_export_details(sql).unwrap_err();
        assert!(
            err.contains("unresolved item name"),
            "wrong error message: {err}"
        );
    }

    // --- parse_connection_details --------------------------------------------

    #[mz_ore::test]
    fn connection_kafka_single_broker_default_progress() {
        // No explicit PROGRESS TOPIC: the helper leaves it null and the view
        // reconstructs the default.
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO KAFKA \
             (BROKER = 'localhost:9092', SECURITY PROTOCOL = plaintext)";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "brokers": ["localhost:9092"],
                "progress_topic": null,
            }),
        );
    }

    #[mz_ore::test]
    fn connection_kafka_explicit_progress_topic() {
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO KAFKA \
             (BROKER = 'localhost:9092', PROGRESS TOPIC = 'override', \
              SECURITY PROTOCOL = plaintext)";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "brokers": ["localhost:9092"],
                "progress_topic": "override",
            }),
        );
    }

    #[mz_ore::test]
    fn connection_kafka_broker_list() {
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO KAFKA \
             (BROKERS ('b1:9092', 'b2:9092'), SECURITY PROTOCOL = plaintext)";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "brokers": ["b1:9092", "b2:9092"],
                "progress_topic": null,
            }),
        );
    }

    #[mz_ore::test]
    fn connection_ssh_public_keys() {
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO SSH TUNNEL \
             (HOST = 'ssh.example.com', PORT = 22, USER = 'mz', \
              PUBLIC KEY 1 = 'ssh-ed25519 AAAA', PUBLIC KEY 2 = 'ssh-ed25519 BBBB')";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "public_key_1": "ssh-ed25519 AAAA",
                "public_key_2": "ssh-ed25519 BBBB",
            }),
        );
    }

    #[mz_ore::test]
    fn connection_aws_credentials_inline_key() {
        // Inline ACCESS KEY ID, secret SECRET ACCESS KEY. Assume-role columns
        // stay null and auth_kind is credentials.
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO AWS \
             (ACCESS KEY ID = 'AKIAEXAMPLE', \
              SECRET ACCESS KEY = SECRET [u1 AS \"materialize\".\"public\".\"sk\"])";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "auth_kind": "credentials",
                "endpoint": null,
                "region": null,
                "access_key_id": "AKIAEXAMPLE",
                "access_key_id_secret_id": null,
                "secret_access_key_secret_id": "u1",
                "session_token": null,
                "session_token_secret_id": null,
                "assume_role_arn": null,
                "assume_role_session_name": null,
            }),
        );
    }

    #[mz_ore::test]
    fn connection_aws_credentials_secret_key_and_session_token() {
        // Every credential provided as a secret reference lands in the matching
        // *_secret_id column as the referenced secret's catalog id.
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO AWS \
             (ENDPOINT = 'http://localhost', REGION = 'us-east-1', \
              ACCESS KEY ID = SECRET [u1 AS \"materialize\".\"public\".\"ak\"], \
              SECRET ACCESS KEY = SECRET [u2 AS \"materialize\".\"public\".\"sk\"], \
              SESSION TOKEN = SECRET [u3 AS \"materialize\".\"public\".\"st\"])";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "auth_kind": "credentials",
                "endpoint": "http://localhost",
                "region": "us-east-1",
                "access_key_id": null,
                "access_key_id_secret_id": "u1",
                "secret_access_key_secret_id": "u2",
                "session_token": null,
                "session_token_secret_id": "u3",
                "assume_role_arn": null,
                "assume_role_session_name": null,
            }),
        );
    }

    #[mz_ore::test]
    fn connection_aws_assume_role() {
        // Assume-role sets auth_kind and the assume-role columns; credential
        // columns stay null.
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO AWS \
             (ASSUME ROLE ARN 'arn:aws:iam::123:role/mz', \
              ASSUME ROLE SESSION NAME 'sess')";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "auth_kind": "assume-role",
                "endpoint": null,
                "region": null,
                "access_key_id": null,
                "access_key_id_secret_id": null,
                "secret_access_key_secret_id": null,
                "session_token": null,
                "session_token_secret_id": null,
                "assume_role_arn": "arn:aws:iam::123:role/mz",
                "assume_role_session_name": "sess",
            }),
        );
    }

    #[mz_ore::test]
    fn connection_kafka_unquoted_progress_topic() {
        // A bare identifier PROGRESS TOPIC persists unquoted in create_sql. The
        // helper must surface it, else the view substitutes the default topic.
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO KAFKA \
             (BROKER = 'localhost:9092', PROGRESS TOPIC = my_topic, \
              SECURITY PROTOCOL = plaintext)";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(out["progress_topic"], json!("my_topic"));
    }

    #[mz_ore::test]
    fn connection_aws_unquoted_option_values() {
        // Planning accepts bare identifiers for these options and persists them
        // unquoted. The helper must surface them, not fall back to NULL.
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO AWS \
             (ENDPOINT = localhost, REGION = useast1, \
              ASSUME ROLE ARN 'arn:aws:iam::123:role/mz', \
              ASSUME ROLE SESSION NAME = mysession)";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(out["endpoint"], json!("localhost"));
        assert_eq!(out["region"], json!("useast1"));
        assert_eq!(out["assume_role_session_name"], json!("mysession"));
    }

    #[mz_ore::test]
    fn connection_aws_empty_endpoint_is_null() {
        // Planning coerces ENDPOINT = '' to None, so the packer wrote NULL. The
        // view must match, not report an empty string.
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO AWS \
             (ENDPOINT = '', ASSUME ROLE ARN 'arn:aws:iam::123:role/mz')";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(out["endpoint"], serde_json::Value::Null);
    }

    #[mz_ore::test]
    fn connection_other_type_returns_null_jsonb() {
        // A connection type without a detail view (postgres) yields null.
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"c\" TO POSTGRES \
             (HOST = 'db', DATABASE = 'postgres', USER = 'mz')";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(out, serde_json::Value::Null);
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn connection_non_connection_returns_null_jsonb() {
        let sql = "CREATE VIEW v AS SELECT 1";
        let out = super::connection_details(sql).expect("ok");
        assert_eq!(out, serde_json::Value::Null);
    }

    // --- parse_catalog_create_sql envelope_type ------------------------------

    #[mz_ore::test]
    fn catalog_kafka_old_syntax_omitted_envelope_defaults_none() {
        // An old-syntax kafka source (carries EXPOSE PROGRESS AS) ingests into
        // its own relation, so an omitted ENVELOPE means the default NONE, which
        // the pre-MV packer reported as 'none'.
        let sql = "CREATE SOURCE \"materialize\".\"public\".\"k\" \
             IN CLUSTER [u42] \
             FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"] \
             (TOPIC 'test') FORMAT TEXT \
             EXPOSE PROGRESS AS [u12 AS \"materialize\".\"public\".\"k_progress\"]";
        let out = super::item_details(sql).expect("ok");
        assert_eq!(out["envelope_type"], json!("none"));
    }

    #[mz_ore::test]
    fn catalog_kafka_old_syntax_explicit_envelope() {
        let sql = "CREATE SOURCE \"materialize\".\"public\".\"k\" \
             IN CLUSTER [u42] \
             FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"] \
             (TOPIC 'test') FORMAT BYTES ENVELOPE UPSERT \
             EXPOSE PROGRESS AS [u12 AS \"materialize\".\"public\".\"k_progress\"]";
        let out = super::item_details(sql).expect("ok");
        assert_eq!(out["envelope_type"], json!("upsert"));
    }

    #[mz_ore::test]
    fn catalog_kafka_new_syntax_source_omits_envelope_type() {
        // A new-syntax kafka source has no progress subsource (no EXPOSE PROGRESS
        // AS). It ingests nothing itself. Envelopes live on the per-table exports,
        // so its own envelope_type stays absent (SQL NULL).
        let sql = "CREATE SOURCE \"materialize\".\"public\".\"k\" \
             IN CLUSTER [u42] \
             FROM KAFKA CONNECTION [u11 AS \"materialize\".\"public\".\"k_conn\"]";
        let out = super::item_details(sql).expect("ok");
        assert_eq!(out.get("envelope_type"), None);
    }

    #[mz_ore::test]
    fn catalog_non_kafka_source_omits_envelope_type() {
        // Non-kafka sources carry no envelope, so envelope_type stays absent
        // (SQL NULL), not 'none'.
        let sql = "CREATE SOURCE \"materialize\".\"public\".\"lg\" \
             IN CLUSTER [u42] FROM LOAD GENERATOR COUNTER";
        let out = super::item_details(sql).expect("ok");
        assert_eq!(out.get("envelope_type"), None);
    }

    // --- parse_catalog_create_sql, CreateSink arm ----------------------------

    const AVRO_FORMAT: &str = "FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION \
         [u12 AS \"materialize\".\"public\".\"csr_conn\"]";

    /// A persisted kafka-sink `create_sql`: resolved names, and a `TOPIC` that
    /// planning guarantees.
    fn kafka_sink_sql(key: Option<&str>, format: &str, envelope: &str) -> String {
        let key_clause = key.map(|k| format!(" KEY ({k})")).unwrap_or_default();
        format!(
            "CREATE SINK \"materialize\".\"public\".\"snk\" \
             IN CLUSTER [u42] \
             FROM [u1 AS \"materialize\".\"public\".\"t\"] \
             INTO KAFKA CONNECTION [u10 AS \"materialize\".\"public\".\"k_conn\"] \
             (TOPIC 'sink-topic'){key_clause} {format} ENVELOPE {envelope}"
        )
    }

    fn iceberg_sink_sql(mode: &str) -> String {
        format!(
            "CREATE SINK \"materialize\".\"public\".\"ice\" \
             IN CLUSTER [u42] \
             FROM [u1 AS \"materialize\".\"public\".\"t\"] \
             INTO ICEBERG CATALOG CONNECTION [u20 AS \"materialize\".\"public\".\"cat_conn\"] \
             (NAMESPACE 'ns', TABLE 'tbl') \
             USING AWS CONNECTION [u21 AS \"materialize\".\"public\".\"aws_conn\"] \
             MODE {mode}"
        )
    }

    #[mz_ore::test]
    fn sink_kafka_bare_format_without_key() {
        let sql = kafka_sink_sql(None, "FORMAT JSON", "DEBEZIUM");
        let out = super::item_details(&sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "type": "sink",
                "sink_type": "kafka",
                "cluster_id": "u42",
                "connection_id": "u10",
                "topic": "sink-topic",
                "envelope_type": "debezium",
                "format": "json",
                "value_format": "json",
            }),
        );
    }

    #[mz_ore::test]
    fn sink_kafka_bare_format_with_key_derives_key_format() {
        // A bare format applies to the key too once the sink has a KEY, which
        // is what makes the deprecated `format` column collapse to `avro`.
        let sql = kafka_sink_sql(Some("a"), AVRO_FORMAT, "UPSERT");
        let out = super::item_details(&sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "type": "sink",
                "sink_type": "kafka",
                "cluster_id": "u42",
                "connection_id": "u10",
                "topic": "sink-topic",
                "envelope_type": "upsert",
                "format": "avro",
                "key_format": "avro",
                "value_format": "avro",
            }),
        );
    }

    #[mz_ore::test]
    fn sink_kafka_bare_text_format_with_key_does_not_collapse() {
        // Only avro/avro and json/json collapse, so a keyed text sink reports
        // the composite form even though both halves are `text`.
        let sql = kafka_sink_sql(Some("a"), "FORMAT TEXT", "UPSERT");
        let out = super::item_details(&sql).expect("ok");
        assert_eq!(out["format"], json!("key-text-value-text"));
        assert_eq!(out["key_format"], json!("text"));
        assert_eq!(out["value_format"], json!("text"));
    }

    #[mz_ore::test]
    fn sink_kafka_key_value_json_collapses() {
        let sql = kafka_sink_sql(Some("a"), "KEY FORMAT JSON VALUE FORMAT JSON", "UPSERT");
        let out = super::item_details(&sql).expect("ok");
        assert_eq!(out["format"], json!("json"));
        assert_eq!(out["key_format"], json!("json"));
        assert_eq!(out["value_format"], json!("json"));
    }

    #[mz_ore::test]
    fn sink_kafka_key_value_mixed_is_composite() {
        let sql = kafka_sink_sql(Some("a"), "KEY FORMAT TEXT VALUE FORMAT BYTES", "UPSERT");
        let out = super::item_details(&sql).expect("ok");
        assert_eq!(out["format"], json!("key-text-value-bytes"));
        assert_eq!(out["key_format"], json!("text"));
        assert_eq!(out["value_format"], json!("bytes"));
    }

    #[mz_ore::test]
    fn sink_kafka_key_format_without_key_is_dropped() {
        // `kafka_sink_builder` ignores the key half of the format spec when the
        // sink has no KEY, so neither `key_format` nor the composite `format`
        // may reflect it.
        let sql = kafka_sink_sql(None, "KEY FORMAT JSON VALUE FORMAT TEXT", "DEBEZIUM");
        let out = super::item_details(&sql).expect("ok");
        assert_eq!(out["format"], json!("text"));
        assert_eq!(out["key_format"], serde_json::Value::Null);
        assert_eq!(out["value_format"], json!("text"));
    }

    #[mz_ore::test]
    fn sink_kafka_missing_topic_errors() {
        let sql = "CREATE SINK \"materialize\".\"public\".\"snk\" \
             IN CLUSTER [u42] \
             FROM [u1 AS \"materialize\".\"public\".\"t\"] \
             INTO KAFKA CONNECTION [u10 AS \"materialize\".\"public\".\"k_conn\"] \
             FORMAT JSON ENVELOPE DEBEZIUM";
        let err = super::item_details(sql).unwrap_err();
        assert!(err.contains("missing TOPIC"), "wrong error message: {err}");
    }

    #[mz_ore::test]
    fn sink_iceberg_upsert_mode() {
        let sql = iceberg_sink_sql("UPSERT");
        let out = super::item_details(&sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "type": "sink",
                "sink_type": "iceberg",
                "cluster_id": "u42",
                // The catalog connection, never the AWS connection (u21).
                "connection_id": "u20",
                "namespace": "ns",
                "table": "tbl",
                "envelope_type": "upsert",
            }),
        );
    }

    #[mz_ore::test]
    fn sink_iceberg_append_mode() {
        let sql = iceberg_sink_sql("APPEND");
        let out = super::item_details(&sql).expect("ok");
        // `append` is reachable only through an iceberg sink's MODE.
        assert_eq!(out["envelope_type"], json!("append"));
    }

    #[mz_ore::test]
    fn sink_iceberg_missing_table_errors() {
        let sql = "CREATE SINK \"materialize\".\"public\".\"ice\" \
             IN CLUSTER [u42] \
             FROM [u1 AS \"materialize\".\"public\".\"t\"] \
             INTO ICEBERG CATALOG CONNECTION [u20 AS \"materialize\".\"public\".\"cat_conn\"] \
             (NAMESPACE 'ns') MODE UPSERT";
        let err = super::item_details(sql).unwrap_err();
        assert!(err.contains("missing TABLE"), "wrong error message: {err}");
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn sink_arm_leaves_other_item_types_alone() {
        let sql = "CREATE VIEW \"materialize\".\"public\".\"v\" AS SELECT 1";
        let out = super::item_details(sql).expect("ok");
        assert_eq!(out, json!({ "type": "view", "definition": "SELECT 1;" }));
    }

    // --- parse_catalog_create_sql --------------------------------------------

    /// `type` for a `create_sql`, or the error message if parsing failed.
    fn item_type(sql: &str) -> Result<String, String> {
        match super::item_details(sql) {
            Ok(out) => match out {
                serde_json::Value::Object(mut m) => match m.remove("type") {
                    Some(serde_json::Value::String(s)) => Ok(s),
                    other => panic!("no string `type` key: {other:?}"),
                },
                other => panic!("not a JSON object: {other:?}"),
            },
            Err(msg) => Err(msg),
        }
    }

    fn view_sql(query: &str) -> String {
        format!("CREATE VIEW \"materialize\".\"public\".\"v\" AS {query}")
    }

    /// `definition` for a `CREATE VIEW` whose query is `query`.
    fn view_definition(query: &str) -> String {
        match super::item_details(&view_sql(query)).expect("ok") {
            serde_json::Value::Object(mut m) => match m.remove("definition") {
                Some(serde_json::Value::String(s)) => s,
                other => panic!("no string `definition` key: {other:?}"),
            },
            other => panic!("not a JSON object: {other:?}"),
        }
    }

    /// `mz_tables` and `mz_views` select rows by
    /// `parse_catalog_create_sql(...)->>'type'`, and the function runs over
    /// every `Item` row in the catalog, so a statement kind that changes its
    /// reported type silently gains or loses rows in those relations. Pin the
    /// type of every kind the catalog can hold.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn catalog_item_type_per_statement_kind() {
        let cases = [
            (
                "CREATE TABLE \"materialize\".\"public\".\"t\" (a int4)",
                "table",
            ),
            // A table created from a source, and a webhook table, are both
            // `table`, so both land in mz_tables.
            (
                "CREATE TABLE \"materialize\".\"public\".\"tbl\" \
                 FROM SOURCE [u1 AS \"materialize\".\"public\".\"src\"] \
                 (REFERENCE = \"topic\") FORMAT TEXT",
                "table",
            ),
            (
                "CREATE TABLE \"materialize\".\"public\".\"wht\" FROM WEBHOOK BODY FORMAT JSON",
                "table",
            ),
            (
                "CREATE VIEW \"materialize\".\"public\".\"v\" AS SELECT 1",
                "view",
            ),
            (
                "CREATE MATERIALIZED VIEW \"materialize\".\"public\".\"mv\" \
                 IN CLUSTER [u1] AS SELECT 1",
                "materialized-view",
            ),
            (
                "CREATE SOURCE \"materialize\".\"public\".\"lg\" \
                 IN CLUSTER [u1] FROM LOAD GENERATOR COUNTER",
                "source",
            ),
            (
                "CREATE SOURCE \"materialize\".\"public\".\"wh\" \
                 IN CLUSTER [u1] FROM WEBHOOK BODY FORMAT JSON",
                "source",
            ),
            (
                "CREATE SUBSOURCE \"materialize\".\"public\".\"sub\" (id int4) \
                 OF SOURCE [u1 AS \"materialize\".\"public\".\"src\"]",
                "subsource",
            ),
            (
                "CREATE SUBSOURCE \"materialize\".\"public\".\"progress\" (id int4) \
                 WITH (PROGRESS)",
                "subsource",
            ),
            (
                "CREATE SINK \"materialize\".\"public\".\"snk\" IN CLUSTER [u1] \
                 FROM [u1 AS \"materialize\".\"public\".\"t\"] \
                 INTO KAFKA CONNECTION [u2 AS \"materialize\".\"public\".\"c\"] \
                 (TOPIC 'tp') FORMAT JSON ENVELOPE DEBEZIUM",
                "sink",
            ),
            (
                "CREATE INDEX \"i\" IN CLUSTER [u1] \
                 ON [u1 AS \"materialize\".\"public\".\"t\"] (\"a\")",
                "index",
            ),
            (
                "CREATE TYPE \"materialize\".\"public\".\"ty\" AS LIST (ELEMENT TYPE = int4)",
                "type",
            ),
            (
                "CREATE SECRET \"materialize\".\"public\".\"s\" AS 'x'",
                "secret",
            ),
            (
                "CREATE CONNECTION \"materialize\".\"public\".\"c\" \
                 TO KAFKA (BROKER 'b', SECURITY PROTOCOL PLAINTEXT)",
                "connection",
            ),
        ];
        for (sql, expected) in cases {
            assert_eq!(item_type(sql).as_deref(), Ok(expected), "for {sql}");
        }
    }

    /// `mz_views.definition` is produced here. It used to be produced by
    /// `pack_view_update` in the adapter, so the exact rendering is a
    /// compatibility surface: `pg_views.definition` reads it.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn catalog_view_definition() {
        // Identifiers and function names come back fully quoted, literals
        // untouched, and PostgreSQL's trailing semicolon is appended.
        assert_eq!(view_definition("SELECT 1"), "SELECT 1;");
        assert_eq!(
            view_definition("WITH c AS (SELECT 1 AS a) SELECT a FROM c"),
            "WITH \"c\" AS (SELECT 1 AS \"a\") SELECT \"a\" FROM \"c\";"
        );
        assert_eq!(
            view_definition("SELECT 1 UNION ALL SELECT 2"),
            "SELECT 1 UNION ALL SELECT 2;"
        );
        assert_eq!(
            view_definition("SELECT (SELECT max(a) FROM [u1 AS \"materialize\".\"public\".\"t\"])"),
            "SELECT (SELECT \"max\"(\"a\") FROM [u1 AS \"materialize\".\"public\".\"t\"]);"
        );
        // Identifiers needing quotes, an embedded double quote, non-ASCII, an
        // embedded single quote in a literal, and ORDER BY all survive.
        assert_eq!(
            view_definition(
                "SELECT \"a b\", \"héllo\", \"q\"\"x\" \
                 FROM [u1 AS \"materialize\".\"public\".\"t\"] \
                 WHERE s = 'lit''eral' AND n = 42 ORDER BY 1"
            ),
            "SELECT \"a b\", \"héllo\", \"q\"\"x\" \
             FROM [u1 AS \"materialize\".\"public\".\"t\"] \
             WHERE \"s\" = 'lit''eral' AND \"n\" = 42 ORDER BY 1;"
        );
    }

    /// The rendering must be a fixed point: `pg_views` consumers re-issue
    /// `definition` as the body of a new view, so a second pass through the
    /// parser has to produce the identical string. The trailing `;` is part of
    /// what gets re-parsed.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn catalog_view_definition_is_idempotent() {
        for query in [
            "SELECT 1",
            "WITH c AS (SELECT 1 AS a) SELECT a FROM c",
            "SELECT 1 UNION ALL SELECT 2",
            "SELECT \"a b\", \"q\"\"x\" FROM [u1 AS \"materialize\".\"public\".\"t\"] \
             WHERE s = 'lit''eral' ORDER BY 1",
        ] {
            let once = view_definition(query);
            assert_eq!(
                view_definition(&once),
                once,
                "not a fixed point for {query}"
            );
        }
    }

    /// `mz_tables.source_id` comes from this key. A table with no source must
    /// omit it entirely, so the MV's `->>'source_id'` yields SQL NULL.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn catalog_table_source_id() {
        let from_source = super::item_details(
            "CREATE TABLE \"materialize\".\"public\".\"tbl\" \
                 FROM SOURCE [u1 AS \"materialize\".\"public\".\"src\"] \
                 (REFERENCE = \"topic\") FORMAT TEXT",
        )
        .expect("ok");
        assert_eq!(from_source, json!({ "type": "table", "source_id": "u1" }));

        for sql in [
            "CREATE TABLE \"materialize\".\"public\".\"t\" (a int4)",
            "CREATE TABLE \"materialize\".\"public\".\"wht\" FROM WEBHOOK BODY FORMAT JSON",
        ] {
            assert_eq!(
                super::item_details(sql).expect("ok"),
                json!({ "type": "table" }),
                "for {sql}"
            );
        }
    }

    /// Every error here is fatal to the whole of `mz_tables`/`mz_views`, not to
    /// one row: the MVs call this function inside their `WHERE` clause, so an
    /// item the parser rejects makes the relation unreadable for everyone.
    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn catalog_create_sql_errors() {
        assert_eq!(
            item_type("this is not sql"),
            Err(
                "failed to parse create_sql: Expected a keyword at the beginning of a statement, \
                 found identifier \"this\""
                    .to_string()
            )
        );
        assert_eq!(
            item_type("CREATE TABLE t (a int4); CREATE TABLE u (b int4)"),
            Err("expected a single statement, found 2".to_string())
        );
        // A statement that is not a CREATE of a catalog item, e.g. if a future
        // change persists something else in an Item record.
        assert_eq!(
            item_type("SELECT 1"),
            Err("not a CREATE item statement".to_string())
        );
        // Catalog `create_sql` always names items by id. An unresolved name
        // means the record was written wrong.
        assert_eq!(
            item_type(
                "CREATE TABLE \"materialize\".\"public\".\"tbl\" \
                 FROM SOURCE src (REFERENCE = \"topic\")"
            ),
            Err("unresolved item name".to_string())
        );
    }

    // --- parse_catalog_item_references ----------------------------------------

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn item_references_view() {
        let sql = "CREATE VIEW \"materialize\".\"public\".\"v\" AS \
             SELECT \"pg_catalog\".\"abs\"(\"a\"::[s20 AS \"pg_catalog\".\"int4\"]) \
             FROM [u1 AS \"materialize\".\"public\".\"t\"]";
        let out = super::item_references(sql).expect("ok");
        assert_eq!(
            out,
            json!({
                "ids": ["s20", "u1"],
                "named_funcs": [{"schema": "pg_catalog", "name": "abs"}],
                "named_types": [],
                "named_relations": [],
            })
        );
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn item_references_connection() {
        let sql = "CREATE CONNECTION \"materialize\".\"public\".\"kc\" TO KAFKA (\
             BROKER 'kafka:9092', \
             SSH TUNNEL [u5 AS \"materialize\".\"public\".\"ssh\"], \
             SASL PASSWORD = SECRET [u7 AS \"materialize\".\"public\".\"pw\"])";
        let out = super::item_references(sql).expect("ok");
        assert_eq!(out["ids"], json!(["u5", "u7"]));
        assert_eq!(out["named_funcs"], json!([]));
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn item_references_cte_shadowing() {
        let sql = "CREATE VIEW \"materialize\".\"public\".\"v\" AS \
             WITH \"c\" AS (SELECT * FROM [u1 AS \"materialize\".\"public\".\"t\"]) \
             SELECT * FROM \"c\"";
        let out = super::item_references(sql).expect("ok");
        assert_eq!(out["ids"], json!(["u1"]));
        assert_eq!(out["named_relations"], json!([]));
    }

    #[mz_ore::test]
    #[cfg_attr(miri, ignore)] // error: unsupported operation: can't call foreign function `rust_psm_stack_pointer` on OS `linux`
    fn item_references_name_based_relation_surfaces() {
        // Stored SQL always prints relations as id references. A name-based
        // relation must surface in "named_relations" rather than vanish.
        let sql = "CREATE VIEW \"materialize\".\"public\".\"v\" AS \
             SELECT * FROM \"mz_catalog\".\"mz_tables\"";
        let out = super::item_references(sql).expect("ok");
        assert_eq!(
            out["named_relations"],
            json!([{"schema": "mz_catalog", "name": "mz_tables"}])
        );
    }

    #[mz_ore::test]
    fn item_references_parse_error() {
        let err = super::item_references("NOT SQL").unwrap_err();
        assert!(
            err.contains("failed to parse"),
            "wrong error message: {err}"
        );
    }

    #[mz_ore::test]
    fn catalog_retain_history_absent_is_omitted() {
        let sql = "CREATE TABLE \"materialize\".\"public\".\"t\" (\"a\" pg_catalog.int4)";
        let out = super::item_details(sql).expect("ok");
        assert_eq!(out.get("retain_history_millis"), None);
    }

    #[mz_ore::test]
    fn catalog_retain_history_spellings() {
        // A string or a bare number parses as an interval, the number as seconds, and an
        // interval literal applies its range qualifier.
        for (option, millis) in [
            ("'1h'", 3_600_000u64),
            ("90", 90_000),
            ("INTERVAL '2' DAY", 172_800_000),
            ("INTERVAL '1:30' MINUTE TO SECOND", 90_000),
        ] {
            let sql = format!(
                "CREATE TABLE \"materialize\".\"public\".\"t\" (\"a\" pg_catalog.int4) \
                 WITH (RETAIN HISTORY = FOR {option})"
            );
            let out = super::item_details(&sql).expect("ok");
            assert_eq!(out["retain_history_millis"], json!(millis), "{option}");
        }
    }

    #[mz_ore::test]
    fn catalog_retain_history_zero_disables_compaction() {
        for option in ["'0'", "0", "NULL"] {
            let sql = format!(
                "CREATE TABLE \"materialize\".\"public\".\"t\" (\"a\" pg_catalog.int4) \
                 WITH (RETAIN HISTORY = FOR {option})"
            );
            let out = super::item_details(&sql).expect("ok");
            assert_eq!(out["retain_history_millis"], json!(u64::MAX), "{option}");
        }
    }

    #[mz_ore::test]
    fn catalog_retain_history_on_each_item_kind() {
        let cases = [
            (
                "CREATE SOURCE \"materialize\".\"public\".\"s\" IN CLUSTER [u1] \
                 FROM LOAD GENERATOR COUNTER WITH (RETAIN HISTORY = FOR '2h')",
                7_200_000u64,
            ),
            (
                "CREATE INDEX \"t_idx\" IN CLUSTER [u1] \
                 ON [u2 AS \"materialize\".\"public\".\"t\"] (\"a\") \
                 WITH (RETAIN HISTORY = FOR '3h')",
                10_800_000,
            ),
            (
                "CREATE MATERIALIZED VIEW \"materialize\".\"public\".\"mv\" IN CLUSTER [u1] \
                 WITH (RETAIN HISTORY = FOR '4h', REFRESH = ON COMMIT) AS SELECT 1",
                14_400_000,
            ),
            (
                "CREATE SUBSOURCE \"materialize\".\"public\".\"sub\" (id int4) \
                 OF SOURCE [u1 AS \"materialize\".\"public\".\"src\"] \
                 WITH (EXTERNAL REFERENCE = \"db\".\"public\".\"t\", RETAIN HISTORY = FOR '5h')",
                18_000_000,
            ),
            (
                "CREATE TABLE \"materialize\".\"public\".\"tbl\" \
                 FROM SOURCE [u1 AS \"materialize\".\"public\".\"src\"] \
                 (REFERENCE = \"db\".\"public\".\"t\") WITH (RETAIN HISTORY = FOR '6h')",
                21_600_000,
            ),
        ];
        for (sql, millis) in cases {
            let out = super::item_details(sql).expect("ok");
            assert_eq!(out["retain_history_millis"], json!(millis), "{sql}");
        }
    }

    const MZ_TIMESTAMP: &str = "[s1 AS \"mz_catalog\".\"mz_timestamp\"]";

    #[mz_ore::test]
    fn catalog_refresh_omitted_means_on_commit() {
        let sql = "CREATE MATERIALIZED VIEW \"materialize\".\"public\".\"mv\" IN CLUSTER [u1] \
                   AS SELECT 1";
        let out = super::item_details(sql).expect("ok");
        assert_eq!(out["refresh"], json!([{"type": "on-commit"}]));
    }

    #[mz_ore::test]
    fn catalog_refresh_one_entry_per_option() {
        let sql = format!(
            "CREATE MATERIALIZED VIEW \"materialize\".\"public\".\"mv\" IN CLUSTER [u1] \
             WITH (REFRESH = ON COMMIT, \
             REFRESH = AT 32472144000000::{MZ_TIMESTAMP}, \
             REFRESH = EVERY '1 day' ALIGNED TO 946684800000::{MZ_TIMESTAMP}, \
             REFRESH = EVERY '90 minutes' ALIGNED TO '2000-01-01 00:00:00+00') \
             AS SELECT 1"
        );
        let out = super::item_details(&sql).expect("ok");
        assert_eq!(
            out["refresh"],
            json!([
                {"type": "on-commit"},
                {"type": "at", "at": format!("32472144000000::{MZ_TIMESTAMP}")},
                {
                    "type": "every",
                    "interval": "24:00:00",
                    "aligned_to": format!("946684800000::{MZ_TIMESTAMP}"),
                },
                {
                    "type": "every",
                    "interval": "01:30:00",
                    "aligned_to": "'2000-01-01 00:00:00+00'",
                },
            ])
        );
    }

    #[mz_ore::test]
    fn catalog_refresh_times_read_as_written() {
        let sql = format!(
            "CREATE MATERIALIZED VIEW \"materialize\".\"public\".\"mv\" IN CLUSTER [u1] \
             WITH (REFRESH = AT ((1790696789484::{MZ_TIMESTAMP})), \
             REFRESH = AT 946684800000::{MZ_TIMESTAMP}::[s46 AS \"pg_catalog\".\"text\"]::[s22 AS \"pg_catalog\".\"int8\"] + 2000, \
             REFRESH = EVERY '1 day' ALIGNED TO '2000-01-01 00:00:00+00'::[s52 AS \"pg_catalog\".\"timestamptz\"] + INTERVAL '1 hour') \
             AS SELECT 1"
        );
        let out = super::item_details(&sql).expect("ok");
        assert_eq!(
            out["refresh"],
            json!([
                {"type": "at", "at": format!("((1790696789484::{MZ_TIMESTAMP}))")},
                {
                    "type": "at",
                    "at": format!(
                        "946684800000::{MZ_TIMESTAMP}::[s46 AS \"pg_catalog\".\"text\"]::[s22 AS \"pg_catalog\".\"int8\"] + 2000"
                    ),
                },
                {
                    "type": "every",
                    "interval": "24:00:00",
                    "aligned_to": "'2000-01-01 00:00:00+00'::[s52 AS \"pg_catalog\".\"timestamptz\"] + INTERVAL '1 hour'",
                },
            ])
        );
    }

    #[mz_ore::test]
    fn catalog_replacement_target() {
        let mv_sql_without_replacement = "CREATE MATERIALIZED VIEW \"materialize\".\"public\".\"mv\" IN CLUSTER [u1] \
             AS SELECT 1";
        let out = super::item_details(mv_sql_without_replacement).expect("ok");
        assert_eq!(out.get("replacement_target"), None);

        let mv_sql_with_replacement = "CREATE REPLACEMENT MATERIALIZED VIEW \"materialize\".\"public\".\"rp\" \
                   FOR [u7 AS \"materialize\".\"public\".\"mv\"] IN CLUSTER [u1] \
                   WITH (REFRESH = ON COMMIT) AS SELECT 1";
        let out = super::item_details(mv_sql_with_replacement).expect("ok");
        assert_eq!(out["replacement_target"], json!("u7"));
    }
}