sql-insight 0.4.0

//! Reference (identity) types shared by SQL inspection features.
//!
//! [`TableReference`] / [`ColumnReference`] are *qualified names* that
//! denote a table / column in a catalog or schema — pure identity, not
//! a relation (no tuples) nor a schema (no attribute types). They carry
//! only enough to name the thing and compare two names for equality.

use core::fmt;

use crate::casing::IdentifierCasing;
use crate::error::Error;
use sqlparser::ast::{
    Expr, GroupByExpr, Ident, Insert, JoinOperator, ObjectName, Query, Select, SelectFlavor,
    SelectItem, SetExpr, TableFactor, TableObject,
};

/// Physical table identity — the `catalog.schema.name` triplet.
///
/// `TableReference` deliberately carries no alias: aliasing is a
/// use-site decoration, not part of a table's identity. Use-site alias
/// information, when needed, is carried by the structures that wrap a
/// `TableReference` (e.g. resolver bindings).
///
/// **Equality has two levels.** The derived `Eq` / `Hash` are
/// *structural* — case- and quote-sensitive, exact segments. That is the
/// right dedup when references come from catalog-backed analysis (matched
/// tables are canonicalized, so equal tables produce equal references) and
/// for direct cross-statement comparison. For catalog-free dedup, where
/// the same table may appear under fold-equivalent spellings (`users` vs
/// `USERS`), use [`identity_key`](Self::identity_key) /
/// [`same_table`](Self::same_table), which fold by a dialect's
/// [`IdentifierCasing`].
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct TableReference {
    #[cfg_attr(
        feature = "serde",
        serde(serialize_with = "crate::serde_support::opt_ident")
    )]
    pub catalog: Option<Ident>,
    #[cfg_attr(
        feature = "serde",
        serde(serialize_with = "crate::serde_support::opt_ident")
    )]
    pub schema: Option<Ident>,
    #[cfg_attr(
        feature = "serde",
        serde(serialize_with = "crate::serde_support::ident")
    )]
    pub name: Ident,
}

/// One read-side occurrence of a [`TableReference`], pairing the
/// identity with how the resolver resolved it ([`ResolutionKind`]).
///
/// The table-granularity mirror of [`ColumnRead`]. Read-side surfaces
/// ([`TableOperation::reads`] and [`TableLineageEdge::source`]) use this
/// wrapper so each occurrence can carry resolution metadata while
/// [`TableReference`] stays identity-only. The write-side counterpart is
/// [`TableWrite`] ([`TableOperation::writes`], [`TableLineageEdge::target`]).
///
/// Unlike [`ColumnRead`], `reference` is **always present**: a table's
/// name is written out in the SQL, so even an
/// [`Ambiguous`](ResolutionKind::Ambiguous) table read (the catalog
/// holds several tables matching an under-qualified name) still surfaces
/// the reference as written. [`Unresolved`](ResolutionKind::Unresolved)
/// therefore never arises at table granularity — it is columns-only.
/// The resolution records how the catalog matched the table:
/// [`Cataloged`](ResolutionKind::Cataloged) for a unique registered hit,
/// [`Ambiguous`](ResolutionKind::Ambiguous) for several, and
/// [`Inferred`](ResolutionKind::Inferred) for a catalog miss or
/// catalog-less mode.
///
/// [`TableOperation::reads`]: crate::extractor::TableOperation::reads
/// [`TableOperation::writes`]: crate::extractor::TableOperation::writes
/// [`TableLineageEdge::source`]: crate::extractor::TableLineageEdge::source
/// [`TableLineageEdge::target`]: crate::extractor::TableLineageEdge::target
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct TableRead {
    pub reference: TableReference,
    pub resolution: ResolutionKind,
}

/// One write-side occurrence of a [`TableReference`] — a DML / DDL write
/// target — pairing the identity with how the catalog matched it
/// ([`ResolutionKind`]).
///
/// The write-role counterpart of [`TableRead`], kept a distinct type so a
/// read can't be passed where a write is meant (and so the write side can
/// diverge later). The `resolution` carries the same catalog-match outcome a
/// scanned source would: [`Cataloged`](ResolutionKind::Cataloged) for a unique
/// registered hit, [`Ambiguous`](ResolutionKind::Ambiguous) for several, and
/// [`Inferred`](ResolutionKind::Inferred) for a catalog miss or catalog-less
/// mode — so the [`Cataloged`](ResolutionKind::Cataloged)-detects-catalog-aware
/// invariant holds on writes too. `reference` is always present (a target's
/// name is written out), so [`Unresolved`](ResolutionKind::Unresolved) never
/// arises here, exactly as for [`TableRead`].
///
/// [`TableOperation::writes`]: crate::extractor::TableOperation::writes
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct TableWrite {
    pub reference: TableReference,
    pub resolution: ResolutionKind,
}

/// A column-level identity reference: an optional owning table plus the
/// column name.
///
/// `table` is `Option` because a column the resolver couldn't pin to a
/// single owning table — [`Ambiguous`](ResolutionKind::Ambiguous) or
/// [`Unresolved`](ResolutionKind::Unresolved) (see
/// [`ColumnRead::resolution`] for *why*) — still surfaces its name with
/// `table: None`. Identity is name-based: two `ColumnReference`s with the
/// same `table` and `name` compare equal, independent of where they
/// appeared in the SQL or how the resolver placed them. (For dialect-aware
/// equality, see [`identity_key`](Self::identity_key).)
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct ColumnReference {
    pub table: Option<TableReference>,
    #[cfg_attr(
        feature = "serde",
        serde(serialize_with = "crate::serde_support::ident")
    )]
    pub name: Ident,
}

/// One read-side occurrence of a [`ColumnReference`], pairing the
/// identity with how the resolver resolved it ([`ResolutionKind`]).
///
/// Read-side surfaces ([`ColumnOperation::reads`] and
/// [`ColumnLineageEdge::source`]) use this wrapper so the same column
/// referenced twice can carry per-occurrence resolution metadata
/// without breaking [`ColumnReference`]'s identity-only contract. The
/// write-side counterpart is [`ColumnWrite`].
///
/// [`ColumnOperation::reads`]: crate::extractor::ColumnOperation::reads
/// [`ColumnLineageEdge::source`]: crate::extractor::ColumnLineageEdge::source
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct ColumnRead {
    pub reference: ColumnReference,
    pub resolution: ResolutionKind,
}

/// One write-side occurrence of a [`ColumnReference`] — a written column —
/// pairing the identity with how the resolver resolved it against the target
/// ([`ResolutionKind`]). The write-role counterpart of [`ColumnRead`], kept a
/// distinct type so a read can't be passed where a write is meant.
///
/// `resolution` is the column's catalog match against its write target:
/// [`Cataloged`](ResolutionKind::Cataloged) when the column is in the target's
/// catalog column list, else [`Inferred`](ResolutionKind::Inferred)
/// (catalog-free, the target's columns aren't known, the column isn't listed,
/// or a freshly created / altered relation).
///
/// The owning table is pinned whenever the statement names the sink — every
/// INSERT / DDL write, a qualified `SET t2.col`, and an unqualified SET with
/// one writable relation. Only an **unqualified SET among several writable
/// relations** (a multi-table `UPDATE t1 JOIN t2 SET col = …`) is *inferred*,
/// with the same rules as a read: a sole candidate pins its owner, several
/// candidates surface [`Ambiguous`](ResolutionKind::Ambiguous) and none
/// [`Unresolved`](ResolutionKind::Unresolved) — `table: None`, the column
/// still named, exactly like an unattributed [`ColumnRead`]. An unattributed
/// write contributes no table-level write.
///
/// [`ColumnOperation::writes`]: crate::extractor::ColumnOperation::writes
/// [`ColumnTarget::Relation`]: crate::extractor::ColumnTarget::Relation
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub struct ColumnWrite {
    pub reference: ColumnReference,
    pub resolution: ResolutionKind,
}

/// How a reference was resolved — "what kind of resolution backs this
/// `(table, name)` placement?".
///
/// Catalog-less mode runs as an *inference mode*: every real-table
/// binding's schema is unknown, so a single-candidate resolution
/// is best-effort, not catalog-backed. CTE and derived bodies do carry
/// known schemas (the resolver derives them from the body's
/// projection), but those refs are synthetic and dropped from the
/// public reads / lineage by the resolver's post-pass.
///
/// `Ambiguous` and `Unresolved` are the two failure modes. Both come
/// with `table: None` on the [`ColumnReference`]; the variant tells
/// the consumer *why* the resolver gave up. (`Unresolved` arises only
/// for columns — a table reference always has a name present.)
///
/// # Invariants
///
/// - **Catalog-less mode → no public `Cataloged`**: every surviving
///   non-synthetic ref points at an unknown real table, so the
///   strongest claim the resolver can make is
///   [`Inferred`](Self::Inferred). Catalog-aware analysis is
///   therefore detectable by the presence of `Cataloged`.
/// - **Catalog-aware mode does not imply `Cataloged`**: catalogs are
///   often partial. Refs against tables the catalog doesn't cover,
///   or against a real unknown table that won a multi-candidate
///   tiebreaker over known ones, both still come back as
///   [`Inferred`](Self::Inferred).
///
/// # How each variant arises
///
/// | Situation | ResolutionKind |
/// |---|---|
/// | catalog-less, real unknown table, sole candidate | [`Inferred`](Self::Inferred) |
/// | catalog-less, two real unknown tables in scope | [`Ambiguous`](Self::Ambiguous) |
/// | catalog-less, CTE known body confirms the column | (internal `Cataloged`; synthetic, dropped) |
/// | catalog-less, CTE known body denies the column (`SELECT typo FROM cte` where cte = `[id]`) | [`Unresolved`](Self::Unresolved) |
/// | catalog-aware, known binding lists the column | [`Cataloged`](Self::Cataloged) |
/// | catalog-aware, known binding *doesn't* list the column | [`Unresolved`](Self::Unresolved) |
/// | catalog-aware, one known confirms + one unknown suspect (known-witness-over-unknown-suspects) | [`Inferred`](Self::Inferred) |
/// | catalog-aware, two or more known schemas confirm | [`Ambiguous`](Self::Ambiguous) |
/// | qualified `t.col` where `t` is unknown | [`Inferred`](Self::Inferred) |
/// | qualified `t.col` where `t` is known and lists `col` | [`Cataloged`](Self::Cataloged) |
///
/// # Consumer guidance
///
/// - **Strict mode validation**: a fully resolved, catalog-confirmed
///   statement satisfies
///   `op.diagnostics.is_empty() && op.reads.iter().all(|r| r.resolution == ResolutionKind::Cataloged)`.
/// - **DFD / CRUD comprehension**: treat
///   [`Cataloged`](Self::Cataloged) and [`Inferred`](Self::Inferred)
///   interchangeably as "resolved" (use the `(table, name)` pair);
///   treat [`Ambiguous`](Self::Ambiguous) and
///   [`Unresolved`](Self::Unresolved) as "incomplete".
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize))]
pub enum ResolutionKind {
    /// Backed by a known schema that lists the column / names the
    /// table. On the public surface this means a catalog (or registry)
    /// entry backed the reference. Internally a CTE / derived body's
    /// known schema also yields this variant on a synthetic ref, but
    /// the post-pass drops those — so consumers only ever see
    /// `Cataloged` for catalog-backed real references.
    Cataloged,
    /// Resolution succeeded by assuming the reference exists where the
    /// resolver placed it: an unknown-schema binding adopted as the
    /// sole candidate, a qualified reference whose qualifier alone
    /// determined the table, or a known witness winning over
    /// unknown suspects in a multi-candidate scope. All defensible
    /// inferences in catalog-less or partial-catalog mode, but not
    /// proven.
    Inferred,
    /// Multiple plausible candidates and the resolver couldn't pick
    /// one: either two-or-more known schemas confirmed the column
    /// (genuine ambiguity), or every candidate was an unknown
    /// suspect with no tiebreaker. `ColumnReference.table` is `None`.
    Ambiguous,
    /// No in-scope binding could plausibly own the column: either
    /// every known schema in scope explicitly denied it, or the
    /// scope chain held no bindings at all. `ColumnReference.table`
    /// is `None`. Columns only.
    Unresolved,
}

impl TableReference {
    pub(crate) fn try_from_name(name: &ObjectName) -> Result<Self, Error> {
        // Every part must be a plain identifier. A non-identifier part — e.g.
        // Snowflake's `IDENTIFIER('t')`, a function-computed name — makes the
        // reference unrepresentable; `as_ident` is `None` there, so the
        // all-or-nothing `collect` yields `None` and we return `Err` rather
        // than `unwrap`-panicking (callers drop it best-effort).
        let parts = name
            .0
            .iter()
            .map(|part| part.as_ident())
            .collect::<Option<Vec<&Ident>>>()
            .ok_or_else(|| {
                Error::AnalysisError(format!(
                    "table name `{name}` is not a plain identifier path"
                ))
            })?;
        match parts.as_slice() {
            [] => Err(Error::AnalysisError(
                "ObjectName has no identifiers".to_string(),
            )),
            [n] => Ok(TableReference {
                catalog: None,
                schema: None,
                name: (*n).clone(),
            }),
            [schema, n] => Ok(TableReference {
                catalog: None,
                schema: Some((*schema).clone()),
                name: (*n).clone(),
            }),
            [catalog, schema, n] => Ok(TableReference {
                catalog: Some((*catalog).clone()),
                schema: Some((*schema).clone()),
                name: (*n).clone(),
            }),
            _ => Err(Error::AnalysisError(
                "Too many identifiers provided".to_string(),
            )),
        }
    }

    /// Format a slice of `TableReference`s as a comma-separated string
    /// (e.g. `"t1, schema.t2, catalog.schema.t3"`). Shared by the
    /// table-extractor `Display` surfaces.
    pub(crate) fn format_list(tables: &[Self]) -> String {
        tables
            .iter()
            .map(|t| t.to_string())
            .collect::<Vec<_>>()
            .join(", ")
    }

    /// Decode an `[Ident]` slice into a `TableReference`. 1 element =
    /// bare name, 2 = `schema.name`, 3 = `catalog.schema.name`. Returns
    /// `None` for 0 or 4+ parts. Use [`Self::try_from_name`] when the
    /// input is an [`ObjectName`] (4+ parts surface as `Error` there).
    pub(crate) fn try_from_parts(parts: &[Ident]) -> Option<Self> {
        match parts {
            [name] => Some(TableReference {
                catalog: None,
                schema: None,
                name: name.clone(),
            }),
            [schema, name] => Some(TableReference {
                catalog: None,
                schema: Some(schema.clone()),
                name: name.clone(),
            }),
            [catalog, schema, name] => Some(TableReference {
                catalog: Some(catalog.clone()),
                schema: Some(schema.clone()),
                name: name.clone(),
            }),
            _ => None,
        }
    }

    /// Parse an INSERT statement's target into (identity, alias) pair. An
    /// Oracle inline-view target (`INSERT INTO (SELECT … FROM t) …`) resolves
    /// through to its single base table; a view over no single base table (a
    /// join, a set operation) surfaces as an `AnalysisError`.
    pub(crate) fn from_insert_with_alias(value: &Insert) -> Result<(Self, Option<Ident>), Error> {
        let name = match &value.table {
            TableObject::TableName(object_name) => object_name,
            TableObject::TableFunction(function) => &function.name,
            // Only a single-table view is shape-determined; a join view's base
            // table needs binder resolution (column attribution), out of reach
            // of a plain identity parse.
            TableObject::TableQuery(query) => {
                match insert_target_view(query)
                    .as_ref()
                    .and_then(insert_target_base)
                {
                    Some(name) => name,
                    None => {
                        return Err(Error::AnalysisError(
                            "INSERT target is a subquery over no single base table".to_string(),
                        ))
                    }
                }
            }
        };
        // `Insert::table_alias` is now a `TableAliasWithoutColumns`; the public
        // pair still exposes just the alias identifier.
        let alias = value.table_alias.as_ref().map(|a| a.alias.clone());
        Ok((Self::try_from_name(name)?, alias))
    }

    /// Parse a `TableFactor::Table` into (identity, alias) pair. Other
    /// `TableFactor` variants (Derived / NestedJoin / Pivot / Unpivot /
    /// MatchRecognize / TableFunction / Function) do not name a stored
    /// table, so they surface as an `AnalysisError`.
    pub(crate) fn from_table_factor_with_alias(
        table: &TableFactor,
    ) -> Result<(Self, Option<Ident>), Error> {
        match table {
            TableFactor::Table { name, alias, .. } => Ok((
                Self::try_from_name(name)?,
                alias.as_ref().map(|a| a.name.clone()),
            )),
            _ => Err(Error::AnalysisError(
                "TableFactor variant other than Table cannot be converted to a TableReference"
                    .to_string(),
            )),
        }
    }
}

impl fmt::Display for TableReference {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let mut parts = Vec::new();
        if let Some(catalog) = &self.catalog {
            parts.push(catalog.to_string());
        }
        if let Some(schema) = &self.schema {
            parts.push(schema.to_string());
        }
        parts.push(self.name.to_string());
        write!(f, "{}", parts.join("."))
    }
}

impl fmt::Display for ColumnReference {
    /// `table.column` when the owning table is known (the table renders as
    /// its own [`TableReference`] path), otherwise just `column`. Mirrors
    /// [`TableReference`]'s `Display` for the column-identity case.
    ///
    /// ```rust
    /// use sql_insight::{ColumnReference, TableReference};
    ///
    /// let qualified = ColumnReference {
    ///     table: Some(TableReference {
    ///         catalog: None,
    ///         schema: Some("public".into()),
    ///         name: "users".into(),
    ///     }),
    ///     name: "id".into(),
    /// };
    /// assert_eq!(qualified.to_string(), "public.users.id");
    ///
    /// let bare = ColumnReference { table: None, name: "id".into() };
    /// assert_eq!(bare.to_string(), "id");
    /// ```
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match &self.table {
            Some(table) => write!(f, "{table}.{}", self.name),
            None => write!(f, "{}", self.name),
        }
    }
}

/// An opaque, dialect-aware identity key for a [`TableReference`].
///
/// Two references whose keys are equal denote the same table *under the
/// given dialect's case-folding* — e.g. `users` and `USERS` share a key in
/// PostgreSQL, but not in a case-sensitive dialect. Use it to deduplicate
/// references catalog-free, where the structural `Eq` / `Hash` on
/// `TableReference` (case-sensitive, quote-sensitive) would over-count
/// fold-equivalent spellings. (With a catalog, matched references are
/// already canonicalized, so structural dedup suffices.)
///
/// The key is **identity**, not wildcard matching: every present segment is
/// significant, so a bare `users` and a qualified `public.users` have
/// *different* keys (they are different identities). The folded text is not
/// observable — only equality / hashing.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct TableIdentityKey {
    catalog: Option<String>,
    schema: Option<String>,
    name: String,
}

/// An opaque, dialect-aware identity key for a [`ColumnReference`] — the
/// [`TableIdentityKey`] of its owning table (if any, folded by the table
/// rule) plus the column name folded by the column rule. See
/// [`TableIdentityKey`] for the identity-vs-matching and opacity notes.
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub struct ColumnIdentityKey {
    table: Option<TableIdentityKey>,
    name: String,
}

impl TableReference {
    /// The dialect-aware [`TableIdentityKey`] for this reference: each
    /// segment folded by `casing`'s table rule. Equal keys denote the same
    /// table under that dialect's casing.
    ///
    /// This is the **catalog-free dedup key**. The structural `Eq` on
    /// `TableReference` is exact (case- and quote-sensitive), so without a
    /// catalog to canonicalize spellings it over-counts `users` and `USERS`
    /// as two tables; folding by the dialect's casing collapses them.
    ///
    /// ```rust
    /// use std::collections::HashSet;
    /// use sql_insight::{CaseRule, IdentifierCasing, TableReference};
    ///
    /// let users = TableReference { catalog: None, schema: None, name: "users".into() };
    /// let upper = TableReference { catalog: None, schema: None, name: "USERS".into() };
    ///
    /// // Structural equality is exact — these read as two different tables.
    /// assert_ne!(users, upper);
    ///
    /// // Under a case-folding dialect (here lower-folding, e.g. PostgreSQL)
    /// // they share one identity.
    /// let casing = IdentifierCasing::uniform(CaseRule::Lower);
    /// assert!(users.same_table(&upper, &casing));
    ///
    /// // So a fold-keyed set counts the table once, where a structural
    /// // `HashSet<TableReference>` would count two.
    /// let distinct: HashSet<_> = [&users, &upper]
    ///     .iter()
    ///     .map(|t| t.identity_key(&casing))
    ///     .collect();
    /// assert_eq!(distinct.len(), 1);
    ///
    /// // Identity, not wildcard: a bare name and a schema-qualified one stay
    /// // distinct (different identities, not a prefix match).
    /// let qualified = TableReference {
    ///     catalog: None,
    ///     schema: Some("public".into()),
    ///     name: "users".into(),
    /// };
    /// assert!(!users.same_table(&qualified, &casing));
    /// ```
    pub fn identity_key(&self, casing: &IdentifierCasing) -> TableIdentityKey {
        let fold = |ident: &Ident| casing.table.normalize(ident);
        TableIdentityKey {
            catalog: self.catalog.as_ref().map(&fold),
            schema: self.schema.as_ref().map(&fold),
            name: fold(&self.name),
        }
    }

    /// Whether `self` and `other` denote the same table under `casing` —
    /// equivalent to comparing their [`identity_key`](Self::identity_key)s.
    pub fn same_table(&self, other: &Self, casing: &IdentifierCasing) -> bool {
        self.identity_key(casing) == other.identity_key(casing)
    }
}

impl ColumnReference {
    /// The dialect-aware [`ColumnIdentityKey`] for this reference: the
    /// owning table folded by the table rule, the column name by the column
    /// rule. Equal keys denote the same column under that dialect's casing.
    ///
    /// Like [`TableReference::identity_key`] this is the catalog-free dedup
    /// key — folding both the owning table and the column name (by their
    /// separate rules, which a dialect can set apart). The owning table is
    /// part of the identity: same column name, different table → different
    /// column.
    ///
    /// ```rust
    /// use sql_insight::{CaseRule, ColumnReference, IdentifierCasing, TableReference};
    ///
    /// let owned = |t: &str| Some(TableReference { catalog: None, schema: None, name: t.into() });
    /// let lower = ColumnReference { table: owned("users"), name: "id".into() };
    /// let upper = ColumnReference { table: owned("USERS"), name: "ID".into() };
    ///
    /// // Structural equality is exact; a case-folding casing merges them.
    /// assert_ne!(lower, upper);
    /// let casing = IdentifierCasing::uniform(CaseRule::Insensitive);
    /// assert!(lower.same_column(&upper, &casing));
    ///
    /// // A different owning table is a different column, same name or not.
    /// let other = ColumnReference { table: owned("accounts"), name: "id".into() };
    /// assert!(!lower.same_column(&other, &casing));
    /// ```
    pub fn identity_key(&self, casing: &IdentifierCasing) -> ColumnIdentityKey {
        ColumnIdentityKey {
            table: self.table.as_ref().map(|t| t.identity_key(casing)),
            name: casing.column.normalize(&self.name),
        }
    }

    /// Whether `self` and `other` denote the same column under `casing` —
    /// equivalent to comparing their [`identity_key`](Self::identity_key)s.
    pub fn same_column(&self, other: &Self, casing: &IdentifierCasing) -> bool {
        self.identity_key(casing) == other.identity_key(casing)
    }
}

impl TryFrom<&Insert> for TableReference {
    type Error = Error;

    fn try_from(value: &Insert) -> Result<Self, Self::Error> {
        Self::from_insert_with_alias(value).map(|(table, _)| table)
    }
}

impl TryFrom<&TableFactor> for TableReference {
    type Error = Error;

    fn try_from(table: &TableFactor) -> Result<Self, Self::Error> {
        Self::from_table_factor_with_alias(table).map(|(table, _)| table)
    }
}

impl TryFrom<&ObjectName> for TableReference {
    type Error = Error;

    fn try_from(obj_name: &ObjectName) -> Result<Self, Self::Error> {
        Self::try_from_name(obj_name)
    }
}

/// A shape-gated Oracle inline-view INSERT target, handed over entirely as
/// **gate-extracted data** — the FROM factors as `(name, alias)` pairs, the
/// projection, the join operators, and the WHERE. The gate proves every
/// factor is a plain table; carrying the proof as data means no downstream
/// re-match (no unreachable fallback arm), and nothing re-reads the `Query`,
/// so no consumer can pair `factors` with a clause it wasn\'t derived from.
pub(crate) struct InsertTargetView<'a> {
    /// Every FROM factor (each `TableWithJoins`' relation and joins, in
    /// source order): its table name and optional alias.
    pub(crate) factors: Vec<(&'a ObjectName, Option<&'a Ident>)>,
    /// The projection items — the insertable target columns' material.
    pub(crate) projection: &'a [SelectItem],
    /// Every join's operator, the constraint carrier (no relation data — that
    /// lives in `factors`): the binder binds each `ON` as filter reads.
    pub(crate) join_operators: Vec<&'a JoinOperator>,
    /// The WHERE predicate — filter reads over the view's relations.
    pub(crate) selection: Option<&'a Expr>,
}

/// Shape-gate an Oracle inline-view INSERT target
/// (`INSERT INTO (SELECT … FROM …) …`): only the minimal insertable-view shape
/// passes — a projection, a FROM of **plain tables** (a single table, or a
/// join / comma list of them; `ARRAY JOIN` operands are not tables), and an
/// optional WHERE. `None` for everything else: a non-table factor, or any
/// other clause (GROUP BY / HAVING / DISTINCT / ORDER BY / FETCH / CONNECT BY
/// / …) makes the view non-insertable — and could carry column references
/// that would otherwise drop silently. Which table the row lands in is
/// [`insert_target_base`] for the single-table shape, and binder-side column
/// attribution for a join view. Both destructures are exhaustive, so a new
/// `Query` / `Select` clause forces a keep-or-reject decision here.
pub(crate) fn insert_target_view(query: &Query) -> Option<InsertTargetView<'_>> {
    let Query {
        with: None,
        body,
        order_by: None,
        limit_clause: None,
        fetch: None,
        locks,
        for_clause: None,
        settings: None,
        format_clause: None,
        pipe_operators,
    } = query
    else {
        return None;
    };
    if !locks.is_empty() || !pipe_operators.is_empty() {
        return None;
    }
    let SetExpr::Select(select) = body.as_ref() else {
        return None;
    };
    let Select {
        select_token: _,
        optimizer_hints: _,
        distinct: None,
        select_modifiers: None,
        top: None,
        top_before_distinct: _,
        projection,
        exclude: None,
        into: None,
        from,
        lateral_views,
        prewhere: None,
        selection,
        connect_by,
        group_by: GroupByExpr::Expressions(group_by, group_by_modifiers),
        cluster_by,
        distribute_by,
        sort_by,
        having: None,
        named_window,
        qualify: None,
        window_before_qualify: _,
        value_table_mode: None,
        flavor: SelectFlavor::Standard,
    } = select.as_ref()
    else {
        return None;
    };
    if !group_by.is_empty()
        || !group_by_modifiers.is_empty()
        || !cluster_by.is_empty()
        || !distribute_by.is_empty()
        || !sort_by.is_empty()
        || !lateral_views.is_empty()
        || !connect_by.is_empty()
        || !named_window.is_empty()
    {
        return None;
    }
    if from.is_empty() {
        return None;
    }
    fn plain_table(factor: &TableFactor) -> Option<(&ObjectName, Option<&Ident>)> {
        match factor {
            TableFactor::Table {
                name,
                alias,
                args: None,
                ..
            } => Some((name, alias.as_ref().map(|a| &a.name))),
            _ => None,
        }
    }
    let mut factors = Vec::new();
    let mut join_operators = Vec::new();
    for twj in from {
        factors.push(plain_table(&twj.relation)?);
        for join in &twj.joins {
            // An ARRAY JOIN operand parses as a table factor but is an array
            // column, not a relation — never a view over base tables.
            if matches!(
                join.join_operator,
                JoinOperator::ArrayJoin
                    | JoinOperator::LeftArrayJoin
                    | JoinOperator::InnerArrayJoin
            ) {
                return None;
            }
            factors.push(plain_table(&join.relation)?);
            join_operators.push(&join.join_operator);
        }
    }
    Some(InsertTargetView {
        factors,
        projection,
        join_operators,
        selection: selection.as_ref(),
    })
}

/// The single base table of a shape-gated inline view
/// ([`insert_target_view`]), when the FROM is exactly one plain table — the
/// shape-determined case a plain identity parse can resolve. A join view
/// returns `None`: its base table is whichever relation the projection's
/// columns attribute to, which needs the binder (qualifier / catalog
/// resolution).
pub(crate) fn insert_target_base<'a>(view: &InsertTargetView<'a>) -> Option<&'a ObjectName> {
    match view.factors.as_slice() {
        [(name, _)] => Some(name),
        _ => None,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use sqlparser::ast::{SetExpr, Statement};
    use sqlparser::dialect::GenericDialect;
    use sqlparser::parser::Parser;

    /// The first FROM factor of `SELECT 1 FROM <from>` — a handle on a parsed
    /// `TableFactor` (and, for a `Table`, its `ObjectName`) to drive the public
    /// `TryFrom` conversions.
    fn first_table_factor(from: &str) -> TableFactor {
        let sql = format!("SELECT 1 FROM {from}");
        let mut stmts = Parser::parse_sql(&GenericDialect {}, &sql).unwrap();
        let Statement::Query(query) = stmts.remove(0) else {
            panic!("expected a query");
        };
        let SetExpr::Select(select) = *query.body else {
            panic!("expected a SELECT");
        };
        select.from.into_iter().next().unwrap().relation
    }

    #[test]
    fn try_from_object_name_keeps_catalog_schema_name_and_displays_all_parts() {
        let factor = first_table_factor("cat.sch.tbl");
        let TableFactor::Table { name, .. } = &factor else {
            panic!("expected a table factor");
        };
        let reference = TableReference::try_from(name).unwrap();
        assert_eq!(reference.catalog.as_ref().unwrap().value, "cat");
        assert_eq!(reference.schema.as_ref().unwrap().value, "sch");
        assert_eq!(reference.name.value, "tbl");
        // Display renders every present part (the three-part / catalog branch).
        assert_eq!(reference.to_string(), "cat.sch.tbl");
    }

    #[test]
    fn try_from_table_factor_converts_a_table_and_rejects_a_derived_factor() {
        let table = first_table_factor("a.b");
        let reference = TableReference::try_from(&table).unwrap();
        assert_eq!(reference.schema.as_ref().unwrap().value, "a");
        assert_eq!(reference.name.value, "b");
        // A non-`Table` factor names no stored table — an analysis error.
        let derived = first_table_factor("(SELECT 1) AS d");
        assert!(matches!(
            TableReference::try_from(&derived),
            Err(Error::AnalysisError(_))
        ));
    }

    #[test]
    fn try_from_insert_takes_the_target_name() {
        let mut stmts =
            Parser::parse_sql(&GenericDialect {}, "INSERT INTO a.b VALUES (1)").unwrap();
        let Statement::Insert(insert) = stmts.remove(0) else {
            panic!("expected an insert");
        };
        let reference = TableReference::try_from(&insert).unwrap();
        assert_eq!(reference.schema.as_ref().unwrap().value, "a");
        assert_eq!(reference.name.value, "b");
    }

    #[test]
    fn from_insert_with_alias_extracts_the_target_alias_identifier() {
        // `INSERT INTO t AS foo …` (PostgreSQL): the target alias is a
        // `TableAliasWithoutColumns`. The pair keeps the base table (`t`) plus
        // just the alias identifier (`foo`), dropping the `explicit`
        // (`AS`-was-written) flag the analysis doesn't need.
        use sqlparser::dialect::PostgreSqlDialect;
        let mut stmts =
            Parser::parse_sql(&PostgreSqlDialect {}, "INSERT INTO t AS foo (a) VALUES (1)")
                .unwrap();
        let Statement::Insert(insert) = stmts.remove(0) else {
            panic!("expected an insert");
        };
        let (reference, alias) = TableReference::from_insert_with_alias(&insert).unwrap();
        assert_eq!(reference.name.value, "t");
        assert_eq!(alias.unwrap().value, "foo");
    }

    #[test]
    fn try_from_insert_sees_through_an_inline_view_target() {
        // Oracle `INSERT INTO (SELECT … FROM s.t) …` resolves to the view's
        // single base table; a join view has no single base table → error.
        use sqlparser::dialect::OracleDialect;
        let parse = |sql: &str| {
            let mut stmts = Parser::parse_sql(&OracleDialect {}, sql).unwrap();
            let Statement::Insert(insert) = stmts.remove(0) else {
                panic!("expected an insert");
            };
            insert
        };
        let insert = parse("INSERT INTO (SELECT a FROM s.t WHERE a > 0) VALUES (1)");
        let reference = TableReference::try_from(&insert).unwrap();
        assert_eq!(reference.schema.as_ref().unwrap().value, "s");
        assert_eq!(reference.name.value, "t");
        let join = parse("INSERT INTO (SELECT a.id FROM a JOIN b ON a.id = b.id) VALUES (1)");
        assert!(matches!(
            TableReference::try_from(&join),
            Err(Error::AnalysisError(_))
        ));
    }

    #[test]
    fn insert_target_view_rejects_an_array_join_operand() {
        // An ARRAY JOIN operand parses as a table factor but is an array
        // column, not a relation — the gate must reject it so it can't
        // masquerade as a companion table. No current dialect parses both an
        // inline-view INSERT target *and* ARRAY JOIN, so this exercises the
        // gate directly on a ClickHouse-parsed SELECT.
        use sqlparser::dialect::ClickHouseDialect;
        let parse = |sql: &str| {
            let mut stmts = Parser::parse_sql(&ClickHouseDialect {}, sql).unwrap();
            let Statement::Query(query) = stmts.remove(0) else {
                panic!("expected a query");
            };
            query
        };
        let plain = parse("SELECT e.id FROM emp e JOIN dept d ON e.id = d.id");
        let view = insert_target_view(&plain).unwrap();
        // The gate hands everything over as extracted data: both factors
        // (with aliases), the join's operator, and the projection.
        assert_eq!(view.factors.len(), 2);
        assert_eq!(view.factors[1].1.unwrap().value, "d");
        assert_eq!(view.join_operators.len(), 1);
        assert_eq!(view.projection.len(), 1);
        assert!(view.selection.is_none());
        let array_join = parse("SELECT e.id FROM emp e ARRAY JOIN arr");
        assert!(insert_target_view(&array_join).is_none());
    }
}