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