sqlmesh.core.model.meta

  1from __future__ import annotations
  2
  3import typing as t
  4from enum import Enum
  5from functools import cached_property
  6from typing_extensions import Self
  7
  8from pydantic import Field
  9from sqlglot import Dialect, exp, parse_one
 10from sqlglot.helper import ensure_collection, ensure_list
 11from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
 12
 13from sqlmesh.core import dialect as d
 14from sqlmesh.core.config.common import VirtualEnvironmentMode
 15from sqlmesh.core.config.linter import LinterConfig
 16from sqlmesh.core.dialect import normalize_model_name
 17from sqlmesh.utils import classproperty
 18from sqlmesh.core.model.common import (
 19    bool_validator,
 20    default_catalog_validator,
 21    depends_on_validator,
 22    properties_validator,
 23    parse_properties,
 24)
 25from sqlmesh.core.model.kind import (
 26    CustomKind,
 27    IncrementalByUniqueKeyKind,
 28    ModelKind,
 29    OnDestructiveChange,
 30    SCDType2ByColumnKind,
 31    SCDType2ByTimeKind,
 32    TimeColumn,
 33    ViewKind,
 34    model_kind_validator,
 35    OnAdditiveChange,
 36)
 37from sqlmesh.core.node import _Node, str_or_exp_to_str
 38from sqlmesh.core.reference import Reference
 39from sqlmesh.utils.date import TimeLike
 40from sqlmesh.utils.errors import ConfigError
 41from sqlmesh.utils.pydantic import (
 42    ValidationInfo,
 43    field_validator,
 44    list_of_fields_validator,
 45    model_validator,
 46    get_dialect,
 47)
 48
 49if t.TYPE_CHECKING:
 50    from sqlmesh.core._typing import CustomMaterializationProperties, SessionProperties
 51    from sqlmesh.core.engine_adapter._typing import GrantsConfig
 52
 53FunctionCall = t.Tuple[str, t.Dict[str, exp.Expression]]
 54
 55
 56class GrantsTargetLayer(str, Enum):
 57    """Target layer(s) where grants should be applied."""
 58
 59    ALL = "all"
 60    PHYSICAL = "physical"
 61    VIRTUAL = "virtual"
 62
 63    @classproperty
 64    def default(cls) -> "GrantsTargetLayer":
 65        return GrantsTargetLayer.VIRTUAL
 66
 67    @property
 68    def is_all(self) -> bool:
 69        return self == GrantsTargetLayer.ALL
 70
 71    @property
 72    def is_physical(self) -> bool:
 73        return self == GrantsTargetLayer.PHYSICAL
 74
 75    @property
 76    def is_virtual(self) -> bool:
 77        return self == GrantsTargetLayer.VIRTUAL
 78
 79    def __str__(self) -> str:
 80        return self.name
 81
 82    def __repr__(self) -> str:
 83        return str(self)
 84
 85
 86class ModelMeta(_Node):
 87    """Metadata for models which can be defined in SQL."""
 88
 89    dialect: str = ""
 90    name: str
 91    kind: ModelKind = ViewKind()
 92    retention: t.Optional[int] = None  # not implemented yet
 93    table_format: t.Optional[str] = None
 94    storage_format: t.Optional[str] = None
 95    partitioned_by_: t.List[exp.Expression] = Field(default=[], alias="partitioned_by")
 96    clustered_by: t.List[exp.Expression] = []
 97    default_catalog: t.Optional[str] = None
 98    depends_on_: t.Optional[t.Set[str]] = Field(default=None, alias="depends_on")
 99    columns_to_types_: t.Optional[t.Dict[str, exp.DataType]] = Field(default=None, alias="columns")
100    column_descriptions_: t.Optional[t.Dict[str, str]] = Field(
101        default=None, alias="column_descriptions"
102    )
103    audits: t.List[FunctionCall] = []
104    grains: t.List[exp.Expression] = []
105    references: t.List[exp.Expression] = []
106    physical_schema_override: t.Optional[str] = None
107    physical_properties_: t.Optional[exp.Tuple] = Field(default=None, alias="physical_properties")
108    virtual_properties_: t.Optional[exp.Tuple] = Field(default=None, alias="virtual_properties")
109    session_properties_: t.Optional[exp.Tuple] = Field(default=None, alias="session_properties")
110    allow_partials: bool = False
111    signals: t.List[FunctionCall] = []
112    enabled: bool = True
113    physical_version: t.Optional[str] = None
114    gateway: t.Optional[str] = None
115    optimize_query: t.Optional[bool] = None
116    ignored_rules_: t.Optional[t.Set[str]] = Field(
117        default=None, exclude=True, alias="ignored_rules"
118    )
119    formatting: t.Optional[bool] = Field(default=None, exclude=True)
120    virtual_environment_mode: VirtualEnvironmentMode = VirtualEnvironmentMode.default
121    grants_: t.Optional[exp.Tuple] = Field(default=None, alias="grants")
122    grants_target_layer: GrantsTargetLayer = GrantsTargetLayer.default
123
124    _bool_validator = bool_validator
125    _model_kind_validator = model_kind_validator
126    _properties_validator = properties_validator
127    _default_catalog_validator = default_catalog_validator
128    _depends_on_validator = depends_on_validator
129
130    @field_validator("audits", "signals", mode="before")
131    def _func_call_validator(cls, v: t.Any, field: t.Any) -> t.Any:
132        is_signal = getattr(field, "name" if hasattr(field, "name") else "field_name") == "signals"
133
134        return d.extract_function_calls(v, allow_tuples=is_signal)
135
136    @field_validator("tags", mode="before")
137    def _value_or_tuple_validator(cls, v: t.Any, info: ValidationInfo) -> t.Any:
138        return ensure_list(cls._validate_value_or_tuple(v, info.data))
139
140    @classmethod
141    def _validate_value_or_tuple(
142        cls, v: t.Dict[str, t.Any], data: t.Dict[str, t.Any], normalize: bool = False
143    ) -> t.Any:
144        dialect = data.get("dialect")
145
146        def _normalize(value: t.Any) -> t.Any:
147            return normalize_identifiers(value, dialect=dialect) if normalize else value
148
149        if isinstance(v, exp.Paren):
150            v = [v.unnest()]
151
152        if isinstance(v, (exp.Tuple, exp.Array)):
153            return [_normalize(e).name for e in v.expressions]
154        if isinstance(v, exp.Expression):
155            return _normalize(v).name
156        if isinstance(v, str):
157            value = _normalize(v)
158            return value.name if isinstance(value, exp.Expression) else value
159        if isinstance(v, (list, tuple)):
160            return [cls._validate_value_or_tuple(elm, data, normalize=normalize) for elm in v]
161
162        return v
163
164    @field_validator("table_format", "storage_format", mode="before")
165    def _format_validator(cls, v: t.Any, info: ValidationInfo) -> t.Optional[str]:
166        if isinstance(v, exp.Expression) and not (isinstance(v, (exp.Literal, exp.Identifier))):
167            return v.sql(info.data.get("dialect"))
168        return str_or_exp_to_str(v)
169
170    @field_validator("dialect", mode="before")
171    def _dialect_validator(cls, v: t.Any) -> t.Optional[str]:
172        # dialects are parsed as identifiers and may get normalized as uppercase,
173        # so this ensures they'll be stored as lowercase
174        dialect = str_or_exp_to_str(v)
175        return dialect and dialect.lower()
176
177    @field_validator("physical_version", mode="before")
178    def _physical_version_validator(cls, v: t.Any) -> t.Optional[str]:
179        if v is None:
180            return v
181        return str_or_exp_to_str(v)
182
183    @field_validator("gateway", mode="before")
184    def _gateway_validator(cls, v: t.Any) -> t.Optional[str]:
185        if v is None:
186            return None
187        gateway = str_or_exp_to_str(v)
188        return gateway and gateway.lower()
189
190    @field_validator("partitioned_by_", "clustered_by", mode="before")
191    def _partition_and_cluster_validator(
192        cls, v: t.Any, info: ValidationInfo
193    ) -> t.List[exp.Expression]:
194        if (
195            isinstance(v, list)
196            and all(isinstance(i, str) for i in v)
197            and info.field_name == "partitioned_by_"
198        ):
199            # this branch gets hit when we are deserializing from json because `partitioned_by` is stored as a List[str]
200            # however, we should only invoke this if the list contains strings because this validator is also
201            # called by Python models which might pass a List[exp.Expression]
202            string_to_parse = (
203                f"({','.join(v)})"  # recreate the (a, b, c) part of "partitioned_by (a, b, c)"
204            )
205            parsed = parse_one(
206                string_to_parse, into=exp.PartitionedByProperty, dialect=get_dialect(info)
207            )
208            v = parsed.this.expressions if isinstance(parsed.this, exp.Schema) else v
209
210        expressions = list_of_fields_validator(v, info.data)
211
212        for expression in expressions:
213            num_cols = len(list(expression.find_all(exp.Column)))
214
215            error_msg: t.Optional[str] = None
216            if num_cols == 0:
217                error_msg = "does not contain a column"
218            elif num_cols > 1:
219                error_msg = "contains multiple columns"
220
221            if error_msg:
222                raise ConfigError(f"Field '{expression}' {error_msg}")
223
224        return expressions
225
226    @field_validator(
227        "columns_to_types_", "derived_columns_to_types", mode="before", check_fields=False
228    )
229    def _columns_validator(
230        cls, v: t.Any, info: ValidationInfo
231    ) -> t.Optional[t.Dict[str, exp.DataType]]:
232        columns_to_types = {}
233        dialect = info.data.get("dialect")
234
235        if isinstance(v, exp.Schema):
236            for column in v.expressions:
237                expr = column.args.get("kind")
238                if not isinstance(expr, exp.DataType):
239                    raise ConfigError(f"Missing data type for column '{column.name}'.")
240
241                expr.meta["dialect"] = dialect
242                columns_to_types[normalize_identifiers(column, dialect=dialect).name] = expr
243
244            return columns_to_types
245
246        if isinstance(v, dict):
247            udt = Dialect.get_or_raise(dialect).SUPPORTS_USER_DEFINED_TYPES
248            for k, data_type in v.items():
249                expr = exp.DataType.build(data_type, dialect=dialect, udt=udt)
250                expr.meta["dialect"] = dialect
251                columns_to_types[normalize_identifiers(k, dialect=dialect).name] = expr
252
253            return columns_to_types
254
255        return v
256
257    @field_validator("column_descriptions_", mode="before")
258    def _column_descriptions_validator(
259        cls, vs: t.Any, info: ValidationInfo
260    ) -> t.Optional[t.Dict[str, str]]:
261        dialect = info.data.get("dialect")
262
263        if vs is None:
264            return None
265
266        if isinstance(vs, exp.Paren):
267            vs = vs.flatten()
268
269        if isinstance(vs, (exp.Tuple, exp.Array)):
270            vs = vs.expressions
271
272        raw_col_descriptions = (
273            vs
274            if isinstance(vs, dict)
275            else {".".join([part.this for part in v.this.parts]): v.expression.name for v in vs}
276        )
277
278        col_descriptions = {
279            normalize_identifiers(k, dialect=dialect).name: v
280            for k, v in raw_col_descriptions.items()
281        }
282
283        columns_to_types = info.data.get("columns_to_types_")
284        if columns_to_types:
285            from sqlmesh.core.console import get_console
286
287            console = get_console()
288            for column_name in list(col_descriptions):
289                if column_name not in columns_to_types:
290                    console.log_warning(
291                        f"In model '{info.data['name']}', a description is provided for column '{column_name}' but it is not a column in the model."
292                    )
293                    del col_descriptions[column_name]
294
295        return col_descriptions
296
297    @field_validator("grains", "references", mode="before")
298    def _refs_validator(cls, vs: t.Any, info: ValidationInfo) -> t.List[exp.Expression]:
299        dialect = info.data.get("dialect")
300
301        if isinstance(vs, exp.Paren):
302            vs = vs.unnest()
303
304        if isinstance(vs, (exp.Tuple, exp.Array)):
305            vs = vs.expressions
306        else:
307            vs = [
308                d.parse_one(v, dialect=dialect) if isinstance(v, str) else v
309                for v in ensure_collection(vs)
310            ]
311
312        refs = []
313
314        for v in vs:
315            v = exp.column(v) if isinstance(v, exp.Identifier) else v
316            v.meta["dialect"] = dialect
317            refs.append(v)
318
319        return refs
320
321    @field_validator("ignored_rules_", mode="before")
322    def ignored_rules_validator(cls, vs: t.Any) -> t.Any:
323        return LinterConfig._validate_rules(vs)
324
325    @field_validator("grants_target_layer", mode="before")
326    def _grants_target_layer_validator(cls, v: t.Any) -> t.Any:
327        if isinstance(v, exp.Identifier):
328            return v.this
329        if isinstance(v, exp.Literal) and v.is_string:
330            return v.this
331        return v
332
333    @field_validator("session_properties_", mode="before")
334    def session_properties_validator(cls, v: t.Any, info: ValidationInfo) -> t.Any:
335        # use the generic properties validator to parse the session properties
336        parsed_session_properties = parse_properties(type(cls), v, info)
337        if not parsed_session_properties:
338            return parsed_session_properties
339
340        for eq in parsed_session_properties:
341            prop_name = eq.left.name
342
343            if prop_name == "query_label":
344                query_label = eq.right
345                if not isinstance(
346                    query_label, (exp.Array, exp.Tuple, exp.Paren, d.MacroFunc, d.MacroVar)
347                ):
348                    raise ConfigError(
349                        "Invalid value for `session_properties.query_label`. Must be an array or tuple."
350                    )
351
352                label_tuples: t.List[exp.Expression] = (
353                    [query_label.unnest()]
354                    if isinstance(query_label, exp.Paren)
355                    else query_label.expressions
356                )
357
358                for label_tuple in label_tuples:
359                    if not (
360                        isinstance(label_tuple, exp.Tuple)
361                        and len(label_tuple.expressions) == 2
362                        and all(isinstance(label, exp.Literal) for label in label_tuple.expressions)
363                    ):
364                        raise ConfigError(
365                            "Invalid entry in `session_properties.query_label`. Must be tuples of string literals with length 2."
366                        )
367            elif prop_name == "authorization":
368                authorization = eq.right
369                if not (
370                    isinstance(authorization, exp.Literal) and authorization.is_string
371                ) and not isinstance(authorization, (d.MacroFunc, d.MacroVar)):
372                    raise ConfigError(
373                        "Invalid value for `session_properties.authorization`. Must be a string literal."
374                    )
375
376        return parsed_session_properties
377
378    @model_validator(mode="before")
379    def _pre_root_validator(cls, data: t.Any) -> t.Any:
380        if not isinstance(data, dict):
381            return data
382
383        grain = data.pop("grain", None)
384        if grain:
385            grains = data.get("grains")
386            if grains:
387                raise ConfigError(
388                    f"Cannot use argument 'grain' ({grain}) with 'grains' ({grains}), use only grains"
389                )
390            data["grains"] = ensure_list(grain)
391
392        table_properties = data.pop("table_properties", None)
393        if table_properties:
394            if not isinstance(table_properties, str):
395                # Do not warn when deserializing from the state.
396                model_name = data["name"]
397                from sqlmesh.core.console import get_console
398
399                get_console().log_warning(
400                    f"Model '{model_name}' is using the `table_properties` attribute which is deprecated. Please use `physical_properties` instead."
401                )
402            physical_properties = data.get("physical_properties")
403            if physical_properties:
404                raise ConfigError(
405                    f"Cannot use argument 'table_properties' ({table_properties}) with 'physical_properties' ({physical_properties}), use only physical_properties."
406                )
407
408            data["physical_properties"] = table_properties
409
410        return data
411
412    @model_validator(mode="after")
413    def _root_validator(self) -> Self:
414        kind: t.Any = self.kind
415
416        for field in ("partitioned_by_", "clustered_by"):
417            if (
418                getattr(self, field, None)
419                and not kind.is_materialized
420                and not (kind.is_view and kind.materialized)
421            ):
422                name = field[:-1] if field.endswith("_") else field
423                raise ValueError(f"{name} field cannot be set for {kind.name} models")
424        if kind.is_incremental_by_partition and not getattr(self, "partitioned_by_", None):
425            raise ValueError(f"partitioned_by field is required for {kind.name} models")
426
427        # needs to be in a mode=after model validator so that the field validators have run to convert from Expression -> str
428        if (storage_format := self.storage_format) and storage_format.lower() in {
429            "iceberg",
430            "hive",
431            "hudi",
432            "delta",
433        }:
434            from sqlmesh.core.console import get_console
435
436            get_console().log_warning(
437                f"Model {self.name} has `storage_format` set to a table format '{storage_format}' which is deprecated. Please use the `table_format` property instead."
438            )
439
440        # Validate grants configuration for model kind support
441        if self.grants is not None and not kind.supports_grants:
442            raise ValueError(f"grants cannot be set for {kind.name} models")
443
444        return self
445
446    @property
447    def time_column(self) -> t.Optional[TimeColumn]:
448        """The time column for incremental models."""
449        return getattr(self.kind, "time_column", None)
450
451    @property
452    def unique_key(self) -> t.List[exp.Expression]:
453        if isinstance(
454            self.kind, (SCDType2ByTimeKind, SCDType2ByColumnKind, IncrementalByUniqueKeyKind)
455        ):
456            return self.kind.unique_key
457        return []
458
459    @property
460    def column_descriptions(self) -> t.Dict[str, str]:
461        """A dictionary of column names to annotation comments."""
462        return self.column_descriptions_ or {}
463
464    @property
465    def lookback(self) -> int:
466        """The incremental lookback window."""
467        return getattr(self.kind, "lookback", 0) or 0
468
469    def lookback_start(self, start: TimeLike) -> TimeLike:
470        if self.lookback == 0:
471            return start
472
473        for _ in range(self.lookback):
474            start = self.interval_unit.cron_prev(start)
475        return start
476
477    @property
478    def batch_size(self) -> t.Optional[int]:
479        """The maximal number of units in a single task for a backfill."""
480        return getattr(self.kind, "batch_size", None)
481
482    @property
483    def batch_concurrency(self) -> t.Optional[int]:
484        """The maximal number of batches that can run concurrently for a backfill."""
485        return getattr(self.kind, "batch_concurrency", None)
486
487    @cached_property
488    def physical_properties(self) -> t.Dict[str, exp.Expression]:
489        """A dictionary of properties that will be applied to the physical layer. It replaces table_properties which is deprecated."""
490        if self.physical_properties_:
491            return {e.this.name: e.expression for e in self.physical_properties_.expressions}
492        return {}
493
494    @cached_property
495    def virtual_properties(self) -> t.Dict[str, exp.Expression]:
496        """A dictionary of properties that will be applied to the virtual layer."""
497        if self.virtual_properties_:
498            return {e.this.name: e.expression for e in self.virtual_properties_.expressions}
499        return {}
500
501    @property
502    def session_properties(self) -> SessionProperties:
503        """A dictionary of session properties."""
504        if not self.session_properties_:
505            return {}
506
507        return d.interpret_key_value_pairs(self.session_properties_)
508
509    @property
510    def custom_materialization_properties(self) -> CustomMaterializationProperties:
511        if isinstance(self.kind, CustomKind):
512            return self.kind.materialization_properties
513        return {}
514
515    @cached_property
516    def grants(self) -> t.Optional[GrantsConfig]:
517        """A dictionary of grants mapping permission names to lists of grantees."""
518
519        if self.grants_ is None:
520            return None
521
522        if not self.grants_.expressions:
523            return {}
524
525        grants_dict = {}
526        for eq_expr in self.grants_.expressions:
527            try:
528                permission_name = self._validate_config_expression(eq_expr.left)
529                grantee_list = self._validate_nested_config_values(eq_expr.expression)
530                grants_dict[permission_name] = grantee_list
531            except ConfigError as e:
532                permission_name = (
533                    eq_expr.left.name if hasattr(eq_expr.left, "name") else str(eq_expr.left)
534                )
535                raise ConfigError(f"Invalid grants configuration for '{permission_name}': {e}")
536
537        return grants_dict if grants_dict else None
538
539    @property
540    def all_references(self) -> t.List[Reference]:
541        """All references including grains."""
542        return [Reference(model_name=self.name, expression=e, unique=True) for e in self.grains] + [
543            Reference(model_name=self.name, expression=e, unique=True) for e in self.references
544        ]
545
546    @property
547    def on(self) -> t.List[str]:
548        """The grains to be used as join condition in table_diff."""
549
550        on: t.List[str] = []
551        for expr in [ref.expression for ref in self.all_references if ref.unique]:
552            if isinstance(expr, exp.Tuple):
553                on.extend([key.this.sql(dialect=self.dialect) for key in expr.expressions])
554            else:
555                # Handle a single Column or Paren expression
556                on.append(expr.this.sql(dialect=self.dialect))
557
558        return on
559
560    @property
561    def managed_columns(self) -> t.Dict[str, exp.DataType]:
562        return getattr(self.kind, "managed_columns", {})
563
564    @property
565    def when_matched(self) -> t.Optional[exp.Whens]:
566        if isinstance(self.kind, IncrementalByUniqueKeyKind):
567            return self.kind.when_matched
568        return None
569
570    @property
571    def merge_filter(self) -> t.Optional[exp.Expression]:
572        if isinstance(self.kind, IncrementalByUniqueKeyKind):
573            return self.kind.merge_filter
574        return None
575
576    @property
577    def catalog(self) -> t.Optional[str]:
578        """Returns the catalog of a model."""
579        return self.fully_qualified_table.catalog
580
581    @cached_property
582    def fully_qualified_table(self) -> exp.Table:
583        return exp.to_table(self.fqn)
584
585    @cached_property
586    def fqn(self) -> str:
587        return normalize_model_name(
588            self.name, default_catalog=self.default_catalog, dialect=self.dialect
589        )
590
591    @property
592    def on_destructive_change(self) -> OnDestructiveChange:
593        return getattr(self.kind, "on_destructive_change", OnDestructiveChange.ALLOW)
594
595    @property
596    def on_additive_change(self) -> OnAdditiveChange:
597        """Return the model's additive change setting if it has one."""
598        return getattr(self.kind, "on_additive_change", OnAdditiveChange.ALLOW)
599
600    @property
601    def ignored_rules(self) -> t.Set[str]:
602        return self.ignored_rules_ or set()
603
604    def _validate_config_expression(self, expr: exp.Expression) -> str:
605        if isinstance(expr, (d.MacroFunc, d.MacroVar)):
606            raise ConfigError(f"Unresolved macro: {expr.sql(dialect=self.dialect)}")
607
608        if isinstance(expr, exp.Null):
609            raise ConfigError("NULL value")
610
611        if isinstance(expr, exp.Literal):
612            return str(expr.this).strip()
613        if isinstance(expr, (exp.Column, exp.Identifier)):
614            return expr.name
615        return expr.sql(dialect=self.dialect).strip()
616
617    def _validate_nested_config_values(self, value_expr: exp.Expression) -> t.List[str]:
618        result = []
619
620        def flatten_expr(expr: exp.Expression) -> None:
621            if isinstance(expr, exp.Array):
622                for elem in expr.expressions:
623                    flatten_expr(elem)
624            elif isinstance(expr, (exp.Tuple, exp.Paren)):
625                expressions = [expr.unnest()] if isinstance(expr, exp.Paren) else expr.expressions
626                for elem in expressions:
627                    flatten_expr(elem)
628            else:
629                result.append(self._validate_config_expression(expr))
630
631        flatten_expr(value_expr)
632        return result