1from __future__ import annotations
   2
   3import logging
   4import re
   5import sqlglot
   6from sqlglot import exp
   7import typing as t
   8
   9from sqlmesh.core.engine_adapter.base import (
  10    InsertOverwriteStrategy,
  11    get_source_columns_to_types,
  12)
  13from sqlmesh.core.engine_adapter.mixins import (
  14    ClusteredByMixin,
  15    LogicalMergeMixin,
  16    PandasNativeFetchDFSupportMixin,
  17)
  18from sqlmesh.core.engine_adapter.shared import (
  19    CommentCreationTable,
  20    CommentCreationView,
  21    DataObject,
  22    DataObjectType,
  23    set_catalog,
  24    to_schema,
  25)
  26from sqlmesh.core.node import IntervalUnit
  27from sqlmesh.utils.errors import SQLMeshError
  28
  29if t.TYPE_CHECKING:
  30    from sqlmesh.core._typing import SchemaName, TableName
  31    from sqlmesh.core.engine_adapter._typing import QueryOrDF
  32
  33logger = logging.getLogger(__name__)
  34
  35
  36###############################################################################
  37# Declarative Type System for Property Validation and Normalization
  38###############################################################################
  39"""
  40Declarative type system for property validation and normalization.
  41
  42This module provides a declarative way to define property types with clear separation
  43between validation (type checking) and normalization (type conversion).
  44"""
  45Validated = t.Any  # validated intermediate value (AST nodes, string, list...)
  46Normalized = t.Any  # final normalized output
  47
  48# Allowed outputs for EnumType normalize / or general property outputs.
  49PROPERTY_OUTPUT_TYPES = {
  50    "str",  # "HASH"
  51    "var",  # exp.Var("ASYNC")
  52    "identifier",  # exp.Identifier
  53    "literal",  # exp.Literal.string("HASH")
  54    "column",  # exp.Column(this="HASH")
  55    "ast_expr",  # generic exp.Expr
  56}
  57
  58
  59# ============================================================
  60# Fragment parser (robust-ish)
  61# ============================================================
  62def parse_fragment(text: str) -> t.Union[exp.Expr, t.List[exp.Expr]]:
  63    """
  64    Try to parse a DSL fragment into SQLGlot AST(s).
  65
  66    Behavior:
  67    1. If parse_one succeeds, return the exp.Expr.
  68    2. If fails but text contains comma, split by commas and parse each part.
  69    3. If it's parenthesized like "(a, b)", parse and return exp.Tuple or list.
  70    4. If it's a simple token like "IDENT", return exp.Identifier.
  71    """
  72    if isinstance(text, exp.Expr):
  73        return text
  74
  75    if not isinstance(text, str):
  76        raise TypeError("parse_fragment expects a string")
  77
  78    s = text.strip()
  79    try:
  80        parsed = sqlglot.parse_one(s)
  81        return parsed
  82    except Exception:
  83        raise ValueError(f"Unable to parse fragment: {s}")
  84
  85
  86# ============================================================
  87# Base Type
  88# ============================================================
  89class DeclarativeType:
  90    """
  91    Base class for declarative type system.
  92
  93    Design Philosophy:
  94    -----------------
  95    - validate(value): Type checking only - returns validated intermediate value or None
  96    - normalize(validated): Type conversion only - transforms to target output format
  97
  98    Methods:
  99    --------
 100    validate(value) -> Optional[Validated]
 101        Check if value conforms to this type, maybe include some tiny different types
 102        Returns: Validated intermediate value if valid, None otherwise.
 103
 104    normalize(validated) -> Normalized
 105        Convert validated intermediate value to final output format.
 106        Returns: Normalized value in target format.
 107
 108    __call__(value) -> Normalized
 109        Convenience method: validate + normalize in one step.
 110    """
 111
 112    def validate(self, value: t.Any) -> t.Optional[Validated]:
 113        """Check if value conforms to this type. Return validated value or None.
 114        String that can be parsed as literal
 115        """
 116        raise NotImplementedError(f"{self.__class__.__name__}.validate() must be implemented")
 117
 118    def normalize(self, validated: Validated) -> Normalized:
 119        """Convert validated intermediate value to final output format."""
 120        # Default: identity transformation
 121        return validated
 122
 123    def __call__(self, value: t.Any) -> Normalized:
 124        """Validate and normalize in one step."""
 125        validated = self.validate(value)
 126        if validated is None:
 127            raise ValueError(f"Value {value!r} does not conform to type {self.__class__.__name__}")
 128        return self.normalize(validated)
 129
 130
 131# ============================================================
 132# Primitive Types
 133# ============================================================
 134class StringType(DeclarativeType):
 135    """
 136    String type validator.
 137
 138    Accepts:
 139    - Python str only
 140
 141    Validation: Returns the string if valid, None otherwise.
 142    Normalization: Returns the string as-is (identity).
 143    """
 144
 145    def __init__(self, normalized_type: str = "str"):
 146        """
 147        Args:
 148            normalized_type: Target type for normalization.
 149                - "literal": Convert to exp.Literal.string()
 150                - "str": Keep as string (default)
 151                - "identifier": Convert to exp.Identifier
 152        """
 153        self.normalized_type = normalized_type
 154
 155    def validate(self, value: t.Any) -> t.Optional[str]:
 156        """Check if value is a Python string. Returns string or None."""
 157        return value if isinstance(value, str) else None
 158
 159    def normalize(self, validated: str) -> str:
 160        """Return string as-is (identity normalization)."""
 161        return validated
 162
 163
 164class LiteralType(DeclarativeType):
 165    """
 166    Literal type validator.
 167
 168    Accepts:
 169    - exp.Literal only (from AST)
 170    - String that can be parsed as literal
 171
 172    Validation: Returns exp.Literal if valid, None otherwise.
 173    Normalization: Converts to target type based on normalized_type parameter.
 174    """
 175
 176    def __init__(self, normalized_type: t.Optional[str] = None):
 177        """
 178        Args:
 179            normalized_type: Target type for normalization.
 180                - None: Keep as exp.Literal (default)
 181                - "literal": Keep as exp.Literal
 182                - "str": Convert to Python string
 183        """
 184        self.normalized_type = normalized_type
 185
 186    def validate(self, value: t.Any) -> t.Optional[exp.Literal]:
 187        """Check if value is a literal type. Returns exp.Literal or None."""
 188        # Try parsing string first
 189        if isinstance(value, str):
 190            try:
 191                value = parse_fragment(value)
 192            except Exception:
 193                return None
 194
 195        # Check if it's a Literal
 196        if isinstance(value, exp.Literal):
 197            return value
 198
 199        return None
 200
 201    def normalize(self, validated: exp.Literal) -> t.Union[exp.Literal, str]:
 202        """Convert to target type based on normalized_type."""
 203        if self.normalized_type == "str":
 204            return validated.this
 205        # None or "literal" - keep as-is
 206        return validated
 207
 208
 209class IdentifierType(DeclarativeType):
 210    """
 211    Identifier type validator.
 212
 213    Accepts:
 214    - exp.Identifier only
 215    - String that can be parsed as identifier
 216
 217    Validation: Returns exp.Identifier if valid, None otherwise.
 218    Normalization: Converts to target type based on normalized_type parameter.
 219    """
 220
 221    def __init__(self, normalized_type: t.Optional[str] = None):
 222        """
 223        Args:
 224            normalized_type: Target type for normalization.
 225                - None: Keep as exp.Identifier (default)
 226                - "literal": Convert to exp.Literal.string()
 227                - "str": Convert to Python string
 228                - "identifier": Keep as exp.Identifier
 229                - "column": Convert to exp.Column
 230        """
 231        self.normalized_type = normalized_type
 232
 233    def validate(self, value: t.Any) -> t.Optional[exp.Identifier]:
 234        """Check if value is an identifier type. Returns exp.Identifier or None."""
 235        # Try parsing string first
 236        if isinstance(value, str):
 237            try:
 238                value = parse_fragment(value)
 239            except Exception:
 240                return None
 241
 242        # Check if it's an Identifier
 243        if isinstance(value, exp.Identifier):
 244            return value
 245
 246        return None
 247
 248    def normalize(
 249        self, validated: exp.Identifier
 250    ) -> t.Union[exp.Identifier, exp.Column, exp.Literal, str]:
 251        """Convert to target type based on normalized_type."""
 252        if self.normalized_type == "column":
 253            return exp.column(validated.this)
 254        if self.normalized_type == "literal":
 255            return exp.Literal.string(validated.this)
 256        if self.normalized_type == "str":
 257            return validated.this
 258        # None or "identifier" - keep as-is
 259        return validated
 260
 261
 262class ColumnType(DeclarativeType):
 263    """
 264    Column type validator.
 265
 266    Accepts:
 267    - exp.Column only
 268    - String that can be parsed as column
 269
 270    Validation: Returns exp.Column if valid, None otherwise.
 271    Normalization: Converts to target type based on normalized_type parameter.
 272    """
 273
 274    def __init__(self, normalized_type: t.Optional[str] = None):
 275        """
 276        Args:
 277            normalized_type: Target type for normalization.
 278                - None: Keep as exp.Column (default)
 279                - "literal": Convert to exp.Literal.string()
 280                - "str": Convert to Python string
 281                - "identifier": Convert to exp.Identifier
 282                - "column": Keep as exp.Column
 283        """
 284        self.normalized_type = normalized_type
 285
 286    def validate(self, value: t.Any) -> t.Optional[exp.Column]:
 287        """Check if value is a column type. Returns exp.Column or None."""
 288        # Try parsing string first
 289        if isinstance(value, str):
 290            try:
 291                value = parse_fragment(value)
 292            except Exception:
 293                return None
 294
 295        # Check if it's a Column
 296        if isinstance(value, exp.Column):
 297            return value
 298
 299        return None
 300
 301    def normalize(
 302        self, validated: exp.Column
 303    ) -> t.Union[exp.Column, exp.Identifier, exp.Literal, str]:
 304        """Convert to target type based on normalized_type."""
 305        if self.normalized_type == "identifier":
 306            return exp.Identifier(this=validated.this)
 307        if self.normalized_type == "literal":
 308            return exp.Literal.string(validated.this)
 309        if self.normalized_type == "str":
 310            return str(validated.this)
 311        # None or "column" - keep as-is
 312        return validated
 313
 314
 315class EqType(DeclarativeType):
 316    """
 317    EQ expression type validator (key=value pairs).
 318
 319    Accepts:
 320    - exp.EQ(left, right)
 321    - String that can be parsed as key=value
 322
 323    Validation: Returns (key_name, value_expr) tuple if valid, None otherwise.
 324    Normalization: Returns the (key, value) tuple as-is.
 325    """
 326
 327    def validate(self, value: t.Any) -> t.Optional[t.Tuple[str, t.Any]]:
 328        """Check if value is an EQ expression. Returns (key, value) tuple or None."""
 329        # Try parsing string first
 330        if isinstance(value, str):
 331            try:
 332                value = parse_fragment(value)
 333            except Exception:
 334                return None
 335
 336        # Check if it's an EQ expression
 337        if isinstance(value, exp.EQ):
 338            # Extract key name from left side
 339            left = value.this
 340            # Extract value from right side
 341            right = value.expression
 342
 343            key_name = None
 344            if isinstance(left, exp.Column):
 345                key_name = left.this.name if hasattr(left.this, "name") else str(left.this)
 346            elif isinstance(left, exp.Identifier):
 347                key_name = left.this
 348            elif isinstance(left, str):
 349                key_name = left
 350            else:
 351                key_name = str(left)
 352
 353            return (key_name, right)
 354
 355        return None
 356
 357    def normalize(self, validated: t.Tuple[str, t.Any]) -> t.Tuple[str, t.Any]:
 358        """Return (key, value) tuple as-is (identity normalization)."""
 359        return validated
 360
 361
 362class EnumType(DeclarativeType):
 363    """
 364    Enumerated value type validator.
 365
 366    Accepts values from a predefined set of allowed values.
 367    Following input types are allowed:
 368    - str
 369    - exp.Literal
 370    - exp.Var
 371    - exp.Identifier
 372    - exp.Column
 373
 374    Parameters:
 375    -----------
 376    valid_values : t.Sequence[str]
 377        List of allowed values (e.g., ["HASH", "RANDOM"])
 378    normalized_type : t.Optional[str]
 379        Target type for normalization:
 380        - "str": Python string (default)
 381        - "identifier": exp.Identifier
 382        - "literal": exp.Literal.string()
 383        - "column": exp.Column
 384        - "ast_expr": generic exp.Expr (defaults to Identifier)
 385    case_sensitive : bool
 386        Whether to perform case-sensitive matching (default: False)
 387
 388    Validation: Checks if value is in allowed set, returns canonical string.
 389    Normalization: Converts to specified target type.
 390    """
 391
 392    def __init__(
 393        self,
 394        valid_values: t.Sequence[str],
 395        normalized_type: str = "str",
 396        case_sensitive: bool = False,
 397    ):
 398        self.valid_values = list(valid_values)
 399        self.case_sensitive = bool(case_sensitive)
 400        self.normalized_type = normalized_type
 401
 402        if self.normalized_type is not None and self.normalized_type not in PROPERTY_OUTPUT_TYPES:
 403            raise ValueError(
 404                f"normalized_type must be one of {PROPERTY_OUTPUT_TYPES}, got {self.normalized_type!r}"
 405            )
 406
 407        # Pre-compute normalized values for efficient lookup
 408        self._values_normalized = [v if case_sensitive else v.upper() for v in self.valid_values]
 409
 410    def _extract_text(self, value: t.Any) -> t.Optional[str]:
 411        """Extract text from various value types."""
 412        if isinstance(value, str):
 413            return value
 414        if isinstance(value, (exp.Literal, exp.Var)):
 415            return str(value.this)
 416        if isinstance(value, (exp.Identifier, exp.Column)):
 417            # For Identifier/Column, this might be another Expression
 418            if isinstance(value.this, str):
 419                return value.this
 420            elif hasattr(value.this, "name"):  # noqa: RET505
 421                return str(value.this.name)
 422            else:
 423                return str(value.this)
 424        return None
 425
 426    def _normalize_text(self, text: str) -> str:
 427        """Normalize text for comparison based on case sensitivity."""
 428        return text if self.case_sensitive else text.upper()
 429
 430    def validate(self, value: t.Any) -> t.Optional[str]:
 431        """Check if value is in the allowed enum set. Returns canonical string or None."""
 432        # Try parsing string first
 433        if isinstance(value, str):
 434            try:
 435                parsed = parse_fragment(value)
 436                # If parsed successfully, extract text from AST node
 437                if isinstance(parsed, (exp.Identifier, exp.Literal, exp.Column)):
 438                    value = parsed
 439            except Exception:
 440                # If parsing fails, treat as plain string
 441                pass
 442
 443        # Extract text from value
 444        text = self._extract_text(value)
 445
 446        if text is None:
 447            return None
 448
 449        # Normalize and check against allowed values
 450        normalized_text = self._normalize_text(text)
 451        if normalized_text in self._values_normalized:
 452            return normalized_text
 453
 454        return None
 455
 456    def normalize(self, validated: str) -> Normalized:
 457        """Convert validated enum string to target type."""
 458        # validated is already canonical (e.g., "HASH")
 459        if self.normalized_type is None or self.normalized_type == "str":
 460            return validated
 461        if self.normalized_type == "var":
 462            return exp.Var(this=validated)
 463        if self.normalized_type == "literal":
 464            return exp.Literal.string(validated)
 465        if self.normalized_type == "identifier":
 466            return exp.Identifier(this=validated)
 467        if self.normalized_type == "column":
 468            return exp.Column(this=validated)
 469        if self.normalized_type == "ast_expr":
 470            return exp.Identifier(this=validated)
 471
 472        # Fallback to string
 473        return validated
 474
 475
 476class FuncType(DeclarativeType):
 477    """
 478    Function type validator.
 479
 480    Accepts:
 481    - exp.Func (built-in functions like date_trunc, CAST, etc.)
 482    - exp.Anonymous (custom/dialect functions like RANGE, LIST)
 483    - String that can be parsed as function call
 484
 485    Validation: Returns exp.Func or exp.Anonymous if valid, None otherwise.
 486    Normalization: Returns the function expression as-is (identity).
 487
 488    Examples:
 489        date_trunc('day', col1)     → exp.Func
 490        RANGE(col1, col2)           → exp.Anonymous
 491        LIST(region, status)        → exp.Anonymous
 492    """
 493
 494    def validate(self, value: t.Any) -> t.Optional[t.Union[exp.Func, exp.Anonymous]]:
 495        """Check if value is a function type. Returns exp.Func/exp.Anonymous or None."""
 496        # Try parsing string first
 497        if isinstance(value, str):
 498            try:
 499                value = parse_fragment(value)
 500            except Exception:
 501                return None
 502
 503        # Check if it's a Func or Anonymous function
 504        if isinstance(value, (exp.Func, exp.Anonymous)):
 505            return value
 506
 507        return None
 508
 509    def normalize(
 510        self, validated: t.Union[exp.Func, exp.Anonymous]
 511    ) -> t.Union[exp.Func, exp.Anonymous]:
 512        """Return function expression as-is (identity normalization)."""
 513        return validated
 514
 515
 516# ============================================================
 517# AnyOf (combinator)
 518# ============================================================
 519class AnyOf(DeclarativeType):
 520    """
 521    Union type - accepts first matching subtype.
 522
 523    This is a combinator type that tries each subtype in order and accepts
 524    the first one that validates successfully.
 525
 526    Validation: Tries each subtype, returns (matched_type, validated_value) tuple.
 527    Normalization: Uses the matched subtype's normalize method.
 528    """
 529
 530    def __init__(self, *types: DeclarativeType):
 531        if not types:
 532            raise ValueError("AnyOf requires at least one type")
 533
 534        # Validate all types are DeclarativeType instances
 535        for type_ in types:
 536            if not isinstance(type_, DeclarativeType):
 537                raise TypeError(f"AnyOf expects DeclarativeType instances, got {type_!r}")
 538
 539        self.types: t.List[DeclarativeType] = list(types)
 540
 541    def validate(self, value: t.Any) -> t.Optional[t.Tuple[DeclarativeType, Validated]]:
 542        """Try each subtype in order, return (matched_type, validated_value) or None."""
 543        for sub_type in self.types:
 544            validated = sub_type.validate(value)
 545            if validated is not None:
 546                # Return both the matched type and validated value
 547                return (sub_type, validated)
 548
 549        # No type matched
 550        return None
 551
 552    def normalize(self, validated: t.Tuple[DeclarativeType, Validated]) -> Normalized:
 553        """Normalize using the matched subtype's normalize method."""
 554        matched_type, validated_value = validated
 555        return matched_type.normalize(validated_value)
 556
 557
 558# ============================================================
 559# SequenceOf (Tuple/List/Paren/Single -> normalized list/tuple)
 560# ============================================================
 561class SequenceOf(DeclarativeType):
 562    """
 563    Sequence/List type validator with built-in union type support.
 564
 565    Accepts various sequence representations and validates each element against
 566    one or more possible types (similar to AnyOf for each element).
 567    Optionally accepts single elements (promoted to single-item lists).
 568
 569    Accepts:
 570    - exp.Tuple: (a, b, c)
 571    - exp.Array: [a, b, c]
 572    - exp.Paren: (a) or ((a, b))
 573    - Python list/tuple: [a, b] or (a, b)
 574    - String: "a, b, c" (parsed)
 575    - Single element: a (if allow_single=True, promoted to [a])
 576
 577    Validation: Returns list of (matched_type, validated_value) tuples or None.
 578    Normalization: Returns list of normalized elements using matched type's normalize.
 579
 580    Examples:
 581        # Single type
 582        SequenceOf(ColumnType())
 583
 584        # Multiple types (union) - each element tries types in order
 585        SequenceOf(ColumnType(), IdentifierType(), LiteralType())
 586
 587        # Allow single element
 588        SequenceOf(ColumnType(), allow_single=True)
 589
 590        # Multiple types + allow single
 591        SequenceOf(ColumnType(), IdentifierType(), allow_single=True)
 592    """
 593
 594    def __init__(
 595        self,
 596        *elem_types: DeclarativeType,
 597        allow_single: bool = False,
 598        output_as: str = "list",
 599    ):
 600        """
 601        Args:
 602            *elem_types: One or more type validators for elements.
 603                        If multiple types provided, each element tries types in order (AnyOf behavior).
 604            allow_single: Whether to accept single elements (promoted to list). Default: False.
 605            output_as: Output format - "list" or "tuple". Default: "list".
 606        """
 607        if not elem_types:
 608            raise ValueError("SequenceOf requires at least one element type")
 609
 610        self.elem_types: t.List[DeclarativeType] = list(elem_types)
 611        self.allow_single = allow_single
 612        self.output_as = output_as
 613
 614    def validate(self, value: t.Any) -> t.Optional[t.List[t.Tuple[DeclarativeType, Validated]]]:
 615        """Validate each element in the sequence. Returns list of (matched_type, validated_value) tuples or None."""
 616        # Extract elements from various container types
 617        elems = self._extract_elements(value)
 618        if elems is None:
 619            return None
 620
 621        # Validate each element against all possible types (AnyOf behavior)
 622        validated_items: t.List[t.Tuple[DeclarativeType, Validated]] = []
 623        for elem in elems:
 624            # Try each type until one matches
 625            matched = False
 626            for elem_type in self.elem_types:
 627                validated = elem_type.validate(elem)
 628                if validated is not None:
 629                    validated_items.append((elem_type, validated))
 630                    matched = True
 631                    break
 632
 633            # If no type matched, the whole sequence fails if any element fails
 634            if not matched:
 635                return None
 636
 637        return validated_items
 638
 639    def normalize(
 640        self, validated: t.List[t.Tuple[DeclarativeType, Validated]]
 641    ) -> t.Union[t.List[Normalized], t.Tuple[Normalized, ...]]:
 642        """Normalize each validated element using its matched type's normalize method."""
 643        normalized_items = [elem_type.normalize(value) for elem_type, value in validated]
 644
 645        # Convert to desired output format
 646        if self.output_as == "tuple":
 647            return tuple(normalized_items)
 648        return normalized_items  # default: list
 649
 650    def _extract_elements(self, value: t.Any) -> t.Optional[t.List[t.Any]]:
 651        """
 652        Extract elements from various container representations.
 653        Returns list of raw elements or None if extraction fails.
 654        """
 655        # Python list/tuple - process first before string parsing
 656        if isinstance(value, (list, tuple)):
 657            return list(value)
 658
 659        # Try parsing string for AST types
 660        if isinstance(value, str):
 661            try:
 662                value = parse_fragment(value)
 663            except Exception:
 664                # If parsing fails and we accept single strings, promote to list
 665                if self.allow_single and any(isinstance(t, StringType) for t in self.elem_types):
 666                    return [value]
 667                return None
 668
 669        # SQL Tuple: (a, b, c)
 670        if isinstance(value, exp.Tuple):
 671            return list(value.expressions)
 672
 673        # SQL Array: [a, b, c]
 674        if isinstance(value, exp.Array):
 675            return list(value.expressions)
 676
 677        # SQL Paren: (a) or ((a, b))
 678        if isinstance(value, exp.Paren):
 679            inner = value.this
 680            if isinstance(inner, exp.Tuple):
 681                return list(inner.expressions)
 682            return [inner]
 683
 684        # Single AST element: promote to list (if allow_single)
 685        if self.allow_single and isinstance(value, exp.Expr):
 686            return [value]
 687
 688        return None
 689
 690
 691# ============================================================
 692# Field Definition for Structured Types
 693# ============================================================
 694class Field:
 695    """
 696    Field specification for StructuredTupleType.
 697
 698    Defines validation rules, types, and metadata for a single field.
 699
 700    Args:
 701        type: DeclarativeType instance for validating field value
 702        required: Whether this field is required (default: False)
 703        aliases: List of alternative field names (default: [])
 704        doc: Documentation string for this field
 705
 706    Example:
 707        Field(
 708            type=EnumType(["HASH", "RANDOM"]),
 709            required=True,
 710            aliases=["distribution_type"],
 711            doc="Distribution kind: HASH or RANDOM"
 712        )
 713    """
 714
 715    def __init__(
 716        self,
 717        type: DeclarativeType,
 718        required: bool = False,
 719        aliases: t.Optional[t.List[str]] = None,
 720        doc: t.Optional[str] = None,
 721    ):
 722        self.type = type
 723        self.required = required
 724        self.aliases = aliases or []
 725        self.doc = doc
 726
 727
 728# ============================================================
 729# StructuredTupleType - Base class for typed tuples
 730# ============================================================
 731class StructuredTupleType(DeclarativeType):
 732    """
 733    Base class for validating tuples with typed fields.
 734
 735    Subclasses define FIELDS dict to specify structure:
 736
 737    FIELDS = {
 738        "field_name": Field(
 739            type=SomeType(),
 740            required=True,
 741            aliases=["alt_name1", "alt_name2"]
 742        ),
 743        ...
 744    }
 745
 746    Validation Process:
 747    1. Parse tuple into key=value pairs (exp.EQ)
 748    2. Match keys against FIELDS (including aliases)
 749    3. Validate each field value with specified type
 750    4. Check required fields are present
 751    5. Handle unknown/invalid fields based on error flags
 752
 753    Returns: Dict[str, Any] with canonical field names as keys
 754
 755    Example:
 756        class DistributionTupleInputType(StructuredTupleType):
 757            FIELDS = {
 758                "kind": Field(type=EnumType(["HASH", "RANDOM"]), required=True),
 759                "columns": Field(type=SequenceOf(ColumnType())),
 760            }
 761
 762    Args:
 763        error_on_unknown_field: If True, raise error when encountering unknown fields.
 764                                If False, silently skip unknown fields (default: False)
 765        error_on_invalid_field: If True, raise error when field value validation fails.
 766                                If False, return None for entire validation (default: True)
 767    """
 768
 769    FIELDS: t.Dict[str, Field] = {}  # Subclasses override this
 770
 771    def __init__(self, error_on_unknown_field: bool = True, error_on_invalid_field: bool = True):
 772        self.error_on_unknown_field = error_on_unknown_field
 773        self.error_on_invalid_field = error_on_invalid_field
 774
 775        # Build alias mapping: alias -> canonical_name
 776        self._alias_map: t.Dict[str, str] = {}
 777        for field_name, field_spec in self.FIELDS.items():
 778            # Map canonical name to itself
 779            self._alias_map[field_name] = field_name
 780            # Map aliases to canonical name
 781            for alias in field_spec.aliases:
 782                self._alias_map[alias] = field_name
 783
 784    def validate(
 785        self, value: t.Any
 786    ) -> t.Optional[t.Dict[str, t.Tuple[DeclarativeType, Validated]]]:
 787        """
 788        Validate structured tuple.
 789
 790        Returns: Dict mapping canonical field names to (matched_type, validated_value) tuples,
 791                 or None if validation fails.
 792
 793        Raises:
 794            ValueError: If error_on_unknown_field=True and unknown field encountered
 795            ValueError: If error_on_invalid_field=True and field validation fails
 796        """
 797        # Try parsing string first
 798        if isinstance(value, str):
 799            try:
 800                value = parse_fragment(value)
 801            except Exception:
 802                return None
 803
 804        # Extract key=value pairs from tuple/paren
 805        pairs = self._extract_pairs(value)
 806        if pairs is None:
 807            return None
 808
 809        # Validate each pair and build result dict
 810        result: t.Dict[str, t.Tuple[DeclarativeType, Validated]] = {}
 811        eq_type = EqType()
 812
 813        for pair_expr in pairs:
 814            # Validate as EQ expression
 815            eq_validated = eq_type.validate(pair_expr)
 816            if eq_validated is None:
 817                continue  # Skip non-EQ expressions
 818
 819            key, value_expr = eq_validated
 820
 821            # Resolve alias to canonical name
 822            canonical_name = self._alias_map.get(key)
 823            if canonical_name is None:
 824                # Unknown field
 825                if self.error_on_unknown_field:
 826                    raise ValueError(
 827                        f"Unknown field '{key}' in {self.__class__.__name__}. "
 828                        f"Valid fields: {list(self.FIELDS.keys())}"
 829                    )
 830                # Skip unknown field
 831                continue
 832
 833            # Get field spec
 834            field_spec = self.FIELDS[canonical_name]
 835
 836            # Validate field value with specified type
 837            validated_value = field_spec.type.validate(value_expr)
 838            if validated_value is None:
 839                # Field validation failed
 840                if self.error_on_invalid_field:
 841                    raise ValueError(
 842                        f"Invalid value for field '{canonical_name}': {value_expr}. "
 843                        f"Expected type: {field_spec.type.__class__.__name__}, "
 844                        f"Actual type: {type(value_expr).__name__}"
 845                    )
 846                # Return None for entire validation
 847                return None
 848
 849            # Store with canonical name
 850            result[canonical_name] = (field_spec.type, validated_value)
 851
 852        # Check required fields
 853        for field_name, field_spec in self.FIELDS.items():
 854            if field_spec.required and field_name not in result:
 855                # Required field missing
 856                if self.error_on_invalid_field:
 857                    raise ValueError(
 858                        f"Required field '{field_name}' is missing in {self.__class__.__name__}"
 859                    )
 860                return None
 861
 862        return result
 863
 864    def normalize(
 865        self, validated: t.Dict[str, t.Tuple[DeclarativeType, Validated]]
 866    ) -> t.Dict[str, Normalized]:
 867        """
 868        Normalize validated fields.
 869
 870        Returns: Dict mapping canonical field names to normalized values.
 871        """
 872        return {
 873            field_name: field_type.normalize(value)
 874            for field_name, (field_type, value) in validated.items()
 875        }
 876
 877    def _extract_pairs(self, value: t.Any) -> t.Optional[t.List[t.Any]]:
 878        """
 879        Extract list of expressions from tuple/paren.
 880        Each expression should be an exp.EQ (key=value).
 881        """
 882        # exp.Tuple: (a=1, b=2)
 883        if isinstance(value, list):
 884            return value
 885        if isinstance(value, exp.Tuple):
 886            return list(value.expressions)
 887
 888        # exp.Paren: (a=1) or ((a=1, b=2))
 889        if isinstance(value, exp.Paren):
 890            inner = value.this
 891            if isinstance(inner, exp.Tuple):
 892                return list(inner.expressions)
 893            return [inner]
 894
 895        return None
 896
 897
 898class DistributionTupleInputType(StructuredTupleType):
 899    """
 900    StarRocks distribution tuple validator.
 901
 902    Accepts:
 903    - (kind='HASH', columns=(id, dt), buckets=10)
 904    - (kind='HASH', expressions=(id, dt), bucket_num=10)
 905    - (kind='RANDOM')
 906
 907    Returns: Dict with fields:
 908        - kind: "HASH" or "RANDOM" (string)
 909        - columns: List[exp.Column] (optional, for HASH)
 910        - buckets: exp.Literal (optional)
 911
 912    Field Aliases:
 913        - columns: expressions
 914        - buckets: bucket, bucket_num
 915
 916    Examples:
 917        Input:  (kind='HASH', columns=(id, dt), buckets=10)
 918        Output: {
 919            'kind': 'HASH',
 920            'columns': [exp.Column('id'), exp.Column('dt')],
 921            'buckets': exp.Literal.number(10)
 922        }
 923
 924        Input:  (kind='RANDOM')
 925        Output: {'kind': 'RANDOM'}
 926
 927    Conversion:
 928        Use factory methods to convert normalized values to unified dict format:
 929        - from_enum(): Convert EnumType normalized value (str) → dict
 930        - from_func(): Convert FuncType normalized value (exp.Func) → dict
 931        - to_unified_dict(): Convert any normalized value → dict
 932    """
 933
 934    FIELDS = {
 935        "kind": Field(
 936            type=EnumType(["HASH", "RANDOM"], normalized_type="str"),
 937            required=True,
 938            doc="Distribution type: HASH or RANDOM",
 939        ),
 940        "columns": Field(
 941            type=SequenceOf(
 942                ColumnType(),
 943                IdentifierType(normalized_type="column"),
 944                allow_single=True,
 945            ),
 946            required=False,
 947            aliases=["expressions"],
 948            doc="Columns for HASH distribution",
 949        ),
 950        "buckets": Field(
 951            type=AnyOf(LiteralType(), StringType(normalized_type="literal")),
 952            required=False,
 953            aliases=["bucket", "bucket_num"],
 954            doc="Number of buckets",
 955        ),
 956    }
 957
 958
 959class DistributionTupleOutputType(StructuredTupleType):
 960    """
 961    Output validator for distribution tuple.
 962
 963    Used to validate normalized distribution values which are already dicts.
 964    Overrides validate() to handle dict input directly (for output validation),
 965    while parent class handles tuple/string input (for input validation).
 966    """
 967
 968    FIELDS = {
 969        "kind": Field(
 970            type=EnumType(["HASH", "RANDOM"]),
 971            required=True,
 972        ),
 973        "columns": Field(
 974            type=SequenceOf(ColumnType(), allow_single=False),
 975            required=False,
 976        ),
 977        "buckets": Field(
 978            type=LiteralType(),
 979            required=False,
 980        ),
 981    }
 982
 983    def validate(self, value: t.Any) -> t.Optional[t.Dict[str, t.Any]]:
 984        """
 985        Validate a distribution value for OUTPUT validation.
 986
 987        For output validation, accepts:
 988        - dict: Validate structure directly (normalized output)
 989        - tuple/string: Delegate to parent class (for completeness)
 990
 991        Returns: The dict if valid, None otherwise
 992        """
 993        # For output validation, handle dict directly
 994        if isinstance(value, dict):
 995            # Validate required 'kind' field
 996            kind = value.get("kind")
 997            if kind is None:
 998                return None
 999
1000            # Validate 'kind' is a valid enum value
1001            kind_spec = self.FIELDS["kind"].type
1002            if kind_spec.validate(kind) is None:
1003                return None
1004
1005            # Validate 'columns' if present
1006            columns = value.get("columns")
1007            if columns is not None:
1008                columns_spec = self.FIELDS["columns"].type
1009                if columns_spec.validate(columns) is None:
1010                    return None
1011
1012            # Validate 'buckets' if present
1013            buckets = value.get("buckets")
1014            if buckets is not None:
1015                buckets_spec = self.FIELDS["buckets"].type
1016                if buckets_spec.validate(buckets) is None:
1017                    return None
1018
1019            return value
1020
1021        # For tuple/string, delegate to parent class
1022        return super().validate(value)
1023
1024    # ============================================================
1025    # Factory methods for conversion from other normalized types
1026    # ============================================================
1027
1028    @staticmethod
1029    def from_enum(enum_value: str, buckets: t.Optional[int] = None) -> t.Dict[str, t.Any]:
1030        """
1031        Create distribution dict from EnumType normalized value.
1032
1033        Args:
1034            enum_value: "RANDOM" (from EnumType)
1035            buckets: Optional bucket count
1036
1037        Returns:
1038            Dict with kind/columns/buckets fields
1039
1040        Example:
1041            >>> DistributionTupleOutputType.from_enum("RANDOM")
1042            {'kind': 'RANDOM', 'columns': [], 'buckets': None}
1043        """
1044        return {"kind": enum_value, "columns": [], "buckets": buckets}
1045
1046    @staticmethod
1047    def from_func(
1048        func: t.Union[exp.Func, exp.Anonymous], buckets: t.Optional[int] = None
1049    ) -> t.Dict[str, t.Any]:
1050        """
1051        Create distribution dict from FuncType normalized value.
1052
1053        Args:
1054            func: HASH(id, dt) or RANDOM() (from FuncType)
1055            buckets: Optional bucket count
1056
1057        Returns:
1058            Dict with kind/columns/buckets fields
1059
1060        Example:
1061            >> func = sqlglot.parse_one("HASH(id, dt)")
1062            >> DistributionTupleOutputType.from_func(func)
1063            {"kind": "HASH", "columns": [exp.Column("id"), exp.Column("dt")], "buckets": None}
1064        """
1065        func_name = func.name.upper() if hasattr(func, "name") else str(func.this).upper()
1066
1067        if func_name == "HASH":
1068            # Extract columns from HASH(col1, col2, ...)
1069            columns: list[exp.Column] = [func.this] if isinstance(func.this, exp.Column) else []
1070            columns.extend(func.expressions)
1071            return {"kind": "HASH", "columns": columns, "buckets": buckets}
1072        elif func_name == "RANDOM":  # noqa: RET505
1073            return {"kind": "RANDOM", "columns": [], "buckets": buckets}
1074        else:
1075            raise ValueError(f"Unknown distribution function: {func_name}")
1076
1077    @staticmethod
1078    def to_unified_dict(
1079        normalized_value: t.Any, buckets: t.Optional[int] = None
1080    ) -> t.Dict[str, t.Any]:
1081        """
1082        Convert any normalized distribution value to unified dict format.
1083
1084        This is a convenience method that dispatches to appropriate factory method.
1085
1086        Args:
1087            normalized_value: Result from DistributedByInputSpec normalization
1088                             (dict | str | exp.Func)
1089            buckets: Optional bucket count override
1090
1091        Returns:
1092            Unified dict with kind/columns/buckets fields
1093
1094        Raises:
1095            TypeError: If value type is not supported
1096
1097        Example:
1098            >>> # From DistributionTupleOutputType
1099            >>> DistributionTupleOutputType.to_unified_dict({"kind": "HASH", "columns": [...]})
1100            {'kind': 'HASH', 'columns': [Ellipsis]}
1101
1102            >>> # From EnumType
1103            >>> DistributionTupleOutputType.to_unified_dict("RANDOM")
1104            {'kind': 'RANDOM', 'columns': [], 'buckets': None}
1105
1106            >> # From FuncType
1107            >> DistributionTupleOutputType.to_unified_dict(sqlglot.parse_one("HASH(id)"))
1108            {'kind': 'HASH', 'columns': [exp.Column('id')], 'buckets': None}
1109        """
1110        if isinstance(normalized_value, dict):
1111            # Already in DistributionTupleInputType format
1112            return normalized_value
1113        elif isinstance(normalized_value, str):  # noqa: RET505
1114            # From EnumType: "RANDOM"
1115            return DistributionTupleOutputType.from_enum(normalized_value, buckets)
1116        elif isinstance(normalized_value, (exp.Func, exp.Anonymous)):
1117            # From FuncType: HASH(id, dt)
1118            return DistributionTupleOutputType.from_func(normalized_value, buckets)
1119        else:
1120            raise TypeError(
1121                f"Cannot convert {type(normalized_value).__name__} to distribution dict. "
1122                f"Expected dict, str, or exp.Func/exp.Anonymous."
1123            )
1124
1125
1126# ============================================================
1127# Type Specifications for StarRocks Properties (INPUT and OUTPUT)
1128# ============================================================
1129class PropertySpecs:
1130    # Accepts:
1131    # - Single column: id
1132    # - Multiple columns: (id, dt)
1133    # - String for string input: "id, dt" (will be auto-wrapped and parsed by preprocess_parentheses)
1134    GeneralColumnListInputSpec = SequenceOf(
1135        ColumnType(),
1136        StringType(normalized_type="column"),
1137        IdentifierType(normalized_type="column"),
1138        allow_single=True,
1139    )
1140
1141    # TableKey: Simple key specification (primary_key, duplicate_key, unique_key, aggregate_key)
1142    # Accepts:
1143    # - Single column: id
1144    # - Multiple columns: (id, dt)
1145    TableKeyInputSpec = GeneralColumnListInputSpec
1146
1147    # Partitioned By: Flexible partition specification
1148    # Accepts:
1149    # - Single column: col1
1150    # - Multiple columns: (col1, col2)
1151    # - Mixed: (col1, "col2") - string will be parsed
1152    # - RANGE(col1) or RANGE(col1, col2)
1153    # - LIST(col1) or LIST(col1, col2)
1154    # - Expression: (date_trunc('day', col1), col2)
1155    PartitionedByInputSpec = SequenceOf(
1156        ColumnType(),
1157        StringType(normalized_type="column"),
1158        IdentifierType(normalized_type="column"),
1159        FuncType(),  # RANGE(), LIST(), date_trunc(), etc.
1160        allow_single=True,
1161    )
1162
1163    # Partitions: List of partition definitions (strings)
1164    # Accepts:
1165    # - Single partition: 'PARTITION p1 VALUES LESS THAN ("2024-01-01")'
1166    # - Multiple partitions: ('PARTITION p1 ...', 'PARTITION p2 ...')
1167    # Note: Single string is auto-promoted to list
1168    PartitionsInputSpec = SequenceOf(
1169        StringType(), LiteralType(normalized_type="str"), allow_single=True
1170    )
1171
1172    # Distribution: StarRocks distribution specification
1173    # Accepts:
1174    # - Structured tuple1: (kind='HASH', columns=(id, dt), buckets=10)
1175    # - Structured tuple2: (kind='RANDOM')
1176    # - String format: "HASH(id)", "RANDOM", or "(kind='HASH', columns=(id), buckets=10)"
1177    # Note: Does NOT accept simple columns like id or (id, dt)
1178    #    And it can't directly accept "HASH(id) BUCKETS 10", you need to split it with "BUCKETS" to two parts.
1179    DistributedByInputSpec = AnyOf(
1180        DistributionTupleInputType(),  # Try structured tuple first (most specific)
1181        EnumType(["RANDOM"], normalized_type="str"),  # "RANDOM"
1182        FuncType(),  # "HASH(id)",
1183    )
1184
1185    # OrderBy: Simple ordering specification
1186    # Accepts:
1187    # - Single column: dt
1188    # - Multiple columns: (dt, id, status)
1189    OrderByInputSpec = GeneralColumnListInputSpec
1190
1191    # Refresh scheme: Accepts various types, normalizes to string
1192    # For properties like refresh_scheme, it can be a string, identifier, or column
1193    RefreshSchemeInputSpec = AnyOf(
1194        EnumType(["ASYNC", "MANUAL"], normalized_type="var"),
1195        ColumnType(normalized_type="str"),  # Columns → will be converted to string
1196        IdentifierType(normalized_type="str"),  # Identifiers → will be converted to string
1197        LiteralType(normalized_type="str"),  # Numbers and string → to string
1198        StringType(),  # Plain strings
1199    )
1200
1201    # Generic property value: Accepts various types, normalizes to string
1202    # For properties like replication_num, storage_medium, etc.
1203    # StarRocks PROPERTIES syntax requires all values to be strings: "value"
1204    # So we normalize everything to string for consistent SQL generation
1205    GenericPropertyInputSpec = AnyOf(
1206        StringType(),  # Plain strings
1207        LiteralType(normalized_type="str"),  # Numbers and string → will be converted to string
1208        IdentifierType(normalized_type="str"),  # Identifiers → will be converted to string
1209        ColumnType(normalized_type="str"),  # Columns → will be converted to string
1210    )
1211
1212    """
1213    Input Property Specification for StarRocks
1214
1215    This specification defines the validation and normalization rules for StarRocks properties.
1216    Properties are specified in the physical_properties block of a SQLMesh model.
1217
1218    Supported properties:
1219    - partitioned_by / partition_by: Partition specification
1220    - partitions: List of partition definitions
1221    - distributed_by: Distribution specification (HASH/RANDOM with structured tuple or string)
1222    - order_by: Ordering specification (simple column list)
1223    - table key:
1224        - primary_key: Primary key columns
1225        - duplicate_key: Duplicate key columns (for DUPLICATE KEY table)
1226        - unique_key: Unique key columns (for UNIQUE KEY table)
1227        - aggregate_key: Aggregate key columns (for AGGREGATE KEY table)
1228    - other properties: Any other properties not listed above will be treated as generic
1229                        string properties (e.g., replication_num, storage_medium, etc.)
1230
1231    Examples:
1232        duplicate_key = dt                             # Single key
1233        primary_key = (id, customer_id)                # Multiple keys
1234
1235        partitioned_by = col1                          # Single column
1236        partitioned_by = (col1, col2)                  # Multiple columns
1237        partitioned_by = (col1, "col2")                # Mixed (string will be parsed)
1238        partitioned_by = date_trunc('day', col1)       # Expression partition with single func
1239        partitioned_by = (date_trunc('day', col1), col2)  # Expression partition with multiple exprs
1240        partitioned_by = RANGE(col1, col2)             # RANGE partition
1241        partitioned_by = LIST(region, status)          # LIST partition
1242
1243        distributed_by = (kind='HASH', columns=(id, dt), buckets=10)  # Structured
1244        distributed_by = (kind='RANDOM')               # RANDOM distribution
1245        distributed_by = "HASH(id)"                    # String format
1246        distributed_by = "RANDOM"                      # String format
1247
1248        order_by = dt                                  # Single column
1249        order_by = (dt, id, status)                    # Multiple columns
1250
1251        replication_num = 3                            # Generic property (auto-handled)
1252        storage_medium = "SSD"                         # Generic property (auto-handled)
1253    """
1254    PROPERTY_INPUT_SPECS: t.Dict[str, DeclarativeType] = {
1255        # Table key properties
1256        "primary_key": TableKeyInputSpec,
1257        "duplicate_key": TableKeyInputSpec,
1258        "unique_key": TableKeyInputSpec,
1259        "aggregate_key": TableKeyInputSpec,
1260        # Partition-related properties
1261        "partitioned_by": PartitionedByInputSpec,
1262        "partitions": PartitionsInputSpec,
1263        # Distribution property
1264        "distributed_by": DistributedByInputSpec,
1265        # Ordering property
1266        "clustered_by": OrderByInputSpec,
1267        # View properties
1268        # StarRocks syntax: SECURITY {NONE | INVOKER | DEFINER}
1269        "security": EnumType(["NONE", "INVOKER", "DEFINER"], normalized_type="str"),
1270        # Materialized view refresh properties (StarRocks uses REFRESH ...)
1271        # - refresh_moment: IMMEDIATE | DEFERRED
1272        "refresh_moment": EnumType(["IMMEDIATE", "DEFERRED"], normalized_type="str"),
1273        # - refresh_scheme: ASYNC | ASYNC [START (...) EVERY (INTERVAL ...)] | MANUAL
1274        #     it should be a string/literal if START/EVERY is present, other than ASYNC
1275        "refresh_scheme": RefreshSchemeInputSpec,
1276        # Note: All other properties not listed here will be handled, an example here
1277        "replication_num": GenericPropertyInputSpec,
1278    }
1279
1280    # Default output spec for properties not in PROPERTY_OUTPUT_SPECS
1281    GenericPropertyOutputSpec = StringType()
1282
1283    """
1284    Output Property Specification for StarRocks after validation+normalization
1285
1286    This specification describes the expected types after normalization.
1287    For most properties, OUTPUT spec is the same as INPUT spec since normalization
1288    preserves the diverse types (dict | str | exp.Func for distribution).
1289
1290    Conversion to unified formats (e.g., all distributions → dict) happens separately
1291    in the usage layer via factory methods like DistributionTupleInputType.to_unified_dict().
1292
1293    Expected Output Types (after normalization):
1294    - table keys: List[exp.Expr] - columns
1295    - partitioned_by: List[exp.Expr] - columns, functions
1296    - partitions: List[str] - partition definition strings
1297    - distributed_by: Dict | str | exp.Func - DistributionTupleInputType, EnumType, or FuncType output
1298    - order_by: List[exp.Expr] - columns
1299    - generic properties: str - normalized string values
1300    """
1301    GeneralColumnListOutputSpec: DeclarativeType = SequenceOf(ColumnType(), allow_single=False)
1302
1303    PROPERTY_OUTPUT_SPECS: t.Dict[str, DeclarativeType] = {
1304        "primary_key": GeneralColumnListOutputSpec,
1305        "duplicate_key": GeneralColumnListOutputSpec,
1306        "unique_key": GeneralColumnListOutputSpec,
1307        "aggregate_key": GeneralColumnListOutputSpec,
1308        "partitioned_by": SequenceOf(ColumnType(), FuncType(), allow_single=False),
1309        "partitions": SequenceOf(StringType(), allow_single=False),
1310        "distributed_by": AnyOf(
1311            DistributionTupleOutputType(),  # Try structured tuple first (most specific)
1312            EnumType(["RANDOM"], normalized_type="str"),  # "RANDOM"
1313            FuncType(),  # "HASH(id)",
1314        ),  # Still dict | str | exp.Func after normalize
1315        "clustered_by": GeneralColumnListOutputSpec,
1316        "security": EnumType(["NONE", "INVOKER", "DEFINER"], normalized_type="str"),
1317        "refresh_moment": EnumType(["IMMEDIATE", "DEFERRED"], normalized_type="str"),
1318        "refresh_scheme": AnyOf(
1319            EnumType(["ASYNC", "MANUAL"], normalized_type="var"),
1320            StringType(),
1321        ),
1322        # Generic properties use GenericPropertyOutputSpec, an example here
1323        "replication_num": GenericPropertyOutputSpec,
1324    }
1325
1326    # ============================================================
1327    # Helper functions
1328    # ============================================================
1329
1330    @staticmethod
1331    def get_property_input_spec(property_name: str) -> DeclarativeType:
1332        """
1333        Get the INPUT type validator for a property.
1334
1335        Returns the specific type from PROPERTY_INPUT_SPECS if defined,
1336        otherwise returns GenericPropertyInputSpec for unknown properties.
1337
1338        This allows any property not explicitly defined to be treated
1339        as a generic string property.
1340        """
1341        return PropertySpecs.PROPERTY_INPUT_SPECS.get(
1342            property_name, PropertySpecs.GenericPropertyInputSpec
1343        )
1344
1345    @staticmethod
1346    def get_property_output_spec(property_name: str) -> DeclarativeType:
1347        """
1348        Get the OUTPUT type validator for a property.
1349
1350        Returns the specific type from PROPERTY_OUTPUT_SPECS if defined,
1351        otherwise returns GenericPropertyOutputSpec for unknown properties.
1352
1353        This allows validating that normalized values conform to expected output types.
1354        """
1355        return PropertySpecs.PROPERTY_OUTPUT_SPECS.get(
1356            property_name, PropertySpecs.GenericPropertyOutputSpec
1357        )
1358
1359
1360# ============================================================
1361# Property Validation Helpers
1362# ============================================================
1363class PropertyValidator:
1364    """
1365    Centralized property validation helpers for table properties.
1366
1367    Provides reusable validation functions to avoid code duplication
1368    and ensure consistent error messages across different property handlers.
1369    """
1370
1371    TABLE_KEY_TYPES = {"primary_key", "duplicate_key", "unique_key", "aggregate_key"}
1372
1373    # All important properties except generic properties
1374    IMPORTANT_PROPERTY_NAMES = {
1375        *TABLE_KEY_TYPES,
1376        "partitioned_by",
1377        "partitions",
1378        "distributed_by",
1379        "clustered_by",
1380    }
1381
1382    # Centralized property alias configuration
1383    # Maps canonical name -> list of valid aliases
1384    PROPERTY_ALIASES: t.Dict[str, t.Set[str]] = {
1385        "partitioned_by": {"partition_by"},
1386        "clustered_by": {"order_by"},
1387    }
1388
1389    EXCLUSIVE_PROPERTY_NAME_MAP: t.Dict[str, t.Set[str]] = {
1390        "key_type": set(TABLE_KEY_TYPES),
1391        **PROPERTY_ALIASES,
1392    }
1393
1394    # Centralized invalid property name configuration
1395    # Maps canonical name -> list of invalid/deprecated names
1396    INVALID_PROPERTY_NAME_MAP: t.Dict[str, t.List[str]] = {
1397        "partitioned_by": ["partition"],
1398        "distributed_by": ["distribution", "distribute"],
1399        "clustered_by": ["order", "ordering"],
1400    }
1401
1402    @staticmethod
1403    def ensure_parenthesized(value: t.Any) -> t.Any:
1404        """
1405        Ensure string value is wrapped in parentheses for parse_fragment compatibility.
1406
1407        For string inputs like 'id1, id2', wraps to '(id1, id2)' so that
1408        parse_fragment can parse it correctly.
1409
1410        Args:
1411            value: Input value (string, expression, or other)
1412
1413        Returns:
1414            - For strings/Literal/Column(quoted): wrapped in parentheses if not already
1415            - For other types: returned unchanged
1416
1417        Example:
1418            >>> PropertyValidator.ensure_parenthesized('id1, id2')
1419            '(id1, id2)'
1420            >>> PropertyValidator.ensure_parenthesized('(id1, id2)')
1421            '(id1, id2)'
1422            >>> PropertyValidator.ensure_parenthesized(exp.Literal.string('id1, id2'))
1423            '(id1, id2)'
1424            >>> PropertyValidator.ensure_parenthesized(exp.Column(quoted=True, name='id1, id2'))
1425            Column(quoted=True, name=id1, id2)
1426        """
1427        # logger.debug("ensure_parenthesized. value: %s, type: %s", value, type(value))
1428
1429        # Extract string content from Literal
1430        if isinstance(value, exp.Literal) and value.is_string:
1431            value = value.this
1432        # Extract string content from Column (quoted)
1433        elif isinstance(value, exp.Column) and hasattr(value.this, "quoted") and value.this.quoted:
1434            value = value.name  # Column.name returns the string
1435        elif not isinstance(value, str):
1436            return value
1437
1438        stripped = value.strip()
1439        if not stripped:
1440            return value
1441
1442        # Check if already wrapped in parentheses
1443        if stripped.startswith("(") and stripped.endswith(")"):
1444            return value
1445
1446        return f"({stripped})"
1447
1448    @staticmethod
1449    def validate_and_normalize_property(
1450        property_name: str, value: t.Any, preprocess_parentheses: bool = False
1451    ) -> t.Any:
1452        """
1453        Complete property processing pipeline using SPEC:
1454        1. Optionally preprocess string with parentheses
1455        2. Get INPUT type validator
1456        3. Validate and normalize input value
1457        4. Get OUTPUT type validator
1458        5. Verify normalized output conforms to expected type
1459        6. Return verified output
1460
1461        After validation, the output type is guaranteed by SPEC.
1462        Unexpected types indicate SPEC configuration errors.
1463
1464        Args:
1465            property_name: Name of the property
1466            value: The property value to validate
1467            preprocess_parentheses: If True, wrap string values in parentheses
1468
1469        Returns:
1470            The normalized value
1471
1472        Raises:
1473            SQLMeshError: If validation fails
1474
1475        Example:
1476            >>> validated = PropertyValidator.validate_and_normalize_property("distributed_by", "RANDOM")
1477            >>> # Result: "RANDOM" (string from EnumType)
1478        """
1479        # logger.debug("validate_and_normalize_property. value: %s, type: %s", value, type(value))
1480
1481        # Step 1: Optionally preprocess string with parentheses
1482        if preprocess_parentheses:
1483            value = PropertyValidator.ensure_parenthesized(value)
1484
1485        # Step 2: Get INPUT type validator
1486        input_spec = PropertySpecs.get_property_input_spec(property_name)
1487        if input_spec is None:
1488            raise SQLMeshError(f"Unknown property '{property_name}'.")
1489
1490        # Step 3: Validate
1491        validated = input_spec.validate(value)
1492        if validated is None:
1493            raise SQLMeshError(f"Invalid value type for property '{property_name}': {value!r}.")
1494
1495        # Step 4: Normalize
1496        normalized = input_spec.normalize(validated)
1497
1498        # Step 5: Check by using output spec
1499        output_spec = PropertySpecs.get_property_output_spec(property_name)
1500        if output_spec is not None:
1501            if output_spec.validate(normalized) is None:
1502                raise SQLMeshError(
1503                    f"Normalized value for property '{property_name}' doesn't match output spec: {normalized!r}."
1504                )
1505
1506        # Step 6: Return
1507        return normalized
1508
1509    @staticmethod
1510    def check_invalid_names(
1511        valid_name: str,
1512        invalid_names: t.List[str],
1513        table_properties: t.Dict[str, t.Any],
1514        suggestion: t.Optional[str] = None,
1515    ) -> None:
1516        """
1517        Check for invalid/deprecated property names and raise error with suggestion.
1518
1519        Args:
1520            valid_name: The correct property name
1521            invalid_names: List of invalid/deprecated names to check for
1522            table_properties: Table properties dictionary to check
1523            suggestion: Optional custom error message suggestion
1524
1525        Raises:
1526            SQLMeshError: If any invalid name is found
1527
1528        Example:
1529            >> PropertyValidator.check_invalid_names(
1530            ...     valid_name="partitioned_by",
1531            ...     invalid_names=["partition_by", "partition"],
1532            ...     table_properties={"partition_by": "dt"}
1533            ... )
1534            SQLMeshError: Invalid property 'partition_by'. Use 'partitioned_by' instead.
1535        """
1536        for invalid_name in invalid_names:
1537            if invalid_name in table_properties:
1538                msg = suggestion or f"Use '{valid_name}' instead"
1539                raise SQLMeshError(f"Invalid property '{invalid_name}'. {msg}.")
1540
1541    @classmethod
1542    def check_all_invalid_names(cls, table_properties: t.Dict[str, t.Any]) -> None:
1543        """
1544        Check all invalid property names at once using INVALID_PROPERTY_NAME_MAP config.
1545
1546        Args:
1547            table_properties: Table properties dictionary to check
1548
1549        Raises:
1550            SQLMeshError: If any invalid name is found
1551        """
1552        for valid_name, invalid_names in cls.INVALID_PROPERTY_NAME_MAP.items():
1553            cls.check_invalid_names(valid_name, invalid_names, table_properties)
1554
1555    @staticmethod
1556    def check_at_most_one(
1557        property_name: str,
1558        property_description: str,
1559        table_properties: t.Dict[str, t.Any],
1560        exclusive_property_names: t.Optional[t.Set[str]] = None,
1561        parameter_value: t.Optional[t.Any] = None,
1562    ) -> t.Optional[str]:
1563        """
1564        Ensure at most one property from a mutually exclusive group is defined.
1565
1566        Args:
1567            property_name: the canonical name
1568            property_description: description of the property group (for error messages)
1569            exclusive_property_names: List of mutually exclusive property names.
1570                Defaults to canonical name and aliases if not provided.
1571            table_properties: Table properties dictionary to check
1572            parameter_value: Optional parameter value (takes priority over table_properties)
1573
1574        Returns:
1575            Name of the active property, or None if none found
1576            NOTE: If the parameter value is provided, it returns None
1577
1578        Raises:
1579            SQLMeshError: If multiple properties from the group are defined
1580
1581        Example:
1582            >> PropertyValidator.check_at_most_one(
1583            ...     property_name="primary_key",
1584            ...     property_description="key type",
1585            ...     exclusive_property_names=["primary_key", "duplicate_key", "unique_key", "aggregate_key"],
1586            ...     table_properties={"primary_key": "(id)", "duplicate_key": "(id)"}
1587            ... )
1588            SQLMeshError: Multiple key type properties defined: ['primary_key', 'duplicate_key'].
1589                         Only one is allowed.
1590        """
1591        if not exclusive_property_names:
1592            exclusive_property_names = PropertyValidator.EXCLUSIVE_PROPERTY_NAME_MAP.get(
1593                property_name, set()
1594            ) | {property_name}
1595        # logger.debug("Checking at most one property for '%s': %s", property_name, exclusive_property_names)
1596        # Check parameter first (highest priority)
1597        if parameter_value is not None:
1598            # Check if any conflicting properties exist in table_properties
1599            conflicts = [name for name in exclusive_property_names if name in table_properties]
1600            if conflicts:
1601                param_display = f"{property_name} (parameter)"
1602                raise SQLMeshError(
1603                    f"Conflicting {property_description} definitions: "
1604                    f"{param_display} provided along with table_properties {conflicts}. "
1605                    f"Only one {property_description} is allowed."
1606                )
1607            return None
1608
1609        # Check table_properties for multiple definitions
1610        present = [name for name in exclusive_property_names if name in table_properties]
1611        # logger.debug("Get table key names for %s from table_properties: %s", property_name, present)
1612
1613        if len(present) > 1:
1614            raise SQLMeshError(
1615                f"Multiple {property_description} properties defined: {present}. "
1616                f"Only one is allowed."
1617            )
1618
1619        return present[0] if present else None
1620
1621
1622###############################################################################
1623# StarRocks Engine Adapter
1624###############################################################################
1625@set_catalog()
1626class StarRocksEngineAdapter(
1627    LogicalMergeMixin,
1628    PandasNativeFetchDFSupportMixin,
1629    ClusteredByMixin,
1630):
1631    """
1632    StarRocks Engine Adapter for SQLMesh.
1633
1634    StarRocks is a high-performance analytical database with its own dialect-specific
1635    behavior. This adapter highlights a few key characteristics:
1636
1637    1. PRIMARY KEY support is native and must be emitted in the post-schema section.
1638    2. DELETE with subqueries is supported on PRIMARY KEY tables, but other key types still
1639       need guard rails (no boolean literals, TRUNCATE for WHERE TRUE, etc.).
1640    3. Partitioning supports RANGE, LIST, and expression-based syntaxes.
1641
1642    Implementation strategy:
1643    - Override only where StarRocks syntax/behavior diverges from the base adapter.
1644    - Keep the rest of the functionality delegated to the shared base implementation.
1645    """
1646
1647    # ==================== Class Attributes (Declarative Configuration) ====================
1648
1649    DIALECT = "starrocks"
1650    """SQLGlot dialect name for SQL generation"""
1651
1652    DEFAULT_BATCH_SIZE = 10000
1653    """Default batch size for bulk operations"""
1654
1655    SUPPORTS_TRANSACTIONS = False
1656    """
1657    StarRocks does not support transactions for multiple DML statements.
1658    - No BEGIN/COMMIT/ROLLBACK (only txn for multiple INSERT statements from v3.5)
1659    - Operations are auto-committed
1660    - Backfill uses partition-level atomicity
1661    """
1662
1663    INSERT_OVERWRITE_STRATEGY = InsertOverwriteStrategy.DELETE_INSERT
1664    """
1665    StarRocks does support INSERT OVERWRITE syntax (and dynamic overwrite from v3.5).
1666    Use DELETE + INSERT pattern:
1667    1. DELETE FROM table WHERE condition
1668    2. INSERT INTO table SELECT ...
1669
1670    Base class automatically handles this strategy without overriding insert methods.
1671
1672    TODO: later, we can add support for INSERT OVERWRITE, even use Primary Key for beter performance
1673    """
1674
1675    COMMENT_CREATION_TABLE = CommentCreationTable.IN_SCHEMA_DEF_NO_CTAS
1676    """Column comments are added inline in a plain CREATE TABLE, but StarRocks CTAS only accepts a
1677    bare column-name list (no types or per-column COMMENT) before AS SELECT. So for CTAS we emit
1678    `CREATE TABLE t COMMENT '...' AS SELECT ...` (table comment only) and register column comments
1679    afterward via ALTER TABLE ... MODIFY COLUMN ... COMMENT (see _build_create_comment_column_exp)."""
1680
1681    COMMENT_CREATION_VIEW = CommentCreationView.IN_SCHEMA_DEF_NO_COMMANDS
1682    """View comments are added in CREATE VIEW statement"""
1683
1684    SUPPORTS_MATERIALIZED_VIEWS = True
1685    """StarRocks supports materialized views with refresh strategies"""
1686
1687    SUPPORTS_MATERIALIZED_VIEW_SCHEMA = True
1688    """
1689    StarRocks materialized views support specifying a column list, but the column definition is
1690    limited (e.g. column name + comment, not full type definitions). We set this to True and
1691    implement custom MV schema rendering in create_view/_create_materialized_view.
1692    """
1693
1694    RECREATE_MATERIALIZED_VIEW_ON_EVALUATION = False
1695    """
1696    StarRocks async materialized views maintain themselves: they revalidate automatically even if the
1697    underlying data is dropped, and the data is kept current either by StarRocks' automatic refresh or
1698    by an explicit `REFRESH MATERIALIZED VIEW` (which also enables partition-level incremental refresh).
1699    """
1700
1701    SUPPORTS_REPLACE_TABLE = False
1702    """No REPLACE TABLE syntax; use DROP + CREATE instead"""
1703
1704    SUPPORTS_CREATE_DROP_CATALOG = False
1705    """StarRocks supports DROPing external catalogs.
1706    TODO: whether it's external catalogs, or includes the internal catalog
1707    """
1708
1709    SUPPORTS_INDEXES = True
1710    """
1711    StarRocks supports PRIMARY KEY in CREATE TABLE, but NOT standalone CREATE INDEX.
1712
1713    We set this to True to enable PRIMARY KEY generation in CREATE TABLE statements.
1714    The create_index() method is overridden to prevent actual CREATE INDEX execution.
1715
1716    Supported (defined in CREATE TABLE):
1717    - PRIMARY KEY: Automatically creates sorted index
1718    - INDEX clause: For bloom filter, bitmap, inverted indexes
1719    NOT supported:
1720        CREATE INDEX idx_name ON t (name);  -- Will be skipped by create_index()
1721    """
1722
1723    SUPPORTS_TUPLE_IN = False
1724    """
1725    StarRocks does NOT support tuple IN syntax: (col1, col2) IN ((val1, val2), (val3, val4))
1726
1727    Instead, use OR with AND conditions:
1728    (col1 = val1 AND col2 = val2) OR (col1 = val3 AND col2 = val4)
1729
1730    This is automatically handled by snapshot_id_filter and snapshot_name_version_filter
1731    in sqlmesh/core/state_sync/db/utils.py when SUPPORTS_TUPLE_IN = False.
1732    """
1733
1734    MAX_TABLE_COMMENT_LENGTH = 2048
1735    """Maximum length for table comments"""
1736
1737    MAX_COLUMN_COMMENT_LENGTH = 255
1738    """Maximum length for column comments"""
1739
1740    MAX_IDENTIFIER_LENGTH = 64
1741    """Maximum length for table/column names"""
1742
1743    RESOLVE_TABLE_REFS_IN_PHYSICAL_PROPERTIES: t.FrozenSet[str] = frozenset(
1744        {"excluded_trigger_tables", "excluded_refresh_tables"}
1745    )
1746    """StarRocks async materialized views accept these properties to exclude certain tables from
1747    triggering or participating in refreshes.  When the value references a managed SQLMesh model,
1748    StarRocks needs the physical table name (db.table), not the logical view name.  Managed-model
1749    physical names carry no catalog prefix (catalog support is UNSUPPORTED), so they are already in
1750    the warehouse-local db.table form StarRocks expects; unmanaged references (e.g. an external
1751    catalog's ext_catalog.db.table) pass through unchanged."""
1752
1753    # ==================== Schema Operations ====================
1754    # StarRocks supports CREATE/DROP SCHEMA the same as CREATE/DROP DATABSE.
1755    # So, no need to implement create_schema / drop_schema
1756
1757    # ==================== Data Object Query ====================
1758    def _get_data_objects(
1759        self, schema_name: SchemaName, object_names: t.Optional[t.Set[str]] = None
1760    ) -> t.List[DataObject]:
1761        """
1762        Returns all the data objects that exist in the given schema.
1763        Uses information_schema tables which are compatible with MySQL protocol.
1764
1765        StarRocks uses the MySQL-compatible information_schema layout, so the same query
1766        works here.
1767        Note: Materialized View is not reliably distinguished from View (both may appear as `VIEW`)
1768        in information_schema.tables. We therefore best-effort detect MVs via
1769        information_schema.materialized_views and upgrade matching objects to `materialized_view`.
1770
1771        Args:
1772            schema_name: The schema (database) to query
1773            object_names: Optional set of specific table names to filter
1774
1775        Returns:
1776            List of DataObject instances representing tables and views
1777        """
1778        schema_db = to_schema(schema_name).db
1779        query = (
1780            exp.select(
1781                exp.column("table_schema").as_("schema_name"),
1782                exp.column("table_name").as_("name"),
1783                exp.case(exp.column("table_type"))
1784                .when(
1785                    exp.Literal.string("BASE TABLE"),
1786                    exp.Literal.string("table"),
1787                )
1788                .when(
1789                    exp.Literal.string("VIEW"),
1790                    exp.Literal.string("view"),
1791                )
1792                .else_("table_type")
1793                .as_("type"),
1794            )
1795            .from_(exp.table_("tables", db="information_schema"))
1796            .where(exp.column("table_schema").eq(schema_db))
1797        )
1798        if object_names:
1799            # StarRocks may treat information_schema table_name comparisons as case-sensitive.
1800            # Use LOWER(table_name) to match case-insensitively.
1801            lowered_names = [name.lower() for name in object_names]
1802            query = query.where(exp.func("LOWER", exp.column("table_name")).isin(*lowered_names))
1803
1804        df = self.fetchdf(query)
1805        objects = [
1806            DataObject(
1807                schema=row.schema_name,
1808                name=row.name,
1809                type=DataObjectType.from_str(str(row.type)),
1810            )
1811            for row in df.itertuples()
1812        ]
1813
1814        # Best-effort upgrade of MV types using information_schema.materialized_views.
1815        # If this fails (unsupported / permissions / version), fall back to information_schema.tables.
1816        try:
1817            mv_query = (
1818                exp.select(
1819                    exp.column("table_schema").as_("schema_name"),
1820                    exp.column("table_name").as_("name"),
1821                )
1822                .from_(exp.table_("materialized_views", db="information_schema"))
1823                .where(exp.column("table_schema").eq(schema_db))
1824            )
1825            if object_names:
1826                lowered_names = [name.lower() for name in object_names]
1827                mv_query = mv_query.where(
1828                    exp.func("LOWER", exp.column("table_name")).isin(*lowered_names)
1829                )
1830
1831            mv_df = self.fetchdf(mv_query)
1832            mv_names: t.Set[str] = {
1833                t.cast(str, r.name).lower() for r in mv_df.itertuples() if r.name
1834            }
1835
1836            if mv_names:
1837                for obj in objects:
1838                    if obj.name.lower() in mv_names:
1839                        obj.type = DataObjectType.MATERIALIZED_VIEW
1840        except Exception:
1841            logger.warning(
1842                f"[StarRocks] Failed to get materialized views from information_schema.materialized_views"
1843            )
1844
1845        return objects
1846
1847    def create_index(
1848        self,
1849        table_name: TableName,
1850        index_name: str,
1851        columns: t.Tuple[str, ...],
1852        exists: bool = True,
1853    ) -> None:
1854        """
1855        Override to prevent CREATE INDEX statements (not supported in StarRocks).
1856
1857        StarRocks does not support standalone CREATE INDEX statements.
1858        Indexes must be defined during CREATE TABLE using INDEX clause.
1859
1860        Since SQLMesh state tables use PRIMARY KEY (which provides efficient indexing),
1861        we simply log and skip additional index creation requests.
1862
1863        This matches upstream StarRocks limitations and prevents accidental CREATE INDEX calls.
1864        """
1865        logger.warning(
1866            f"[StarRocks] Skipping CREATE INDEX {index_name} on {table_name} - "
1867            f"StarRocks does not support standalone CREATE INDEX statements. "
1868            f"PRIMARY KEY provides equivalent indexing for columns: {columns}"
1869        )
1870        return
1871
1872    def _create_table_like(
1873        self,
1874        target_table_name: TableName,
1875        source_table_name: TableName,
1876        exists: bool,
1877        **kwargs: t.Any,
1878    ) -> None:
1879        """Create a new table using StarRocks' native `CREATE TABLE ... LIKE ...` syntax.
1880
1881        The base implementation re-creates the target table from `columns(source)` which can
1882        lose non-column metadata. Using LIKE lets the engine preserve more of the original
1883        table definition (engine-defined behavior).
1884        """
1885        self.execute(
1886            exp.Create(
1887                this=exp.to_table(target_table_name),
1888                kind="TABLE",
1889                exists=exists,
1890                properties=exp.Properties(
1891                    expressions=[
1892                        exp.LikeProperty(
1893                            this=exp.to_table(source_table_name),
1894                        ),
1895                    ],
1896                ),
1897            )
1898        )
1899
1900    def delete_from(
1901        self,
1902        table_name: TableName,
1903        where: t.Optional[t.Union[str, exp.Expr]] = None,
1904    ) -> None:
1905        """
1906        Delete from a table.
1907
1908        StarRocks DELETE limitations by table type:
1909
1910        PRIMARY KEY tables:
1911        - Support complex WHERE conditions (subqueries, BETWEEN, etc.)
1912        - No special handling needed
1913
1914        Other table types (DUPLICATE/UNIQUE/AGGREGATE KEY):
1915        - WHERE TRUE not supported → use TRUNCATE TABLE
1916        - Boolean literals (TRUE/FALSE) not supported
1917        - BETWEEN not supported → convert to >= AND <=
1918        - Others not supported:
1919            - CAST() not supported in WHERE
1920            - Subqueries not supported
1921            - ...
1922
1923        But, I don't know what the table type is.
1924
1925        Args:
1926            table_name: The table to delete from
1927            where: The where clause to filter rows to delete
1928        """
1929        # Parse where clause if it's a string
1930        where_expr: t.Optional[exp.Expr]
1931        if isinstance(where, str):
1932            from sqlglot import parse_one
1933
1934            where_expr = parse_one(where, dialect=self.dialect)
1935        else:
1936            where_expr = where
1937
1938        # If no where clause or WHERE TRUE, use TRUNCATE TABLE (for all table types)
1939        if not where_expr or where_expr == exp.true():
1940            table_expr = exp.to_table(table_name) if isinstance(table_name, str) else table_name
1941            logger.info(
1942                f"Converting DELETE FROM {table_name} WHERE TRUE to TRUNCATE TABLE "
1943                "(StarRocks does not support WHERE TRUE in DELETE)"
1944            )
1945            self.execute(f"TRUNCATE TABLE {table_expr.sql(dialect=self.dialect, identify=True)}")
1946            return
1947
1948        # For non-PRIMARY KEY tables, apply WHERE clause restrictions
1949        # Note: We conservatively apply restrictions to all tables since we can't easily
1950        # determine table type at DELETE time. PRIMARY KEY tables will still work with
1951        # simplified conditions, while non-PRIMARY KEY tables require them.
1952        if isinstance(where_expr, exp.Expr):
1953            original_where = where_expr
1954            # Remove boolean literals (not supported in any table type)
1955            where_expr = self._where_clause_remove_boolean_literals(where_expr)
1956            # Convert BETWEEN to >= AND <= (required for DUPLICATE/UNIQUE/AGGREGATE KEY tables)
1957            where_expr = self._where_clause_convert_between_to_comparison(where_expr)
1958
1959            if where_expr != original_where:
1960                logger.debug(
1961                    f"Converted WHERE clause for StarRocks compatibility, table: {table_name}.\n"
1962                    f"  Original: {original_where.sql(dialect=self.dialect)}\n"
1963                    f"  Converted: {where_expr.sql(dialect=self.dialect)}"
1964                )
1965
1966        # Use parent implementation
1967        super().delete_from(table_name, where_expr)
1968
1969    def _where_clause_remove_boolean_literals(self, expression: exp.Expr) -> exp.Expr:
1970        """
1971        Remove TRUE/FALSE boolean literals from WHERE expressions.
1972
1973        StarRocks Limitation (except PRIMARY KEY tables):
1974        Boolean literals (TRUE/FALSE) are not supported in WHERE clauses.
1975
1976        This method simplifies expressions:
1977        - (condition) AND TRUE / TRUE AND (condition) → condition
1978        - (condition) OR FALSE / FALSE OR (condition) → condition
1979        - WHERE TRUE → 1=1 (though TRUNCATE is used instead)
1980        - WHERE FALSE → 1=0
1981
1982        Args:
1983            expression: The expression to clean
1984
1985        Returns:
1986            Cleaned expression without boolean literals
1987        """
1988
1989        def transform(node: exp.Expr) -> exp.Expr:
1990            # Handle standalone TRUE/FALSE at the top level
1991            if node == exp.true():
1992                # Convert TRUE to 1=1
1993                return exp.EQ(this=exp.Literal.number(1), expression=exp.Literal.number(1))
1994            elif node == exp.false():  # noqa: RET505
1995                # Convert FALSE to 1=0
1996                return exp.EQ(this=exp.Literal.number(1), expression=exp.Literal.number(0))
1997
1998            # Handle AND expressions
1999            elif isinstance(node, exp.And):
2000                left = node.this
2001                right = node.expression
2002
2003                # Remove TRUE from AND
2004                if left == exp.true():
2005                    return right
2006                if right == exp.true():
2007                    return left
2008
2009            # Handle OR expressions
2010            elif isinstance(node, exp.Or):
2011                left = node.this
2012                right = node.expression
2013
2014                # Remove FALSE from OR
2015                if left == exp.false():
2016                    return right
2017                if right == exp.false():
2018                    return left
2019
2020            return node
2021
2022        # Transform the expression tree
2023        return expression.transform(transform, copy=True)
2024
2025    def _where_clause_convert_between_to_comparison(self, expression: exp.Expr) -> exp.Expr:
2026        """
2027        Convert BETWEEN expressions to >= AND <= comparisons.
2028
2029        StarRocks Limitation (DUPLICATE/UNIQUE/AGGREGATE KEY Tables):
2030        BETWEEN is not supported in DELETE WHERE clauses for non-PRIMARY KEY tables.
2031
2032        PRIMARY KEY tables support BETWEEN, but this conversion is safe for all table types
2033        since the converted form (>= AND <=) is semantically equivalent.
2034
2035        This method converts:
2036        - col BETWEEN a AND b  →  col >= a AND col <= b
2037
2038        Args:
2039            expression: The expression potentially containing BETWEEN
2040
2041        Returns:
2042            Expression with BETWEEN converted to comparisons
2043        """
2044
2045        def transform(node: exp.Expr) -> exp.Expr:
2046            if isinstance(node, exp.Between):
2047                # Extract components: col BETWEEN low AND high
2048                column = node.this  # The column being tested
2049                low = node.args.get("low")  # Lower bound
2050                high = node.args.get("high")  # Upper bound
2051
2052                if column and low and high:
2053                    # Build: column >= low AND column <= high
2054                    gte = exp.GTE(this=column.copy(), expression=low.copy())
2055                    lte = exp.LTE(this=column.copy(), expression=high.copy())
2056                    return exp.And(this=gte, expression=lte)
2057
2058            return node
2059
2060        # Transform the expression tree
2061        return expression.transform(transform, copy=True)
2062
2063    def execute(
2064        self,
2065        expressions: t.Union[str, exp.Expr, t.Sequence[exp.Expr]],
2066        ignore_unsupported_errors: bool = False,
2067        quote_identifiers: bool = True,
2068        track_rows_processed: bool = False,
2069        **kwargs: t.Any,
2070    ) -> None:
2071        """
2072        Override execute to strip FOR UPDATE from queries (not supported in StarRocks).
2073
2074        StarRocks is an OLAP database and does not support row-level locking via
2075        SELECT ... FOR UPDATE. This method removes lock expressions before execution.
2076
2077        Args:
2078            expressions: SQL expression(s) to execute
2079            ignore_unsupported_errors: Whether to ignore unsupported errors
2080            quote_identifiers: Whether to quote identifiers
2081            track_rows_processed: Whether to track rows processed
2082            **kwargs: Additional arguments
2083        """
2084        from sqlglot.helper import ensure_list
2085
2086        if isinstance(expressions, str):
2087            super().execute(
2088                expressions,
2089                ignore_unsupported_errors=ignore_unsupported_errors,
2090                quote_identifiers=quote_identifiers,
2091                track_rows_processed=track_rows_processed,
2092                **kwargs,
2093            )
2094            return
2095
2096        # Process expressions to remove FOR UPDATE
2097        processed_expressions: t.List[exp.Expr] = []
2098        for e in ensure_list(expressions):
2099            if not isinstance(e, exp.Expr):
2100                super().execute(
2101                    expressions,
2102                    ignore_unsupported_errors=ignore_unsupported_errors,
2103                    quote_identifiers=quote_identifiers,
2104                    track_rows_processed=track_rows_processed,
2105                    **kwargs,
2106                )
2107                return
2108
2109            # Remove lock (FOR UPDATE) from SELECT statements
2110            if isinstance(e, exp.Select) and e.args.get("locks"):
2111                e = e.copy()
2112                e.set("locks", None)
2113                logger.warning(
2114                    f"[StarRocks] Removed FOR UPDATE from SELECT statement: "
2115                    f"{e.sql(dialect=self.dialect, identify=quote_identifiers)}"
2116                )
2117            processed_expressions.append(e)
2118
2119        # Call parent execute with processed expressions
2120        super().execute(
2121            processed_expressions,
2122            ignore_unsupported_errors=ignore_unsupported_errors,
2123            quote_identifiers=quote_identifiers,
2124            track_rows_processed=track_rows_processed,
2125            **kwargs,
2126        )
2127
2128    def adjust_physical_properties_for_incremental(
2129        self,
2130        physical_properties: t.Dict[str, t.Any],
2131        *,
2132        requires_delete_capable_table: bool,
2133        unique_key: t.Optional[t.List[exp.Expr]],
2134        model_name: str,
2135    ) -> t.Dict[str, t.Any]:
2136        """Enforce that StarRocks incremental models use a PRIMARY KEY table.
2137
2138        Incremental kinds rely on DELETE/MERGE statements that StarRocks only supports on PRIMARY
2139        KEY tables; DUPLICATE/UNIQUE/AGGREGATE KEY tables reject the predicates SQLMesh generates
2140        (e.g. a time-range DELETE with a CAST bound, or any non-key-column predicate). When a
2141        unique_key is available (INCREMENTAL_BY_UNIQUE_KEY) we promote it to a PRIMARY KEY;
2142        otherwise a PRIMARY KEY must be specified explicitly via physical_properties, and we raise
2143        so the failure is clear at creation time rather than producing a broken table.
2144
2145        The caller owns ``physical_properties`` (it is already a defensive copy), so we mutate and
2146        return it in place.
2147        """
2148        if not requires_delete_capable_table or "primary_key" in physical_properties:
2149            return physical_properties
2150
2151        # Promote the model's unique_key to a PRIMARY KEY table so that complex DELETE/MERGE
2152        # statements remain supported.
2153        if unique_key:
2154            physical_properties["primary_key"] = (
2155                unique_key[0] if len(unique_key) == 1 else exp.Tuple(expressions=unique_key)
2156            )
2157            logger.info(
2158                "Model '%s' promoted to PRIMARY KEY table on StarRocks to support rich DELETE operations.",
2159                model_name,
2160            )
2161            return physical_properties
2162
2163        raise SQLMeshError(
2164            f"StarRocks incremental model '{model_name}' requires a PRIMARY KEY table. "
2165            "Incremental kinds use DELETE/MERGE operations that StarRocks only supports on PRIMARY KEY "
2166            "tables; DUPLICATE/UNIQUE/AGGREGATE KEY tables are not sufficient. "
2167            "Specify `physical_properties (primary_key = (...))`, or set `unique_key` on the model."
2168        )
2169
2170    # ==================== Table Creation (CORE IMPLEMENTATION) ====================
2171    def _create_table_from_columns(
2172        self,
2173        table_name: TableName,
2174        target_columns_to_types: t.Dict[str, exp.DataType],
2175        primary_key: t.Optional[t.Tuple[str, ...]] = None,
2176        exists: bool = True,
2177        table_description: t.Optional[str] = None,
2178        column_descriptions: t.Optional[t.Dict[str, str]] = None,
2179        **kwargs: t.Any,
2180    ) -> None:
2181        """
2182        Create a table using column definitions.
2183
2184        Unified Model Parameter vs Physical Properties Handling:
2185        For properties that can be defined both as model parameters and in physical_properties,
2186        this method implements a unified priority strategy:
2187        1. Model parameter takes priority if present
2188        2. Otherwise, use value from physical_properties
2189        3. Ensure at most one definition exists
2190
2191        Supported unified properties:
2192        - primary_key: Model parameter OR physical_properties.primary_key
2193        - partitioned_by: Model parameter OR physical_properties.partitioned_by/partition_by
2194        - clustered_by: Model parameter OR physical_properties.clustered_by/order_by
2195
2196        Other key types (duplicate_key, aggregate_key, unique_key) only support physical_properties.
2197
2198        StarRocks Key Column Ordering Constraint:
2199        ALL key types (PRIMARY KEY, UNIQUE KEY, DUPLICATE KEY, AGGREGATE KEY) require:
2200        - Key columns MUST be the first N columns in CREATE TABLE
2201        - Column order MUST match the KEY clause order
2202
2203        Implementation Strategy:
2204        1. Normalize model parameters into table_properties with priority handling
2205        2. Extract and validate key columns from unified table_properties
2206        3. Validate no conflicts between different key types
2207        4. Reorder columns to place key columns first
2208        5. For PRIMARY KEY: Pass to base class (sets SUPPORTS_INDEXES=True)
2209        6. For other keys: Handle in _build_table_key_property
2210
2211        Args:
2212            table_name: Fully qualified table name
2213            target_columns_to_types: Column definitions {name: DataType}
2214            primary_key: Primary key column names (model parameter, takes priority)
2215            exists: Add IF NOT EXISTS clause
2216            table_description: Table comment
2217            column_descriptions: Column comments {column_name: comment}
2218            kwargs: Additional properties including:
2219                - partitioned_by: Partition columns (model parameter)
2220                - clustered_by: Clustering columns (model parameter)
2221                - table_properties: Physical properties dict
2222
2223        Example:
2224            # Model parameter (priority):
2225            partitioned_by=dt,
2226            clustered_by=(dt, id))
2227            physical_properties(
2228                primary_key=(id, dt)
2229            )
2230
2231            # Or physical_properties only:
2232            physical_properties(
2233                duplicate_key=(id, dt),
2234                partitioned_by=dt,
2235                order_by=(dt, id)
2236            )
2237        """
2238        # Use setdefault to simplify table_properties access
2239        table_properties = kwargs.setdefault("table_properties", {})
2240
2241        # Extract and validate key columns from table_properties
2242        # Priority: parameter primary_key > table_properties (already handled above)
2243        key_type, key_columns = self._extract_and_validate_key_columns(
2244            table_properties, primary_key
2245        )
2246        # logger.debug(
2247        #     "_create_table_from_columns: extracted key_type=%s, key_columns=%s",
2248        #     key_type,
2249        #     key_columns,
2250        # )
2251
2252        # IMPORTANT: Normalize parameter primary_key into table_properties for unified handling
2253        # This ensures _build_table_properties_exp() can access primary_key even when
2254        # it's passed as a model parameter rather than in physical_properties
2255        if primary_key:
2256            table_properties["primary_key"] = primary_key
2257            logger.debug("_create_table_from_columns: unified primary_key into table_properties")
2258        elif key_type:
2259            # logger.debug(
2260            #     "table key type '%s' may be handled in _build_table_key_property", key_type
2261            # )
2262            pass
2263
2264        # StarRocks key column ordering constraint: All key types need reordering
2265        if key_columns:
2266            target_columns_to_types = self._reorder_columns_for_key(
2267                target_columns_to_types, key_columns, key_type or "key"
2268            )
2269
2270        # IMPORTANT: Do NOT pass primary_key to base class!
2271        # Unlike other databases, StarRocks requires PRIMARY KEY to be in POST_SCHEMA location
2272        # (in properties section after columns), not inside schema (inside column definitions).
2273        # We handle ALL key types (including PRIMARY KEY) in _build_table_key_property.
2274        # logger.debug(
2275        #     "_create_table_from_columns: NOT passing primary_key to base class (handled in _build_table_key_property)"
2276        # )
2277        super()._create_table_from_columns(
2278            table_name=table_name,
2279            target_columns_to_types=target_columns_to_types,
2280            primary_key=None,  # StarRocks handles PRIMARY KEY in properties, not schema
2281            exists=exists,
2282            table_description=table_description,
2283            column_descriptions=column_descriptions,
2284            **kwargs,
2285        )
2286
2287    # ==================== View / Materialized View ====================
2288    def create_view(
2289        self,
2290        view_name: TableName,
2291        query_or_df: QueryOrDF,
2292        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
2293        replace: bool = True,
2294        materialized: bool = False,
2295        materialized_properties: t.Optional[t.Dict[str, t.Any]] = None,
2296        table_description: t.Optional[str] = None,
2297        column_descriptions: t.Optional[t.Dict[str, str]] = None,
2298        view_properties: t.Optional[t.Dict[str, exp.Expr]] = None,
2299        source_columns: t.Optional[t.List[str]] = None,
2300        **create_kwargs: t.Any,
2301    ) -> None:
2302        """
2303        StarRocks behavior:
2304        - Regular VIEW: supports CREATE OR REPLACE (base behavior)
2305        - MATERIALIZED VIEW: does NOT support CREATE OR REPLACE, so replace=True => DROP + CREATE
2306        """
2307        if not materialized:
2308            return super().create_view(
2309                view_name=view_name,
2310                query_or_df=query_or_df,
2311                target_columns_to_types=target_columns_to_types,
2312                replace=replace,
2313                materialized=False,
2314                materialized_properties=materialized_properties,
2315                table_description=table_description,
2316                column_descriptions=column_descriptions,
2317                view_properties=view_properties,
2318                source_columns=source_columns,
2319                **create_kwargs,
2320            )
2321
2322        # MATERIALIZED VIEW path
2323        # MVs with audits get a synchronous refresh after creation (see _create_materialized_view),
2324        # which requires REFRESH DEFERRED. Validate before the drop so we fail without destroying
2325        # an existing MV.
2326        has_audits = bool((materialized_properties or {}).get("has_audits"))
2327        if has_audits:
2328            self._validate_deferred_refresh_for_audits(view_name, view_properties)
2329
2330        if replace:
2331            # Avoid DROP MATERIALIZED VIEW failure when an object with the same name exists but is not an MV.
2332            self.drop_data_object_on_type_mismatch(
2333                self.get_data_object(view_name), DataObjectType.MATERIALIZED_VIEW
2334            )
2335            self.drop_view(view_name, ignore_if_not_exists=True, materialized=True)
2336        # logger.debug(
2337        #     f"Creating materialized view: {view_name}, materialized: {materialized}, "
2338        #     f"materialized_properties: {materialized_properties}, "
2339        #     f"view_properties: {view_properties}, create_kwargs: {create_kwargs}, "
2340        # )
2341
2342        return self._create_materialized_view(
2343            view_name=view_name,
2344            query_or_df=query_or_df,
2345            target_columns_to_types=target_columns_to_types,
2346            materialized_properties=materialized_properties,
2347            table_description=table_description,
2348            column_descriptions=column_descriptions,
2349            view_properties=view_properties,
2350            source_columns=source_columns,
2351            **create_kwargs,
2352        )
2353
2354    def _create_materialized_view(
2355        self,
2356        view_name: TableName,
2357        query_or_df: QueryOrDF,
2358        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
2359        materialized_properties: t.Optional[t.Dict[str, t.Any]] = None,
2360        table_description: t.Optional[str] = None,
2361        column_descriptions: t.Optional[t.Dict[str, str]] = None,
2362        view_properties: t.Optional[t.Dict[str, exp.Expr]] = None,
2363        source_columns: t.Optional[t.List[str]] = None,
2364        **create_kwargs: t.Any,
2365    ) -> None:
2366        """
2367        Create a StarRocks materialized view.
2368
2369        StarRocks MV schema supports a column list but does NOT support explicit data types in that list.
2370        We therefore build a schema with column names + optional COMMENT only.
2371        """
2372        import pandas as pd
2373
2374        query_or_df = self._native_df_to_pandas_df(query_or_df)
2375
2376        if isinstance(query_or_df, pd.DataFrame):
2377            values: t.List[t.Tuple[t.Any, ...]] = list(
2378                query_or_df.itertuples(index=False, name=None)
2379            )
2380            target_columns_to_types, source_columns = self._columns_to_types(
2381                query_or_df, target_columns_to_types, source_columns
2382            )
2383            if not target_columns_to_types:
2384                raise SQLMeshError("columns_to_types must be provided for dataframes")
2385            source_columns_to_types = get_source_columns_to_types(
2386                target_columns_to_types, source_columns
2387            )
2388            query_or_df = self._values_to_sql(
2389                values,
2390                source_columns_to_types,
2391                batch_start=0,
2392                batch_end=len(values),
2393            )
2394
2395        source_queries, target_columns_to_types = self._get_source_queries_and_columns_to_types(
2396            query_or_df,
2397            target_columns_to_types,
2398            batch_size=0,
2399            target_table=view_name,
2400            source_columns=source_columns,
2401        )
2402        if len(source_queries) != 1:
2403            raise SQLMeshError("Only one source query is supported for creating materialized views")
2404
2405        target_table = exp.to_table(view_name)
2406        schema: t.Union[exp.Table, exp.Schema] = self._build_materialized_view_schema_exp(
2407            target_table,
2408            target_columns_to_types=target_columns_to_types,
2409            column_descriptions=column_descriptions,
2410        )
2411
2412        # Pass model materialized properties through the existing properties builder
2413        partitioned_by = None
2414        clustered_by = None
2415        partition_interval_unit = None
2416        if materialized_properties:
2417            partitioned_by = materialized_properties.get("partitioned_by")
2418            clustered_by = materialized_properties.get("clustered_by")
2419            partition_interval_unit = materialized_properties.get("partition_interval_unit")
2420            # logger.debug(
2421            #     f"Get info from materialized_properties: {materialized_properties}, "
2422            #     f"partitioned_by: {partitioned_by}, "
2423            #     f"clustered_by: {clustered_by}, "
2424            #     f"partition_interval_unit: {partition_interval_unit}"
2425            # )
2426
2427        properties_exp = self._build_table_properties_exp(
2428            catalog_name=target_table.catalog,
2429            table_properties=view_properties,
2430            target_columns_to_types=target_columns_to_types,
2431            table_description=table_description,
2432            partitioned_by=partitioned_by,
2433            clustered_by=clustered_by,
2434            partition_interval_unit=partition_interval_unit,
2435            table_kind="MATERIALIZED_VIEW",
2436        )
2437
2438        with source_queries[0] as query:
2439            self.execute(
2440                exp.Create(
2441                    this=schema,
2442                    kind="VIEW",
2443                    replace=False,
2444                    expression=query,
2445                    properties=properties_exp,
2446                    **create_kwargs,
2447                ),
2448                quote_identifiers=self.QUOTE_IDENTIFIERS_IN_VIEWS,
2449            )
2450
2451        # MVs with audits are created with REFRESH DEFERRED (enforced in create_view), so StarRocks
2452        # does not populate them on creation. Audits need data, so block on a synchronous refresh.
2453        if bool((materialized_properties or {}).get("has_audits")):
2454            refresh_sql = (
2455                f"REFRESH MATERIALIZED VIEW "
2456                f"{exp.to_table(view_name).sql(dialect=self.dialect, identify=True)} "
2457                f"WITH SYNC MODE"
2458            )
2459            self.execute(refresh_sql)
2460
2461        self._clear_data_object_cache(view_name)
2462
2463    def _build_materialized_view_schema_exp(
2464        self,
2465        table: exp.Table,
2466        *,
2467        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
2468        column_descriptions: t.Optional[t.Dict[str, str]] = None,
2469    ) -> t.Union[exp.Table, exp.Schema]:
2470        """
2471        Build a StarRocks MV schema with column names + optional COMMENT only (no types).
2472        """
2473        columns: t.List[str] = []
2474        if target_columns_to_types:
2475            columns = list(target_columns_to_types)
2476        elif column_descriptions:
2477            columns = list(column_descriptions)
2478
2479        if not columns:
2480            return table
2481
2482        column_descriptions = column_descriptions or {}
2483        expressions: t.List[exp.Expr] = []
2484        for col in columns:
2485            constraints: t.List[exp.ColumnConstraint] = []
2486            comment = column_descriptions.get(col)
2487            if comment:
2488                constraints.append(
2489                    exp.ColumnConstraint(
2490                        kind=exp.CommentColumnConstraint(
2491                            this=exp.Literal.string(self._truncate_column_comment(comment))
2492                        )
2493                    )
2494                )
2495            expressions.append(
2496                exp.ColumnDef(
2497                    this=exp.to_identifier(col),
2498                    constraints=constraints,
2499                )
2500            )
2501
2502        return exp.Schema(this=table, expressions=expressions)
2503
2504    # ==================== Table Properties Builder (for Table and MV/VIew) ====================
2505    def _build_table_properties_exp(
2506        self,
2507        catalog_name: t.Optional[str] = None,
2508        table_format: t.Optional[str] = None,
2509        storage_format: t.Optional[str] = None,
2510        partitioned_by: t.Optional[t.List[exp.Expr]] = None,
2511        partition_interval_unit: t.Optional[IntervalUnit] = None,
2512        clustered_by: t.Optional[t.List[exp.Expr]] = None,
2513        table_properties: t.Optional[t.Dict[str, exp.Expr]] = None,
2514        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
2515        table_description: t.Optional[str] = None,
2516        table_kind: t.Optional[str] = None,
2517        **kwargs: t.Any,
2518    ) -> t.Optional[exp.Properties]:
2519        """
2520        Build table properties for StarRocks CREATE TABLE statement.
2521
2522        Unified Model Parameter vs Physical Properties Handling:
2523        This method receives both model parameters (partitioned_by, clustered_by) and
2524        physical_properties (table_properties dict). Priority is handled as follows:
2525
2526        1. primary_key / partitioned_by / clustered_by (ORDER BY)
2527           - Model parameter takes priority
2528           - Falls back to physical_properties.xxx
2529           - Handled in _build_partition_property
2530
2531        2. special for primary_key:
2532           - Still need to be processed in _build_table_key_property
2533
2534        3. Other key types (duplicate_key, unique_key, aggregate_key):
2535           - Only available via physical_properties
2536           - Handled in _build_table_key_property
2537
2538        Handles:
2539        - Key constraints (PRIMARY KEY, DUPLICATE KEY, UNIQUE KEY)
2540        - Partition expressions (RANGE/LIST/EXPRESSION)
2541        - Distribution (HASH/RANDOM)
2542        - Order by (clustering)
2543        - Table comment
2544        - Other properties (replication_num, storage_medium, etc.)
2545
2546        Args:
2547            partitioned_by: Partition columns/expression from model parameter (takes priority)
2548            clustered_by: Clustering columns from model parameter (takes priority)
2549            table_properties: Dictionary containing physical_properties:
2550                - primary_key/duplicate_key/unique_key/aggregate_key: Tuple/list of column names
2551                - partitioned_by(partition_by): Partition definition (fallback)
2552                - distributed_by: Tuple of EQ expressions (kind, expressions, buckets) or string
2553                - clustered_by(order_by): Clustering definition (fallback)
2554                - replication_num, storage_medium, etc.: Literal values
2555            table_description: Table comment
2556        """
2557        properties: t.List[exp.Expr] = []
2558        table_properties_copy = dict(table_properties) if table_properties else {}
2559        # logger.debug(
2560        #     "_build_table_properties_exp: table_properties=%s",
2561        #     table_properties.keys() if table_properties else [],
2562        # )
2563
2564        is_mv = table_kind == "MATERIALIZED_VIEW"
2565        if is_mv:
2566            # Required for CREATE MATERIALIZED VIEW (SQLGlot uses this property to switch the keyword)
2567            properties.append(exp.MaterializedProperty())
2568
2569        # Validate all property names at once
2570        PropertyValidator.check_all_invalid_names(table_properties_copy)
2571
2572        # Check for mutually exclusive key types
2573        # Note: primary_key is already set into table_properties if model param is set
2574        active_key_type = PropertyValidator.check_at_most_one(
2575            property_name="key_type",
2576            property_description="key type",
2577            table_properties=table_properties_copy,
2578        )
2579        if is_mv and active_key_type:
2580            raise SQLMeshError(
2581                f"You can't specify the table type when the table is a materialized view. "
2582                f"Current specified key type '{active_key_type}'."
2583            )
2584
2585        # 0. Extract key columns for partition/distribution validation (read-only, don't pop yet)
2586        key_type, key_columns = None, None
2587        if active_key_type:
2588            key_type = active_key_type
2589            key_expr = table_properties_copy[key_type]
2590            # Use validate_and_normalize_property to get List[exp.Column], then extract names
2591            normalized = PropertyValidator.validate_and_normalize_property(
2592                key_type, key_expr, preprocess_parentheses=True
2593            )
2594            key_columns = tuple(col.name for col in normalized)
2595
2596        # 1. Handle key constraints (ALL types including PRIMARY KEY)
2597        key_prop = self._build_table_key_property(table_properties_copy, active_key_type)
2598        if key_prop:
2599            properties.append(key_prop)
2600
2601        # 2. Add table comment (it must be ahead of other properties except the talbe key/type)
2602        if table_description:
2603            properties.append(
2604                exp.SchemaCommentProperty(
2605                    this=exp.Literal.string(self._truncate_table_comment(table_description))
2606                )
2607            )
2608
2609        # 3. Handle partitioned_by (PARTITION BY RANGE/LIST/EXPRESSION)
2610        partition_prop = self._build_partition_property(
2611            partitioned_by,
2612            partition_interval_unit,
2613            target_columns_to_types,
2614            catalog_name,
2615            table_properties_copy,
2616            key_type,
2617            key_columns,
2618        )
2619        if partition_prop:
2620            properties.append(partition_prop)
2621
2622        # 4. Handle distributed_by (DISTRIBUTED BY HASH/RANDOM)
2623        distributed_prop = self._build_distributed_by_property(table_properties_copy, key_columns)
2624        if distributed_prop:
2625            properties.append(distributed_prop)
2626
2627        # 5. Handle refresh_property (REFRESH ...)
2628        # StarRocks only supports ASYNC materialized views, which require a REFRESH clause.
2629        # Synchronous MVs are not supported, so a missing refresh is a hard error rather than
2630        # a silent fallback (which would create an undetectable sync MV).
2631        if is_mv:
2632            refresh_prop = self._build_refresh_property(table_properties_copy)
2633            if refresh_prop is None:
2634                raise SQLMeshError(
2635                    "StarRocks materialized views require a REFRESH clause. "
2636                    "Specify at least one of 'refresh_moment' or 'refresh_scheme' in the model's "
2637                    "physical_properties (e.g. refresh_scheme = 'ASYNC')."
2638                )
2639            properties.append(refresh_prop)
2640
2641        # 6. Handle order_by/clustered_by (ORDER BY ...)
2642        order_prop = self._build_order_by_property(table_properties_copy, clustered_by or None)
2643        if order_prop:
2644            properties.append(order_prop)
2645
2646        # 5. Handle other properties (replication_num, storage_medium, etc.)
2647        other_props = self._build_other_properties(table_properties_copy)
2648        properties.extend(other_props)
2649
2650        return exp.Properties(expressions=properties) if properties else None
2651
2652    def _build_view_properties_exp(
2653        self,
2654        view_properties: t.Optional[t.Dict[str, exp.Expr]] = None,
2655        table_description: t.Optional[str] = None,
2656        **kwargs: t.Any,
2657    ) -> t.Optional[exp.Properties]:
2658        """
2659        Build CREATE VIEW properties for StarRocks.
2660
2661        Supports StarRocks view SECURITY syntax: SECURITY {NONE | INVOKER}
2662        via exp.SqlSecurityProperty (renders as `SECURITY <value>`).
2663        """
2664        properties: t.List[exp.Expr] = []
2665
2666        if table_description:
2667            properties.append(
2668                exp.SchemaCommentProperty(
2669                    this=exp.Literal.string(self._truncate_table_comment(table_description))
2670                )
2671            )
2672
2673        if view_properties:
2674            view_properties_copy = dict(view_properties)
2675            security = view_properties_copy.pop("security", None)
2676            if security is not None:
2677                security_text = PropertyValidator.validate_and_normalize_property(
2678                    "security", security
2679                )
2680                # exp.SqlSecurityProperty renders as `SECURITY <value>` (no '=')
2681                properties.append(exp.SqlSecurityProperty(this=exp.Var(this=security_text)))
2682
2683            properties.extend(self._table_or_view_properties_to_expressions(view_properties_copy))
2684
2685        if properties:
2686            return exp.Properties(expressions=properties)
2687        return None
2688
2689    def _build_table_key_property(
2690        self, table_properties: t.Dict[str, t.Any], active_key_type: t.Optional[str]
2691    ) -> t.Optional[exp.Expr]:
2692        """
2693        Build key constraint property for ALL key types including PRIMARY KEY.
2694
2695        Unlike other databases where PRIMARY KEY is handled by base class in schema,
2696        StarRocks requires ALL key types (PRIMARY KEY, DUPLICATE KEY, UNIQUE KEY, AGGREGATE KEY)
2697        to be in POST_SCHEMA location (properties section after columns).
2698
2699        Handles:
2700        - PRIMARY KEY
2701        - DUPLICATE KEY
2702        - UNIQUE KEY
2703        - AGGREGATE KEY (when implemented)
2704
2705        Args:
2706            table_properties: Dictionary containing key definitions (will be modified)
2707            active_key_type: The active key type or None
2708
2709        Returns:
2710            Key property expression for the active key type, or None
2711        """
2712        if not active_key_type:
2713            return None
2714
2715        # Configuration: key_name -> Property class (excluding primary_key)
2716        KEY_PROPERTY_CLASSES: t.Dict[str, t.Type[exp.Expr]] = {
2717            "primary_key": exp.PrimaryKey,
2718            "duplicate_key": exp.DuplicateKeyProperty,
2719            "unique_key": exp.UniqueKeyProperty,
2720            # "aggregate_key": exp.AggregateKeyProperty,  # Not implemented yet
2721        }
2722
2723        property_class = KEY_PROPERTY_CLASSES.get(active_key_type)
2724        key_value = table_properties.pop(active_key_type, None)
2725        if not property_class:
2726            # Aggregate key requires special handling
2727            if active_key_type == "aggregate_key":
2728                raise SQLMeshError(
2729                    "AGGREGATE KEY tables are not currently supported. "
2730                    "AGGREGATE KEY requires specifying aggregation functions (SUM/MAX/MIN/REPLACE) "
2731                    "for value columns, which is not supported in the current model configuration syntax. "
2732                    "Please use PRIMARY KEY, UNIQUE KEY, or DUPLICATE KEY instead."
2733                )
2734            # Unknown key type
2735            logger.warning(f"[StarRocks] Unknown key type: {active_key_type}")
2736            return None
2737        if key_value is None:
2738            logger.error(f"Failed to get the parameter value for {active_key_type!r}")
2739            return None
2740
2741        logger.debug(
2742            "_build_table_key_property: input key=%s value=%s",
2743            active_key_type,
2744            key_value,
2745        )
2746
2747        # Validate and normalize
2748        # preprocess_parentheses=True handles string preprocessing like 'id, dt' -> '(id, dt)'
2749        normalized = PropertyValidator.validate_and_normalize_property(
2750            active_key_type, key_value, preprocess_parentheses=True
2751        )
2752        # normalized is List[exp.Column] as defined in TableKeyInputSpec
2753        result = property_class(expressions=list(normalized))
2754        return result
2755
2756    def _build_partition_property(
2757        self,
2758        partitioned_by: t.Optional[t.List[exp.Expr]],
2759        partition_interval_unit: t.Optional["IntervalUnit"],
2760        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]],
2761        catalog_name: t.Optional[str],
2762        table_properties: t.Dict[str, t.Any],
2763        key_type: t.Optional[str],
2764        key_columns: t.Optional[t.Tuple[str, ...]],
2765    ) -> t.Optional[exp.Expr]:
2766        """
2767        Build partition property expression.
2768
2769        StarRocks supports:
2770        - PARTITION BY RANGE (cols) - for time-based partitions
2771        - PARTITION BY LIST (cols) - for categorical partitions
2772        - PARTITION BY (exprs) - for expression partitions, can also be `exprs` (without `(`, and `)`)
2773
2774        Args:
2775            partitioned_by: Partition column expressions from parameter
2776            partition_interval_unit: Optional time unit for automatic partitioning
2777            target_columns_to_types: Column definitions
2778            catalog_name: Catalog name (if applicable)
2779            table_properties: Dictionary containing partitioned_by/partitions (will be modified)
2780            key_type: Table key type (for validation)
2781            key_columns: Table key columns (partition columns must be subset)
2782
2783        Returns:
2784            Partition property expression or None
2785        """
2786        # Priority: parameter > partition_by (alias) > partitioned_by
2787        # Use PropertyValidator to check mutual exclusion between parameter and properties
2788        partition_param_name = PropertyValidator.check_at_most_one(
2789            property_name="partitioned_by",
2790            property_description="partition definition",
2791            table_properties=table_properties,
2792            parameter_value=partitioned_by or None,
2793        )
2794
2795        # If parameter was provided, it takes priority
2796        if not partitioned_by and partition_param_name:
2797            # Get from table_properties
2798            partitioned_by = table_properties.pop(partition_param_name, None)
2799        if not partitioned_by:
2800            return None
2801
2802        # Parse partition expressions to extract columns and kind (RANGE/LIST)
2803        partition_kind, partition_cols = self._parse_partition_expressions(partitioned_by)
2804        logger.debug(
2805            "_build_partition_property: partition_kind=%s, partition_cols=%s",
2806            partition_kind,
2807            partition_cols,
2808        )
2809
2810        def extract_column_name(expr: exp.Expr) -> t.Optional[str]:
2811            if isinstance(expr, exp.Column):
2812                return str(expr.name)
2813            elif isinstance(expr, (exp.Anonymous, exp.Func)):  # noqa: RET505
2814                return None  # not implemented
2815            else:
2816                return str(expr)
2817
2818        # Validate partition columns are in key columns (StarRocks requirement)
2819        if key_columns:
2820            partition_col_names = set(extract_column_name(expr) for expr in partition_cols) - {None}
2821            key_cols_set = set(key_columns)
2822            not_in_key = partition_col_names - key_cols_set
2823            if not_in_key:
2824                logger.warning(
2825                    f"[StarRocks] Partition columns {not_in_key} not in {key_type} columns {key_cols_set}. "
2826                    "StarRocks requires partition columns to be part of the table key."
2827                )
2828
2829        # Get partition definitions (RANGE/LIST partitions)
2830        # Note: Expression-based partitioning (partition_kind=None) does not support pre-created partitions
2831        if partitions := table_properties.pop("partitions", None):
2832            if partition_kind is None:
2833                logger.warning(
2834                    "[StarRocks] 'partitions' parameter is ignored for expression-based partitioning. "
2835                    "Expression partitioning creates partitions automatically and does not support "
2836                    "pre-created partition definitions."
2837                )
2838                partitions = None  # Ignore partitions for expression-based partitioning
2839            else:
2840                partitions = PropertyValidator.validate_and_normalize_property(
2841                    "partitions", partitions
2842                )
2843
2844        # Build partition expression using base class method
2845        result = self._build_partitioned_by_exp(
2846            partition_cols,
2847            partition_interval_unit=partition_interval_unit,
2848            target_columns_to_types=target_columns_to_types,
2849            catalog_name=catalog_name,
2850            partitions=partitions,
2851            partition_kind=partition_kind,
2852        )
2853        return result
2854
2855    def _parse_partition_expressions(
2856        self, partitioned_by: t.List[exp.Expr]
2857    ) -> t.Tuple[t.Optional[str], t.List[exp.Expr]]:
2858        """
2859        Parse partition expressions and extract partition kind (RANGE/LIST).
2860
2861        Uses PartitionedByInputSpec to validate and normalize the entire list,
2862        then extracts RANGE/LIST kind from function expressions.
2863
2864        The SPEC output is List[exp.Column | exp.Anonymous | exp.Func], where:
2865        - exp.Column: Regular column reference
2866        - exp.Anonymous: Function call like RANGE(col), LIST(col), and other datetime related functions
2867        - exp.Func: date_trunc(), and other built-in functions
2868
2869        Args:
2870            partitioned_by: List of partition expressions
2871
2872        Returns:
2873            Tuple of (partition_kind, normalized_columns)
2874            - partition_kind: "RANGE", "LIST", or None
2875            - normalized_columns: List of Column expressions, or function expressions
2876        """
2877        parsed_cols: t.List[exp.Expr] = []
2878        partition_kind: t.Optional[str] = None
2879
2880        normalized = PropertyValidator.validate_and_normalize_property(
2881            "partitioned_by", partitioned_by, preprocess_parentheses=True
2882        )
2883        # Process each normalized expression
2884        for norm_expr in normalized:
2885            # Check if it's a RANGE function (exp.Anonymous)
2886            if isinstance(norm_expr, exp.Anonymous) and norm_expr.this:
2887                func_name = str(norm_expr.this).upper()
2888                if func_name in ("RANGE", "LIST"):
2889                    partition_kind = func_name
2890                    # Extract column expressions from function arguments
2891                    for arg in norm_expr.expressions:
2892                        if isinstance(arg, exp.Column):
2893                            parsed_cols.append(arg)
2894                        else:
2895                            parsed_cols.append(exp.to_column(str(arg)))
2896                    continue
2897
2898            # Check if it's a LIST expression (SQLGlot parses LIST(...) as exp.List)
2899            if isinstance(norm_expr, exp.List):
2900                partition_kind = "LIST"
2901                # Extract column expressions from list items
2902                for item in norm_expr.expressions:
2903                    if isinstance(item, exp.Column):
2904                        parsed_cols.append(item)
2905                    else:
2906                        parsed_cols.append(exp.to_column(str(item)))
2907                continue
2908
2909            # Regular column or other function (date_trunc, etc.)
2910            parsed_cols.append(norm_expr)
2911
2912        return partition_kind, parsed_cols
2913
2914    def _build_partitioned_by_exp(
2915        self,
2916        partitioned_by: t.List[exp.Expr],
2917        *,
2918        partition_interval_unit: t.Optional["IntervalUnit"] = None,
2919        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
2920        catalog_name: t.Optional[str] = None,
2921        **kwargs: t.Any,
2922    ) -> t.Optional[
2923        t.Union[
2924            exp.PartitionedByProperty,
2925            exp.PartitionByRangeProperty,
2926            exp.PartitionByListProperty,
2927            exp.Property,
2928        ]
2929    ]:
2930        """
2931        Build StarRocks partitioning expression.
2932
2933        - partition_kind: RANGE/LIST/None (passed via kwargs, None as expression partitioning)
2934        - partitioned_by: normalized partition column/func/anonymous expressions
2935        - partitions: partition definitions as List[str] (passed via kwargs)
2936
2937        Supports both RANGE and LIST partition syntaxes, and expression partition syntax.
2938
2939        Args:
2940            partitioned_by: List of partition column expressions
2941            partition_interval_unit: Optional time unit (unused for now)
2942            target_columns_to_types: Column definitions (unused for now)
2943            catalog_name: Catalog name (unused for now)
2944            **kwargs: Must contain 'partition_kind' and optionally 'partitions'
2945
2946        Returns:
2947            PartitionByRangeProperty, PartitionByListProperty, or None
2948        """
2949        partition_kind = kwargs.get("partition_kind")
2950        partitions: t.Optional[t.List[str]] = kwargs.get("partitions")
2951
2952        # Process partitions to create_expressions
2953        # partitions is already List[str] after SPEC normalization
2954        create_expressions: t.Optional[t.List[exp.Var]] = None
2955        if partitions:
2956            create_expressions = [exp.Var(this=p, quoted=False) for p in partitions]
2957
2958        # Build partition expression
2959        if partition_kind == "LIST":
2960            return exp.PartitionByListProperty(
2961                partition_expressions=partitioned_by,
2962                create_expressions=create_expressions,
2963            )
2964        elif partition_kind == "RANGE":  # noqa: RET505
2965            return exp.PartitionByRangeProperty(
2966                partition_expressions=partitioned_by,
2967                create_expressions=create_expressions,
2968            )
2969        elif partition_kind is None:
2970            return exp.PartitionedByProperty(this=exp.Schema(expressions=partitioned_by))
2971
2972        return None
2973
2974    def _build_distributed_by_property(
2975        self,
2976        table_properties: t.Dict[str, t.Any],
2977        key_columns: t.Optional[t.Tuple[str, ...]],
2978    ) -> t.Optional[exp.DistributedByProperty]:
2979        """
2980        Build DISTRIBUTED BY property from table_properties.
2981
2982        Supports:
2983        1. Structured tuple: (kind='HASH', columns=(id, dt), buckets=10)
2984        2. String format: "HASH(id)", "RANDOM", "HASH(id) BUCKETS 10"
2985        3. None: Returns None (no default distribution)
2986
2987        For complex string like "HASH(id) BUCKETS 10", uses split-and-combine:
2988        - Split on 'BUCKETS' to separate HASH part and bucket count
2989        - Parse HASH part via DistributedByInputSpec
2990        - Parse bucket count as number
2991        - Combine into unified dict
2992
2993        Args:
2994            table_properties: Dictionary containing distributed_by (will be modified)
2995            key_columns: Table key columns (used for default distribution)
2996
2997        Returns:
2998            DistributedByProperty or None
2999        """
3000        distributed_by = table_properties.pop("distributed_by", None)
3001
3002        # No default - if not set, return None
3003        if distributed_by is None:
3004            return None
3005
3006        # Try to parse complex string with BUCKETS first
3007        unified = self._parse_distribution_with_buckets(distributed_by)
3008        if unified is None:
3009            # Fall back to SPEC-based parsing
3010            normalized = PropertyValidator.validate_and_normalize_property(
3011                "distributed_by", distributed_by
3012            )
3013            # Convert to unified dict format
3014            unified = DistributionTupleOutputType.to_unified_dict(normalized)
3015
3016        logger.debug(
3017            "_build_distributed_by_property: normalized to kind=%s, columns=%s, buckets=%s",
3018            unified.get("kind"),
3019            unified.get("columns"),
3020            unified.get("buckets"),
3021        )
3022
3023        # Build expression
3024        kind_expr = exp.Var(this=unified["kind"])
3025        # Convert columns to expressions
3026        columns: t.List[exp.Column] = unified.get("columns", [])
3027        expressions_list: t.List[exp.Expr] = []
3028        for col in columns:
3029            if isinstance(col, exp.Expr):
3030                expressions_list.append(col)
3031            else:
3032                expressions_list.append(exp.to_column(str(col)))
3033        # Build buckets expression
3034        buckets: t.Optional[t.Any] = unified.get("buckets")
3035        buckets_expr: t.Optional[exp.Expr] = None
3036        if buckets is not None:
3037            if isinstance(buckets, exp.Literal):
3038                buckets_expr = buckets
3039            else:
3040                buckets_expr = exp.Literal.number(int(buckets))
3041
3042        result = exp.DistributedByProperty(
3043            kind=kind_expr,
3044            expressions=expressions_list,
3045            buckets=buckets_expr,
3046            order=None,
3047        )
3048        return result
3049
3050    def _validate_deferred_refresh_for_audits(
3051        self,
3052        view_name: TableName,
3053        view_properties: t.Optional[t.Dict[str, exp.Expr]],
3054    ) -> None:
3055        """
3056        Ensure a materialized view with audits uses REFRESH DEFERRED.
3057
3058        StarRocks audits require data to exist in the MV, so SQLMesh issues an explicit synchronous
3059        `REFRESH MATERIALIZED VIEW ... WITH SYNC MODE` right after creating the MV. For that to be
3060        deterministic, the MV must use `refresh_moment = 'DEFERRED'`; otherwise StarRocks' automatic
3061        (IMMEDIATE) refresh would run concurrently and race with the explicit one. A missing
3062        refresh_moment defaults to IMMEDIATE in StarRocks, so it is rejected as well.
3063        """
3064        refresh_moment = (view_properties or {}).get("refresh_moment")
3065        normalized = (
3066            PropertyValidator.validate_and_normalize_property("refresh_moment", refresh_moment)
3067            if refresh_moment is not None
3068            else None
3069        )
3070        if normalized != "DEFERRED":
3071            raise SQLMeshError(
3072                f"[StarRocks] Materialized view '{exp.to_table(view_name).sql(dialect=self.dialect)}' "
3073                "has audits, which require a synchronous refresh after creation. This is only "
3074                "supported with deferred refresh, so the model must set "
3075                "`refresh_moment = 'DEFERRED'` in its physical_properties "
3076                f"(got {normalized or 'no refresh_moment; StarRocks defaults to IMMEDIATE'}). "
3077                "DEFERRED prevents StarRocks' "
3078                "automatic refresh from racing with the synchronous refresh SQLMesh issues."
3079            )
3080
3081    def _build_refresh_property(
3082        self, table_properties: t.Dict[str, t.Any]
3083    ) -> t.Optional[exp.RefreshTriggerProperty]:
3084        """
3085        Build StarRocks MV REFRESH clause as exp.RefreshTriggerProperty.
3086
3087        Input (from physical_properties):
3088        - refresh_moment: IMMEDIATE | DEFERRED (optional)
3089        - refresh_scheme: MANUAL | ASYNC [START (<start_time>)] EVERY (INTERVAL <n> <unit>) (optional)
3090
3091        Output mapping (to match sqlglot StarRocks generator refreshtriggerproperty_sql):
3092        - method: refresh_moment when provided; otherwise a sentinel that won't render
3093        - kind: ASYNC | MANUAL
3094        - starts/every/unit: parsed from refresh_scheme if present
3095        """
3096        refresh_moment = table_properties.pop("refresh_moment", None)
3097        refresh_scheme = table_properties.pop("refresh_scheme", None)
3098        if refresh_moment is None and refresh_scheme is None:
3099            return None
3100
3101        # method is required by exp.RefreshTriggerProperty, but StarRocks syntax does NOT support AUTO.
3102        # We use a sentinel value that the StarRocks generator will not render (it only renders
3103        # IMMEDIATE/DEFERRED).
3104        method_expr = None
3105        if refresh_moment is not None:
3106            refresh_moment_text = PropertyValidator.validate_and_normalize_property(
3107                "refresh_moment", refresh_moment
3108            )
3109            method_expr = exp.Var(this=refresh_moment_text)
3110
3111        kind_expr: t.Optional[exp.Expr] = None
3112        starts_expr: t.Optional[exp.Expr] = None
3113        every_expr: t.Optional[exp.Expr] = None
3114        unit_expr: t.Optional[exp.Expr] = None
3115
3116        if refresh_scheme is not None:
3117            scheme_text = PropertyValidator.validate_and_normalize_property(
3118                "refresh_scheme", refresh_scheme
3119            )
3120            if isinstance(scheme_text, exp.Var):
3121                kind_expr = scheme_text
3122            else:
3123                kind_expr, starts_expr, every_expr, unit_expr = self._parse_refresh_scheme(
3124                    scheme_text
3125                )
3126
3127        return exp.RefreshTriggerProperty(
3128            method=method_expr,
3129            kind=kind_expr,
3130            starts=starts_expr,
3131            every=every_expr,
3132            unit=unit_expr,
3133        )
3134
3135    def _parse_refresh_scheme(
3136        self, refresh_scheme: str
3137    ) -> t.Tuple[
3138        t.Optional[exp.Expr],
3139        t.Optional[exp.Expr],
3140        t.Optional[exp.Expr],
3141        t.Optional[exp.Expr],
3142    ]:
3143        """
3144        Parse StarRocks refresh_scheme text into (kind, starts, every, unit).
3145
3146        parsing simple and robust. We only extract:
3147        - kind: ASYNC | MANUAL (must appear at the beginning), None if not provided
3148        - starts: START (<start_time>) where <start_time> is treated as a raw string
3149        - every/unit: EVERY (INTERVAL <n> <unit>)
3150        """
3151        text = (refresh_scheme or "").strip()
3152        if not text:
3153            return None, None, None, None
3154
3155        m_kind = re.match(r"^(MANUAL|ASYNC)\b", text, flags=re.IGNORECASE)
3156        if not m_kind:
3157            raise SQLMeshError(
3158                f"[StarRocks] Invalid refresh_scheme {refresh_scheme!r}. Expected to start with MANUAL or ASYNC."
3159            )
3160        kind = m_kind.group(1).upper()
3161        kind_expr: t.Optional[exp.Expr] = exp.Var(this=kind)
3162
3163        starts_expr: t.Optional[exp.Expr] = None
3164        every_expr: t.Optional[exp.Expr] = None
3165        unit_expr: t.Optional[exp.Expr] = None
3166        m_start = re.search(
3167            r"\bSTART\s*\(\s*(?:'([^']*)'|\"([^\"]*)\"|([^)]*))\s*\)", text, flags=re.IGNORECASE
3168        )
3169        if m_start:
3170            start_inner = (m_start.group(1) or m_start.group(2) or m_start.group(3) or "").strip()
3171            starts_expr = exp.Literal.string(start_inner)
3172        m_every = re.search(
3173            r"\bEVERY\s*\(\s*INTERVAL\s+(\d+)\s+(\w+)\s*\)", text, flags=re.IGNORECASE
3174        )
3175        if m_every:
3176            every_expr = exp.Literal.number(int(m_every.group(1)))
3177            unit_expr = exp.Var(this=m_every.group(2).upper())
3178        return kind_expr, starts_expr, every_expr, unit_expr
3179
3180    def _parse_distribution_with_buckets(
3181        self, distributed_by: t.Any
3182    ) -> t.Optional[t.Dict[str, t.Any]]:
3183        """
3184        Parse complex distribution expressions like 'HASH(id) BUCKETS 10'.
3185
3186        Since SQLGlot cannot parse 'HASH(id) BUCKETS 10' directly, we:
3187        1. Detect if input is a string containing 'BUCKETS'
3188        2. Split into HASH part and BUCKETS part
3189        3. Parse HASH part via DistributedByInputSpec
3190        4. Extract bucket count as number
3191        5. Combine into unified dict
3192
3193        Args:
3194            distributed_by: The distribution value (may be string, expression, etc.)
3195
3196        Returns:
3197            Unified dict with keys: kind, columns, buckets
3198            Returns None if not a complex BUCKETS expression
3199            (The output function will still handle "HASH(id)" without BUCKETS)
3200        """
3201        # Only handle string or Literal string values
3202        if isinstance(distributed_by, str):
3203            text = distributed_by
3204        elif isinstance(distributed_by, exp.Literal) and distributed_by.is_string:
3205            text = str(distributed_by.this)
3206        else:
3207            return None
3208
3209        # Check if contains BUCKETS keyword (case-insensitive)
3210        if "BUCKETS" not in text.upper():
3211            return None
3212
3213        # Split on BUCKETS (case-insensitive)
3214        match = re.match(r"^(.+?)\s+BUCKETS\s+(\d+)\s*$", text.strip(), flags=re.IGNORECASE)
3215        if not match:
3216            return None
3217
3218        hash_part = match.group(1).strip()
3219        buckets_str = match.group(2)
3220
3221        # Parse the HASH/RANDOM part via SPEC
3222        normalized = PropertyValidator.validate_and_normalize_property("distributed_by", hash_part)
3223
3224        return DistributionTupleOutputType.to_unified_dict(normalized, int(buckets_str))
3225
3226    def _build_order_by_property(
3227        self,
3228        table_properties: t.Dict[str, t.Any],
3229        clustered_by: t.Optional[t.List[exp.Expr]],
3230    ) -> t.Optional[exp.Cluster]:
3231        """
3232        Build ORDER BY (clustering) property.
3233
3234        Supports both:
3235        - clustered_by parameter (from create_table call)
3236        - order_by in table_properties (backward compatibility alias)
3237
3238        Priority: clustered_by parameter > order_by in table_properties
3239
3240        Args:
3241            table_properties: Dictionary containing optional order_by (will be modified)
3242            clustered_by: Clustering columns from parameter
3243
3244        Returns:
3245            Cluster expression (generates ORDER BY) or None
3246        """
3247        # Priority: clustered_by parameter > order_by in table_properties
3248        # Use PropertyValidator to check mutual exclusion between parameter and property
3249        order_by_param_name = PropertyValidator.check_at_most_one(
3250            property_name="clustered_by",
3251            property_description="clustering definition",
3252            table_properties=table_properties,
3253            parameter_value=clustered_by,
3254        )
3255
3256        # If parameter was provided, it takes priority
3257        if clustered_by is None and order_by_param_name:
3258            # Get order_by from table_properties (already validated by check_at_most_one)
3259            order_by = table_properties.pop(order_by_param_name, None)
3260            if order_by is not None:
3261                normalized = PropertyValidator.validate_and_normalize_property(
3262                    "clustered_by", order_by, preprocess_parentheses=True
3263                )
3264                clustered_by = list(normalized)
3265
3266        if clustered_by:
3267            result = exp.Cluster(expressions=clustered_by)
3268            return result
3269        else:  # noqa: RET505
3270            return None
3271
3272    def _build_other_properties(self, table_properties: t.Dict[str, t.Any]) -> t.List[exp.Property]:
3273        """
3274        Build other literal properties (replication_num, storage_medium, etc.).
3275
3276        Uses validate_and_normalize_property for validation and ensures output is string,
3277        as StarRocks PROPERTIES syntax requires all values to be strings.
3278
3279        Args:
3280            table_properties: Dictionary containing properties (will be modified)
3281
3282        Returns:
3283            List of Property expressions
3284        """
3285        other_props = []
3286
3287        for key, value in list(table_properties.items()):
3288            # Skip special keys handled elsewhere
3289            if key in PropertyValidator.IMPORTANT_PROPERTY_NAMES:
3290                logger.warning(f"[StarRocks] {key!r} should have been processed already, skipping")
3291                continue
3292
3293            # Remove from properties
3294            table_properties.pop(key)
3295
3296            # Validate and normalize to string
3297            # All other properties are treated as generic string properties
3298            try:
3299                normalized = PropertyValidator.validate_and_normalize_property(key, value)
3300                other_props.append(
3301                    exp.Property(
3302                        this=exp.to_identifier(key),
3303                        value=exp.Literal.string(str(normalized)),
3304                    )
3305                )
3306            except SQLMeshError as e:
3307                logger.warning("[StarRocks] skipping property %s due to error: %s", key, e)
3308
3309        return other_props
3310
3311    def _extract_and_validate_key_columns(
3312        self,
3313        table_properties: t.Dict[str, t.Any],
3314        primary_key: t.Optional[t.Tuple[str, ...]] = None,
3315    ) -> t.Tuple[t.Optional[str], t.Optional[t.Tuple[str, ...]]]:
3316        """
3317        Extract and validate key columns from table_properties.
3318
3319        All key types require:
3320        - Key columns must be the first N columns in CREATE TABLE
3321        - Column order must match the KEY clause order
3322
3323        Priority:
3324        - Parameter primary_key > table_properties primary_key
3325        - Only one key type allowed per table
3326
3327        Args:
3328            table_properties: Table properties dictionary (lowercase keys expected)
3329            primary_key: Primary key from method parameter (highest priority)
3330
3331        Returns:
3332            Tuple of (key_type, key_columns)
3333            - key_type: One of 'primary_key', 'unique_key', 'duplicate_key', 'aggregate_key', None
3334            - key_columns: Tuple of column names, or None
3335
3336        Raises:
3337            SQLMeshError: If multiple key types are defined or column extraction fails
3338        """
3339        # Use PropertyValidator to check mutual exclusion
3340        active_key_type = PropertyValidator.check_at_most_one(
3341            property_name="key_type",  # dummy
3342            property_description="table key type",
3343            table_properties=table_properties,
3344            parameter_value=primary_key,
3345        )
3346
3347        # If parameter primary_key was provided, return it
3348        if primary_key:
3349            return ("primary_key", primary_key)
3350
3351        # Extract from table_properties
3352        if not active_key_type:
3353            return (None, None)
3354
3355        # Get the key expression and normalize via SPEC
3356        key_expr = table_properties[active_key_type]  # Read without popping
3357        # Use validate_and_normalize_property to get List[exp.Column], then extract names
3358        normalized = PropertyValidator.validate_and_normalize_property(
3359            active_key_type, key_expr, preprocess_parentheses=True
3360        )
3361        key_columns = tuple(col.name for col in normalized)
3362
3363        return (active_key_type, key_columns)
3364
3365    def _reorder_columns_for_key(
3366        self,
3367        target_columns_to_types: t.Dict[str, exp.DataType],
3368        key_columns: t.Tuple[str, ...],
3369        key_type: str = "key",
3370    ) -> t.Dict[str, exp.DataType]:
3371        """
3372        Reorder columns to place key columns first.
3373
3374        StarRocks Constraint (ALL Table Types):
3375        Key columns (PRIMARY/UNIQUE/DUPLICATE/AGGREGATE) MUST be the first N columns
3376        in the CREATE TABLE statement, in the same order as defined in the KEY clause.
3377
3378        Example:
3379            Input:
3380                columns = {"customer_id": INT, "order_id": BIGINT, "event_date": DATE}
3381                key_columns = ("order_id", "event_date")
3382                key_type = "primary_key"
3383
3384            Output:
3385                {"order_id": BIGINT, "event_date": DATE, "customer_id": INT}
3386
3387        Args:
3388            target_columns_to_types: Original column order (from SELECT)
3389            key_columns: Key column names in desired order
3390            key_type: Type of key for logging (primary_key, unique_key, etc.)
3391
3392        Returns:
3393            Reordered columns with key columns first
3394
3395        Raises:
3396            SQLMeshError: If a key column is not found in target_columns_to_types
3397        """
3398        # Validate that all key columns exist
3399        missing_key_cols = set(key_columns) - set(target_columns_to_types.keys())
3400        if missing_key_cols:
3401            raise SQLMeshError(
3402                f"{key_type} columns {missing_key_cols} not found in table columns. "
3403                f"Available columns: {list(target_columns_to_types.keys())}"
3404            )
3405
3406        # Build new ordered dict: key columns first, then remaining columns
3407        reordered = {}
3408
3409        # 1. Add key columns in key order
3410        for key_col in key_columns:
3411            reordered[key_col] = target_columns_to_types[key_col]
3412
3413        # 2. Add remaining columns (preserve original order)
3414        for col_name, col_type in target_columns_to_types.items():
3415            if col_name not in key_columns:
3416                reordered[col_name] = col_type
3417
3418        logger.info(
3419            f"Reordered columns for {key_type.upper()}: "
3420            f"Original order: {list(target_columns_to_types.keys())}, "
3421            f"New order: {list(reordered.keys())}"
3422        )
3423
3424        return reordered
3425
3426    def _build_create_comment_table_exp(
3427        self, table: exp.Table, table_comment: str, table_kind: str = "TABLE"
3428    ) -> str:
3429        """
3430        Build ALTER TABLE COMMENT SQL for table comment modification.
3431
3432        StarRocks uses non-standard syntax for table comments:
3433            ALTER TABLE {table} COMMENT = '{comment}'
3434
3435        Note: This method is typically NOT called for StarRocks because the table comment is
3436        included directly in CREATE TABLE (and CTAS) via SchemaCommentProperty, which StarRocks
3437        accepts even for `CREATE TABLE ... COMMENT '...' AS SELECT`.
3438
3439        However, this override is provided for potential future use cases:
3440        - Modifying comments on existing tables via ALTER TABLE
3441        - View comments (if COMMENT_CREATION_VIEW changes)
3442
3443        Args:
3444            table: Table expression
3445            table_comment: The comment to add
3446            table_kind: Type of object (TABLE, VIEW, etc.)
3447
3448        Returns:
3449            SQL string for ALTER TABLE COMMENT
3450        """
3451        table_sql = table.sql(dialect=self.dialect, identify=True)
3452        comment_sql = exp.Literal.string(self._truncate_table_comment(table_comment)).sql(
3453            dialect=self.dialect
3454        )
3455        return f"ALTER TABLE {table_sql} COMMENT = {comment_sql}"
3456
3457    def _build_create_comment_column_exp(
3458        self,
3459        table: exp.Table,
3460        column_name: str,
3461        column_comment: str,
3462        table_kind: str = "TABLE",
3463    ) -> str:
3464        """
3465        Build ALTER TABLE MODIFY COLUMN SQL for column comment modification.
3466
3467        StarRocks accepts the comment without re-stating the column type:
3468            ALTER TABLE {table} MODIFY COLUMN {column} COMMENT '{comment}'
3469
3470        Because COMMENT_CREATION_TABLE = IN_SCHEMA_DEF_NO_CTAS, column comments are inlined for a
3471        plain CREATE TABLE but NOT for CTAS (StarRocks rejects types/comments in a CTAS column
3472        list). This method is therefore the fallback used to register column comments after a CTAS,
3473        and to modify column comments on existing tables.
3474
3475        Args:
3476            table: Table expression
3477            column_name: Name of the column
3478            column_comment: The comment to add
3479            table_kind: Type of object (TABLE, VIEW, etc.)
3480
3481        Returns:
3482            SQL string for ALTER TABLE MODIFY COLUMN with COMMENT
3483        """
3484        table_sql = table.sql(dialect=self.dialect, identify=True)
3485        column_sql = exp.to_identifier(column_name).sql(dialect=self.dialect, identify=True)
3486
3487        comment_sql = exp.Literal.string(self._truncate_column_comment(column_comment)).sql(
3488            dialect=self.dialect
3489        )
3490
3491        return f"ALTER TABLE {table_sql} MODIFY COLUMN {column_sql} COMMENT {comment_sql}"
3492
3493    # ==================== Methods NOT Needing Override (Base Class Works) ====================
3494    # The following methods work correctly with base class implementation:
3495    # - columns(): Query column definitions via DESCRIBE TABLE
3496    # - table_exists(): Check if table exists via information_schema
3497    # - insert_append(): Standard INSERT INTO ... SELECT
3498    # - insert_overwrite_by_time_partition(): Uses DELETE_INSERT strategy (handled by base)
3499    # - fetchall() / fetchone(): Standard query execution
3500    # - execute(): Base SQL execution. (Modifyed for `FOR UPDATE` lock operation only)
3501    # - create_table_properties(): Delegate to _build_table_properties_exp()