sqlmesh.core.model.kind

   1from __future__ import annotations
   2
   3import typing as t
   4from enum import Enum
   5from typing_extensions import Self
   6
   7from pydantic import Field
   8from sqlglot import exp
   9from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
  10from sqlglot.optimizer.qualify_columns import quote_identifiers
  11from sqlglot.optimizer.simplify import gen
  12from sqlglot.time import format_time
  13
  14from sqlmesh.core import dialect as d
  15from sqlmesh.core.model.common import (
  16    parse_properties,
  17    properties_validator,
  18    validate_extra_and_required_fields,
  19)
  20from sqlmesh.core.model.seed import CsvSettings
  21from sqlmesh.utils.errors import ConfigError
  22from sqlmesh.utils.pydantic import (
  23    PydanticModel,
  24    SQLGlotBool,
  25    SQLGlotColumn,
  26    SQLGlotListOfFieldsOrStar,
  27    SQLGlotListOfFields,
  28    SQLGlotPositiveInt,
  29    SQLGlotString,
  30    SQLGlotCron,
  31    ValidationInfo,
  32    column_validator,
  33    field_validator,
  34    get_dialect,
  35    validate_string,
  36    validate_expression,
  37)
  38
  39
  40if t.TYPE_CHECKING:
  41    from sqlmesh.core._typing import CustomMaterializationProperties
  42
  43    MODEL_KIND = t.TypeVar("MODEL_KIND", bound="_ModelKind")
  44
  45
  46class ModelKindMixin:
  47    @property
  48    def model_kind_name(self) -> t.Optional[ModelKindName]:
  49        """Returns the model kind name."""
  50        raise NotImplementedError
  51
  52    @property
  53    def is_incremental_by_time_range(self) -> bool:
  54        return self.model_kind_name == ModelKindName.INCREMENTAL_BY_TIME_RANGE
  55
  56    @property
  57    def is_incremental_by_unique_key(self) -> bool:
  58        return self.model_kind_name == ModelKindName.INCREMENTAL_BY_UNIQUE_KEY
  59
  60    @property
  61    def is_incremental_by_partition(self) -> bool:
  62        return self.model_kind_name == ModelKindName.INCREMENTAL_BY_PARTITION
  63
  64    @property
  65    def is_incremental_unmanaged(self) -> bool:
  66        return self.model_kind_name == ModelKindName.INCREMENTAL_UNMANAGED
  67
  68    @property
  69    def is_incremental(self) -> bool:
  70        return (
  71            self.is_incremental_by_time_range
  72            or self.is_incremental_by_unique_key
  73            or self.is_incremental_by_partition
  74            or self.is_incremental_unmanaged
  75            or self.is_scd_type_2
  76        )
  77
  78    @property
  79    def is_full(self) -> bool:
  80        return self.model_kind_name == ModelKindName.FULL
  81
  82    @property
  83    def is_view(self) -> bool:
  84        return self.model_kind_name == ModelKindName.VIEW
  85
  86    @property
  87    def is_embedded(self) -> bool:
  88        return self.model_kind_name == ModelKindName.EMBEDDED
  89
  90    @property
  91    def is_seed(self) -> bool:
  92        return self.model_kind_name == ModelKindName.SEED
  93
  94    @property
  95    def is_external(self) -> bool:
  96        return self.model_kind_name == ModelKindName.EXTERNAL
  97
  98    @property
  99    def is_scd_type_2(self) -> bool:
 100        return self.model_kind_name in {
 101            ModelKindName.SCD_TYPE_2,
 102            ModelKindName.SCD_TYPE_2_BY_TIME,
 103            ModelKindName.SCD_TYPE_2_BY_COLUMN,
 104        }
 105
 106    @property
 107    def is_scd_type_2_by_time(self) -> bool:
 108        return self.model_kind_name in {ModelKindName.SCD_TYPE_2, ModelKindName.SCD_TYPE_2_BY_TIME}
 109
 110    @property
 111    def is_scd_type_2_by_column(self) -> bool:
 112        return self.model_kind_name == ModelKindName.SCD_TYPE_2_BY_COLUMN
 113
 114    @property
 115    def is_custom(self) -> bool:
 116        return self.model_kind_name == ModelKindName.CUSTOM
 117
 118    @property
 119    def is_managed(self) -> bool:
 120        return self.model_kind_name == ModelKindName.MANAGED
 121
 122    @property
 123    def is_dbt_custom(self) -> bool:
 124        return self.model_kind_name == ModelKindName.DBT_CUSTOM
 125
 126    @property
 127    def is_symbolic(self) -> bool:
 128        """A symbolic model is one that doesn't execute at all."""
 129        return self.model_kind_name in (ModelKindName.EMBEDDED, ModelKindName.EXTERNAL)
 130
 131    @property
 132    def is_materialized(self) -> bool:
 133        return self.model_kind_name is not None and not (self.is_symbolic or self.is_view)
 134
 135    @property
 136    def only_execution_time(self) -> bool:
 137        """Whether or not this model only cares about execution time to render."""
 138        return self.is_view or self.is_full
 139
 140    @property
 141    def full_history_restatement_only(self) -> bool:
 142        """Whether or not this model only supports restatement of full history."""
 143        return (
 144            self.is_incremental_unmanaged
 145            or self.is_incremental_by_unique_key
 146            or self.is_incremental_by_partition
 147            or self.is_scd_type_2
 148            or self.is_managed
 149            or self.is_full
 150            or self.is_view
 151        )
 152
 153    @property
 154    def supports_python_models(self) -> bool:
 155        return True
 156
 157    @property
 158    def supports_grants(self) -> bool:
 159        """Whether this model kind supports grants configuration."""
 160        return self.is_materialized or self.is_view
 161
 162
 163class ModelKindName(str, ModelKindMixin, Enum):
 164    """The kind of model, determining how this data is computed and stored in the warehouse."""
 165
 166    INCREMENTAL_BY_TIME_RANGE = "INCREMENTAL_BY_TIME_RANGE"
 167    INCREMENTAL_BY_UNIQUE_KEY = "INCREMENTAL_BY_UNIQUE_KEY"
 168    INCREMENTAL_BY_PARTITION = "INCREMENTAL_BY_PARTITION"
 169    INCREMENTAL_UNMANAGED = "INCREMENTAL_UNMANAGED"
 170    FULL = "FULL"
 171    # Legacy alias to SCD Type 2 By Time
 172    # Only used for Parsing and mapping name to SCD Type 2 By Time
 173    SCD_TYPE_2 = "SCD_TYPE_2"
 174    SCD_TYPE_2_BY_TIME = "SCD_TYPE_2_BY_TIME"
 175    SCD_TYPE_2_BY_COLUMN = "SCD_TYPE_2_BY_COLUMN"
 176    VIEW = "VIEW"
 177    EMBEDDED = "EMBEDDED"
 178    SEED = "SEED"
 179    EXTERNAL = "EXTERNAL"
 180    CUSTOM = "CUSTOM"
 181    MANAGED = "MANAGED"
 182    DBT_CUSTOM = "DBT_CUSTOM"
 183
 184    @property
 185    def model_kind_name(self) -> t.Optional[ModelKindName]:
 186        return self
 187
 188    def __str__(self) -> str:
 189        return self.name
 190
 191    def __repr__(self) -> str:
 192        return str(self)
 193
 194
 195class OnDestructiveChange(str, Enum):
 196    """What should happen when a forward-only model change requires a destructive schema change."""
 197
 198    ERROR = "ERROR"
 199    WARN = "WARN"
 200    ALLOW = "ALLOW"
 201    IGNORE = "IGNORE"
 202
 203    @property
 204    def is_error(self) -> bool:
 205        return self == OnDestructiveChange.ERROR
 206
 207    @property
 208    def is_warn(self) -> bool:
 209        return self == OnDestructiveChange.WARN
 210
 211    @property
 212    def is_allow(self) -> bool:
 213        return self == OnDestructiveChange.ALLOW
 214
 215    @property
 216    def is_ignore(self) -> bool:
 217        return self == OnDestructiveChange.IGNORE
 218
 219
 220class OnAdditiveChange(str, Enum):
 221    """What should happen when a forward-only model change requires an additive schema change."""
 222
 223    ERROR = "ERROR"
 224    WARN = "WARN"
 225    ALLOW = "ALLOW"
 226    IGNORE = "IGNORE"
 227
 228    @property
 229    def is_error(self) -> bool:
 230        return self == OnAdditiveChange.ERROR
 231
 232    @property
 233    def is_warn(self) -> bool:
 234        return self == OnAdditiveChange.WARN
 235
 236    @property
 237    def is_allow(self) -> bool:
 238        return self == OnAdditiveChange.ALLOW
 239
 240    @property
 241    def is_ignore(self) -> bool:
 242        return self == OnAdditiveChange.IGNORE
 243
 244
 245def _on_destructive_change_validator(
 246    cls: t.Type, v: t.Union[OnDestructiveChange, str, exp.Identifier]
 247) -> t.Any:
 248    if v and not isinstance(v, OnDestructiveChange):
 249        return OnDestructiveChange(
 250            v.this.upper() if isinstance(v, (exp.Identifier, exp.Literal)) else v.upper()
 251        )
 252    return v
 253
 254
 255def _on_additive_change_validator(
 256    cls: t.Type, v: t.Union[OnAdditiveChange, str, exp.Identifier]
 257) -> t.Any:
 258    if v and not isinstance(v, OnAdditiveChange):
 259        return OnAdditiveChange(
 260            v.this.upper() if isinstance(v, (exp.Identifier, exp.Literal)) else v.upper()
 261        )
 262    return v
 263
 264
 265on_additive_change_validator = field_validator("on_additive_change", mode="before")(
 266    _on_additive_change_validator
 267)
 268
 269on_destructive_change_validator = field_validator("on_destructive_change", mode="before")(
 270    _on_destructive_change_validator
 271)
 272
 273
 274class _ModelKind(PydanticModel, ModelKindMixin):
 275    name: ModelKindName
 276
 277    @property
 278    def model_kind_name(self) -> t.Optional[ModelKindName]:
 279        return self.name
 280
 281    def to_expression(
 282        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 283    ) -> d.ModelKind:
 284        kwargs["expressions"] = expressions
 285        return d.ModelKind(this=self.name.value.upper(), **kwargs)
 286
 287    @property
 288    def data_hash_values(self) -> t.List[t.Optional[str]]:
 289        return [self.name.value]
 290
 291    @property
 292    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 293        return []
 294
 295
 296class TimeColumn(PydanticModel):
 297    column: exp.Expr
 298    format: t.Optional[str] = None
 299
 300    @classmethod
 301    def validator(cls) -> classmethod:
 302        def _time_column_validator(v: t.Any, info: ValidationInfo) -> TimeColumn:
 303            return TimeColumn.create(v, get_dialect(info.data))
 304
 305        return field_validator("time_column", mode="before")(_time_column_validator)
 306
 307    @field_validator("column", mode="before")
 308    @classmethod
 309    def _column_validator(cls, v: t.Union[str, exp.Expr]) -> exp.Expr:
 310        if not v:
 311            raise ConfigError("Time Column cannot be empty.")
 312        if isinstance(v, str):
 313            return exp.to_column(v)
 314        return v
 315
 316    @property
 317    def expression(self) -> exp.Expr:
 318        """Convert this pydantic model into a time_column SQLGlot expression."""
 319        if not self.format:
 320            return self.column
 321
 322        return exp.Tuple(expressions=[self.column, exp.Literal.string(self.format)])
 323
 324    def to_expression(self, dialect: str) -> exp.Expr:
 325        """Convert this pydantic model into a time_column SQLGlot expression."""
 326        if not self.format:
 327            return self.column
 328
 329        return exp.Tuple(
 330            expressions=[
 331                self.column,
 332                exp.Literal.string(
 333                    format_time(self.format, d.Dialect.get_or_raise(dialect).INVERSE_TIME_MAPPING)
 334                ),
 335            ]
 336        )
 337
 338    def to_property(self, dialect: str = "") -> exp.Property:
 339        return exp.Property(this="time_column", value=self.to_expression(dialect))
 340
 341    @classmethod
 342    def create(cls, v: t.Any, dialect: str) -> Self:
 343        if isinstance(v, exp.Tuple):
 344            column_expr = v.expressions[0]
 345            column = (
 346                exp.column(column_expr) if isinstance(column_expr, exp.Identifier) else column_expr
 347            )
 348            format = v.expressions[1].name if len(v.expressions) > 1 else None
 349        elif isinstance(v, exp.Expr):
 350            column = exp.column(v) if isinstance(v, exp.Identifier) else v
 351            format = None
 352        elif isinstance(v, str):
 353            column = d.parse_one(v, dialect=dialect)
 354            column.meta.pop("sql")
 355            format = None
 356        elif isinstance(v, dict):
 357            column_raw = v["column"]
 358            column = (
 359                d.parse_one(column_raw, dialect=dialect)
 360                if isinstance(column_raw, str)
 361                else column_raw
 362            )
 363            format = v.get("format")
 364        elif isinstance(v, TimeColumn):
 365            column = v.column
 366            format = v.format
 367        else:
 368            raise ConfigError(f"Invalid time_column: '{v}'.")
 369
 370        column = quote_identifiers(normalize_identifiers(column, dialect=dialect), dialect=dialect)
 371        column.meta["dialect"] = dialect
 372
 373        return cls(column=column, format=format)
 374
 375
 376def _kind_dialect_validator(cls: t.Type, v: t.Optional[str]) -> str:
 377    if v is None:
 378        return get_dialect({})
 379    return v
 380
 381
 382kind_dialect_validator = field_validator("dialect", mode="before")(_kind_dialect_validator)
 383
 384
 385class _Incremental(_ModelKind):
 386    on_destructive_change: OnDestructiveChange = OnDestructiveChange.ERROR
 387    on_additive_change: OnAdditiveChange = OnAdditiveChange.ALLOW
 388    auto_restatement_cron: t.Optional[SQLGlotCron] = None
 389
 390    _on_destructive_change_validator = on_destructive_change_validator
 391    _on_additive_change_validator = on_additive_change_validator
 392
 393    @property
 394    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 395        return [
 396            *super().metadata_hash_values,
 397            str(self.on_destructive_change),
 398            str(self.on_additive_change),
 399            self.auto_restatement_cron,
 400        ]
 401
 402    def to_expression(
 403        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 404    ) -> d.ModelKind:
 405        return super().to_expression(
 406            expressions=[
 407                *(expressions or []),
 408                *_properties(
 409                    {
 410                        "on_destructive_change": self.on_destructive_change.value,
 411                        "on_additive_change": self.on_additive_change.value,
 412                        "auto_restatement_cron": self.auto_restatement_cron,
 413                    }
 414                ),
 415            ],
 416        )
 417
 418
 419class _IncrementalBy(_Incremental):
 420    dialect: t.Optional[str] = Field(None, validate_default=True)
 421    batch_size: t.Optional[SQLGlotPositiveInt] = None
 422    batch_concurrency: t.Optional[SQLGlotPositiveInt] = None
 423    lookback: t.Optional[SQLGlotPositiveInt] = None
 424    forward_only: SQLGlotBool = False
 425    disable_restatement: SQLGlotBool = False
 426
 427    _dialect_validator = kind_dialect_validator
 428
 429    @property
 430    def data_hash_values(self) -> t.List[t.Optional[str]]:
 431        return [
 432            *super().data_hash_values,
 433            self.dialect,
 434            str(self.lookback) if self.lookback is not None else None,
 435        ]
 436
 437    @property
 438    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 439        return [
 440            *super().metadata_hash_values,
 441            str(self.batch_size) if self.batch_size is not None else None,
 442            str(self.forward_only),
 443            str(self.disable_restatement),
 444        ]
 445
 446    def to_expression(
 447        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 448    ) -> d.ModelKind:
 449        return super().to_expression(
 450            expressions=[
 451                *(expressions or []),
 452                *_properties(
 453                    {
 454                        "batch_size": self.batch_size,
 455                        "batch_concurrency": self.batch_concurrency,
 456                        "lookback": self.lookback,
 457                        "forward_only": self.forward_only,
 458                        "disable_restatement": self.disable_restatement,
 459                    }
 460                ),
 461            ],
 462        )
 463
 464
 465class IncrementalByTimeRangeKind(_IncrementalBy):
 466    name: t.Literal[ModelKindName.INCREMENTAL_BY_TIME_RANGE] = (
 467        ModelKindName.INCREMENTAL_BY_TIME_RANGE
 468    )
 469    time_column: TimeColumn
 470    auto_restatement_intervals: t.Optional[SQLGlotPositiveInt] = None
 471    partition_by_time_column: SQLGlotBool = True
 472
 473    _time_column_validator = TimeColumn.validator()
 474
 475    def to_expression(
 476        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 477    ) -> d.ModelKind:
 478        return super().to_expression(
 479            expressions=[
 480                *(expressions or []),
 481                self.time_column.to_property(kwargs.get("dialect") or ""),
 482                *_properties(
 483                    {
 484                        "partition_by_time_column": self.partition_by_time_column,
 485                    }
 486                ),
 487                *(
 488                    [_property("auto_restatement_intervals", self.auto_restatement_intervals)]
 489                    if self.auto_restatement_intervals is not None
 490                    else []
 491                ),
 492            ]
 493        )
 494
 495    @property
 496    def data_hash_values(self) -> t.List[t.Optional[str]]:
 497        return [*super().data_hash_values, gen(self.time_column.column), self.time_column.format]
 498
 499    @property
 500    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 501        return [
 502            *super().metadata_hash_values,
 503            str(self.partition_by_time_column),
 504            str(self.auto_restatement_intervals)
 505            if self.auto_restatement_intervals is not None
 506            else None,
 507        ]
 508
 509
 510class IncrementalByUniqueKeyKind(_IncrementalBy):
 511    name: t.Literal[ModelKindName.INCREMENTAL_BY_UNIQUE_KEY] = (
 512        ModelKindName.INCREMENTAL_BY_UNIQUE_KEY
 513    )
 514    unique_key: SQLGlotListOfFields
 515    when_matched: t.Optional[exp.Whens] = None
 516    merge_filter: t.Optional[exp.Expr] = None
 517    batch_concurrency: t.Literal[1] = 1
 518
 519    @field_validator("when_matched", mode="before")
 520    def _when_matched_validator(
 521        cls,
 522        v: t.Optional[t.Union[str, list, exp.Whens]],
 523        info: ValidationInfo,
 524    ) -> t.Optional[exp.Whens]:
 525        if v is None:
 526            return v
 527        if isinstance(v, list):
 528            v = " ".join(v)
 529
 530        dialect = get_dialect(info.data)
 531
 532        if isinstance(v, str):
 533            # Whens wrap the WHEN clauses, but the parentheses aren't parsed by sqlglot
 534            v = v.strip()
 535            if v.startswith("("):
 536                v = v[1:-1]
 537
 538            v = t.cast(exp.Whens, d.parse_one(v, into=exp.Whens, dialect=dialect))
 539
 540        v = validate_expression(v, dialect=dialect)
 541        return t.cast(exp.Whens, v.transform(d.replace_merge_table_aliases, dialect=dialect))
 542
 543    @field_validator("merge_filter", mode="before")
 544    def _merge_filter_validator(
 545        cls,
 546        v: t.Optional[exp.Expr],
 547        info: ValidationInfo,
 548    ) -> t.Optional[exp.Expr]:
 549        if v is None:
 550            return v
 551
 552        dialect = get_dialect(info.data)
 553
 554        if isinstance(v, str):
 555            v = v.strip()
 556            v = d.parse_one(v, dialect=dialect)
 557
 558        v = validate_expression(v, dialect=dialect)
 559        return v.transform(d.replace_merge_table_aliases, dialect=dialect)
 560
 561    @property
 562    def data_hash_values(self) -> t.List[t.Optional[str]]:
 563        return [
 564            *super().data_hash_values,
 565            *(gen(k) for k in self.unique_key),
 566            gen(self.when_matched) if self.when_matched is not None else None,
 567            gen(self.merge_filter) if self.merge_filter is not None else None,
 568        ]
 569
 570    def to_expression(
 571        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 572    ) -> d.ModelKind:
 573        return super().to_expression(
 574            expressions=[
 575                *(expressions or []),
 576                *_properties(
 577                    {
 578                        "unique_key": exp.Tuple(expressions=self.unique_key),
 579                        "when_matched": self.when_matched,
 580                        "merge_filter": self.merge_filter,
 581                    }
 582                ),
 583            ],
 584        )
 585
 586
 587class IncrementalByPartitionKind(_Incremental):
 588    name: t.Literal[ModelKindName.INCREMENTAL_BY_PARTITION] = ModelKindName.INCREMENTAL_BY_PARTITION
 589    forward_only: t.Literal[True] = True
 590    disable_restatement: SQLGlotBool = False
 591
 592    @field_validator("forward_only", mode="before")
 593    def _forward_only_validator(cls, v: t.Union[bool, exp.Expr]) -> t.Literal[True]:
 594        if v is not True:
 595            raise ConfigError(
 596                "Do not specify the `forward_only` configuration key - INCREMENTAL_BY_PARTITION models are always forward_only."
 597            )
 598        return v
 599
 600    @property
 601    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 602        return [
 603            *super().metadata_hash_values,
 604            str(self.forward_only),
 605            str(self.disable_restatement),
 606        ]
 607
 608    def to_expression(
 609        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 610    ) -> d.ModelKind:
 611        return super().to_expression(
 612            expressions=[
 613                *(expressions or []),
 614                *_properties(
 615                    {
 616                        "forward_only": self.forward_only,
 617                        "disable_restatement": self.disable_restatement,
 618                    }
 619                ),
 620            ],
 621        )
 622
 623
 624class IncrementalUnmanagedKind(_Incremental):
 625    name: t.Literal[ModelKindName.INCREMENTAL_UNMANAGED] = ModelKindName.INCREMENTAL_UNMANAGED
 626    insert_overwrite: SQLGlotBool = False
 627    forward_only: SQLGlotBool = True
 628    disable_restatement: SQLGlotBool = True
 629
 630    @property
 631    def data_hash_values(self) -> t.List[t.Optional[str]]:
 632        return [*super().data_hash_values, str(self.insert_overwrite)]
 633
 634    @property
 635    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 636        return [
 637            *super().metadata_hash_values,
 638            str(self.forward_only),
 639            str(self.disable_restatement),
 640        ]
 641
 642    def to_expression(
 643        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 644    ) -> d.ModelKind:
 645        return super().to_expression(
 646            expressions=[
 647                *(expressions or []),
 648                *_properties(
 649                    {
 650                        "insert_overwrite": self.insert_overwrite,
 651                        "forward_only": self.forward_only,
 652                        "disable_restatement": self.disable_restatement,
 653                    }
 654                ),
 655            ],
 656        )
 657
 658
 659class ViewKind(_ModelKind):
 660    name: t.Literal[ModelKindName.VIEW] = ModelKindName.VIEW
 661    materialized: SQLGlotBool = False
 662
 663    @property
 664    def data_hash_values(self) -> t.List[t.Optional[str]]:
 665        return [*super().data_hash_values, str(self.materialized)]
 666
 667    @property
 668    def supports_python_models(self) -> bool:
 669        return False
 670
 671    def to_expression(
 672        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 673    ) -> d.ModelKind:
 674        return super().to_expression(
 675            expressions=[
 676                *(expressions or []),
 677                _property("materialized", self.materialized),
 678            ],
 679        )
 680
 681
 682class SeedKind(_ModelKind):
 683    name: t.Literal[ModelKindName.SEED] = ModelKindName.SEED
 684    path: SQLGlotString
 685    batch_size: SQLGlotPositiveInt = 1000
 686    csv_settings: t.Optional[CsvSettings] = None
 687
 688    @field_validator("csv_settings", mode="before")
 689    @classmethod
 690    def _parse_csv_settings(cls, v: t.Any) -> t.Optional[CsvSettings]:
 691        if v is None or isinstance(v, CsvSettings):
 692            return v
 693        if isinstance(v, exp.Expr):
 694            tuple_exp = parse_properties(cls, v, None)
 695            if not tuple_exp:
 696                return None
 697            return CsvSettings(**{e.left.name: e.right for e in tuple_exp.expressions})
 698        if isinstance(v, dict):
 699            return CsvSettings(**v)
 700        return v
 701
 702    def to_expression(
 703        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 704    ) -> d.ModelKind:
 705        """Convert the seed kind into a SQLGlot expression."""
 706        return super().to_expression(
 707            expressions=[
 708                *(expressions or []),
 709                *_properties(
 710                    {
 711                        "path": exp.Literal.string(self.path),
 712                        "batch_size": self.batch_size,
 713                    }
 714                ),
 715            ],
 716        )
 717
 718    @property
 719    def data_hash_values(self) -> t.List[t.Optional[str]]:
 720        csv_setting_values = (self.csv_settings or CsvSettings()).dict().values()
 721        return [
 722            *super().data_hash_values,
 723            *(v if isinstance(v, (str, type(None))) else str(v) for v in csv_setting_values),
 724        ]
 725
 726    @property
 727    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 728        return [*super().metadata_hash_values, str(self.batch_size)]
 729
 730    @property
 731    def supports_python_models(self) -> bool:
 732        return False
 733
 734
 735class FullKind(_ModelKind):
 736    name: t.Literal[ModelKindName.FULL] = ModelKindName.FULL
 737
 738
 739class _SCDType2Kind(_Incremental):
 740    dialect: t.Optional[str] = Field(None, validate_default=True)
 741    unique_key: SQLGlotListOfFields
 742    valid_from_name: SQLGlotColumn = Field(exp.column("valid_from"), validate_default=True)
 743    valid_to_name: SQLGlotColumn = Field(exp.column("valid_to"), validate_default=True)
 744    invalidate_hard_deletes: SQLGlotBool = False
 745    time_data_type: exp.DataType = Field(exp.DataType.build("TIMESTAMP"), validate_default=True)
 746    batch_size: t.Optional[SQLGlotPositiveInt] = None
 747
 748    forward_only: SQLGlotBool = True
 749    disable_restatement: SQLGlotBool = True
 750
 751    _dialect_validator = kind_dialect_validator
 752
 753    _always_validate_column = field_validator("valid_from_name", "valid_to_name", mode="before")(
 754        column_validator
 755    )
 756
 757    @field_validator("time_data_type", mode="before")
 758    @classmethod
 759    def _time_data_type_validator(cls, v: t.Union[str, exp.Expr], values: t.Any) -> exp.Expr:
 760        if isinstance(v, exp.Expr) and not isinstance(v, exp.DataType):
 761            v = v.name
 762        dialect = get_dialect(values)
 763        data_type = exp.DataType.build(v, dialect=dialect)
 764        # Clear meta["sql"] (set by our parser extension) so the pydantic encoder
 765        # uses dialect-aware rendering: e.sql(dialect=meta["dialect"]). Without this,
 766        # the raw SQL text takes priority, which can be wrong for dialect-normalized
 767        # types (e.g., default "TIMESTAMP" should render as "DATETIME" in BigQuery).
 768        data_type.meta.pop("sql", None)
 769        data_type.meta["dialect"] = dialect
 770        return data_type
 771
 772    @property
 773    def managed_columns(self) -> t.Dict[str, exp.DataType]:
 774        return {
 775            self.valid_from_name.name: self.time_data_type,
 776            self.valid_to_name.name: self.time_data_type,
 777        }
 778
 779    @property
 780    def data_hash_values(self) -> t.List[t.Optional[str]]:
 781        return [
 782            *super().data_hash_values,
 783            self.dialect,
 784            *(gen(k) for k in self.unique_key),
 785            gen(self.valid_from_name),
 786            gen(self.valid_to_name),
 787            str(self.invalidate_hard_deletes),
 788            self.time_data_type.sql(self.dialect),
 789            gen(self.batch_size) if self.batch_size is not None else None,
 790        ]
 791
 792    @property
 793    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 794        return [
 795            *super().metadata_hash_values,
 796            str(self.forward_only),
 797            str(self.disable_restatement),
 798        ]
 799
 800    def to_expression(
 801        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 802    ) -> d.ModelKind:
 803        return super().to_expression(
 804            expressions=[
 805                *(expressions or []),
 806                *_properties(
 807                    {
 808                        "unique_key": exp.Tuple(expressions=self.unique_key),
 809                        "valid_from_name": self.valid_from_name,
 810                        "valid_to_name": self.valid_to_name,
 811                        "invalidate_hard_deletes": self.invalidate_hard_deletes,
 812                        "time_data_type": self.time_data_type,
 813                        "forward_only": self.forward_only,
 814                        "disable_restatement": self.disable_restatement,
 815                    }
 816                ),
 817            ],
 818        )
 819
 820
 821class SCDType2ByTimeKind(_SCDType2Kind):
 822    name: t.Literal[ModelKindName.SCD_TYPE_2, ModelKindName.SCD_TYPE_2_BY_TIME] = (
 823        ModelKindName.SCD_TYPE_2_BY_TIME
 824    )
 825    updated_at_name: SQLGlotColumn = Field(exp.column("updated_at"), validate_default=True)
 826    updated_at_as_valid_from: SQLGlotBool = False
 827
 828    _always_validate_updated_at = field_validator("updated_at_name", mode="before")(
 829        column_validator
 830    )
 831
 832    @property
 833    def data_hash_values(self) -> t.List[t.Optional[str]]:
 834        return [
 835            *super().data_hash_values,
 836            gen(self.updated_at_name),
 837            str(self.updated_at_as_valid_from),
 838        ]
 839
 840    def to_expression(
 841        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 842    ) -> d.ModelKind:
 843        return super().to_expression(
 844            expressions=[
 845                *(expressions or []),
 846                *_properties(
 847                    {
 848                        "updated_at_name": self.updated_at_name,
 849                        "updated_at_as_valid_from": self.updated_at_as_valid_from,
 850                    }
 851                ),
 852            ],
 853        )
 854
 855
 856class SCDType2ByColumnKind(_SCDType2Kind):
 857    name: t.Literal[ModelKindName.SCD_TYPE_2_BY_COLUMN] = ModelKindName.SCD_TYPE_2_BY_COLUMN
 858    columns: SQLGlotListOfFieldsOrStar
 859    execution_time_as_valid_from: SQLGlotBool = False
 860    updated_at_name: t.Optional[SQLGlotColumn] = None
 861
 862    @property
 863    def data_hash_values(self) -> t.List[t.Optional[str]]:
 864        columns_sql = (
 865            [gen(c) for c in self.columns]
 866            if isinstance(self.columns, list)
 867            else [gen(self.columns)]
 868        )
 869        return [
 870            *super().data_hash_values,
 871            *columns_sql,
 872            str(self.execution_time_as_valid_from),
 873            gen(self.updated_at_name) if self.updated_at_name is not None else None,
 874        ]
 875
 876    def to_expression(
 877        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 878    ) -> d.ModelKind:
 879        return super().to_expression(
 880            expressions=[
 881                *(expressions or []),
 882                *_properties(
 883                    {
 884                        "columns": exp.Tuple(expressions=self.columns)
 885                        if isinstance(self.columns, list)
 886                        else self.columns,
 887                        "execution_time_as_valid_from": self.execution_time_as_valid_from,
 888                    }
 889                ),
 890            ],
 891        )
 892
 893
 894class ManagedKind(_ModelKind):
 895    name: t.Literal[ModelKindName.MANAGED] = ModelKindName.MANAGED
 896    disable_restatement: t.Literal[True] = True
 897
 898    @property
 899    def supports_python_models(self) -> bool:
 900        return False
 901
 902
 903class DbtCustomKind(_ModelKind):
 904    name: t.Literal[ModelKindName.DBT_CUSTOM] = ModelKindName.DBT_CUSTOM
 905    materialization: str
 906    adapter: str = "default"
 907    definition: str
 908    dialect: t.Optional[str] = Field(None, validate_default=True)
 909
 910    _dialect_validator = kind_dialect_validator
 911
 912    @field_validator("materialization", "adapter", "definition", mode="before")
 913    @classmethod
 914    def _validate_fields(cls, v: t.Any) -> str:
 915        return validate_string(v)
 916
 917    @property
 918    def data_hash_values(self) -> t.List[t.Optional[str]]:
 919        return [
 920            *super().data_hash_values,
 921            self.materialization,
 922            self.definition,
 923            self.adapter,
 924            self.dialect,
 925        ]
 926
 927    def to_expression(
 928        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 929    ) -> d.ModelKind:
 930        return super().to_expression(
 931            expressions=[
 932                *(expressions or []),
 933                *_properties(
 934                    {
 935                        "materialization": exp.Literal.string(self.materialization),
 936                        "adapter": exp.Literal.string(self.adapter),
 937                    }
 938                ),
 939            ],
 940        )
 941
 942
 943class EmbeddedKind(_ModelKind):
 944    name: t.Literal[ModelKindName.EMBEDDED] = ModelKindName.EMBEDDED
 945
 946    @property
 947    def supports_python_models(self) -> bool:
 948        return False
 949
 950
 951class ExternalKind(_ModelKind):
 952    name: t.Literal[ModelKindName.EXTERNAL] = ModelKindName.EXTERNAL
 953
 954
 955class CustomKind(_ModelKind):
 956    name: t.Literal[ModelKindName.CUSTOM] = ModelKindName.CUSTOM
 957    materialization: str
 958    materialization_properties_: t.Optional[exp.Tuple] = Field(
 959        default=None, alias="materialization_properties"
 960    )
 961    forward_only: SQLGlotBool = False
 962    disable_restatement: SQLGlotBool = False
 963    batch_size: t.Optional[SQLGlotPositiveInt] = None
 964    batch_concurrency: t.Optional[SQLGlotPositiveInt] = None
 965    lookback: t.Optional[SQLGlotPositiveInt] = None
 966    auto_restatement_cron: t.Optional[SQLGlotCron] = None
 967    auto_restatement_intervals: t.Optional[SQLGlotPositiveInt] = None
 968
 969    # so that CustomKind subclasses know the dialect when validating / normalizing / interpreting values in `materialization_properties`
 970    dialect: str = Field(exclude=True)
 971
 972    _properties_validator = properties_validator
 973
 974    @field_validator("materialization", mode="before")
 975    @classmethod
 976    def _validate_materialization(cls, v: t.Any) -> str:
 977        # note: create_model_kind() validates the custom materialization class
 978        return validate_string(v)
 979
 980    @property
 981    def materialization_properties(self) -> CustomMaterializationProperties:
 982        """A dictionary of materialization properties."""
 983        if not self.materialization_properties_:
 984            return {}
 985        return d.interpret_key_value_pairs(self.materialization_properties_)
 986
 987    @property
 988    def data_hash_values(self) -> t.List[t.Optional[str]]:
 989        return [
 990            *super().data_hash_values,
 991            self.materialization,
 992            gen(self.materialization_properties_) if self.materialization_properties_ else None,
 993            str(self.lookback) if self.lookback is not None else None,
 994        ]
 995
 996    @property
 997    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 998        return [
 999            *super().metadata_hash_values,
1000            str(self.batch_size) if self.batch_size is not None else None,
1001            str(self.batch_concurrency) if self.batch_concurrency is not None else None,
1002            str(self.forward_only),
1003            str(self.disable_restatement),
1004            self.auto_restatement_cron,
1005            str(self.auto_restatement_intervals)
1006            if self.auto_restatement_intervals is not None
1007            else None,
1008        ]
1009
1010    def to_expression(
1011        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
1012    ) -> d.ModelKind:
1013        return super().to_expression(
1014            expressions=[
1015                *(expressions or []),
1016                *_properties(
1017                    {
1018                        "materialization": exp.Literal.string(self.materialization),
1019                        "materialization_properties": self.materialization_properties_,
1020                        "forward_only": self.forward_only,
1021                        "disable_restatement": self.disable_restatement,
1022                        "batch_size": self.batch_size,
1023                        "batch_concurrency": self.batch_concurrency,
1024                        "lookback": self.lookback,
1025                        "auto_restatement_cron": self.auto_restatement_cron,
1026                        "auto_restatement_intervals": self.auto_restatement_intervals,
1027                    }
1028                ),
1029            ],
1030        )
1031
1032
1033ModelKind = t.Annotated[
1034    t.Union[
1035        EmbeddedKind,
1036        ExternalKind,
1037        FullKind,
1038        IncrementalByTimeRangeKind,
1039        IncrementalByUniqueKeyKind,
1040        IncrementalByPartitionKind,
1041        IncrementalUnmanagedKind,
1042        SeedKind,
1043        ViewKind,
1044        SCDType2ByTimeKind,
1045        SCDType2ByColumnKind,
1046        CustomKind,
1047        ManagedKind,
1048        DbtCustomKind,
1049    ],
1050    Field(discriminator="name"),
1051]
1052
1053MODEL_KIND_NAME_TO_TYPE: t.Dict[str, t.Type[ModelKind]] = {
1054    ModelKindName.EMBEDDED: EmbeddedKind,
1055    ModelKindName.EXTERNAL: ExternalKind,
1056    ModelKindName.FULL: FullKind,
1057    ModelKindName.INCREMENTAL_BY_TIME_RANGE: IncrementalByTimeRangeKind,
1058    ModelKindName.INCREMENTAL_BY_UNIQUE_KEY: IncrementalByUniqueKeyKind,
1059    ModelKindName.INCREMENTAL_BY_PARTITION: IncrementalByPartitionKind,
1060    ModelKindName.INCREMENTAL_UNMANAGED: IncrementalUnmanagedKind,
1061    ModelKindName.SEED: SeedKind,
1062    ModelKindName.VIEW: ViewKind,
1063    ModelKindName.SCD_TYPE_2: SCDType2ByTimeKind,
1064    ModelKindName.SCD_TYPE_2_BY_TIME: SCDType2ByTimeKind,
1065    ModelKindName.SCD_TYPE_2_BY_COLUMN: SCDType2ByColumnKind,
1066    ModelKindName.CUSTOM: CustomKind,
1067    ModelKindName.MANAGED: ManagedKind,
1068    ModelKindName.DBT_CUSTOM: DbtCustomKind,
1069}
1070
1071
1072def model_kind_type_from_name(name: t.Optional[str]) -> t.Type[ModelKind]:
1073    klass = MODEL_KIND_NAME_TO_TYPE.get(name) if name else None
1074    if not klass:
1075        raise ConfigError(f"Invalid model kind '{name}'")
1076    return t.cast(t.Type[ModelKind], klass)
1077
1078
1079def create_model_kind(v: t.Any, dialect: str, defaults: t.Dict[str, t.Any]) -> ModelKind:
1080    if isinstance(v, _ModelKind):
1081        return t.cast(ModelKind, v)
1082
1083    if isinstance(v, (d.ModelKind, dict)):
1084        props = (
1085            {prop.name: prop.args.get("value") for prop in v.expressions}
1086            if isinstance(v, d.ModelKind)
1087            else v
1088        )
1089        name = v.this if isinstance(v, d.ModelKind) else props.get("name")
1090
1091        # We want to ensure whatever name is provided to construct the class is the same name that will be
1092        # found inside the class itself in order to avoid a change during plan/apply for legacy aliases.
1093        # Ex: Pass in `SCD_TYPE_2` then we want to ensure we get `SCD_TYPE_2` as the kind name
1094        # instead of `SCD_TYPE_2_BY_TIME`.
1095        props["name"] = name
1096        kind_type = model_kind_type_from_name(name)
1097
1098        if "dialect" in kind_type.all_fields() and props.get("dialect") is None:
1099            props["dialect"] = dialect
1100
1101        # only pass the on_destructive_change or on_additive_change user default to models inheriting from _Incremental
1102        # that don't explicitly set it in the model definition
1103        if issubclass(kind_type, _Incremental):
1104            for on_change_property in ("on_additive_change", "on_destructive_change"):
1105                if (
1106                    props.get(on_change_property) is None
1107                    and defaults.get(on_change_property) is not None
1108                ):
1109                    props[on_change_property] = defaults.get(on_change_property)
1110
1111        # only pass the batch_concurrency user default to models inheriting from _IncrementalBy
1112        # that don't explicitly set it in the model definition, but ignore subclasses of _IncrementalBy
1113        # that hardcode a specific batch_concurrency
1114        if issubclass(kind_type, _IncrementalBy):
1115            BATCH_CONCURRENCY: t.Final = "batch_concurrency"
1116            if (
1117                props.get(BATCH_CONCURRENCY) is None
1118                and defaults.get(BATCH_CONCURRENCY) is not None
1119                and kind_type.all_field_infos()[BATCH_CONCURRENCY].default is None
1120            ):
1121                props[BATCH_CONCURRENCY] = defaults.get(BATCH_CONCURRENCY)
1122
1123        if kind_type == CustomKind:
1124            # load the custom materialization class and check if it uses a custom kind type
1125            from sqlmesh.core.snapshot.evaluator import get_custom_materialization_type
1126
1127            if "materialization" not in props:
1128                raise ConfigError(
1129                    "The 'materialization' property is required for models of the CUSTOM kind"
1130                )
1131
1132            # The below call will print a warning if a materialization with the given name doesn't exist
1133            # we dont want to throw an error here because we still want Models with a CustomKind to be able
1134            # to be serialized / deserialized in contexts where the custom materialization class may not be available,
1135            # such as in HTTP request handlers
1136            custom_materialization = get_custom_materialization_type(
1137                validate_string(props.get("materialization")), raise_errors=False
1138            )
1139            if custom_materialization is not None:
1140                actual_kind_type, _ = custom_materialization
1141                return actual_kind_type(**props)
1142
1143        validate_extra_and_required_fields(
1144            kind_type, set(props), f"MODEL block 'kind {name}' field"
1145        )
1146        return kind_type(**props)
1147
1148    name = (v.name if isinstance(v, exp.Expr) else str(v)).upper()
1149    return model_kind_type_from_name(name)(name=name)  # type: ignore
1150
1151
1152def _model_kind_validator(cls: t.Type, v: t.Any, info: t.Optional[ValidationInfo]) -> ModelKind:
1153    dialect = get_dialect(info.data) if info else ""
1154    return create_model_kind(v, dialect, {})
1155
1156
1157model_kind_validator: t.Callable = field_validator("kind", mode="before")(_model_kind_validator)
1158
1159
1160def _property(name: str, value: t.Any) -> exp.Property:
1161    return exp.Property(this=exp.var(name), value=exp.convert(value))
1162
1163
1164def _properties(name_value_pairs: t.Dict[str, t.Any]) -> t.List[exp.Property]:
1165    return [_property(k, v) for k, v in name_value_pairs.items() if v is not None]