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.Expr]]
 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.Expr] = Field(default=[], alias="partitioned_by")
 96    clustered_by: t.List[exp.Expr] = []
 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.Expr] = []
105    references: t.List[exp.Expr] = []
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.Expr):
155            return _normalize(v).name
156        if isinstance(v, str):
157            value = _normalize(v)
158            return value.name if isinstance(value, exp.Expr) 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.Expr) 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(cls, v: t.Any, info: ValidationInfo) -> t.List[exp.Expr]:
192        if (
193            isinstance(v, list)
194            and all(isinstance(i, str) for i in v)
195            and info.field_name == "partitioned_by_"
196        ):
197            # this branch gets hit when we are deserializing from json because `partitioned_by` is stored as a List[str]
198            # however, we should only invoke this if the list contains strings because this validator is also
199            # called by Python models which might pass a List[exp.Expression]
200            string_to_parse = (
201                f"({','.join(v)})"  # recreate the (a, b, c) part of "partitioned_by (a, b, c)"
202            )
203            parsed = parse_one(
204                string_to_parse, into=exp.PartitionedByProperty, dialect=get_dialect(info)
205            )
206            v = parsed.this.expressions if isinstance(parsed.this, exp.Schema) else v
207
208        expressions = list_of_fields_validator(v, info.data)
209
210        for expression in expressions:
211            num_cols = len(list(expression.find_all(exp.Column)))
212
213            error_msg: t.Optional[str] = None
214            if num_cols == 0:
215                error_msg = "does not contain a column"
216            elif num_cols > 1:
217                error_msg = "contains multiple columns"
218
219            if error_msg:
220                raise ConfigError(f"Field '{expression}' {error_msg}")
221
222        return expressions
223
224    @field_validator(
225        "columns_to_types_", "derived_columns_to_types", mode="before", check_fields=False
226    )
227    def _columns_validator(
228        cls, v: t.Any, info: ValidationInfo
229    ) -> t.Optional[t.Dict[str, exp.DataType]]:
230        columns_to_types = {}
231        dialect = info.data.get("dialect")
232
233        if isinstance(v, exp.Schema):
234            for column in v.expressions:
235                expr = column.args.get("kind")
236                if not isinstance(expr, exp.DataType):
237                    raise ConfigError(f"Missing data type for column '{column.name}'.")
238
239                expr.meta["dialect"] = dialect
240                columns_to_types[normalize_identifiers(column, dialect=dialect).name] = expr
241
242            return columns_to_types
243
244        if isinstance(v, dict):
245            dialect_obj = Dialect.get_or_raise(dialect)
246            udt = dialect_obj.SUPPORTS_USER_DEFINED_TYPES
247            for k, data_type in v.items():
248                is_string_type = isinstance(data_type, str)
249                expr = exp.DataType.build(data_type, dialect=dialect, udt=udt)
250                # When deserializing from a string (e.g. JSON roundtrip), normalize the type
251                # through the dialect's type system so that aliases (e.g. INT in BigQuery,
252                # which is an alias for INT64/BIGINT) are resolved to their canonical form.
253                # This ensures stable data hash computation across serialization/deserialization
254                # roundtrips. We skip this for DataType objects passed directly (Python API)
255                # since those should be used as-is.
256                if (
257                    is_string_type
258                    and dialect
259                    and expr.this
260                    not in (
261                        exp.DataType.Type.USERDEFINED,
262                        exp.DataType.Type.UNKNOWN,
263                    )
264                ):
265                    sql_repr = expr.sql(dialect=dialect)
266                    try:
267                        normalized = parse_one(sql_repr, read=dialect, into=exp.DataType)
268                        if normalized is not None:
269                            expr = normalized
270                    except Exception:
271                        pass
272                expr.meta["dialect"] = dialect
273                columns_to_types[normalize_identifiers(k, dialect=dialect).name] = expr
274
275            return columns_to_types
276
277        return v
278
279    @field_validator("column_descriptions_", mode="before")
280    def _column_descriptions_validator(
281        cls, vs: t.Any, info: ValidationInfo
282    ) -> t.Optional[t.Dict[str, str]]:
283        dialect = info.data.get("dialect")
284
285        if vs is None:
286            return None
287
288        if isinstance(vs, exp.Paren):
289            vs = vs.flatten()
290
291        if isinstance(vs, (exp.Tuple, exp.Array)):
292            vs = vs.expressions
293
294        raw_col_descriptions = (
295            vs
296            if isinstance(vs, dict)
297            else {".".join([part.this for part in v.this.parts]): v.expression.name for v in vs}
298        )
299
300        col_descriptions = {
301            normalize_identifiers(k, dialect=dialect).name: v
302            for k, v in raw_col_descriptions.items()
303        }
304
305        columns_to_types = info.data.get("columns_to_types_")
306        if columns_to_types:
307            from sqlmesh.core.console import get_console
308
309            console = get_console()
310            for column_name in list(col_descriptions):
311                if column_name not in columns_to_types:
312                    console.log_warning(
313                        f"In model '{info.data['name']}', a description is provided for column '{column_name}' but it is not a column in the model."
314                    )
315                    del col_descriptions[column_name]
316
317        return col_descriptions
318
319    @field_validator("grains", "references", mode="before")
320    def _refs_validator(cls, vs: t.Any, info: ValidationInfo) -> t.List[exp.Expr]:
321        dialect = info.data.get("dialect")
322
323        if isinstance(vs, exp.Paren):
324            vs = vs.unnest()
325
326        if isinstance(vs, (exp.Tuple, exp.Array)):
327            vs = vs.expressions
328        else:
329            vs = [
330                d.parse_one(v, dialect=dialect) if isinstance(v, str) else v
331                for v in ensure_collection(vs)
332            ]
333
334        refs = []
335
336        for v in vs:
337            v = exp.column(v) if isinstance(v, exp.Identifier) else v
338            v.meta["dialect"] = dialect
339            refs.append(v)
340
341        return refs
342
343    @field_validator("ignored_rules_", mode="before")
344    def ignored_rules_validator(cls, vs: t.Any) -> t.Any:
345        return LinterConfig._validate_rules(vs)
346
347    @field_validator("grants_target_layer", mode="before")
348    def _grants_target_layer_validator(cls, v: t.Any) -> t.Any:
349        if isinstance(v, exp.Identifier):
350            return v.this
351        if isinstance(v, exp.Literal) and v.is_string:
352            return v.this
353        return v
354
355    @field_validator("session_properties_", mode="before")
356    def session_properties_validator(cls, v: t.Any, info: ValidationInfo) -> t.Any:
357        # use the generic properties validator to parse the session properties
358        parsed_session_properties = parse_properties(type(cls), v, info)
359        if not parsed_session_properties:
360            return parsed_session_properties
361
362        for eq in parsed_session_properties:
363            prop_name = eq.left.name
364
365            if prop_name == "query_label":
366                query_label = eq.right
367                if not isinstance(
368                    query_label, (exp.Array, exp.Tuple, exp.Paren, d.MacroFunc, d.MacroVar)
369                ):
370                    raise ConfigError(
371                        "Invalid value for `session_properties.query_label`. Must be an array or tuple."
372                    )
373
374                label_tuples: t.List[exp.Expr] = (
375                    [query_label.unnest()]
376                    if isinstance(query_label, exp.Paren)
377                    else query_label.expressions
378                )
379
380                for label_tuple in label_tuples:
381                    if not (
382                        isinstance(label_tuple, exp.Tuple)
383                        and len(label_tuple.expressions) == 2
384                        and all(isinstance(label, exp.Literal) for label in label_tuple.expressions)
385                    ):
386                        raise ConfigError(
387                            "Invalid entry in `session_properties.query_label`. Must be tuples of string literals with length 2."
388                        )
389            elif prop_name == "authorization":
390                authorization = eq.right
391                if not (
392                    isinstance(authorization, exp.Literal) and authorization.is_string
393                ) and not isinstance(authorization, (d.MacroFunc, d.MacroVar)):
394                    raise ConfigError(
395                        "Invalid value for `session_properties.authorization`. Must be a string literal."
396                    )
397
398        return parsed_session_properties
399
400    @model_validator(mode="before")
401    def _pre_root_validator(cls, data: t.Any) -> t.Any:
402        if not isinstance(data, dict):
403            return data
404
405        grain = data.pop("grain", None)
406        if grain:
407            grains = data.get("grains")
408            if grains:
409                raise ConfigError(
410                    f"Cannot use argument 'grain' ({grain}) with 'grains' ({grains}), use only grains"
411                )
412            data["grains"] = ensure_list(grain)
413
414        table_properties = data.pop("table_properties", None)
415        if table_properties:
416            if not isinstance(table_properties, str):
417                # Do not warn when deserializing from the state.
418                model_name = data["name"]
419                from sqlmesh.core.console import get_console
420
421                get_console().log_warning(
422                    f"Model '{model_name}' is using the `table_properties` attribute which is deprecated. Please use `physical_properties` instead."
423                )
424            physical_properties = data.get("physical_properties")
425            if physical_properties:
426                raise ConfigError(
427                    f"Cannot use argument 'table_properties' ({table_properties}) with 'physical_properties' ({physical_properties}), use only physical_properties."
428                )
429
430            data["physical_properties"] = table_properties
431
432        return data
433
434    @model_validator(mode="after")
435    def _root_validator(self) -> Self:
436        kind: t.Any = self.kind
437
438        for field in ("partitioned_by_", "clustered_by"):
439            if (
440                getattr(self, field, None)
441                and not kind.is_materialized
442                and not (kind.is_view and kind.materialized)
443            ):
444                name = field[:-1] if field.endswith("_") else field
445                raise ValueError(f"{name} field cannot be set for {kind.name} models")
446        if kind.is_incremental_by_partition and not getattr(self, "partitioned_by_", None):
447            raise ValueError(f"partitioned_by field is required for {kind.name} models")
448
449        # needs to be in a mode=after model validator so that the field validators have run to convert from Expression -> str
450        if (storage_format := self.storage_format) and storage_format.lower() in {
451            "iceberg",
452            "hive",
453            "hudi",
454            "delta",
455        }:
456            from sqlmesh.core.console import get_console
457
458            get_console().log_warning(
459                f"Model {self.name} has `storage_format` set to a table format '{storage_format}' which is deprecated. Please use the `table_format` property instead."
460            )
461
462        # Validate grants configuration for model kind support
463        if self.grants is not None and not kind.supports_grants:
464            raise ValueError(f"grants cannot be set for {kind.name} models")
465
466        return self
467
468    @property
469    def time_column(self) -> t.Optional[TimeColumn]:
470        """The time column for incremental models."""
471        return getattr(self.kind, "time_column", None)
472
473    @property
474    def unique_key(self) -> t.List[exp.Expr]:
475        if isinstance(
476            self.kind, (SCDType2ByTimeKind, SCDType2ByColumnKind, IncrementalByUniqueKeyKind)
477        ):
478            return self.kind.unique_key
479        return []
480
481    @property
482    def column_descriptions(self) -> t.Dict[str, str]:
483        """A dictionary of column names to annotation comments."""
484        return self.column_descriptions_ or {}
485
486    @property
487    def lookback(self) -> int:
488        """The incremental lookback window."""
489        return getattr(self.kind, "lookback", 0) or 0
490
491    def lookback_start(self, start: TimeLike) -> TimeLike:
492        if self.lookback == 0:
493            return start
494
495        for _ in range(self.lookback):
496            start = self.interval_unit.cron_prev(start)
497        return start
498
499    @property
500    def batch_size(self) -> t.Optional[int]:
501        """The maximal number of units in a single task for a backfill."""
502        return getattr(self.kind, "batch_size", None)
503
504    @property
505    def batch_concurrency(self) -> t.Optional[int]:
506        """The maximal number of batches that can run concurrently for a backfill."""
507        return getattr(self.kind, "batch_concurrency", None)
508
509    @cached_property
510    def physical_properties(self) -> t.Dict[str, exp.Expr]:
511        """A dictionary of properties that will be applied to the physical layer. It replaces table_properties which is deprecated."""
512        if self.physical_properties_:
513            return {e.this.name: e.expression for e in self.physical_properties_.expressions}
514        return {}
515
516    @cached_property
517    def virtual_properties(self) -> t.Dict[str, exp.Expr]:
518        """A dictionary of properties that will be applied to the virtual layer."""
519        if self.virtual_properties_:
520            return {e.this.name: e.expression for e in self.virtual_properties_.expressions}
521        return {}
522
523    @property
524    def session_properties(self) -> SessionProperties:
525        """A dictionary of session properties."""
526        if not self.session_properties_:
527            return {}
528
529        return d.interpret_key_value_pairs(self.session_properties_)
530
531    @property
532    def custom_materialization_properties(self) -> CustomMaterializationProperties:
533        if isinstance(self.kind, CustomKind):
534            return self.kind.materialization_properties
535        return {}
536
537    @cached_property
538    def grants(self) -> t.Optional[GrantsConfig]:
539        """A dictionary of grants mapping permission names to lists of grantees."""
540
541        if self.grants_ is None:
542            return None
543
544        if not self.grants_.expressions:
545            return {}
546
547        grants_dict = {}
548        for eq_expr in self.grants_.expressions:
549            try:
550                permission_name = self._validate_config_expression(eq_expr.left)
551                grantee_list = self._validate_nested_config_values(eq_expr.expression)
552                grants_dict[permission_name] = grantee_list
553            except ConfigError as e:
554                permission_name = (
555                    eq_expr.left.name if hasattr(eq_expr.left, "name") else str(eq_expr.left)
556                )
557                raise ConfigError(f"Invalid grants configuration for '{permission_name}': {e}")
558
559        return grants_dict if grants_dict else None
560
561    @property
562    def all_references(self) -> t.List[Reference]:
563        """All references including grains."""
564        return [Reference(model_name=self.name, expression=e, unique=True) for e in self.grains] + [
565            Reference(model_name=self.name, expression=e, unique=True) for e in self.references
566        ]
567
568    @property
569    def on(self) -> t.List[str]:
570        """The grains to be used as join condition in table_diff."""
571
572        on: t.List[str] = []
573        for expr in [ref.expression for ref in self.all_references if ref.unique]:
574            if isinstance(expr, exp.Tuple):
575                on.extend([key.this.sql(dialect=self.dialect) for key in expr.expressions])
576            else:
577                # Handle a single Column or Paren expression
578                on.append(expr.this.sql(dialect=self.dialect))
579
580        return on
581
582    @property
583    def managed_columns(self) -> t.Dict[str, exp.DataType]:
584        return getattr(self.kind, "managed_columns", {})
585
586    @property
587    def when_matched(self) -> t.Optional[exp.Whens]:
588        if isinstance(self.kind, IncrementalByUniqueKeyKind):
589            return self.kind.when_matched
590        return None
591
592    @property
593    def merge_filter(self) -> t.Optional[exp.Expr]:
594        if isinstance(self.kind, IncrementalByUniqueKeyKind):
595            return self.kind.merge_filter
596        return None
597
598    @property
599    def catalog(self) -> t.Optional[str]:
600        """Returns the catalog of a model."""
601        return self.fully_qualified_table.catalog
602
603    @cached_property
604    def fully_qualified_table(self) -> exp.Table:
605        return exp.to_table(self.fqn)
606
607    @cached_property
608    def fqn(self) -> str:
609        return normalize_model_name(
610            self.name, default_catalog=self.default_catalog, dialect=self.dialect
611        )
612
613    @property
614    def on_destructive_change(self) -> OnDestructiveChange:
615        return getattr(self.kind, "on_destructive_change", OnDestructiveChange.ALLOW)
616
617    @property
618    def on_additive_change(self) -> OnAdditiveChange:
619        """Return the model's additive change setting if it has one."""
620        return getattr(self.kind, "on_additive_change", OnAdditiveChange.ALLOW)
621
622    @property
623    def ignored_rules(self) -> t.Set[str]:
624        return self.ignored_rules_ or set()
625
626    def _validate_config_expression(self, expr: exp.Expr) -> str:
627        if isinstance(expr, (d.MacroFunc, d.MacroVar)):
628            raise ConfigError(f"Unresolved macro: {expr.sql(dialect=self.dialect)}")
629
630        if isinstance(expr, exp.Null):
631            raise ConfigError("NULL value")
632
633        if isinstance(expr, exp.Literal):
634            return str(expr.this).strip()
635        if isinstance(expr, (exp.Column, exp.Identifier)):
636            return expr.name
637        return expr.sql(dialect=self.dialect).strip()
638
639    def _validate_nested_config_values(self, value_expr: exp.Expr) -> t.List[str]:
640        result = []
641
642        def flatten_expr(expr: exp.Expr) -> None:
643            if isinstance(expr, exp.Array):
644                for elem in expr.expressions:
645                    flatten_expr(elem)
646            elif isinstance(expr, (exp.Tuple, exp.Paren)):
647                expressions = [expr.unnest()] if isinstance(expr, exp.Paren) else expr.expressions
648                for elem in expressions:
649                    flatten_expr(elem)
650            else:
651                result.append(self._validate_config_expression(expr))
652
653        flatten_expr(value_expr)
654        return result