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            if not v.expressions:
 345                raise ConfigError("Time Column cannot be empty.")
 346            column_expr = v.expressions[0]
 347            column = (
 348                exp.column(column_expr) if isinstance(column_expr, exp.Identifier) else column_expr
 349            )
 350            format = v.expressions[1].name if len(v.expressions) > 1 else None
 351        elif isinstance(v, exp.Expr):
 352            column = exp.column(v) if isinstance(v, exp.Identifier) else v
 353            format = None
 354        elif isinstance(v, str):
 355            column = d.parse_one(v, dialect=dialect)
 356            column.meta.pop("sql")
 357            format = None
 358        elif isinstance(v, dict):
 359            column_raw = v["column"]
 360            column = (
 361                d.parse_one(column_raw, dialect=dialect)
 362                if isinstance(column_raw, str)
 363                else column_raw
 364            )
 365            format = v.get("format")
 366        elif isinstance(v, TimeColumn):
 367            column = v.column
 368            format = v.format
 369        else:
 370            raise ConfigError(f"Invalid time_column: '{v}'.")
 371
 372        column = quote_identifiers(normalize_identifiers(column, dialect=dialect), dialect=dialect)
 373        column.meta["dialect"] = dialect
 374
 375        return cls(column=column, format=format)
 376
 377
 378def _kind_dialect_validator(cls: t.Type, v: t.Optional[str]) -> str:
 379    if v is None:
 380        return get_dialect({})
 381    return v
 382
 383
 384kind_dialect_validator = field_validator("dialect", mode="before")(_kind_dialect_validator)
 385
 386
 387class _Incremental(_ModelKind):
 388    on_destructive_change: OnDestructiveChange = OnDestructiveChange.ERROR
 389    on_additive_change: OnAdditiveChange = OnAdditiveChange.ALLOW
 390    auto_restatement_cron: t.Optional[SQLGlotCron] = None
 391
 392    _on_destructive_change_validator = on_destructive_change_validator
 393    _on_additive_change_validator = on_additive_change_validator
 394
 395    @property
 396    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 397        return [
 398            *super().metadata_hash_values,
 399            str(self.on_destructive_change),
 400            str(self.on_additive_change),
 401            self.auto_restatement_cron,
 402        ]
 403
 404    def to_expression(
 405        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 406    ) -> d.ModelKind:
 407        return super().to_expression(
 408            expressions=[
 409                *(expressions or []),
 410                *_properties(
 411                    {
 412                        "on_destructive_change": self.on_destructive_change.value,
 413                        "on_additive_change": self.on_additive_change.value,
 414                        "auto_restatement_cron": self.auto_restatement_cron,
 415                    }
 416                ),
 417            ],
 418        )
 419
 420
 421class _IncrementalBy(_Incremental):
 422    dialect: t.Optional[str] = Field(None, validate_default=True)
 423    batch_size: t.Optional[SQLGlotPositiveInt] = None
 424    batch_concurrency: t.Optional[SQLGlotPositiveInt] = None
 425    lookback: t.Optional[SQLGlotPositiveInt] = None
 426    forward_only: SQLGlotBool = False
 427    disable_restatement: SQLGlotBool = False
 428
 429    _dialect_validator = kind_dialect_validator
 430
 431    @property
 432    def data_hash_values(self) -> t.List[t.Optional[str]]:
 433        return [
 434            *super().data_hash_values,
 435            self.dialect,
 436            str(self.lookback) if self.lookback is not None else None,
 437        ]
 438
 439    @property
 440    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 441        return [
 442            *super().metadata_hash_values,
 443            str(self.batch_size) if self.batch_size is not None else None,
 444            str(self.forward_only),
 445            str(self.disable_restatement),
 446        ]
 447
 448    def to_expression(
 449        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 450    ) -> d.ModelKind:
 451        return super().to_expression(
 452            expressions=[
 453                *(expressions or []),
 454                *_properties(
 455                    {
 456                        "batch_size": self.batch_size,
 457                        "batch_concurrency": self.batch_concurrency,
 458                        "lookback": self.lookback,
 459                        "forward_only": self.forward_only,
 460                        "disable_restatement": self.disable_restatement,
 461                    }
 462                ),
 463            ],
 464        )
 465
 466
 467class IncrementalByTimeRangeKind(_IncrementalBy):
 468    name: t.Literal[ModelKindName.INCREMENTAL_BY_TIME_RANGE] = (
 469        ModelKindName.INCREMENTAL_BY_TIME_RANGE
 470    )
 471    time_column: TimeColumn
 472    auto_restatement_intervals: t.Optional[SQLGlotPositiveInt] = None
 473    partition_by_time_column: SQLGlotBool = True
 474
 475    _time_column_validator = TimeColumn.validator()
 476
 477    def to_expression(
 478        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 479    ) -> d.ModelKind:
 480        return super().to_expression(
 481            expressions=[
 482                *(expressions or []),
 483                self.time_column.to_property(kwargs.get("dialect") or ""),
 484                *_properties(
 485                    {
 486                        "partition_by_time_column": self.partition_by_time_column,
 487                    }
 488                ),
 489                *(
 490                    [_property("auto_restatement_intervals", self.auto_restatement_intervals)]
 491                    if self.auto_restatement_intervals is not None
 492                    else []
 493                ),
 494            ]
 495        )
 496
 497    @property
 498    def data_hash_values(self) -> t.List[t.Optional[str]]:
 499        return [*super().data_hash_values, gen(self.time_column.column), self.time_column.format]
 500
 501    @property
 502    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 503        return [
 504            *super().metadata_hash_values,
 505            str(self.partition_by_time_column),
 506            str(self.auto_restatement_intervals)
 507            if self.auto_restatement_intervals is not None
 508            else None,
 509        ]
 510
 511
 512class IncrementalByUniqueKeyKind(_IncrementalBy):
 513    name: t.Literal[ModelKindName.INCREMENTAL_BY_UNIQUE_KEY] = (
 514        ModelKindName.INCREMENTAL_BY_UNIQUE_KEY
 515    )
 516    unique_key: SQLGlotListOfFields
 517    when_matched: t.Optional[exp.Whens] = None
 518    merge_filter: t.Optional[exp.Expr] = None
 519    batch_concurrency: t.Literal[1] = 1
 520
 521    @field_validator("when_matched", mode="before")
 522    def _when_matched_validator(
 523        cls,
 524        v: t.Optional[t.Union[str, list, exp.Whens]],
 525        info: ValidationInfo,
 526    ) -> t.Optional[exp.Whens]:
 527        if v is None:
 528            return v
 529        if isinstance(v, list):
 530            v = " ".join(v)
 531
 532        dialect = get_dialect(info.data)
 533
 534        if isinstance(v, str):
 535            # Whens wrap the WHEN clauses, but the parentheses aren't parsed by sqlglot
 536            v = v.strip()
 537            if v.startswith("("):
 538                v = v[1:-1]
 539
 540            v = t.cast(exp.Whens, d.parse_one(v, into=exp.Whens, dialect=dialect))
 541
 542        v = validate_expression(v, dialect=dialect)
 543        return t.cast(exp.Whens, v.transform(d.replace_merge_table_aliases, dialect=dialect))
 544
 545    @field_validator("merge_filter", mode="before")
 546    def _merge_filter_validator(
 547        cls,
 548        v: t.Optional[exp.Expr],
 549        info: ValidationInfo,
 550    ) -> t.Optional[exp.Expr]:
 551        if v is None:
 552            return v
 553
 554        dialect = get_dialect(info.data)
 555
 556        if isinstance(v, str):
 557            v = v.strip()
 558            v = d.parse_one(v, dialect=dialect)
 559
 560        v = validate_expression(v, dialect=dialect)
 561        return v.transform(d.replace_merge_table_aliases, dialect=dialect)
 562
 563    @property
 564    def data_hash_values(self) -> t.List[t.Optional[str]]:
 565        return [
 566            *super().data_hash_values,
 567            *(gen(k) for k in self.unique_key),
 568            gen(self.when_matched) if self.when_matched is not None else None,
 569            gen(self.merge_filter) if self.merge_filter is not None else None,
 570        ]
 571
 572    def to_expression(
 573        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 574    ) -> d.ModelKind:
 575        return super().to_expression(
 576            expressions=[
 577                *(expressions or []),
 578                *_properties(
 579                    {
 580                        "unique_key": exp.Tuple(expressions=self.unique_key),
 581                        "when_matched": self.when_matched,
 582                        "merge_filter": self.merge_filter,
 583                    }
 584                ),
 585            ],
 586        )
 587
 588
 589class IncrementalByPartitionKind(_Incremental):
 590    name: t.Literal[ModelKindName.INCREMENTAL_BY_PARTITION] = ModelKindName.INCREMENTAL_BY_PARTITION
 591    forward_only: t.Literal[True] = True
 592    disable_restatement: SQLGlotBool = False
 593
 594    @field_validator("forward_only", mode="before")
 595    def _forward_only_validator(cls, v: t.Union[bool, exp.Expr]) -> t.Literal[True]:
 596        if v is not True:
 597            raise ConfigError(
 598                "Do not specify the `forward_only` configuration key - INCREMENTAL_BY_PARTITION models are always forward_only."
 599            )
 600        return v
 601
 602    @property
 603    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 604        return [
 605            *super().metadata_hash_values,
 606            str(self.forward_only),
 607            str(self.disable_restatement),
 608        ]
 609
 610    def to_expression(
 611        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 612    ) -> d.ModelKind:
 613        return super().to_expression(
 614            expressions=[
 615                *(expressions or []),
 616                *_properties(
 617                    {
 618                        "forward_only": self.forward_only,
 619                        "disable_restatement": self.disable_restatement,
 620                    }
 621                ),
 622            ],
 623        )
 624
 625
 626class IncrementalUnmanagedKind(_Incremental):
 627    name: t.Literal[ModelKindName.INCREMENTAL_UNMANAGED] = ModelKindName.INCREMENTAL_UNMANAGED
 628    insert_overwrite: SQLGlotBool = False
 629    forward_only: SQLGlotBool = True
 630    disable_restatement: SQLGlotBool = True
 631
 632    @property
 633    def data_hash_values(self) -> t.List[t.Optional[str]]:
 634        return [*super().data_hash_values, str(self.insert_overwrite)]
 635
 636    @property
 637    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 638        return [
 639            *super().metadata_hash_values,
 640            str(self.forward_only),
 641            str(self.disable_restatement),
 642        ]
 643
 644    def to_expression(
 645        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 646    ) -> d.ModelKind:
 647        return super().to_expression(
 648            expressions=[
 649                *(expressions or []),
 650                *_properties(
 651                    {
 652                        "insert_overwrite": self.insert_overwrite,
 653                        "forward_only": self.forward_only,
 654                        "disable_restatement": self.disable_restatement,
 655                    }
 656                ),
 657            ],
 658        )
 659
 660
 661class ViewKind(_ModelKind):
 662    name: t.Literal[ModelKindName.VIEW] = ModelKindName.VIEW
 663    materialized: SQLGlotBool = False
 664
 665    @property
 666    def data_hash_values(self) -> t.List[t.Optional[str]]:
 667        return [*super().data_hash_values, str(self.materialized)]
 668
 669    @property
 670    def supports_python_models(self) -> bool:
 671        return False
 672
 673    def to_expression(
 674        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 675    ) -> d.ModelKind:
 676        return super().to_expression(
 677            expressions=[
 678                *(expressions or []),
 679                _property("materialized", self.materialized),
 680            ],
 681        )
 682
 683
 684class SeedKind(_ModelKind):
 685    name: t.Literal[ModelKindName.SEED] = ModelKindName.SEED
 686    path: SQLGlotString
 687    batch_size: SQLGlotPositiveInt = 1000
 688    csv_settings: t.Optional[CsvSettings] = None
 689
 690    @field_validator("csv_settings", mode="before")
 691    @classmethod
 692    def _parse_csv_settings(cls, v: t.Any) -> t.Optional[CsvSettings]:
 693        if v is None or isinstance(v, CsvSettings):
 694            return v
 695        if isinstance(v, exp.Expr):
 696            tuple_exp = parse_properties(cls, v, None)
 697            if not tuple_exp:
 698                return None
 699            return CsvSettings(**{e.left.name: e.right for e in tuple_exp.expressions})
 700        if isinstance(v, dict):
 701            return CsvSettings(**v)
 702        return v
 703
 704    def to_expression(
 705        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 706    ) -> d.ModelKind:
 707        """Convert the seed kind into a SQLGlot expression."""
 708        return super().to_expression(
 709            expressions=[
 710                *(expressions or []),
 711                *_properties(
 712                    {
 713                        "path": exp.Literal.string(self.path),
 714                        "batch_size": self.batch_size,
 715                    }
 716                ),
 717            ],
 718        )
 719
 720    @property
 721    def data_hash_values(self) -> t.List[t.Optional[str]]:
 722        csv_setting_values = (self.csv_settings or CsvSettings()).dict().values()
 723        return [
 724            *super().data_hash_values,
 725            *(v if isinstance(v, (str, type(None))) else str(v) for v in csv_setting_values),
 726        ]
 727
 728    @property
 729    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 730        return [*super().metadata_hash_values, str(self.batch_size)]
 731
 732    @property
 733    def supports_python_models(self) -> bool:
 734        return False
 735
 736
 737class FullKind(_ModelKind):
 738    name: t.Literal[ModelKindName.FULL] = ModelKindName.FULL
 739
 740
 741class _SCDType2Kind(_Incremental):
 742    dialect: t.Optional[str] = Field(None, validate_default=True)
 743    unique_key: SQLGlotListOfFields
 744    valid_from_name: SQLGlotColumn = Field(exp.column("valid_from"), validate_default=True)
 745    valid_to_name: SQLGlotColumn = Field(exp.column("valid_to"), validate_default=True)
 746    invalidate_hard_deletes: SQLGlotBool = False
 747    time_data_type: exp.DataType = Field(exp.DataType.build("TIMESTAMP"), validate_default=True)
 748    batch_size: t.Optional[SQLGlotPositiveInt] = None
 749
 750    forward_only: SQLGlotBool = True
 751    disable_restatement: SQLGlotBool = True
 752
 753    _dialect_validator = kind_dialect_validator
 754
 755    _always_validate_column = field_validator("valid_from_name", "valid_to_name", mode="before")(
 756        column_validator
 757    )
 758
 759    @field_validator("time_data_type", mode="before")
 760    @classmethod
 761    def _time_data_type_validator(cls, v: t.Union[str, exp.Expr], values: t.Any) -> exp.Expr:
 762        if isinstance(v, exp.Expr) and not isinstance(v, exp.DataType):
 763            v = v.name
 764        dialect = get_dialect(values)
 765        data_type = exp.DataType.build(v, dialect=dialect)
 766        # Clear meta["sql"] (set by our parser extension) so the pydantic encoder
 767        # uses dialect-aware rendering: e.sql(dialect=meta["dialect"]). Without this,
 768        # the raw SQL text takes priority, which can be wrong for dialect-normalized
 769        # types (e.g., default "TIMESTAMP" should render as "DATETIME" in BigQuery).
 770        data_type.meta.pop("sql", None)
 771        data_type.meta["dialect"] = dialect
 772        return data_type
 773
 774    @property
 775    def managed_columns(self) -> t.Dict[str, exp.DataType]:
 776        return {
 777            self.valid_from_name.name: self.time_data_type,
 778            self.valid_to_name.name: self.time_data_type,
 779        }
 780
 781    @property
 782    def data_hash_values(self) -> t.List[t.Optional[str]]:
 783        return [
 784            *super().data_hash_values,
 785            self.dialect,
 786            *(gen(k) for k in self.unique_key),
 787            gen(self.valid_from_name),
 788            gen(self.valid_to_name),
 789            str(self.invalidate_hard_deletes),
 790            self.time_data_type.sql(self.dialect),
 791            str(self.batch_size) if self.batch_size is not None else None,
 792        ]
 793
 794    @property
 795    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 796        return [
 797            *super().metadata_hash_values,
 798            str(self.forward_only),
 799            str(self.disable_restatement),
 800        ]
 801
 802    def to_expression(
 803        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 804    ) -> d.ModelKind:
 805        return super().to_expression(
 806            expressions=[
 807                *(expressions or []),
 808                *_properties(
 809                    {
 810                        "unique_key": exp.Tuple(expressions=self.unique_key),
 811                        "valid_from_name": self.valid_from_name,
 812                        "valid_to_name": self.valid_to_name,
 813                        "invalidate_hard_deletes": self.invalidate_hard_deletes,
 814                        "time_data_type": self.time_data_type,
 815                        "forward_only": self.forward_only,
 816                        "disable_restatement": self.disable_restatement,
 817                    }
 818                ),
 819            ],
 820        )
 821
 822
 823class SCDType2ByTimeKind(_SCDType2Kind):
 824    name: t.Literal[ModelKindName.SCD_TYPE_2, ModelKindName.SCD_TYPE_2_BY_TIME] = (
 825        ModelKindName.SCD_TYPE_2_BY_TIME
 826    )
 827    updated_at_name: SQLGlotColumn = Field(exp.column("updated_at"), validate_default=True)
 828    updated_at_as_valid_from: SQLGlotBool = False
 829
 830    _always_validate_updated_at = field_validator("updated_at_name", mode="before")(
 831        column_validator
 832    )
 833
 834    @property
 835    def data_hash_values(self) -> t.List[t.Optional[str]]:
 836        return [
 837            *super().data_hash_values,
 838            gen(self.updated_at_name),
 839            str(self.updated_at_as_valid_from),
 840        ]
 841
 842    def to_expression(
 843        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 844    ) -> d.ModelKind:
 845        return super().to_expression(
 846            expressions=[
 847                *(expressions or []),
 848                *_properties(
 849                    {
 850                        "updated_at_name": self.updated_at_name,
 851                        "updated_at_as_valid_from": self.updated_at_as_valid_from,
 852                    }
 853                ),
 854            ],
 855        )
 856
 857
 858class SCDType2ByColumnKind(_SCDType2Kind):
 859    name: t.Literal[ModelKindName.SCD_TYPE_2_BY_COLUMN] = ModelKindName.SCD_TYPE_2_BY_COLUMN
 860    columns: SQLGlotListOfFieldsOrStar
 861    execution_time_as_valid_from: SQLGlotBool = False
 862    updated_at_name: t.Optional[SQLGlotColumn] = None
 863
 864    @property
 865    def data_hash_values(self) -> t.List[t.Optional[str]]:
 866        columns_sql = (
 867            [gen(c) for c in self.columns]
 868            if isinstance(self.columns, list)
 869            else [gen(self.columns)]
 870        )
 871        return [
 872            *super().data_hash_values,
 873            *columns_sql,
 874            str(self.execution_time_as_valid_from),
 875            gen(self.updated_at_name) if self.updated_at_name is not None else None,
 876        ]
 877
 878    def to_expression(
 879        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 880    ) -> d.ModelKind:
 881        return super().to_expression(
 882            expressions=[
 883                *(expressions or []),
 884                *_properties(
 885                    {
 886                        "columns": exp.Tuple(expressions=self.columns)
 887                        if isinstance(self.columns, list)
 888                        else self.columns,
 889                        "execution_time_as_valid_from": self.execution_time_as_valid_from,
 890                    }
 891                ),
 892            ],
 893        )
 894
 895
 896class ManagedKind(_ModelKind):
 897    name: t.Literal[ModelKindName.MANAGED] = ModelKindName.MANAGED
 898    disable_restatement: t.Literal[True] = True
 899
 900    @property
 901    def supports_python_models(self) -> bool:
 902        return False
 903
 904
 905class DbtCustomKind(_ModelKind):
 906    name: t.Literal[ModelKindName.DBT_CUSTOM] = ModelKindName.DBT_CUSTOM
 907    materialization: str
 908    adapter: str = "default"
 909    definition: str
 910    dialect: t.Optional[str] = Field(None, validate_default=True)
 911
 912    _dialect_validator = kind_dialect_validator
 913
 914    @field_validator("materialization", "adapter", "definition", mode="before")
 915    @classmethod
 916    def _validate_fields(cls, v: t.Any) -> str:
 917        return validate_string(v)
 918
 919    @property
 920    def data_hash_values(self) -> t.List[t.Optional[str]]:
 921        return [
 922            *super().data_hash_values,
 923            self.materialization,
 924            self.definition,
 925            self.adapter,
 926            self.dialect,
 927        ]
 928
 929    def to_expression(
 930        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
 931    ) -> d.ModelKind:
 932        return super().to_expression(
 933            expressions=[
 934                *(expressions or []),
 935                *_properties(
 936                    {
 937                        "materialization": exp.Literal.string(self.materialization),
 938                        "adapter": exp.Literal.string(self.adapter),
 939                    }
 940                ),
 941            ],
 942        )
 943
 944
 945class EmbeddedKind(_ModelKind):
 946    name: t.Literal[ModelKindName.EMBEDDED] = ModelKindName.EMBEDDED
 947
 948    @property
 949    def supports_python_models(self) -> bool:
 950        return False
 951
 952
 953class ExternalKind(_ModelKind):
 954    name: t.Literal[ModelKindName.EXTERNAL] = ModelKindName.EXTERNAL
 955
 956
 957class CustomKind(_ModelKind):
 958    name: t.Literal[ModelKindName.CUSTOM] = ModelKindName.CUSTOM
 959    materialization: str
 960    materialization_properties_: t.Optional[exp.Tuple] = Field(
 961        default=None, alias="materialization_properties"
 962    )
 963    forward_only: SQLGlotBool = False
 964    disable_restatement: SQLGlotBool = False
 965    batch_size: t.Optional[SQLGlotPositiveInt] = None
 966    batch_concurrency: t.Optional[SQLGlotPositiveInt] = None
 967    lookback: t.Optional[SQLGlotPositiveInt] = None
 968    auto_restatement_cron: t.Optional[SQLGlotCron] = None
 969    auto_restatement_intervals: t.Optional[SQLGlotPositiveInt] = None
 970
 971    # so that CustomKind subclasses know the dialect when validating / normalizing / interpreting values in `materialization_properties`
 972    dialect: str = Field(exclude=True)
 973
 974    _properties_validator = properties_validator
 975
 976    @field_validator("materialization", mode="before")
 977    @classmethod
 978    def _validate_materialization(cls, v: t.Any) -> str:
 979        # note: create_model_kind() validates the custom materialization class
 980        return validate_string(v)
 981
 982    @property
 983    def materialization_properties(self) -> CustomMaterializationProperties:
 984        """A dictionary of materialization properties."""
 985        if not self.materialization_properties_:
 986            return {}
 987        return d.interpret_key_value_pairs(self.materialization_properties_)
 988
 989    @property
 990    def data_hash_values(self) -> t.List[t.Optional[str]]:
 991        return [
 992            *super().data_hash_values,
 993            self.materialization,
 994            gen(self.materialization_properties_) if self.materialization_properties_ else None,
 995            str(self.lookback) if self.lookback is not None else None,
 996        ]
 997
 998    @property
 999    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
1000        return [
1001            *super().metadata_hash_values,
1002            str(self.batch_size) if self.batch_size is not None else None,
1003            str(self.batch_concurrency) if self.batch_concurrency is not None else None,
1004            str(self.forward_only),
1005            str(self.disable_restatement),
1006            self.auto_restatement_cron,
1007            str(self.auto_restatement_intervals)
1008            if self.auto_restatement_intervals is not None
1009            else None,
1010        ]
1011
1012    def to_expression(
1013        self, expressions: t.Optional[t.List[exp.Expr]] = None, **kwargs: t.Any
1014    ) -> d.ModelKind:
1015        return super().to_expression(
1016            expressions=[
1017                *(expressions or []),
1018                *_properties(
1019                    {
1020                        "materialization": exp.Literal.string(self.materialization),
1021                        "materialization_properties": self.materialization_properties_,
1022                        "forward_only": self.forward_only,
1023                        "disable_restatement": self.disable_restatement,
1024                        "batch_size": self.batch_size,
1025                        "batch_concurrency": self.batch_concurrency,
1026                        "lookback": self.lookback,
1027                        "auto_restatement_cron": self.auto_restatement_cron,
1028                        "auto_restatement_intervals": self.auto_restatement_intervals,
1029                    }
1030                ),
1031            ],
1032        )
1033
1034
1035ModelKind = t.Annotated[
1036    t.Union[
1037        EmbeddedKind,
1038        ExternalKind,
1039        FullKind,
1040        IncrementalByTimeRangeKind,
1041        IncrementalByUniqueKeyKind,
1042        IncrementalByPartitionKind,
1043        IncrementalUnmanagedKind,
1044        SeedKind,
1045        ViewKind,
1046        SCDType2ByTimeKind,
1047        SCDType2ByColumnKind,
1048        CustomKind,
1049        ManagedKind,
1050        DbtCustomKind,
1051    ],
1052    Field(discriminator="name"),
1053]
1054
1055MODEL_KIND_NAME_TO_TYPE: t.Dict[str, t.Type[ModelKind]] = {
1056    ModelKindName.EMBEDDED: EmbeddedKind,
1057    ModelKindName.EXTERNAL: ExternalKind,
1058    ModelKindName.FULL: FullKind,
1059    ModelKindName.INCREMENTAL_BY_TIME_RANGE: IncrementalByTimeRangeKind,
1060    ModelKindName.INCREMENTAL_BY_UNIQUE_KEY: IncrementalByUniqueKeyKind,
1061    ModelKindName.INCREMENTAL_BY_PARTITION: IncrementalByPartitionKind,
1062    ModelKindName.INCREMENTAL_UNMANAGED: IncrementalUnmanagedKind,
1063    ModelKindName.SEED: SeedKind,
1064    ModelKindName.VIEW: ViewKind,
1065    ModelKindName.SCD_TYPE_2: SCDType2ByTimeKind,
1066    ModelKindName.SCD_TYPE_2_BY_TIME: SCDType2ByTimeKind,
1067    ModelKindName.SCD_TYPE_2_BY_COLUMN: SCDType2ByColumnKind,
1068    ModelKindName.CUSTOM: CustomKind,
1069    ModelKindName.MANAGED: ManagedKind,
1070    ModelKindName.DBT_CUSTOM: DbtCustomKind,
1071}
1072
1073
1074def model_kind_type_from_name(name: t.Optional[str]) -> t.Type[ModelKind]:
1075    klass = MODEL_KIND_NAME_TO_TYPE.get(name) if name else None
1076    if not klass:
1077        raise ConfigError(f"Invalid model kind '{name}'")
1078    return t.cast(t.Type[ModelKind], klass)
1079
1080
1081def create_model_kind(v: t.Any, dialect: str, defaults: t.Dict[str, t.Any]) -> ModelKind:
1082    if isinstance(v, _ModelKind):
1083        return t.cast(ModelKind, v)
1084
1085    if isinstance(v, (d.ModelKind, dict)):
1086        props = (
1087            {prop.name: prop.args.get("value") for prop in v.expressions}
1088            if isinstance(v, d.ModelKind)
1089            else v
1090        )
1091        name = v.this if isinstance(v, d.ModelKind) else props.get("name")
1092
1093        # We want to ensure whatever name is provided to construct the class is the same name that will be
1094        # found inside the class itself in order to avoid a change during plan/apply for legacy aliases.
1095        # Ex: Pass in `SCD_TYPE_2` then we want to ensure we get `SCD_TYPE_2` as the kind name
1096        # instead of `SCD_TYPE_2_BY_TIME`.
1097        props["name"] = name
1098        kind_type = model_kind_type_from_name(name)
1099
1100        if "dialect" in kind_type.all_fields() and props.get("dialect") is None:
1101            props["dialect"] = dialect
1102
1103        # only pass the on_destructive_change or on_additive_change user default to models inheriting from _Incremental
1104        # that don't explicitly set it in the model definition
1105        if issubclass(kind_type, _Incremental):
1106            for on_change_property in ("on_additive_change", "on_destructive_change"):
1107                if (
1108                    props.get(on_change_property) is None
1109                    and defaults.get(on_change_property) is not None
1110                ):
1111                    props[on_change_property] = defaults.get(on_change_property)
1112
1113        # only pass the batch_concurrency user default to models inheriting from _IncrementalBy
1114        # that don't explicitly set it in the model definition, but ignore subclasses of _IncrementalBy
1115        # that hardcode a specific batch_concurrency
1116        if issubclass(kind_type, _IncrementalBy):
1117            BATCH_CONCURRENCY: t.Final = "batch_concurrency"
1118            if (
1119                props.get(BATCH_CONCURRENCY) is None
1120                and defaults.get(BATCH_CONCURRENCY) is not None
1121                and kind_type.all_field_infos()[BATCH_CONCURRENCY].default is None
1122            ):
1123                props[BATCH_CONCURRENCY] = defaults.get(BATCH_CONCURRENCY)
1124
1125        if kind_type == CustomKind:
1126            # load the custom materialization class and check if it uses a custom kind type
1127            from sqlmesh.core.snapshot.evaluator import get_custom_materialization_type
1128
1129            if "materialization" not in props:
1130                raise ConfigError(
1131                    "The 'materialization' property is required for models of the CUSTOM kind"
1132                )
1133
1134            # The below call will print a warning if a materialization with the given name doesn't exist
1135            # we dont want to throw an error here because we still want Models with a CustomKind to be able
1136            # to be serialized / deserialized in contexts where the custom materialization class may not be available,
1137            # such as in HTTP request handlers
1138            custom_materialization = get_custom_materialization_type(
1139                validate_string(props.get("materialization")), raise_errors=False
1140            )
1141            if custom_materialization is not None:
1142                actual_kind_type, _ = custom_materialization
1143                return actual_kind_type(**props)
1144
1145        validate_extra_and_required_fields(
1146            kind_type, set(props), f"MODEL block 'kind {name}' field"
1147        )
1148        return kind_type(**props)
1149
1150    name = (v.name if isinstance(v, exp.Expr) else str(v)).upper()
1151    return model_kind_type_from_name(name)(name=name)  # type: ignore
1152
1153
1154def _model_kind_validator(cls: t.Type, v: t.Any, info: t.Optional[ValidationInfo]) -> ModelKind:
1155    dialect = get_dialect(info.data) if info else ""
1156    return create_model_kind(v, dialect, {})
1157
1158
1159model_kind_validator: t.Callable = field_validator("kind", mode="before")(_model_kind_validator)
1160
1161
1162def _property(name: str, value: t.Any) -> exp.Property:
1163    return exp.Property(this=exp.var(name), value=exp.convert(value))
1164
1165
1166def _properties(name_value_pairs: t.Dict[str, t.Any]) -> t.List[exp.Property]:
1167    return [_property(k, v) for k, v in name_value_pairs.items() if v is not None]