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.Expression]] = 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.Expression
 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.Expression]) -> exp.Expression:
 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.Expression:
 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.Expression:
 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.Expression):
 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.Expression]] = 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.Expression]] = 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.Expression]] = 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.Expression] = 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.Expression],
 547        info: ValidationInfo,
 548    ) -> t.Optional[exp.Expression]:
 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.Expression]] = 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.Expression]) -> 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.Expression]] = 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.Expression]] = 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.Expression]] = 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.Expression):
 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.Expression]] = 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(
 760        cls, v: t.Union[str, exp.Expression], values: t.Any
 761    ) -> exp.Expression:
 762        if isinstance(v, exp.Expression) 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        data_type.meta["dialect"] = dialect
 767        return data_type
 768
 769    @property
 770    def managed_columns(self) -> t.Dict[str, exp.DataType]:
 771        return {
 772            self.valid_from_name.name: self.time_data_type,
 773            self.valid_to_name.name: self.time_data_type,
 774        }
 775
 776    @property
 777    def data_hash_values(self) -> t.List[t.Optional[str]]:
 778        return [
 779            *super().data_hash_values,
 780            self.dialect,
 781            *(gen(k) for k in self.unique_key),
 782            gen(self.valid_from_name),
 783            gen(self.valid_to_name),
 784            str(self.invalidate_hard_deletes),
 785            self.time_data_type.sql(self.dialect),
 786            gen(self.batch_size) if self.batch_size is not None else None,
 787        ]
 788
 789    @property
 790    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 791        return [
 792            *super().metadata_hash_values,
 793            str(self.forward_only),
 794            str(self.disable_restatement),
 795        ]
 796
 797    def to_expression(
 798        self, expressions: t.Optional[t.List[exp.Expression]] = None, **kwargs: t.Any
 799    ) -> d.ModelKind:
 800        return super().to_expression(
 801            expressions=[
 802                *(expressions or []),
 803                *_properties(
 804                    {
 805                        "unique_key": exp.Tuple(expressions=self.unique_key),
 806                        "valid_from_name": self.valid_from_name,
 807                        "valid_to_name": self.valid_to_name,
 808                        "invalidate_hard_deletes": self.invalidate_hard_deletes,
 809                        "time_data_type": self.time_data_type,
 810                        "forward_only": self.forward_only,
 811                        "disable_restatement": self.disable_restatement,
 812                    }
 813                ),
 814            ],
 815        )
 816
 817
 818class SCDType2ByTimeKind(_SCDType2Kind):
 819    name: t.Literal[ModelKindName.SCD_TYPE_2, ModelKindName.SCD_TYPE_2_BY_TIME] = (
 820        ModelKindName.SCD_TYPE_2_BY_TIME
 821    )
 822    updated_at_name: SQLGlotColumn = Field(exp.column("updated_at"), validate_default=True)
 823    updated_at_as_valid_from: SQLGlotBool = False
 824
 825    _always_validate_updated_at = field_validator("updated_at_name", mode="before")(
 826        column_validator
 827    )
 828
 829    @property
 830    def data_hash_values(self) -> t.List[t.Optional[str]]:
 831        return [
 832            *super().data_hash_values,
 833            gen(self.updated_at_name),
 834            str(self.updated_at_as_valid_from),
 835        ]
 836
 837    def to_expression(
 838        self, expressions: t.Optional[t.List[exp.Expression]] = None, **kwargs: t.Any
 839    ) -> d.ModelKind:
 840        return super().to_expression(
 841            expressions=[
 842                *(expressions or []),
 843                *_properties(
 844                    {
 845                        "updated_at_name": self.updated_at_name,
 846                        "updated_at_as_valid_from": self.updated_at_as_valid_from,
 847                    }
 848                ),
 849            ],
 850        )
 851
 852
 853class SCDType2ByColumnKind(_SCDType2Kind):
 854    name: t.Literal[ModelKindName.SCD_TYPE_2_BY_COLUMN] = ModelKindName.SCD_TYPE_2_BY_COLUMN
 855    columns: SQLGlotListOfFieldsOrStar
 856    execution_time_as_valid_from: SQLGlotBool = False
 857    updated_at_name: t.Optional[SQLGlotColumn] = None
 858
 859    @property
 860    def data_hash_values(self) -> t.List[t.Optional[str]]:
 861        columns_sql = (
 862            [gen(c) for c in self.columns]
 863            if isinstance(self.columns, list)
 864            else [gen(self.columns)]
 865        )
 866        return [
 867            *super().data_hash_values,
 868            *columns_sql,
 869            str(self.execution_time_as_valid_from),
 870            gen(self.updated_at_name) if self.updated_at_name is not None else None,
 871        ]
 872
 873    def to_expression(
 874        self, expressions: t.Optional[t.List[exp.Expression]] = None, **kwargs: t.Any
 875    ) -> d.ModelKind:
 876        return super().to_expression(
 877            expressions=[
 878                *(expressions or []),
 879                *_properties(
 880                    {
 881                        "columns": exp.Tuple(expressions=self.columns)
 882                        if isinstance(self.columns, list)
 883                        else self.columns,
 884                        "execution_time_as_valid_from": self.execution_time_as_valid_from,
 885                    }
 886                ),
 887            ],
 888        )
 889
 890
 891class ManagedKind(_ModelKind):
 892    name: t.Literal[ModelKindName.MANAGED] = ModelKindName.MANAGED
 893    disable_restatement: t.Literal[True] = True
 894
 895    @property
 896    def supports_python_models(self) -> bool:
 897        return False
 898
 899
 900class DbtCustomKind(_ModelKind):
 901    name: t.Literal[ModelKindName.DBT_CUSTOM] = ModelKindName.DBT_CUSTOM
 902    materialization: str
 903    adapter: str = "default"
 904    definition: str
 905    dialect: t.Optional[str] = Field(None, validate_default=True)
 906
 907    _dialect_validator = kind_dialect_validator
 908
 909    @field_validator("materialization", "adapter", "definition", mode="before")
 910    @classmethod
 911    def _validate_fields(cls, v: t.Any) -> str:
 912        return validate_string(v)
 913
 914    @property
 915    def data_hash_values(self) -> t.List[t.Optional[str]]:
 916        return [
 917            *super().data_hash_values,
 918            self.materialization,
 919            self.definition,
 920            self.adapter,
 921            self.dialect,
 922        ]
 923
 924    def to_expression(
 925        self, expressions: t.Optional[t.List[exp.Expression]] = None, **kwargs: t.Any
 926    ) -> d.ModelKind:
 927        return super().to_expression(
 928            expressions=[
 929                *(expressions or []),
 930                *_properties(
 931                    {
 932                        "materialization": exp.Literal.string(self.materialization),
 933                        "adapter": exp.Literal.string(self.adapter),
 934                    }
 935                ),
 936            ],
 937        )
 938
 939
 940class EmbeddedKind(_ModelKind):
 941    name: t.Literal[ModelKindName.EMBEDDED] = ModelKindName.EMBEDDED
 942
 943    @property
 944    def supports_python_models(self) -> bool:
 945        return False
 946
 947
 948class ExternalKind(_ModelKind):
 949    name: t.Literal[ModelKindName.EXTERNAL] = ModelKindName.EXTERNAL
 950
 951
 952class CustomKind(_ModelKind):
 953    name: t.Literal[ModelKindName.CUSTOM] = ModelKindName.CUSTOM
 954    materialization: str
 955    materialization_properties_: t.Optional[exp.Tuple] = Field(
 956        default=None, alias="materialization_properties"
 957    )
 958    forward_only: SQLGlotBool = False
 959    disable_restatement: SQLGlotBool = False
 960    batch_size: t.Optional[SQLGlotPositiveInt] = None
 961    batch_concurrency: t.Optional[SQLGlotPositiveInt] = None
 962    lookback: t.Optional[SQLGlotPositiveInt] = None
 963    auto_restatement_cron: t.Optional[SQLGlotCron] = None
 964    auto_restatement_intervals: t.Optional[SQLGlotPositiveInt] = None
 965
 966    # so that CustomKind subclasses know the dialect when validating / normalizing / interpreting values in `materialization_properties`
 967    dialect: str = Field(exclude=True)
 968
 969    _properties_validator = properties_validator
 970
 971    @field_validator("materialization", mode="before")
 972    @classmethod
 973    def _validate_materialization(cls, v: t.Any) -> str:
 974        # note: create_model_kind() validates the custom materialization class
 975        return validate_string(v)
 976
 977    @property
 978    def materialization_properties(self) -> CustomMaterializationProperties:
 979        """A dictionary of materialization properties."""
 980        if not self.materialization_properties_:
 981            return {}
 982        return d.interpret_key_value_pairs(self.materialization_properties_)
 983
 984    @property
 985    def data_hash_values(self) -> t.List[t.Optional[str]]:
 986        return [
 987            *super().data_hash_values,
 988            self.materialization,
 989            gen(self.materialization_properties_) if self.materialization_properties_ else None,
 990            str(self.lookback) if self.lookback is not None else None,
 991        ]
 992
 993    @property
 994    def metadata_hash_values(self) -> t.List[t.Optional[str]]:
 995        return [
 996            *super().metadata_hash_values,
 997            str(self.batch_size) if self.batch_size is not None else None,
 998            str(self.batch_concurrency) if self.batch_concurrency is not None else None,
 999            str(self.forward_only),
1000            str(self.disable_restatement),
1001            self.auto_restatement_cron,
1002            str(self.auto_restatement_intervals)
1003            if self.auto_restatement_intervals is not None
1004            else None,
1005        ]
1006
1007    def to_expression(
1008        self, expressions: t.Optional[t.List[exp.Expression]] = None, **kwargs: t.Any
1009    ) -> d.ModelKind:
1010        return super().to_expression(
1011            expressions=[
1012                *(expressions or []),
1013                *_properties(
1014                    {
1015                        "materialization": exp.Literal.string(self.materialization),
1016                        "materialization_properties": self.materialization_properties_,
1017                        "forward_only": self.forward_only,
1018                        "disable_restatement": self.disable_restatement,
1019                        "batch_size": self.batch_size,
1020                        "batch_concurrency": self.batch_concurrency,
1021                        "lookback": self.lookback,
1022                        "auto_restatement_cron": self.auto_restatement_cron,
1023                        "auto_restatement_intervals": self.auto_restatement_intervals,
1024                    }
1025                ),
1026            ],
1027        )
1028
1029
1030ModelKind = t.Annotated[
1031    t.Union[
1032        EmbeddedKind,
1033        ExternalKind,
1034        FullKind,
1035        IncrementalByTimeRangeKind,
1036        IncrementalByUniqueKeyKind,
1037        IncrementalByPartitionKind,
1038        IncrementalUnmanagedKind,
1039        SeedKind,
1040        ViewKind,
1041        SCDType2ByTimeKind,
1042        SCDType2ByColumnKind,
1043        CustomKind,
1044        ManagedKind,
1045        DbtCustomKind,
1046    ],
1047    Field(discriminator="name"),
1048]
1049
1050MODEL_KIND_NAME_TO_TYPE: t.Dict[str, t.Type[ModelKind]] = {
1051    ModelKindName.EMBEDDED: EmbeddedKind,
1052    ModelKindName.EXTERNAL: ExternalKind,
1053    ModelKindName.FULL: FullKind,
1054    ModelKindName.INCREMENTAL_BY_TIME_RANGE: IncrementalByTimeRangeKind,
1055    ModelKindName.INCREMENTAL_BY_UNIQUE_KEY: IncrementalByUniqueKeyKind,
1056    ModelKindName.INCREMENTAL_BY_PARTITION: IncrementalByPartitionKind,
1057    ModelKindName.INCREMENTAL_UNMANAGED: IncrementalUnmanagedKind,
1058    ModelKindName.SEED: SeedKind,
1059    ModelKindName.VIEW: ViewKind,
1060    ModelKindName.SCD_TYPE_2: SCDType2ByTimeKind,
1061    ModelKindName.SCD_TYPE_2_BY_TIME: SCDType2ByTimeKind,
1062    ModelKindName.SCD_TYPE_2_BY_COLUMN: SCDType2ByColumnKind,
1063    ModelKindName.CUSTOM: CustomKind,
1064    ModelKindName.MANAGED: ManagedKind,
1065    ModelKindName.DBT_CUSTOM: DbtCustomKind,
1066}
1067
1068
1069def model_kind_type_from_name(name: t.Optional[str]) -> t.Type[ModelKind]:
1070    klass = MODEL_KIND_NAME_TO_TYPE.get(name) if name else None
1071    if not klass:
1072        raise ConfigError(f"Invalid model kind '{name}'")
1073    return t.cast(t.Type[ModelKind], klass)
1074
1075
1076def create_model_kind(v: t.Any, dialect: str, defaults: t.Dict[str, t.Any]) -> ModelKind:
1077    if isinstance(v, _ModelKind):
1078        return t.cast(ModelKind, v)
1079
1080    if isinstance(v, (d.ModelKind, dict)):
1081        props = (
1082            {prop.name: prop.args.get("value") for prop in v.expressions}
1083            if isinstance(v, d.ModelKind)
1084            else v
1085        )
1086        name = v.this if isinstance(v, d.ModelKind) else props.get("name")
1087
1088        # We want to ensure whatever name is provided to construct the class is the same name that will be
1089        # found inside the class itself in order to avoid a change during plan/apply for legacy aliases.
1090        # Ex: Pass in `SCD_TYPE_2` then we want to ensure we get `SCD_TYPE_2` as the kind name
1091        # instead of `SCD_TYPE_2_BY_TIME`.
1092        props["name"] = name
1093        kind_type = model_kind_type_from_name(name)
1094
1095        if "dialect" in kind_type.all_fields() and props.get("dialect") is None:
1096            props["dialect"] = dialect
1097
1098        # only pass the on_destructive_change or on_additive_change user default to models inheriting from _Incremental
1099        # that don't explicitly set it in the model definition
1100        if issubclass(kind_type, _Incremental):
1101            for on_change_property in ("on_additive_change", "on_destructive_change"):
1102                if (
1103                    props.get(on_change_property) is None
1104                    and defaults.get(on_change_property) is not None
1105                ):
1106                    props[on_change_property] = defaults.get(on_change_property)
1107
1108        # only pass the batch_concurrency user default to models inheriting from _IncrementalBy
1109        # that don't explicitly set it in the model definition, but ignore subclasses of _IncrementalBy
1110        # that hardcode a specific batch_concurrency
1111        if issubclass(kind_type, _IncrementalBy):
1112            BATCH_CONCURRENCY: t.Final = "batch_concurrency"
1113            if (
1114                props.get(BATCH_CONCURRENCY) is None
1115                and defaults.get(BATCH_CONCURRENCY) is not None
1116                and kind_type.all_field_infos()[BATCH_CONCURRENCY].default is None
1117            ):
1118                props[BATCH_CONCURRENCY] = defaults.get(BATCH_CONCURRENCY)
1119
1120        if kind_type == CustomKind:
1121            # load the custom materialization class and check if it uses a custom kind type
1122            from sqlmesh.core.snapshot.evaluator import get_custom_materialization_type
1123
1124            if "materialization" not in props:
1125                raise ConfigError(
1126                    "The 'materialization' property is required for models of the CUSTOM kind"
1127                )
1128
1129            # The below call will print a warning if a materialization with the given name doesn't exist
1130            # we dont want to throw an error here because we still want Models with a CustomKind to be able
1131            # to be serialized / deserialized in contexts where the custom materialization class may not be available,
1132            # such as in HTTP request handlers
1133            custom_materialization = get_custom_materialization_type(
1134                validate_string(props.get("materialization")), raise_errors=False
1135            )
1136            if custom_materialization is not None:
1137                actual_kind_type, _ = custom_materialization
1138                return actual_kind_type(**props)
1139
1140        validate_extra_and_required_fields(
1141            kind_type, set(props), f"MODEL block 'kind {name}' field"
1142        )
1143        return kind_type(**props)
1144
1145    name = (v.name if isinstance(v, exp.Expression) else str(v)).upper()
1146    return model_kind_type_from_name(name)(name=name)  # type: ignore
1147
1148
1149def _model_kind_validator(cls: t.Type, v: t.Any, info: t.Optional[ValidationInfo]) -> ModelKind:
1150    dialect = get_dialect(info.data) if info else ""
1151    return create_model_kind(v, dialect, {})
1152
1153
1154model_kind_validator: t.Callable = field_validator("kind", mode="before")(_model_kind_validator)
1155
1156
1157def _property(name: str, value: t.Any) -> exp.Property:
1158    return exp.Property(this=exp.var(name), value=exp.convert(value))
1159
1160
1161def _properties(name_value_pairs: t.Dict[str, t.Any]) -> t.List[exp.Property]:
1162    return [_property(k, v) for k, v in name_value_pairs.items() if v is not None]