reddb-io-server 1.1.2

//! DDL SQL Parser: CREATE TABLE, DROP TABLE, ALTER TABLE

use super::super::ast::{
    AlterOperation, AlterTableQuery, CreateCollectionQuery, CreateColumnDef, CreateTableQuery,
    CreateVectorQuery, DropCollectionQuery, DropDocumentQuery, DropGraphQuery, DropKvQuery,
    DropTableQuery, DropVectorQuery, ExplainAlterQuery, ExplainFormat, PartitionKind,
    PartitionSpec, QueryExpr, TruncateQuery,
};
use super::super::lexer::Token;
use super::error::ParseError;
use super::Parser;
use crate::catalog::{CollectionModel, SubscriptionDescriptor, SubscriptionOperation};
use crate::storage::schema::{SqlTypeName, TypeModifier, Value};

impl<'a> Parser<'a> {
    /// Parse: CREATE TABLE [IF NOT EXISTS] name (col1 TYPE [modifiers], ...)
    pub fn parse_create_table_query(&mut self) -> Result<QueryExpr, ParseError> {
        self.expect(Token::Create)?;
        self.expect(Token::Table)?;

        let if_not_exists = self.match_if_not_exists()?;
        let name = self.expect_ident()?;

        self.expect(Token::LParen)?;
        let mut columns = Vec::new();
        loop {
            let col = self.parse_column_def()?;
            columns.push(col);
            if !self.consume(&Token::Comma)? {
                break;
            }
        }
        self.expect(Token::RParen)?;

        let mut default_ttl_ms = None;
        let mut context_index_fields = Vec::new();
        let mut context_index_enabled = false;
        let mut timestamps = false;
        let mut subscriptions = Vec::new();

        while self.consume(&Token::With)? {
            if self.consume_ident_ci("EVENTS")? {
                subscriptions.push(self.parse_subscription_descriptor(name.clone())?);
            } else if self.consume_ident_ci("CONTEXT_INDEX")? {
                context_index_enabled = self.parse_bool_assign()?;
            } else if self.consume_ident_ci("CONTEXT")? {
                // Consume INDEX token (reserved keyword)
                if !self.consume(&Token::Index)? {
                    return Err(ParseError::expected(
                        vec!["INDEX"],
                        self.peek(),
                        self.position(),
                    ));
                }
                self.expect(Token::On)?;
                self.expect(Token::LParen)?;
                loop {
                    context_index_fields.push(self.expect_ident()?);
                    if !self.consume(&Token::Comma)? {
                        break;
                    }
                }
                self.expect(Token::RParen)?;
                context_index_enabled = true;
            } else if self.consume_ident_ci("TIMESTAMPS")? {
                timestamps = self.parse_bool_assign()?;
            } else {
                default_ttl_ms = self.parse_create_table_ttl_clause()?;
            }
        }

        Ok(QueryExpr::CreateTable(CreateTableQuery {
            collection_model: CollectionModel::Table,
            name,
            columns,
            if_not_exists,
            default_ttl_ms,
            context_index_fields,
            context_index_enabled,
            timestamps,
            partition_by: None,
            tenant_by: None,
            append_only: false,
            subscriptions,
            vault_own_master_key: false,
        }))
    }

    /// Parse: DROP TABLE [IF EXISTS] name
    pub fn parse_drop_table_query(&mut self) -> Result<QueryExpr, ParseError> {
        self.expect(Token::Drop)?;
        self.expect(Token::Table)?;
        self.parse_drop_table_body()
    }

    /// Parse the body of CREATE TABLE after CREATE TABLE has been consumed
    pub fn parse_create_table_body(&mut self) -> Result<QueryExpr, ParseError> {
        let if_not_exists = self.match_if_not_exists()?;
        let name = self.expect_ident()?;

        self.expect(Token::LParen)?;
        let mut columns = Vec::new();
        loop {
            let col = self.parse_column_def()?;
            columns.push(col);
            if !self.consume(&Token::Comma)? {
                break;
            }
        }
        self.expect(Token::RParen)?;

        let mut default_ttl_ms = None;
        let mut context_index_fields = Vec::new();
        let mut context_index_enabled = false;
        let mut timestamps = false;
        let mut tenant_by: Option<String> = None;
        let mut append_only = false;
        let mut subscriptions = Vec::new();

        while self.consume(&Token::With)? {
            if self.consume_ident_ci("EVENTS")? {
                subscriptions.push(self.parse_subscription_descriptor(name.clone())?);
                continue;
            }
            // Accept both spellings:
            //   WITH key = value
            //   WITH (key = value, key = value)
            // Postgres / ClickHouse use the parenthesised form; the
            // bare form is our legacy shorthand. The parenthesised
            // form collects options separated by commas until `)`.
            let has_parens = self.consume(&Token::LParen)?;

            loop {
                if self.consume_ident_ci("CONTEXT_INDEX")? {
                    context_index_enabled = self.parse_bool_assign()?;
                } else if self.consume_ident_ci("CONTEXT")? {
                    if !self.consume(&Token::Index)? {
                        return Err(ParseError::expected(
                            vec!["INDEX"],
                            self.peek(),
                            self.position(),
                        ));
                    }
                    self.expect(Token::On)?;
                    self.expect(Token::LParen)?;
                    loop {
                        context_index_fields.push(self.expect_ident()?);
                        if !self.consume(&Token::Comma)? {
                            break;
                        }
                    }
                    self.expect(Token::RParen)?;
                    context_index_enabled = true;
                } else if self.consume_ident_ci("TIMESTAMPS")? {
                    timestamps = self.parse_bool_assign()?;
                } else if self.consume_ident_ci("APPEND_ONLY")? {
                    append_only = self.parse_bool_assign()?;
                } else if self.consume_ident_ci("TENANT_BY")? {
                    // `WITH (tenant_by = 'col')` form — accepts `=` optional
                    // and expects a string literal column name.
                    let _ = self.consume(&Token::Eq)?;
                    let value = self.parse_literal_value()?;
                    match value {
                        Value::Text(col) => tenant_by = Some(col.to_string()),
                        other => {
                            return Err(ParseError::new(
                                format!("WITH tenant_by expects a text literal, got {other:?}"),
                                self.position(),
                            ));
                        }
                    }
                } else {
                    default_ttl_ms = self.parse_create_table_ttl_clause()?;
                }
                if has_parens {
                    if self.consume(&Token::Comma)? {
                        continue;
                    }
                    self.expect(Token::RParen)?;
                }
                break;
            }
        }

        // Optional `PARTITION BY RANGE|LIST|HASH (col)` clause (Phase 2.2).
        let partition_by = if self.consume(&Token::Partition)? {
            self.expect(Token::By)?;
            let kind = if self.consume(&Token::Range)? {
                PartitionKind::Range
            } else if self.consume(&Token::List)? {
                PartitionKind::List
            } else if self.consume(&Token::Hash)? {
                PartitionKind::Hash
            } else {
                return Err(ParseError::expected(
                    vec!["RANGE", "LIST", "HASH"],
                    self.peek(),
                    self.position(),
                ));
            };
            self.expect(Token::LParen)?;
            let column = self.expect_ident()?;
            self.expect(Token::RParen)?;
            Some(PartitionSpec { kind, column })
        } else {
            None
        };

        // Shorthand: trailing `APPEND ONLY` keyword pair (PG / ClickHouse
        // style). Accepted after partition spec / tenant spec / or on
        // its own. `WITH (append_only = true)` is the other form and
        // handled above.
        if !append_only && self.consume_ident_ci("APPEND")? {
            if !self.consume_ident_ci("ONLY")? {
                return Err(ParseError::expected(
                    vec!["ONLY"],
                    self.peek(),
                    self.position(),
                ));
            }
            append_only = true;
        }

        // Shorthand: `TENANT BY (col)` or `TENANT BY (root.sub.path)`
        // trailing clause (after partition spec if both are used).
        //
        // Dotted paths let non-table models declare tenancy over their
        // natural nested structures — `metadata.tenant` for vectors,
        // `payload.tenant` for queue messages, `tags.cluster` for
        // timeseries, `properties.org` for graphs. The read-path
        // resolver already navigates these paths via
        // `resolve_runtime_document_path`; here we just store the
        // dotted string and let the policy evaluator do the rest.
        if tenant_by.is_none() && self.consume_ident_ci("TENANT")? {
            self.expect(Token::By)?;
            self.expect(Token::LParen)?;
            // Allow keyword-idents (`metadata`, `type`, `data`) as
            // column names — SQL treats them as bare identifiers in
            // this context.
            let mut path = self.expect_ident_or_keyword()?;
            while self.consume(&Token::Dot)? {
                let next = self.expect_ident_or_keyword()?;
                path = format!("{path}.{next}");
            }
            self.expect(Token::RParen)?;
            tenant_by = Some(path);
        }

        Ok(QueryExpr::CreateTable(CreateTableQuery {
            collection_model: CollectionModel::Table,
            name,
            columns,
            if_not_exists,
            default_ttl_ms,
            context_index_fields,
            context_index_enabled,
            timestamps,
            partition_by,
            tenant_by,
            append_only,
            subscriptions,
            vault_own_master_key: false,
        }))
    }

    /// Parse: EXPLAIN ALTER FOR CREATE TABLE name (...) [FORMAT JSON|SQL]
    ///
    /// Pure read: does not execute DDL. Returns a schema-diff rendering of the
    /// difference between the table's current contract and the target CREATE
    /// TABLE body.
    pub fn parse_explain_alter_query(&mut self) -> Result<QueryExpr, ParseError> {
        self.expect(Token::Explain)?;
        self.expect(Token::Alter)?;
        self.expect(Token::For)?;
        self.expect(Token::Create)?;
        self.expect(Token::Table)?;

        let body = self.parse_create_table_body()?;
        let target = match body {
            QueryExpr::CreateTable(t) => t,
            _ => {
                return Err(ParseError::new(
                    "EXPLAIN ALTER FOR CREATE TABLE body must be a CREATE TABLE statement"
                        .to_string(),
                    self.position(),
                ));
            }
        };

        let format = if self.consume(&Token::Format)? {
            if self.consume(&Token::Json)? {
                ExplainFormat::Json
            } else if self.consume_ident_ci("SQL")? {
                ExplainFormat::Sql
            } else {
                return Err(ParseError::expected(
                    vec!["JSON", "SQL"],
                    self.peek(),
                    self.position(),
                ));
            }
        } else {
            ExplainFormat::Sql
        };

        Ok(QueryExpr::ExplainAlter(ExplainAlterQuery {
            target,
            format,
        }))
    }

    /// Parse the body of DROP TABLE after DROP TABLE has been consumed
    pub fn parse_drop_table_body(&mut self) -> Result<QueryExpr, ParseError> {
        let if_exists = self.match_if_exists()?;
        let name = self.parse_drop_collection_name()?;
        Ok(QueryExpr::DropTable(DropTableQuery { name, if_exists }))
    }

    pub fn parse_drop_graph_body(&mut self) -> Result<QueryExpr, ParseError> {
        let if_exists = self.match_if_exists()?;
        let name = self.parse_drop_collection_name()?;
        Ok(QueryExpr::DropGraph(DropGraphQuery { name, if_exists }))
    }

    pub fn parse_drop_vector_body(&mut self) -> Result<QueryExpr, ParseError> {
        let if_exists = self.match_if_exists()?;
        let name = self.parse_drop_collection_name()?;
        Ok(QueryExpr::DropVector(DropVectorQuery { name, if_exists }))
    }

    pub fn parse_drop_document_body(&mut self) -> Result<QueryExpr, ParseError> {
        let if_exists = self.match_if_exists()?;
        let name = self.parse_drop_collection_name()?;
        Ok(QueryExpr::DropDocument(DropDocumentQuery {
            name,
            if_exists,
        }))
    }

    pub fn parse_create_keyed_body(
        &mut self,
        model: CollectionModel,
    ) -> Result<QueryExpr, ParseError> {
        let if_not_exists = self.match_if_not_exists()?;
        let name = self.parse_drop_collection_name()?;
        let vault_own_master_key =
            if model == CollectionModel::Vault && self.consume(&Token::With)? {
                if !self.consume_ident_ci("OWN")? {
                    return Err(ParseError::expected(
                        vec!["OWN"],
                        self.peek(),
                        self.position(),
                    ));
                }
                if !self.consume_ident_ci("MASTER")? {
                    return Err(ParseError::expected(
                        vec!["MASTER"],
                        self.peek(),
                        self.position(),
                    ));
                }
                if !self.consume(&Token::Key)? && !self.consume_ident_ci("KEY")? {
                    return Err(ParseError::expected(
                        vec!["KEY"],
                        self.peek(),
                        self.position(),
                    ));
                }
                true
            } else {
                false
            };
        Ok(QueryExpr::CreateTable(CreateTableQuery {
            collection_model: model,
            name,
            columns: Vec::new(),
            if_not_exists,
            default_ttl_ms: None,
            context_index_fields: Vec::new(),
            context_index_enabled: false,
            timestamps: false,
            partition_by: None,
            tenant_by: None,
            append_only: false,
            subscriptions: Vec::new(),
            vault_own_master_key,
        }))
    }

    pub fn parse_create_collection_model_body(
        &mut self,
        model: CollectionModel,
    ) -> Result<QueryExpr, ParseError> {
        self.parse_create_keyed_body(model)
    }

    pub fn parse_create_collection_body(&mut self) -> Result<QueryExpr, ParseError> {
        let if_not_exists = self.match_if_not_exists()?;
        let name = self.parse_drop_collection_name()?;
        if !self.consume_ident_ci("KIND")? {
            return Err(ParseError::expected(
                vec!["KIND"],
                self.peek(),
                self.position(),
            ));
        }
        let kind = self.expect_ident_or_keyword()?.to_ascii_lowercase();
        Ok(QueryExpr::CreateCollection(CreateCollectionQuery {
            name,
            kind,
            if_not_exists,
        }))
    }

    pub fn parse_create_vector_body(&mut self) -> Result<QueryExpr, ParseError> {
        let if_not_exists = self.match_if_not_exists()?;
        let name = self.parse_drop_collection_name()?;
        if !self.consume_ident_ci("DIM")? {
            return Err(ParseError::expected(
                vec!["DIM"],
                self.peek(),
                self.position(),
            ));
        }
        let dimension = self.parse_integer()?;
        if dimension <= 0 {
            return Err(ParseError::new(
                "VECTOR DIM must be a positive integer".to_string(),
                self.position(),
            ));
        }
        let metric = if self.consume(&Token::Metric)? {
            self.parse_distance_metric()?
        } else {
            crate::storage::engine::distance::DistanceMetric::Cosine
        };
        Ok(QueryExpr::CreateVector(CreateVectorQuery {
            name,
            dimension: dimension as usize,
            metric,
            if_not_exists,
        }))
    }

    pub fn parse_drop_keyed_body(
        &mut self,
        model: CollectionModel,
    ) -> Result<QueryExpr, ParseError> {
        let if_exists = self.match_if_exists()?;
        let name = self.parse_drop_collection_name()?;
        Ok(QueryExpr::DropKv(DropKvQuery {
            name,
            if_exists,
            model,
        }))
    }

    pub fn parse_drop_kv_body(&mut self) -> Result<QueryExpr, ParseError> {
        self.parse_drop_keyed_body(CollectionModel::Kv)
    }

    pub fn parse_drop_collection_body(&mut self) -> Result<QueryExpr, ParseError> {
        let if_exists = self.match_if_exists()?;
        let name = self.parse_drop_collection_name()?;
        Ok(QueryExpr::DropCollection(DropCollectionQuery {
            name,
            if_exists,
        }))
    }

    pub fn parse_truncate_body(
        &mut self,
        model: Option<CollectionModel>,
    ) -> Result<QueryExpr, ParseError> {
        let if_exists = self.match_if_exists()?;
        let name = self.parse_drop_collection_name()?;
        Ok(QueryExpr::Truncate(TruncateQuery {
            name,
            model,
            if_exists,
        }))
    }

    pub(crate) fn parse_drop_collection_name(&mut self) -> Result<String, ParseError> {
        let mut name = self.expect_ident()?;
        while self.consume(&Token::Dot)? {
            if self.consume(&Token::Star)? {
                name.push_str(".*");
                break;
            }
            let next = self.expect_ident_or_keyword()?;
            name = format!("{name}.{next}");
        }
        Ok(name)
    }

    /// Parse: ALTER TABLE name ADD/DROP/RENAME COLUMN ...
    pub fn parse_alter_table_query(&mut self) -> Result<QueryExpr, ParseError> {
        self.expect(Token::Alter)?;
        self.expect(Token::Table)?;
        let name = self.expect_ident()?;

        let mut operations = Vec::new();
        loop {
            let op = self.parse_alter_operation(&name)?;
            operations.push(op);
            if !self.consume(&Token::Comma)? {
                break;
            }
        }

        Ok(QueryExpr::AlterTable(AlterTableQuery { name, operations }))
    }

    /// Parse a single ALTER TABLE operation
    fn parse_alter_operation(&mut self, table_name: &str) -> Result<AlterOperation, ParseError> {
        if self.consume(&Token::Add)? {
            if self.consume_ident_ci("SUBSCRIPTION")? {
                // ADD SUBSCRIPTION name TO queue [REDACT (...)] [WHERE ...]
                let sub_name = self.expect_ident()?;
                let descriptor = self.parse_subscription_descriptor(table_name.to_string())?;
                Ok(AlterOperation::AddSubscription {
                    name: sub_name,
                    descriptor,
                })
            } else {
                // ADD COLUMN definition (COLUMN keyword is optional)
                let _ = self.consume(&Token::Column)?;
                let col_def = self.parse_column_def()?;
                Ok(AlterOperation::AddColumn(col_def))
            }
        } else if self.consume(&Token::Drop)? {
            if self.consume_ident_ci("SUBSCRIPTION")? {
                // DROP SUBSCRIPTION name
                let sub_name = self.expect_ident()?;
                Ok(AlterOperation::DropSubscription { name: sub_name })
            } else {
                // DROP COLUMN name (COLUMN keyword is optional)
                let _ = self.consume(&Token::Column)?;
                let col_name = self.expect_ident()?;
                Ok(AlterOperation::DropColumn(col_name))
            }
        } else if self.consume(&Token::Rename)? {
            // RENAME COLUMN from TO to
            let _ = self.consume(&Token::Column)?; // COLUMN keyword is optional
            let from = self.expect_ident()?;
            self.expect(Token::To)?;
            let to = self.expect_ident()?;
            Ok(AlterOperation::RenameColumn { from, to })
        } else if self.consume(&Token::Attach)? {
            // ATTACH PARTITION child FOR VALUES ...
            self.expect(Token::Partition)?;
            let child = self.expect_ident()?;
            self.expect(Token::For)?;
            // Accept `VALUES` as an ident since the grammar doesn't have it
            // as a reserved keyword everywhere. Collect the remaining tokens
            // as a raw bound string for round-trip persistence.
            if !self.consume_ident_ci("VALUES")? && !self.consume(&Token::Values)? {
                return Err(ParseError::expected(
                    vec!["VALUES"],
                    self.peek(),
                    self.position(),
                ));
            }
            let bound = self.collect_remaining_tokens_as_string()?;
            Ok(AlterOperation::AttachPartition { child, bound })
        } else if self.consume(&Token::Detach)? {
            // DETACH PARTITION child
            self.expect(Token::Partition)?;
            let child = self.expect_ident()?;
            Ok(AlterOperation::DetachPartition { child })
        } else if self.consume(&Token::Enable)? {
            // ENABLE EVENTS | ENABLE ROW LEVEL SECURITY | ENABLE TENANCY ON (col)
            if self.consume_ident_ci("EVENTS")? {
                Ok(AlterOperation::EnableEvents(
                    self.parse_subscription_descriptor(table_name.to_string())?,
                ))
            } else if self.consume_ident_ci("TENANCY")? {
                self.expect(Token::On)?;
                self.expect(Token::LParen)?;
                // Dotted paths allowed (`metadata.tenant`, `payload.org`).
                let mut path = self.expect_ident_or_keyword()?;
                while self.consume(&Token::Dot)? {
                    let next = self.expect_ident_or_keyword()?;
                    path = format!("{path}.{next}");
                }
                self.expect(Token::RParen)?;
                Ok(AlterOperation::EnableTenancy { column: path })
            } else {
                self.expect(Token::Row)?;
                self.expect(Token::Level)?;
                self.expect(Token::Security)?;
                Ok(AlterOperation::EnableRowLevelSecurity)
            }
        } else if self.consume(&Token::Disable)? {
            // DISABLE EVENTS | DISABLE ROW LEVEL SECURITY | DISABLE TENANCY
            if self.consume_ident_ci("EVENTS")? {
                Ok(AlterOperation::DisableEvents)
            } else if self.consume_ident_ci("TENANCY")? {
                Ok(AlterOperation::DisableTenancy)
            } else {
                self.expect(Token::Row)?;
                self.expect(Token::Level)?;
                self.expect(Token::Security)?;
                Ok(AlterOperation::DisableRowLevelSecurity)
            }
        } else if self.consume(&Token::Set)? || self.consume_ident_ci("SET")? {
            // SET APPEND_ONLY = true|false | SET VERSIONED = true|false
            if self.consume_ident_ci("APPEND_ONLY")? {
                let on = self.parse_bool_assign()?;
                Ok(AlterOperation::SetAppendOnly(on))
            } else if self.consume_ident_ci("VERSIONED")? {
                let on = self.parse_bool_assign()?;
                Ok(AlterOperation::SetVersioned(on))
            } else {
                Err(ParseError::expected(
                    vec!["APPEND_ONLY", "VERSIONED"],
                    self.peek(),
                    self.position(),
                ))
            }
        } else {
            Err(ParseError::expected(
                vec![
                    "ADD", "DROP", "RENAME", "ATTACH", "DETACH", "ENABLE", "DISABLE", "SET",
                ],
                self.peek(),
                self.position(),
            ))
        }
    }

    fn parse_subscription_descriptor(
        &mut self,
        source: String,
    ) -> Result<SubscriptionDescriptor, ParseError> {
        let mut ops_filter = Vec::new();
        if self.consume(&Token::LParen)? {
            loop {
                let op = if self.consume(&Token::Insert)? {
                    SubscriptionOperation::Insert
                } else if self.consume(&Token::Update)? {
                    SubscriptionOperation::Update
                } else if self.consume(&Token::Delete)? {
                    SubscriptionOperation::Delete
                } else {
                    return Err(ParseError::expected(
                        vec!["INSERT", "UPDATE", "DELETE"],
                        self.peek(),
                        self.position(),
                    ));
                };
                ops_filter.push(op);
                if !self.consume(&Token::Comma)? {
                    break;
                }
            }
            self.expect(Token::RParen)?;
        }

        let target_queue = if self.consume(&Token::To)? {
            self.expect_ident()?
        } else {
            format!("{source}_events")
        };

        let mut redact_fields = Vec::new();
        if self.consume_ident_ci("REDACT")? {
            self.expect(Token::LParen)?;
            loop {
                redact_fields.push(self.parse_dotted_redact_path()?);
                if !self.consume(&Token::Comma)? {
                    break;
                }
            }
            self.expect(Token::RParen)?;
        }

        let where_filter = if self.consume(&Token::Where)? {
            Some(self.collect_subscription_where_filter()?)
        } else {
            None
        };

        // ON ALL TENANTS: opt-in cluster-wide subscription (requires capability check at execution)
        let all_tenants = if self.consume(&Token::On)? {
            self.expect(Token::All)?;
            if !self.consume_ident_ci("TENANTS")? {
                return Err(ParseError::expected(
                    vec!["TENANTS"],
                    self.peek(),
                    self.position(),
                ));
            }
            true
        } else {
            false
        };

        // REQUIRES CAPABILITY '...' — parsed and discarded; enforcement is at execution time
        if self.consume_ident_ci("REQUIRES")? {
            self.consume_ident_ci("CAPABILITY")?;
            // consume the capability string literal token
            self.advance()?;
        }

        Ok(SubscriptionDescriptor {
            name: String::new(),
            source,
            target_queue,
            ops_filter,
            where_filter,
            redact_fields,
            enabled: true,
            all_tenants,
        })
    }

    /// Parse a dotted redact path: `field`, `obj.field`, `obj.*.field`, etc.
    fn parse_dotted_redact_path(&mut self) -> Result<String, ParseError> {
        let mut parts = Vec::new();
        if self.consume(&Token::Star)? {
            parts.push("*".to_string());
        } else {
            parts.push(self.expect_ident_or_keyword()?);
        }
        while self.consume(&Token::Dot)? {
            if self.consume(&Token::Star)? {
                parts.push("*".to_string());
            } else {
                parts.push(self.expect_ident_or_keyword()?);
            }
        }
        Ok(parts.join("."))
    }

    fn collect_subscription_where_filter(&mut self) -> Result<String, ParseError> {
        let mut parts = Vec::new();
        while !self.check(&Token::Eof) && !self.check(&Token::Comma) {
            parts.push(self.peek().to_string());
            self.advance()?;
        }
        if parts.is_empty() {
            return Err(ParseError::expected(
                vec!["predicate"],
                self.peek(),
                self.position(),
            ));
        }
        Ok(parts.join(" "))
    }

    /// Capture remaining tokens as a display-joined string.
    ///
    /// Used by `ATTACH PARTITION ... FOR VALUES <bound>` to round-trip the
    /// bound clause into storage without needing a dedicated per-kind AST.
    fn collect_remaining_tokens_as_string(&mut self) -> Result<String, ParseError> {
        let mut parts: Vec<String> = Vec::new();
        while !self.check(&Token::Eof) && !self.check(&Token::Comma) {
            parts.push(self.peek().to_string());
            self.advance()?;
        }
        Ok(parts.join(" "))
    }

    /// Parse a single column definition: name TYPE [NOT NULL] [DEFAULT=val] [COMPRESS:N] [UNIQUE] [PRIMARY KEY]
    fn parse_column_def(&mut self) -> Result<CreateColumnDef, ParseError> {
        let name = self.expect_column_ident()?;
        let sql_type = self.parse_column_type()?;
        let data_type = sql_type.to_string();

        let mut def = CreateColumnDef {
            name,
            data_type,
            sql_type: sql_type.clone(),
            not_null: false,
            default: None,
            compress: None,
            unique: false,
            primary_key: false,
            enum_variants: sql_type.enum_variants().unwrap_or_default(),
            array_element: sql_type.array_element_type(),
            decimal_precision: sql_type.decimal_precision(),
        };

        // Parse modifiers in any order
        loop {
            if self.match_not_null()? {
                def.not_null = true;
            } else if self.consume(&Token::Default)? {
                self.expect(Token::Eq)?;
                def.default = Some(self.parse_literal_string_for_ddl()?);
            } else if self.consume(&Token::Compress)? {
                self.expect(Token::Colon)?;
                def.compress = Some(self.parse_integer()? as u8);
            } else if self.consume(&Token::Unique)? {
                def.unique = true;
            } else if self.match_primary_key()? {
                def.primary_key = true;
            } else {
                break;
            }
        }

        Ok(def)
    }

    /// Parse column type: TEXT, INTEGER, EMAIL, ENUM('a','b','c'), ARRAY(TEXT), DECIMAL(2)
    fn parse_column_type(&mut self) -> Result<SqlTypeName, ParseError> {
        let type_name = self.expect_ident_or_keyword()?;
        if self.consume(&Token::LParen)? {
            let inner = self.parse_type_params()?;
            self.expect(Token::RParen)?;
            Ok(SqlTypeName::new(type_name).with_modifiers(inner))
        } else {
            Ok(SqlTypeName::new(type_name))
        }
    }

    /// Parse type parameters inside parentheses: 'a','b' or TEXT or 2
    fn parse_type_params(&mut self) -> Result<Vec<TypeModifier>, ParseError> {
        let mut parts = Vec::new();
        loop {
            match self.peek().clone() {
                Token::String(s) => {
                    let s = s.clone();
                    self.advance()?;
                    parts.push(TypeModifier::StringLiteral(s));
                }
                Token::Integer(n) => {
                    self.advance()?;
                    parts.push(TypeModifier::Number(n as u32));
                }
                _ => {
                    parts.push(TypeModifier::Type(Box::new(self.parse_column_type()?)));
                }
            }
            if !self.consume(&Token::Comma)? {
                break;
            }
        }
        Ok(parts)
    }

    /// Parse a literal string value for DDL DEFAULT expressions
    fn parse_literal_string_for_ddl(&mut self) -> Result<String, ParseError> {
        match self.peek().clone() {
            Token::String(s) => {
                let s = s.clone();
                self.advance()?;
                Ok(s)
            }
            Token::Integer(n) => {
                self.advance()?;
                Ok(n.to_string())
            }
            Token::Float(n) => {
                self.advance()?;
                Ok(n.to_string())
            }
            Token::True => {
                self.advance()?;
                Ok("true".to_string())
            }
            Token::False => {
                self.advance()?;
                Ok("false".to_string())
            }
            Token::Null => {
                self.advance()?;
                Ok("null".to_string())
            }
            ref other => Err(ParseError::expected(
                vec!["string", "number", "true", "false", "null"],
                other,
                self.position(),
            )),
        }
    }

    fn check_ttl_keyword(&self) -> bool {
        matches!(self.peek(), Token::Ident(name) if name.eq_ignore_ascii_case("ttl"))
    }

    /// Parse `= true` / `= false` after a `WITH <option>` keyword.
    /// Used for boolean table options like `WITH TIMESTAMPS = true`.
    fn parse_bool_assign(&mut self) -> Result<bool, ParseError> {
        self.expect(Token::Eq)?;
        match self.peek() {
            Token::True => {
                self.advance()?;
                Ok(true)
            }
            Token::False => {
                self.advance()?;
                Ok(false)
            }
            other => Err(ParseError::expected(
                vec!["true", "false"],
                other,
                self.position(),
            )),
        }
    }

    fn expect_ident_ci_ddl(&mut self, expected: &str) -> Result<(), ParseError> {
        if self.consume_ident_ci(expected)? {
            Ok(())
        } else {
            Err(ParseError::expected(
                vec![expected],
                self.peek(),
                self.position(),
            ))
        }
    }

    fn parse_create_table_ttl_clause(&mut self) -> Result<Option<u64>, ParseError> {
        let option_name = self.expect_ident_or_keyword()?;
        if !option_name.eq_ignore_ascii_case("ttl") {
            return Err(ParseError::new(
                // F-05: `option_name` is caller-controlled identifier text.
                // Render via `{:?}` so embedded CR/LF/NUL/quotes are escaped
                // before the message reaches downstream serialization sinks.
                format!("unsupported CREATE TABLE option {option_name:?}, expected TTL"),
                self.position(),
            ));
        }

        let ttl_value = self.parse_float()?;
        let ttl_unit = match self.peek() {
            Token::Ident(unit) => {
                let unit = unit.clone();
                self.advance()?;
                unit
            }
            _ => "s".to_string(),
        };

        let multiplier_ms = match ttl_unit.to_ascii_lowercase().as_str() {
            "ms" | "msec" | "millisecond" | "milliseconds" => 1.0,
            "s" | "sec" | "secs" | "second" | "seconds" => 1_000.0,
            "m" | "min" | "mins" | "minute" | "minutes" => 60_000.0,
            "h" | "hr" | "hrs" | "hour" | "hours" => 3_600_000.0,
            "d" | "day" | "days" => 86_400_000.0,
            other => {
                return Err(ParseError::new(
                    // F-05: render `other` via `{:?}` so caller-controlled
                    // bytes (CR / LF / NUL / quotes) are escaped before
                    // reaching downstream serialization sinks.
                    format!("unsupported TTL unit {other:?}"),
                    self.position(),
                ));
            }
        };

        if !ttl_value.is_finite() || ttl_value < 0.0 {
            return Err(ParseError::new(
                "TTL must be a finite, non-negative duration".to_string(),
                self.position(),
            ));
        }

        let ttl_ms = ttl_value * multiplier_ms;
        if ttl_ms > u64::MAX as f64 {
            return Err(ParseError::new(
                "TTL duration is too large".to_string(),
                self.position(),
            ));
        }
        if ttl_ms.fract().abs() >= f64::EPSILON {
            return Err(ParseError::new(
                "TTL duration must resolve to a whole number of milliseconds".to_string(),
                self.position(),
            ));
        }

        Ok(Some(ttl_ms as u64))
    }

    /// Try to match IF NOT EXISTS sequence
    pub(crate) fn match_if_not_exists(&mut self) -> Result<bool, ParseError> {
        if self.check(&Token::If) {
            self.advance()?;
            self.expect(Token::Not)?;
            self.expect(Token::Exists)?;
            Ok(true)
        } else {
            Ok(false)
        }
    }

    /// Try to match IF EXISTS sequence
    pub(crate) fn match_if_exists(&mut self) -> Result<bool, ParseError> {
        if self.check(&Token::If) {
            self.advance()?;
            self.expect(Token::Exists)?;
            Ok(true)
        } else {
            Ok(false)
        }
    }

    /// Try to match NOT NULL sequence
    fn match_not_null(&mut self) -> Result<bool, ParseError> {
        if self.check(&Token::Not) {
            // Peek ahead - only consume if followed by NULL
            // We need to be careful: save state and try
            self.advance()?; // consume NOT
            if self.check(&Token::Null) {
                self.advance()?; // consume NULL
                Ok(true)
            } else {
                // This is tricky - NOT was consumed but next isn't NULL.
                // In column modifier context, NOT should only appear before NULL.
                // Return error for clarity.
                Err(ParseError::expected(
                    vec!["NULL (after NOT)"],
                    self.peek(),
                    self.position(),
                ))
            }
        } else {
            Ok(false)
        }
    }

    /// Try to match PRIMARY KEY sequence
    fn match_primary_key(&mut self) -> Result<bool, ParseError> {
        if self.check(&Token::Primary) {
            self.advance()?;
            self.expect(Token::Key)?;
            Ok(true)
        } else {
            Ok(false)
        }
    }
}