uqa_sql/semantics/view_rewrite/

validation.rs

//
// Unified Query Algebra
//
// Copyright (c) 2023-2026 Cognica, Inc.
//

//! Public-view and mapped-target validation for automatic DML rewriting.

use super::{
    display_relation, instead_of_trigger_definition, non_writable_column,
    not_automatically_updatable, view_updatability, AutomaticViewLayer, BTreeSet,
    ConflictActionPlan, ConflictPlan, DeletePlan, InsertPlan, MergePlan, MergeWhenPlan,
    ReturningAliases, RowSchema, SQLError, ScalarExpr, TriggerEvent, UpdatePlan, ViewColumn,
    ViewMutationCapabilities, ViewRewriteContext,
};

pub(super) fn layer_column<'a>(
    layer: &'a AutomaticViewLayer,
    name: &str,
) -> Option<&'a ViewColumn> {
    layer.columns.iter().find(|column| column.name == name)
}

fn unknown_view_column(layer: &AutomaticViewLayer, column: &str) -> SQLError {
    SQLError::UnknownColumn(format!("{}.{column}", layer.canonical_name))
}

pub(super) fn duplicate_insert_column(column: &str) -> SQLError {
    SQLError::Routine {
        sqlstate: "42701".into(),
        message: format!("column \"{column}\" specified more than once"),
    }
}

pub(super) fn duplicate_assignment(column: &str) -> SQLError {
    SQLError::Routine {
        sqlstate: "42601".into(),
        message: format!("multiple assignments to same column \"{column}\""),
    }
}

fn validate_view_target_columns<'a>(
    layer: &AutomaticViewLayer,
    targets: impl IntoIterator<Item = &'a crate::ast::AssignmentTarget<ScalarExpr>>,
    duplicate: fn(&str) -> SQLError,
) -> Result<(), SQLError> {
    let mut seen = std::collections::BTreeMap::new();
    for target in targets {
        if layer_column(layer, &target.column).is_none() {
            return Err(unknown_view_column(layer, &target.column));
        }
        let whole = target.is_whole_column();
        if seen
            .insert(&target.column, whole)
            .is_some_and(|previous| previous || whole)
        {
            return Err(duplicate(&target.column));
        }
    }
    Ok(())
}

pub(super) fn validate_mapped_columns(
    targets: &[crate::ast::AssignmentTarget<ScalarExpr>],
    duplicate: fn(&str) -> SQLError,
) -> Result<(), SQLError> {
    let mut seen = std::collections::BTreeMap::new();
    for target in targets {
        let whole = target.is_whole_column();
        if seen
            .insert(&target.column, whole)
            .is_some_and(|previous| previous || whole)
        {
            return Err(duplicate(&target.column));
        }
    }
    Ok(())
}

pub(super) fn validate_insert_targets(
    layer: &AutomaticViewLayer,
    plan: &InsertPlan,
) -> Result<(), SQLError> {
    validate_view_target_columns(layer, plan.columns.iter(), duplicate_insert_column)?;
    let Some(conflict) = &plan.on_conflict else {
        return Ok(());
    };
    for column in &conflict.conflict_columns {
        if layer_column(layer, column).is_none() {
            return Err(unknown_view_column(layer, column));
        }
    }
    if let ConflictActionPlan::Update { assignments, .. } = &conflict.action {
        validate_view_target_columns(
            layer,
            assignments.iter().map(|assignment| &assignment.target),
            duplicate_assignment,
        )?;
    }
    Ok(())
}

pub(super) fn validate_update_targets(
    layer: &AutomaticViewLayer,
    plan: &UpdatePlan,
) -> Result<(), SQLError> {
    validate_view_target_columns(
        layer,
        plan.assignments.iter().map(|assignment| &assignment.target),
        duplicate_assignment,
    )
}

pub(super) fn writable_column(
    layer: &AutomaticViewLayer,
    name: &str,
    operation: &str,
) -> Result<String, SQLError> {
    let column = layer_column(layer, name)
        .ok_or_else(|| SQLError::UnknownColumn(format!("{}.{name}", layer.canonical_name)))?;
    column
        .writable_source_column
        .clone()
        .ok_or_else(|| non_writable_column(&layer.canonical_name, name, operation))
}

fn ambiguous_column(column: &str) -> SQLError {
    SQLError::Routine {
        sqlstate: "42702".into(),
        message: format!("column reference \"{column}\" is ambiguous"),
    }
}

#[derive(Clone, Copy)]
pub(super) struct ExpressionScope<'a> {
    pub(super) target_qualifier: &'a str,
    pub(super) returning_aliases: Option<&'a ReturningAliases>,
    pub(super) source: Option<&'a RowSchema>,
    pub(super) include_excluded: bool,
}

impl ExpressionScope<'_> {
    pub(super) fn row_image_qualifier(self, qualifier: &str) -> bool {
        self.returning_aliases.is_some_and(|aliases| {
            [
                (aliases.old.as_str(), aliases.old_explicit),
                (aliases.new.as_str(), aliases.new_explicit),
            ]
            .into_iter()
            .any(|(alias, explicit)| {
                qualifier == alias
                    && (explicit
                        || !self
                            .source
                            .is_some_and(|source| source.has_qualifier(alias)))
            })
        })
    }

    pub(super) fn target_qualifier(self, qualifier: &str) -> bool {
        qualifier == self.target_qualifier
            || (self.include_excluded && qualifier == "excluded")
            || self.row_image_qualifier(qualifier)
    }
}

pub(super) fn validate_view_expression(
    expression: &ScalarExpr,
    layer: &AutomaticViewLayer,
    scope: ExpressionScope<'_>,
) -> Result<(), SQLError> {
    let mut expression = expression.clone();
    let mut error = None;
    crate::plan::rewrite_scalar_expression(&mut expression, &mut |node| {
        if error.is_some() {
            return;
        }
        match node {
            ScalarExpr::Column(column) => {
                let target = layer_column(layer, column).is_some();
                let source = scope
                    .source
                    .is_some_and(|source| source.has_unqualified_column(column));
                if target && (source || scope.include_excluded) {
                    error = Some(ambiguous_column(column));
                } else if !target && !source {
                    error = Some(SQLError::UnknownColumn(column.clone()));
                }
            }
            ScalarExpr::QualifiedColumn { qualifier, column }
                if scope.target_qualifier(qualifier) && layer_column(layer, column).is_none() =>
            {
                error = Some(SQLError::UnknownColumn(format!("{qualifier}.{column}")));
            }
            _ => {}
        }
    });
    error.map_or(Ok(()), Err)
}

fn public_view_columns(
    services: ViewRewriteContext<'_>,
    view: &str,
) -> Result<BTreeSet<String>, SQLError> {
    let definition = services
        .catalog
        .view_definition(view)?
        .ok_or_else(|| SQLError::UnknownTable(view.to_string()))?;
    let schema = stored_view_schema(services, &definition)?;
    Ok(schema
        .columns()
        .iter()
        .enumerate()
        .map(|(position, column)| schema.public_name(position).unwrap_or(column).to_string())
        .collect())
}

fn validate_public_view_expression(
    expression: &ScalarExpr,
    public_columns: &BTreeSet<String>,
    scope: ExpressionScope<'_>,
) -> Result<(), SQLError> {
    let mut expression = expression.clone();
    let mut error = None;
    crate::plan::rewrite_scalar_expression(&mut expression, &mut |node| {
        if error.is_some() {
            return;
        }
        match node {
            ScalarExpr::Column(column) => {
                let target = public_columns.contains(column);
                let source = scope
                    .source
                    .is_some_and(|source| source.has_unqualified_column(column));
                if target && (source || scope.include_excluded) {
                    error = Some(ambiguous_column(column));
                } else if !target && !source {
                    error = Some(SQLError::UnknownColumn(column.clone()));
                }
            }
            ScalarExpr::QualifiedColumn { qualifier, column }
                if scope.target_qualifier(qualifier) && !public_columns.contains(column) =>
            {
                error = Some(SQLError::UnknownColumn(format!("{qualifier}.{column}")));
            }
            _ => {}
        }
    });
    error.map_or(Ok(()), Err)
}

pub(super) fn validate_public_update_contract(
    services: ViewRewriteContext<'_>,
    plan: &UpdatePlan,
    source: Option<&RowSchema>,
) -> Result<(), SQLError> {
    let columns = public_view_columns(services, &plan.table)?;
    let ordinary_scope = ExpressionScope {
        target_qualifier: &plan.target_qualifier,
        returning_aliases: None,
        source,
        include_excluded: false,
    };
    for assignment in &plan.assignments {
        for expression in assignment.expressions() {
            validate_public_view_expression(expression, &columns, ordinary_scope)?;
        }
    }
    if let Some(predicate) = plan.predicate.as_ref() {
        validate_public_view_expression(predicate, &columns, ordinary_scope)?;
    }
    let returning_scope = ExpressionScope {
        returning_aliases: Some(&plan.returning_aliases),
        ..ordinary_scope
    };
    for projection in &plan.returning {
        validate_public_view_expression(&projection.expr, &columns, returning_scope)?;
    }
    Ok(())
}

pub(super) fn validate_public_delete_contract(
    services: ViewRewriteContext<'_>,
    plan: &DeletePlan,
    source: Option<&RowSchema>,
) -> Result<(), SQLError> {
    let columns = public_view_columns(services, &plan.table)?;
    let ordinary_scope = ExpressionScope {
        target_qualifier: &plan.target_qualifier,
        returning_aliases: None,
        source,
        include_excluded: false,
    };
    if let Some(predicate) = plan.predicate.as_ref() {
        validate_public_view_expression(predicate, &columns, ordinary_scope)?;
    }
    let returning_scope = ExpressionScope {
        returning_aliases: Some(&plan.returning_aliases),
        ..ordinary_scope
    };
    for projection in &plan.returning {
        validate_public_view_expression(&projection.expr, &columns, returning_scope)?;
    }
    Ok(())
}

pub(super) fn validate_public_insert_contract(
    services: ViewRewriteContext<'_>,
    plan: &InsertPlan,
) -> Result<(), SQLError> {
    let columns = public_view_columns(services, &plan.table)?;
    for predicate in plan.on_conflict.iter().flat_map(|conflict| {
        conflict
            .expressions
            .iter()
            .chain(conflict.predicate.iter().map(Box::as_ref))
    }) {
        validate_public_view_expression(
            predicate,
            &columns,
            ExpressionScope {
                target_qualifier: &plan.target_qualifier,
                returning_aliases: None,
                source: None,
                include_excluded: false,
            },
        )?;
    }
    if let Some(ConflictPlan {
        action:
            ConflictActionPlan::Update {
                assignments,
                predicate,
            },
        ..
    }) = &plan.on_conflict
    {
        let scope = ExpressionScope {
            target_qualifier: &plan.target_qualifier,
            returning_aliases: None,
            source: None,
            include_excluded: true,
        };
        for assignment in assignments {
            for expression in assignment.expressions() {
                validate_public_view_expression(expression, &columns, scope)?;
            }
        }
        if let Some(predicate) = predicate {
            validate_public_view_expression(predicate, &columns, scope)?;
        }
    }
    let scope = ExpressionScope {
        target_qualifier: &plan.target_qualifier,
        returning_aliases: Some(&plan.returning_aliases),
        source: None,
        include_excluded: false,
    };
    for projection in &plan.returning {
        validate_public_view_expression(&projection.expr, &columns, scope)?;
    }
    Ok(())
}

pub fn validate_public_merge_contract(
    services: ViewRewriteContext<'_>,
    plan: &MergePlan,
    source: &RowSchema,
) -> Result<(), SQLError> {
    let columns = public_view_columns(services, &plan.target)?;
    let matched_scope = ExpressionScope {
        target_qualifier: &plan.target_qualifier,
        returning_aliases: None,
        source: Some(source),
        include_excluded: false,
    };
    let target_only_scope = ExpressionScope {
        source: None,
        ..matched_scope
    };
    validate_public_view_expression(&plan.join_condition, &columns, matched_scope)?;
    for clause in &plan.when_clauses {
        match clause {
            MergeWhenPlan::UpdateMatched {
                condition,
                assignments,
            } => {
                if let Some(condition) = condition {
                    validate_public_view_expression(condition, &columns, matched_scope)?;
                }
                for assignment in assignments {
                    for expression in assignment.expressions() {
                        validate_public_view_expression(expression, &columns, matched_scope)?;
                    }
                }
            }
            MergeWhenPlan::DeleteMatched { condition }
            | MergeWhenPlan::NothingMatched { condition } => {
                if let Some(condition) = condition {
                    validate_public_view_expression(condition, &columns, matched_scope)?;
                }
            }
            MergeWhenPlan::UpdateNotMatchedBySource {
                condition,
                assignments,
            } => {
                if let Some(condition) = condition {
                    validate_public_view_expression(condition, &columns, target_only_scope)?;
                }
                for assignment in assignments {
                    validate_public_view_expression(
                        &assignment.value,
                        &columns,
                        target_only_scope,
                    )?;
                }
            }
            MergeWhenPlan::DeleteNotMatchedBySource { condition }
            | MergeWhenPlan::NothingNotMatchedBySource { condition } => {
                if let Some(condition) = condition {
                    validate_public_view_expression(condition, &columns, target_only_scope)?;
                }
            }
            MergeWhenPlan::InsertNotMatched { .. } | MergeWhenPlan::NothingNotMatched { .. } => {}
        }
    }
    let returning_scope = ExpressionScope {
        returning_aliases: Some(&plan.returning_aliases),
        ..matched_scope
    };
    for projection in &plan.returning {
        validate_public_view_expression(&projection.expr, &columns, returning_scope)?;
    }
    Ok(())
}

pub(super) fn validate_merge_targets(
    layer: &AutomaticViewLayer,
    plan: &MergePlan,
) -> Result<(), SQLError> {
    for clause in &plan.when_clauses {
        match clause {
            MergeWhenPlan::UpdateMatched { assignments, .. }
            | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
                validate_view_target_columns(
                    layer,
                    assignments.iter().map(|assignment| &assignment.target),
                    duplicate_assignment,
                )?;
            }
            MergeWhenPlan::InsertNotMatched { columns, .. } if !columns.is_empty() => {
                validate_view_target_columns(layer, columns.iter(), duplicate_insert_column)?;
            }
            _ => {}
        }
    }
    Ok(())
}

pub fn validate_public_merge_targets(
    services: ViewRewriteContext<'_>,
    plan: &MergePlan,
) -> Result<(), SQLError> {
    for clause in &plan.when_clauses {
        match clause {
            MergeWhenPlan::UpdateMatched { assignments, .. }
            | MergeWhenPlan::UpdateNotMatchedBySource { assignments, .. } => {
                validate_public_view_targets(
                    services,
                    &plan.target,
                    assignments
                        .iter()
                        .map(|assignment| assignment.target.column.as_str()),
                )?;
                validate_mapped_columns(
                    &assignments
                        .iter()
                        .map(|assignment| assignment.target.clone())
                        .collect::<Vec<_>>(),
                    duplicate_assignment,
                )?;
            }
            MergeWhenPlan::InsertNotMatched { columns, .. } if !columns.is_empty() => {
                validate_public_view_targets(
                    services,
                    &plan.target,
                    columns.iter().map(|target| target.column.as_str()),
                )?;
                validate_mapped_columns(columns, duplicate_insert_column)?;
            }
            _ => {}
        }
    }
    Ok(())
}

pub(super) fn merge_action_capability_error(
    view: &str,
    clauses: &[MergeWhenPlan],
    capabilities: ViewMutationCapabilities,
) -> Option<SQLError> {
    clauses.iter().find_map(|clause| match clause {
        MergeWhenPlan::UpdateMatched { .. } | MergeWhenPlan::UpdateNotMatchedBySource { .. }
            if !capabilities.updatable =>
        {
            Some(not_automatically_updatable(view, "UPDATE"))
        }
        MergeWhenPlan::DeleteMatched { .. } | MergeWhenPlan::DeleteNotMatchedBySource { .. }
            if !capabilities.deletable =>
        {
            Some(not_automatically_updatable(view, "DELETE FROM"))
        }
        MergeWhenPlan::InsertNotMatched { .. } if !capabilities.insertable => {
            Some(not_automatically_updatable(view, "INSERT INTO"))
        }
        _ => None,
    })
}

fn validate_merge_rule_free(
    services: ViewRewriteContext<'_>,
    relation: &str,
) -> Result<(), SQLError> {
    let has_rules = [
        crate::ast::RuleEvent::Insert,
        crate::ast::RuleEvent::Update,
        crate::ast::RuleEvent::Delete,
    ]
    .into_iter()
    .map(|event| services.catalog.rules_for(relation, event))
    .collect::<Result<Vec<_>, SQLError>>()?
    .iter()
    .any(|rules| !rules.is_empty());
    if !has_rules {
        return Ok(());
    }
    Err(SQLError::Routine {
        sqlstate: "0A000".into(),
        message: format!(
            "cannot execute MERGE on relation \"{}\"",
            display_relation(relation)
        ),
    })
}

fn merge_uses_event(plan: &MergePlan, event: TriggerEvent) -> bool {
    plan.when_clauses.iter().any(|clause| match event {
        TriggerEvent::Insert => matches!(clause, MergeWhenPlan::InsertNotMatched { .. }),
        TriggerEvent::Update => matches!(
            clause,
            MergeWhenPlan::UpdateMatched { .. } | MergeWhenPlan::UpdateNotMatchedBySource { .. }
        ),
        TriggerEvent::Delete => matches!(
            clause,
            MergeWhenPlan::DeleteMatched { .. } | MergeWhenPlan::DeleteNotMatchedBySource { .. }
        ),
        TriggerEvent::Truncate => false,
    })
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MergeViewTargetPath {
    AutomaticRewrite,
    ViewTriggers,
}

pub fn merge_view_target_path(
    services: ViewRewriteContext<'_>,
    plan: &MergePlan,
) -> Result<MergeViewTargetPath, SQLError> {
    let canonical = services
        .catalog
        .try_resolve_view_name(&plan.target)
        .map_err(|error| SQLError::Internal(format!("resolve MERGE view: {error}")))?
        .ok_or_else(|| SQLError::UnknownTable(plan.target.clone()))?;
    let definition = services
        .catalog
        .view_definition(&canonical)?
        .ok_or_else(|| SQLError::UnknownTable(plan.target.clone()))?;
    if definition.kind == crate::catalog::view::StoredViewKind::Materialized {
        return Err(SQLError::Routine {
            sqlstate: "0A000".into(),
            message: format!(
                "cannot execute MERGE on relation \"{}\"",
                display_relation(&canonical)
            ),
        });
    }
    validate_merge_rule_free(services, &canonical)?;
    let automatic = view_updatability(services, &canonical)?.automatic;
    let insert_trigger = instead_of_trigger_definition(services, &canonical, TriggerEvent::Insert)?;
    let update_trigger = instead_of_trigger_definition(services, &canonical, TriggerEvent::Update)?;
    let delete_trigger = instead_of_trigger_definition(services, &canonical, TriggerEvent::Delete)?;
    let supported = ViewMutationCapabilities {
        insertable: automatic.insertable || insert_trigger,
        updatable: automatic.updatable || update_trigger,
        deletable: automatic.deletable || delete_trigger,
    };
    if let Some(error) = merge_action_capability_error(&canonical, &plan.when_clauses, supported) {
        return Err(error);
    }
    let mut uses_automatic = false;
    let mut uses_trigger = false;
    let mut has_action = false;
    for (event, trigger) in [
        (TriggerEvent::Insert, insert_trigger),
        (TriggerEvent::Update, update_trigger),
        (TriggerEvent::Delete, delete_trigger),
    ] {
        if !merge_uses_event(plan, event) {
            continue;
        }
        has_action = true;
        uses_trigger |= trigger;
        uses_automatic |= !trigger;
    }
    if uses_trigger && uses_automatic {
        return Err(SQLError::Routine {
            sqlstate: "0A000".into(),
            message: format!(
                "cannot merge into view \"{}\"",
                display_relation(&canonical)
            ),
        });
    }
    if uses_trigger || !has_action {
        Ok(MergeViewTargetPath::ViewTriggers)
    } else {
        Ok(MergeViewTargetPath::AutomaticRewrite)
    }
}

pub(super) fn validate_public_view_targets<'a>(
    services: ViewRewriteContext<'_>,
    view: &str,
    columns: impl IntoIterator<Item = &'a str>,
) -> Result<(), SQLError> {
    let definition = services
        .catalog
        .view_definition(view)?
        .ok_or_else(|| SQLError::UnknownTable(view.to_string()))?;
    let schema = stored_view_schema(services, &definition)?;
    for column in columns {
        if !schema.has_unqualified_column(column) {
            return Err(SQLError::UnknownColumn(format!("{view}.{column}")));
        }
    }
    Ok(())
}

pub(super) fn validate_direct_view_rule_path(
    services: ViewRewriteContext<'_>,
    view: &str,
    event: crate::ast::RuleEvent,
    operation: &str,
) -> Result<(), SQLError> {
    let rules = services.catalog.rules_for(view, event)?;
    let has_conditional_instead = rules
        .iter()
        .any(|rule| rule.definition.instead && rule.definition.condition.is_some());
    let has_unconditional_instead = rules
        .iter()
        .any(|rule| rule.definition.instead && rule.definition.condition.is_none());
    if has_conditional_instead && !has_unconditional_instead {
        return Err(not_automatically_updatable(view, operation));
    }
    Ok(())
}

use super::stored_view_schema;