hawdb-sql 0.13.0

// Copyright 2026 Nowledge
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use super::{
    lower_column_expr, lower_expression, lower_table_name, normalize_ident, object_name_parts,
    ExpressionPosition,
};
use crate::ast::*;
use hawdb_core::{HawDBError, Result};
use sqlparser::ast::Spanned;
use sqlparser::ast::{
    AlterTableOperation, ColumnOption, CreateTableOptions, DataType, Expr, HiveDistributionStyle,
    ReferentialAction, SqlOption, TableConstraint, ValueWithSpan,
};
use sqlparser::tokenizer::{Token, Tokenizer};

/// Custom-type modifiers lose quotes and commas in sqlparser's AST. Check
/// only AST-bound vector type locations against the original tokens.
pub(super) fn validate_vector_type_syntax(
    input: &str,
    statement: &sqlparser::ast::Statement,
) -> Result<()> {
    let columns: Vec<_> = match statement {
        sqlparser::ast::Statement::CreateTable(create) => create.columns.iter().collect(),
        sqlparser::ast::Statement::AlterTable(alter) => alter
            .operations
            .iter()
            .filter_map(|operation| match operation {
                AlterTableOperation::AddColumn { column_def, .. } => Some(column_def),
                _ => None,
            })
            .collect(),
        _ => return Ok(()),
    };
    let mut locations = std::collections::BTreeSet::new();
    for column in columns {
        if let DataType::Custom(name, _) = &column.data_type
            && object_name_parts(name)?.as_slice() == ["vector"]
        {
            locations.insert(name.span().start);
        }
    }
    if locations.is_empty() {
        return Ok(());
    }
    let dialect = sqlparser::dialect::PostgreSqlDialect {};
    let tokens = Tokenizer::new(&dialect, input)
        .tokenize_with_location()
        .map_err(|error| HawDBError::Parse(format!("invalid VECTOR type syntax: {error}")))?;
    let mut tokens = tokens
        .into_iter()
        .filter(|token| !matches!(token.token, Token::Whitespace(_)));
    while let Some(token) = tokens.next() {
        if !locations.remove(&token.span.start) {
            continue;
        }
        if !matches!(tokens.next().map(|t| t.token), Some(Token::LParen))
            || !matches!(tokens.next().map(|t| t.token), Some(Token::Number(value, false)) if !value.is_empty() && value.bytes().all(|byte| byte.is_ascii_digit()))
            || !matches!(tokens.next().map(|t| t.token), Some(Token::RParen))
        {
            return Err(HawDBError::Semantic(
                "VECTOR requires exactly one decimal integer dimension".into(),
            ));
        }
    }
    if !locations.is_empty() {
        return Err(HawDBError::Semantic(
            "VECTOR type location is missing from the source".into(),
        ));
    }
    Ok(())
}

pub(super) fn lower_create_table_statement(
    create: &sqlparser::ast::CreateTable,
) -> Result<SqlStatement> {
    super::reject_unsupported_clauses(
        "CREATE TABLE",
        &[
            ("OR REPLACE", create.or_replace),
            ("TEMPORARY", create.temporary),
            ("EXTERNAL", create.external),
            ("DYNAMIC", create.dynamic),
            (
                if create.global == Some(false) {
                    "LOCAL"
                } else {
                    "GLOBAL"
                },
                create.global.is_some(),
            ),
            ("TRANSIENT", create.transient),
            ("VOLATILE", create.volatile),
            ("ICEBERG", create.iceberg),
            (
                "Hive distribution",
                !matches!(create.hive_distribution, HiveDistributionStyle::NONE),
            ),
            ("Hive format", create.hive_formats.is_some()),
            ("STORED AS", create.file_format.is_some()),
            ("LOCATION", create.location.is_some()),
            ("AS query", create.query.is_some()),
            ("WITHOUT ROWID", create.without_rowid),
            ("LIKE", create.like.is_some()),
            ("CLONE", create.clone.is_some()),
            ("VERSION", create.version.is_some()),
            ("COMMENT", create.comment.is_some()),
            ("ON COMMIT", create.on_commit.is_some()),
            ("ON CLUSTER", create.on_cluster.is_some()),
            ("PRIMARY KEY expression", create.primary_key.is_some()),
            ("ORDER BY", create.order_by.is_some()),
            ("PARTITION BY", create.partition_by.is_some()),
            ("CLUSTER BY", create.cluster_by.is_some()),
            ("CLUSTERED BY", create.clustered_by.is_some()),
            ("INHERITS", create.inherits.is_some()),
            ("PARTITION OF", create.partition_of.is_some()),
            ("FOR VALUES", create.for_values.is_some()),
            ("STRICT", create.strict),
            ("COPY GRANTS", create.copy_grants),
            (
                "ENABLE_SCHEMA_EVOLUTION",
                create.enable_schema_evolution.is_some(),
            ),
            ("CHANGE_TRACKING", create.change_tracking.is_some()),
            (
                "DATA_RETENTION_TIME_IN_DAYS",
                create.data_retention_time_in_days.is_some(),
            ),
            (
                "MAX_DATA_EXTENSION_TIME_IN_DAYS",
                create.max_data_extension_time_in_days.is_some(),
            ),
            (
                "DEFAULT_DDL_COLLATION",
                create.default_ddl_collation.is_some(),
            ),
            (
                "WITH AGGREGATION POLICY",
                create.with_aggregation_policy.is_some(),
            ),
            (
                "WITH ROW ACCESS POLICY",
                create.with_row_access_policy.is_some(),
            ),
            ("WITH TAG", create.with_tags.is_some()),
            ("EXTERNAL_VOLUME", create.external_volume.is_some()),
            ("BASE_LOCATION", create.base_location.is_some()),
            ("CATALOG", create.catalog.is_some()),
            ("CATALOG_SYNC", create.catalog_sync.is_some()),
            (
                "STORAGE_SERIALIZATION_POLICY",
                create.storage_serialization_policy.is_some(),
            ),
            ("TARGET_LAG", create.target_lag.is_some()),
            ("WAREHOUSE", create.warehouse.is_some()),
            ("REFRESH_MODE", create.refresh_mode.is_some()),
            ("INITIALIZE", create.initialize.is_some()),
            ("REQUIRE USER", create.require_user),
            ("UNLOGGED", create.unlogged),
            ("SNAPSHOT", create.snapshot),
            (
                "WITH STORAGE LIFECYCLE POLICY",
                create.with_storage_lifecycle_policy.is_some(),
            ),
            ("WITH CONNECTION", create.with_connection.is_some()),
            ("DISTSTYLE", create.diststyle.is_some()),
            ("DISTKEY", create.distkey.is_some()),
            ("SORTKEY", create.sortkey.is_some()),
            ("BACKUP", create.backup.is_some()),
            ("MULTISET", create.multiset.is_some()),
            ("FALLBACK", create.fallback.is_some()),
            ("WITH DATA", create.with_data.is_some()),
        ],
    )?;
    Ok(SqlStatement::CreateTable(CreateTableStatement {
        table: lower_table_name(&create.name)?,
        if_not_exists: create.if_not_exists,
        columns: create
            .columns
            .iter()
            .map(lower_column_definition)
            .collect::<Result<_>>()?,
        constraints: create
            .constraints
            .iter()
            .map(lower_table_constraint)
            .collect::<Result<_>>()?,
        storage: lower_table_storage(&create.table_options)?,
    }))
}

fn lower_table_storage(options: &CreateTableOptions) -> Result<SqlTableStorage> {
    let options = match options {
        CreateTableOptions::None => return Ok(SqlTableStorage::RowPage),
        CreateTableOptions::With(options) => options,
        _ => {
            return Err(HawDBError::Semantic(
                "PostgreSQL CREATE TABLE storage options require WITH (...)".to_string(),
            ));
        }
    };
    let mut storage_mode = None;
    let mut partition_key = None;
    let mut order_key = None;
    let mut generated_order = None;
    for option in options {
        let SqlOption::KeyValue { key, value } = option else {
            return Err(HawDBError::Semantic(
                "PostgreSQL CREATE TABLE storage options require key = value entries".to_string(),
            ));
        };
        let name = normalize_ident(key);
        match name.as_str() {
            "storage_mode" => set_once(
                &mut storage_mode,
                lower_storage_mode(value)?,
                "storage_mode",
            )?,
            "partition_key" => set_once(
                &mut partition_key,
                lower_storage_key(value, "partition_key")?,
                "partition_key",
            )?,
            "order_key" => set_once(
                &mut order_key,
                lower_storage_key(value, "order_key")?,
                "order_key",
            )?,
            "generated_order" => set_once(
                &mut generated_order,
                lower_generated_order(value)?,
                "generated_order",
            )?,
            _ => {
                return Err(HawDBError::Semantic(format!(
                    "unsupported PostgreSQL CREATE TABLE storage option {name}"
                )));
            }
        }
    }
    match storage_mode.as_deref() {
        Some("strict_append") => Ok(SqlTableStorage::StrictAppend {
            partition_key: partition_key.ok_or_else(|| {
                HawDBError::Semantic("strict_append storage requires partition_key".to_string())
            })?,
            order_key: order_key.ok_or_else(|| {
                HawDBError::Semantic("strict_append storage requires order_key".to_string())
            })?,
            generated_order: generated_order.unwrap_or_default(),
        }),
        Some(mode) => Err(HawDBError::Semantic(format!(
            "unsupported PostgreSQL CREATE TABLE storage_mode {mode}"
        ))),
        None if partition_key.is_none() && order_key.is_none() && generated_order.is_none() => {
            Ok(SqlTableStorage::RowPage)
        }
        None => Err(HawDBError::Semantic(
            "partition_key, order_key, and generated_order require storage_mode = 'strict_append'"
                .to_string(),
        )),
    }
}

fn set_once<T>(slot: &mut Option<T>, value: T, name: &str) -> Result<()> {
    if slot.replace(value).is_some() {
        return Err(HawDBError::Semantic(format!(
            "PostgreSQL CREATE TABLE storage option {name} is specified more than once"
        )));
    }
    Ok(())
}

fn lower_storage_mode(value: &Expr) -> Result<String> {
    let Expr::Value(ValueWithSpan {
        value: sqlparser::ast::Value::SingleQuotedString(value),
        ..
    }) = value
    else {
        return Err(HawDBError::Semantic(
            "CREATE TABLE storage_mode must be a string literal".to_string(),
        ));
    };
    Ok(value.to_ascii_lowercase())
}

fn lower_generated_order(value: &Expr) -> Result<crate::SqlGeneratedOrder> {
    let Expr::Value(ValueWithSpan {
        value: sqlparser::ast::Value::SingleQuotedString(value),
        ..
    }) = value
    else {
        return Err(HawDBError::Semantic(
            "CREATE TABLE generated_order must be a string literal".to_string(),
        ));
    };
    match value.to_ascii_lowercase().as_str() {
        "commit_sequence" => Ok(crate::SqlGeneratedOrder::CommitSequence),
        value => Err(HawDBError::Semantic(format!(
            "unsupported strict_append generated_order {value}"
        ))),
    }
}

fn lower_storage_key(value: &Expr, name: &str) -> Result<Vec<String>> {
    let Expr::Value(ValueWithSpan {
        value: sqlparser::ast::Value::SingleQuotedString(value),
        ..
    }) = value
    else {
        return Err(HawDBError::Semantic(format!(
            "CREATE TABLE {name} must be a string literal"
        )));
    };
    if value.is_empty() {
        return Err(HawDBError::Semantic(format!(
            "CREATE TABLE {name} must not contain an empty column name"
        )));
    }
    Ok(vec![value.clone()])
}

fn lower_column_definition(column: &sqlparser::ast::ColumnDef) -> Result<SqlColumnDefinition> {
    let mut nullable = true;
    let mut default = None;
    let mut primary_key = false;
    let mut unique = false;
    let mut references = None;
    for option in &column.options {
        if option.name.is_some() {
            return Err(HawDBError::Semantic(
                "named column constraints are not supported".to_string(),
            ));
        }
        match &option.option {
            ColumnOption::Null => nullable = true,
            ColumnOption::NotNull => nullable = false,
            ColumnOption::Default(expr) => default = Some(lower_column_default(expr)?),
            ColumnOption::PrimaryKey(_) => {
                primary_key = true;
                nullable = false;
            }
            ColumnOption::Unique(_) => unique = true,
            ColumnOption::ForeignKey(reference) => {
                references = Some(lower_foreign_key_reference(reference)?);
            }
            other => {
                return Err(HawDBError::Semantic(format!(
                    "unsupported PostgreSQL column option {other}"
                )));
            }
        }
    }
    Ok(SqlColumnDefinition {
        name: normalize_ident(&column.name),
        data_type: lower_data_type(&column.data_type)?,
        nullable,
        default,
        primary_key,
        unique,
        references,
    })
}

fn lower_column_default(expr: &Expr) -> Result<SqlColumnDefault> {
    match lower_expression(expr, ExpressionPosition::Scalar)?.kind {
        ExprKind::Value(value) => Ok(SqlColumnDefault::Literal(value)),
        ExprKind::Function {
            name,
            arguments,
            distinct: false,
            filter: None,
        } if name == "uuidv7" && arguments.is_empty() => Ok(SqlColumnDefault::UuidV7),
        ExprKind::Function { name, .. } => Err(HawDBError::Semantic(format!(
            "unsupported PostgreSQL column default function {name}"
        ))),
        ExprKind::Column(_) => Err(HawDBError::Semantic(
            "column defaults cannot reference a column".to_string(),
        )),
        _ => Err(HawDBError::Semantic(
            "unsupported PostgreSQL column default expression".to_owned(),
        )),
    }
}

fn lower_data_type(data_type: &DataType) -> Result<SqlDataType> {
    match data_type {
        DataType::Boolean | DataType::Bool => Ok(SqlDataType::Boolean),
        DataType::BigInt(_) | DataType::Int8(_) => Ok(SqlDataType::BigInt),
        DataType::DoublePrecision | DataType::Float8 => Ok(SqlDataType::DoublePrecision),
        DataType::Text => Ok(SqlDataType::Text),
        DataType::Bytea => Ok(SqlDataType::Bytea),
        DataType::Uuid => Ok(SqlDataType::Uuid),
        DataType::Custom(name, modifiers) if object_name_parts(name)?.as_slice() == ["vector"] => {
            let dimension = match modifiers.as_slice() {
                [dimension] if dimension.bytes().all(|byte| byte.is_ascii_digit()) => dimension
                    .parse::<u32>()
                    .ok()
                    .and_then(hawdb_core::VectorDimension::new),
                _ => None,
            };
            dimension.map(SqlDataType::Vector).ok_or_else(|| {
                HawDBError::Semantic(
                    "VECTOR requires one dimension between 1 and 4294967295".into(),
                )
            })
        }
        _ => Err(HawDBError::Semantic(format!(
            "unsupported PostgreSQL data type {data_type}"
        ))),
    }
}

fn lower_table_constraint(constraint: &TableConstraint) -> Result<SqlTableConstraint> {
    match constraint {
        TableConstraint::PrimaryKey(constraint) => {
            if constraint.name.is_some()
                || constraint.index_name.is_some()
                || constraint.index_type.is_some()
                || !constraint.index_options.is_empty()
                || constraint.characteristics.is_some()
            {
                return Err(HawDBError::Semantic(
                    "unsupported PRIMARY KEY clause".to_string(),
                ));
            }
            Ok(SqlTableConstraint::PrimaryKey(lower_index_columns(
                &constraint.columns,
            )?))
        }
        TableConstraint::Unique(constraint) => {
            if constraint.name.is_some()
                || constraint.index_name.is_some()
                || constraint.index_type.is_some()
                || !constraint.index_options.is_empty()
                || constraint.characteristics.is_some()
                || !matches!(
                    constraint.nulls_distinct,
                    sqlparser::ast::NullsDistinctOption::None
                )
            {
                return Err(HawDBError::Semantic(
                    "unsupported UNIQUE clause".to_string(),
                ));
            }
            Ok(SqlTableConstraint::Unique(lower_index_columns(
                &constraint.columns,
            )?))
        }
        TableConstraint::ForeignKey(constraint) => {
            if constraint.name.is_some()
                || constraint.index_name.is_some()
                || constraint.match_kind.is_some()
                || constraint.characteristics.is_some()
            {
                return Err(HawDBError::Semantic(
                    "unsupported FOREIGN KEY clause".to_string(),
                ));
            }
            Ok(SqlTableConstraint::ForeignKey {
                columns: constraint.columns.iter().map(normalize_ident).collect(),
                reference: lower_foreign_key_reference(constraint)?,
            })
        }
        _ => Err(HawDBError::Semantic(
            "CHECK and inline index constraints are not supported".to_string(),
        )),
    }
}

fn lower_foreign_key_reference(
    reference: &sqlparser::ast::ForeignKeyConstraint,
) -> Result<SqlForeignKeyReference> {
    Ok(SqlForeignKeyReference {
        table: lower_table_name(&reference.foreign_table)?,
        columns: reference
            .referred_columns
            .iter()
            .map(normalize_ident)
            .collect(),
        on_delete: lower_referential_action(reference.on_delete)?,
        on_update: lower_referential_action(reference.on_update)?,
    })
}

fn lower_referential_action(action: Option<ReferentialAction>) -> Result<SqlReferentialAction> {
    match action.unwrap_or(ReferentialAction::NoAction) {
        ReferentialAction::NoAction => Ok(SqlReferentialAction::NoAction),
        ReferentialAction::Restrict => Ok(SqlReferentialAction::Restrict),
        ReferentialAction::Cascade => Ok(SqlReferentialAction::Cascade),
        ReferentialAction::SetNull => Ok(SqlReferentialAction::SetNull),
        ReferentialAction::SetDefault => Err(HawDBError::Semantic(
            "ON DELETE/UPDATE SET DEFAULT is not supported".to_string(),
        )),
    }
}

pub(super) fn lower_create_index_statement(
    create: &sqlparser::ast::CreateIndex,
) -> Result<SqlStatement> {
    if create.using.is_some()
        || create.concurrently
        || !create.include.is_empty()
        || create.nulls_distinct.is_some()
        || !create.with.is_empty()
        || create.predicate.is_some()
        || !create.index_options.is_empty()
        || !create.alter_options.is_empty()
    {
        return Err(HawDBError::Semantic(
            "unsupported PostgreSQL CREATE INDEX clause".to_string(),
        ));
    }
    let Some(name) = &create.name else {
        return Err(HawDBError::Semantic(
            "CREATE INDEX requires an explicit name".to_string(),
        ));
    };
    let name_parts = object_name_parts(name)?;
    let [name] = name_parts.as_slice() else {
        return Err(HawDBError::Semantic(
            "CREATE INDEX names must be unqualified".to_string(),
        ));
    };
    Ok(SqlStatement::CreateIndex(CreateIndexStatement {
        name: name.clone(),
        table: lower_table_name(&create.table_name)?,
        columns: create
            .columns
            .iter()
            .map(lower_index_order_item)
            .collect::<Result<_>>()?,
        unique: create.unique,
        if_not_exists: create.if_not_exists,
    }))
}

fn lower_index_columns(columns: &[sqlparser::ast::IndexColumn]) -> Result<Vec<String>> {
    columns
        .iter()
        .map(|column| {
            let item = lower_index_order_item(column)?;
            if item.direction != SqlOrderDirection::Asc
                || item.nulls != SqlNullOrder::DialectDefault
            {
                return Err(HawDBError::Semantic(
                    "constraint columns do not support ordering options".to_string(),
                ));
            }
            Ok(item.column.name)
        })
        .collect()
}

fn lower_index_order_item(column: &sqlparser::ast::IndexColumn) -> Result<SqlIndexColumn> {
    if column.operator_class.is_some() {
        return Err(HawDBError::Semantic(
            "index operator classes are not supported".to_string(),
        ));
    }
    Ok(SqlIndexColumn {
        column: lower_column_expr(&column.column.expr)?,
        direction: super::lower_order_direction(column.column.options.sort.as_ref())?,
        nulls: match column.column.options.nulls_first {
            Some(true) => SqlNullOrder::First,
            Some(false) => SqlNullOrder::Last,
            None => SqlNullOrder::DialectDefault,
        },
    })
}

pub(super) fn lower_alter_table_statement(
    alter: &sqlparser::ast::AlterTable,
) -> Result<SqlStatement> {
    if alter.if_exists
        || alter.only
        || alter.location.is_some()
        || alter.on_cluster.is_some()
        || alter.table_type.is_some()
    {
        return Err(HawDBError::Semantic(
            "unsupported PostgreSQL ALTER TABLE clause".to_string(),
        ));
    }
    let [AlterTableOperation::AddColumn {
        if_not_exists,
        column_def,
        column_position,
        ..
    }] = alter.operations.as_slice()
    else {
        return Err(HawDBError::Semantic(
            "ALTER TABLE supports one ADD COLUMN operation".to_string(),
        ));
    };
    if column_position.is_some() {
        return Err(HawDBError::Semantic(
            "ALTER TABLE column positioning is not supported".to_string(),
        ));
    }
    Ok(SqlStatement::AlterTableAddColumn(
        AlterTableAddColumnStatement {
            table: lower_table_name(&alter.name)?,
            if_not_exists: *if_not_exists,
            column: lower_column_definition(column_def)?,
        },
    ))
}