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