sqlmesh.core.dialect

   1from __future__ import annotations
   2
   3import functools
   4import logging
   5import re
   6import sys
   7import typing as t
   8from contextlib import contextmanager
   9from difflib import unified_diff
  10from enum import Enum, auto
  11from functools import lru_cache
  12
  13from sqlglot import Dialect, Generator, ParseError, Parser, Tokenizer, TokenType, exp
  14from sqlglot.dialects.dialect import DialectType
  15from sqlglot.dialects import DuckDB, Snowflake, TSQL
  16import sqlglot.dialects.athena as athena
  17import sqlglot.generators.athena as athena_generators
  18from sqlglot.parsers.athena import AthenaTrinoParser
  19from sqlglot.helper import seq_get
  20from sqlglot.optimizer.normalize_identifiers import normalize_identifiers
  21from sqlglot.optimizer.qualify_columns import quote_identifiers
  22from sqlglot.optimizer.qualify_tables import qualify_tables
  23from sqlglot.optimizer.scope import traverse_scope
  24from sqlglot.schema import MappingSchema
  25from sqlglot.tokens import Token
  26
  27from sqlmesh.core.constants import LIQUID_CLUSTERING_KEYWORDS, MAX_MODEL_DEFINITION_SIZE
  28from sqlmesh.utils import get_source_columns_to_types
  29from sqlmesh.utils.errors import SQLMeshError, ConfigError
  30from sqlmesh.utils.pandas import columns_to_types_from_df
  31
  32if t.TYPE_CHECKING:
  33    import pandas as pd
  34
  35    from sqlglot._typing import E
  36
  37
  38SQLMESH_MACRO_PREFIX = "@"
  39
  40TABLES_META = "sqlmesh.tables"
  41
  42logger = logging.getLogger(__name__)
  43
  44
  45class Model(exp.Expression):
  46    arg_types = {"expressions": True}
  47
  48
  49class Audit(exp.Expression):
  50    arg_types = {"expressions": True}
  51
  52
  53class Metric(exp.Expression):
  54    arg_types = {"expressions": True}
  55
  56
  57class Jinja(exp.Expression, exp.Func):
  58    arg_types = {"this": True}
  59
  60
  61class JinjaQuery(Jinja):
  62    pass
  63
  64
  65class JinjaStatement(Jinja):
  66    pass
  67
  68
  69class VirtualUpdateStatement(exp.Expression):
  70    arg_types = {"expressions": True}
  71
  72
  73class ModelKind(exp.Expression):
  74    arg_types = {"this": True, "expressions": False}
  75
  76
  77class MacroVar(exp.Var):
  78    pass
  79
  80
  81class MacroFunc(exp.Expression, exp.Func):
  82    @property
  83    def name(self) -> str:
  84        return self.this.name
  85
  86
  87class MacroDef(MacroFunc):
  88    arg_types = {"this": True, "expression": True}
  89
  90
  91class MacroSQL(MacroFunc):
  92    arg_types = {"this": True, "into": False}
  93
  94
  95class MacroStrReplace(MacroFunc):
  96    pass
  97
  98
  99class PythonCode(exp.Expression):
 100    arg_types = {"expressions": True}
 101
 102
 103class DColonCast(exp.Cast):
 104    pass
 105
 106
 107class MetricAgg(exp.Expression, exp.AggFunc):
 108    """Used for computing metrics."""
 109
 110    arg_types = {"this": True}
 111
 112    @property
 113    def output_name(self) -> str:
 114        return self.this.name
 115
 116
 117class StagedFilePath(exp.Expression):
 118    """Represents paths to "staged files" in Snowflake."""
 119
 120    arg_types = exp.Table.arg_types.copy()
 121
 122
 123def _parse_statement(self: Parser) -> t.Optional[exp.Expr]:
 124    if self._curr is None:
 125        return None
 126
 127    parser = PARSERS.get(self._curr.text.upper())
 128    error_msg = None
 129
 130    if parser:
 131        # Capture any available description in the form of a comment
 132        comments = self._curr.comments
 133
 134        index = self._index
 135        try:
 136            self._advance()
 137            meta = self._parse_wrapped(lambda: t.cast(t.Callable, parser)(self))
 138        except ParseError as parse_error:
 139            error_msg = parse_error.args[0]
 140            self._retreat(index)
 141
 142        # Only return the DDL expression if we actually managed to parse one. This is
 143        # done in order to allow parsing standalone identifiers / function calls like
 144        # "metric", or "model(1, 2, 3)", which collide with SQLMesh's DDL syntax.
 145        if self._index != index:
 146            meta.comments = comments
 147            return meta
 148
 149    try:
 150        return self.__parse_statement()  # type: ignore
 151    except ParseError:
 152        if error_msg:
 153            raise ParseError(error_msg)
 154        raise
 155
 156
 157def _parse_lambda(self: Parser, alias: bool = False) -> t.Optional[exp.Expr]:
 158    node = self.__parse_lambda(alias=alias)  # type: ignore
 159    if isinstance(node, exp.Lambda):
 160        node.set("this", self._parse_alias(node.this))
 161    return node
 162
 163
 164def _parse_id_var(
 165    self: Parser,
 166    any_token: bool = True,
 167    tokens: t.Optional[t.Collection[TokenType]] = None,
 168) -> t.Optional[exp.Expr]:
 169    if self._prev and self._prev.text == SQLMESH_MACRO_PREFIX and self._match(TokenType.L_BRACE):
 170        identifier = self.__parse_id_var(any_token=any_token, tokens=tokens)  # type: ignore
 171        if not self._match(TokenType.R_BRACE):
 172            self.raise_error("Expecting }")
 173        identifier.args["this"] = f"@{{{identifier.name}}}"
 174    else:
 175        identifier = self.__parse_id_var(any_token=any_token, tokens=tokens)  # type: ignore
 176
 177    while (
 178        identifier
 179        and not identifier.args.get("quoted")
 180        and self._is_connected()
 181        and (
 182            self._match_texts(("{", SQLMESH_MACRO_PREFIX))
 183            or self._curr.token_type not in self.RESERVED_TOKENS
 184        )
 185    ):
 186        this = identifier.name
 187        brace = False
 188
 189        if self._prev.text == "{":
 190            this += "{"
 191            brace = True
 192        else:
 193            if self._prev.text == SQLMESH_MACRO_PREFIX:
 194                this += "@"
 195            if self._match(TokenType.L_BRACE):
 196                this += "{"
 197                brace = True
 198
 199        next_id = self._parse_id_var(any_token=False)
 200
 201        if next_id:
 202            this += next_id.name
 203        else:
 204            return identifier
 205
 206        if brace:
 207            if self._match(TokenType.R_BRACE):
 208                this += "}"
 209            else:
 210                self.raise_error("Expecting }")
 211
 212        identifier = self.expression(exp.Identifier(this=this, quoted=identifier.quoted))
 213
 214    return identifier
 215
 216
 217def _parse_macro(self: Parser, keyword_macro: str = "") -> t.Optional[exp.Expr]:
 218    if self._prev.text != SQLMESH_MACRO_PREFIX:
 219        return self._parse_parameter()
 220
 221    comments = self._prev.comments
 222    index = self._index
 223    field = self._parse_primary() or self._parse_function(functions={}) or self._parse_id_var()
 224
 225    def _build_macro(field: t.Optional[exp.Expr]) -> t.Optional[exp.Expr]:
 226        if isinstance(field, exp.Func):
 227            macro_name = field.name.upper()
 228            if macro_name != keyword_macro and macro_name in KEYWORD_MACROS:
 229                self._retreat(index)
 230                return None
 231
 232            if isinstance(field, exp.Anonymous):
 233                if macro_name == "DEF":
 234                    return self.expression(
 235                        MacroDef(
 236                            this=field.expressions[0],
 237                            expression=field.expressions[1],
 238                        ),
 239                        comments=comments,
 240                    )
 241                if macro_name == "SQL":
 242                    into = field.expressions[1].this.lower() if len(field.expressions) > 1 else None
 243                    return self.expression(
 244                        MacroSQL(this=field.expressions[0], into=into), comments=comments
 245                    )
 246            else:
 247                field = self.expression(
 248                    exp.Anonymous(
 249                        this=field.sql_name(),
 250                        expressions=list(field.args.values()),
 251                    ),
 252                    comments=comments,
 253                )
 254
 255            return self.expression(MacroFunc(this=field), comments=comments)
 256
 257        if field is None:
 258            return None
 259
 260        if field.is_string or (isinstance(field, exp.Identifier) and field.quoted):
 261            return self.expression(
 262                MacroStrReplace(this=exp.Literal.string(field.this)), comments=comments
 263            )
 264
 265        if "@" in field.this:
 266            return field  # type: ignore[return-value]
 267        return self.expression(MacroVar(this=field.this), comments=comments)
 268
 269    if isinstance(field, (exp.Window, exp.IgnoreNulls, exp.RespectNulls)):
 270        field.set("this", _build_macro(field.this))
 271    else:
 272        field = _build_macro(field)
 273
 274    return field
 275
 276
 277KEYWORD_MACROS = {"WITH", "JOIN", "WHERE", "GROUP_BY", "HAVING", "ORDER_BY", "LIMIT"}
 278
 279
 280def _parse_matching_macro(self: Parser, name: str) -> t.Optional[exp.Expr]:
 281    if not self._match_pair(TokenType.PARAMETER, TokenType.VAR, advance=False) or (
 282        self._next and self._next.text.upper() != name.upper()
 283    ):
 284        return None
 285
 286    self._advance()
 287    return _parse_macro(self, keyword_macro=name)
 288
 289
 290def _parse_body_macro(self: Parser) -> t.Tuple[str, t.Optional[exp.Expr]]:
 291    name = self._next and self._next.text.upper()
 292
 293    if name == "JOIN":
 294        return ("joins", self._parse_join())
 295    if name == "WHERE":
 296        return ("where", self._parse_where())
 297    if name == "GROUP_BY":
 298        return ("group", self._parse_group())
 299    if name == "HAVING":
 300        return ("having", self._parse_having())
 301    if name == "ORDER_BY":
 302        return ("order", self._parse_order())
 303    if name == "LIMIT":
 304        return ("limit", self._parse_limit())
 305    return ("", None)
 306
 307
 308def _parse_with(self: Parser, skip_with_token: bool = False) -> t.Optional[exp.Expr]:
 309    macro = _parse_matching_macro(self, "WITH")
 310    if not macro:
 311        return self.__parse_with(skip_with_token=skip_with_token)  # type: ignore
 312
 313    macro.this.append("expressions", self.__parse_with(skip_with_token=True))  # type: ignore
 314    return macro
 315
 316
 317def _parse_join(
 318    self: Parser, skip_join_token: bool = False, parse_bracket: bool = False
 319) -> t.Optional[exp.Expr]:
 320    index = self._index
 321    method, side, kind = self._parse_join_parts()
 322    macro = _parse_matching_macro(self, "JOIN")
 323    if not macro:
 324        self._retreat(index)
 325        return self.__parse_join(skip_join_token=skip_join_token, parse_bracket=parse_bracket)  # type: ignore
 326
 327    join = self.__parse_join(skip_join_token=True)  # type: ignore
 328    if method:
 329        join.set("method", method.text)
 330    if side:
 331        join.set("side", side.text)
 332    if kind:
 333        join.set("kind", kind.text)
 334
 335    macro.this.append("expressions", join)
 336    return macro
 337
 338
 339def _warn_unsupported(self: Parser) -> None:
 340    from sqlmesh.core.console import get_console
 341
 342    sql = self._find_sql(self._tokens[0], self._tokens[-1])[: self.error_message_context]
 343
 344    get_console().log_warning(
 345        f"'{sql}' could not be semantically understood as it contains unsupported syntax, SQLMesh will treat the command as is. Note that any references to the model's "
 346        "underlying physical table can't be resolved in this case, consider using Jinja as explained here https://sqlmesh.readthedocs.io/en/stable/concepts/macros/macro_variables/#audit-only-variables"
 347    )
 348
 349
 350def _parse_select(
 351    self: Parser,
 352    nested: bool = False,
 353    table: bool = False,
 354    parse_subquery_alias: bool = True,
 355    parse_set_operation: bool = True,
 356    consume_pipe: bool = True,
 357    from_: t.Optional[exp.From] = None,
 358) -> t.Optional[exp.Expr]:
 359    select = self.__parse_select(  # type: ignore
 360        nested=nested,
 361        table=table,
 362        parse_subquery_alias=parse_subquery_alias,
 363        parse_set_operation=parse_set_operation,
 364        consume_pipe=consume_pipe,
 365        from_=from_,
 366    )
 367
 368    if (
 369        not select
 370        and not parse_set_operation
 371        and self._match_pair(TokenType.PARAMETER, TokenType.VAR, advance=False)
 372    ):
 373        self._advance()
 374        return _parse_macro(self)
 375
 376    return select
 377
 378
 379def _parse_where(self: Parser, skip_where_token: bool = False) -> t.Optional[exp.Expr]:
 380    macro = _parse_matching_macro(self, "WHERE")
 381    if not macro:
 382        return self.__parse_where(skip_where_token=skip_where_token)  # type: ignore
 383
 384    macro.this.append("expressions", self.__parse_where(skip_where_token=True))  # type: ignore
 385    return macro
 386
 387
 388def _parse_group(self: Parser, skip_group_by_token: bool = False) -> t.Optional[exp.Expr]:
 389    macro = _parse_matching_macro(self, "GROUP_BY")
 390    if not macro:
 391        return self.__parse_group(skip_group_by_token=skip_group_by_token)  # type: ignore
 392
 393    macro.this.append("expressions", self.__parse_group(skip_group_by_token=True))  # type: ignore
 394    return macro
 395
 396
 397def _parse_having(self: Parser, skip_having_token: bool = False) -> t.Optional[exp.Expr]:
 398    macro = _parse_matching_macro(self, "HAVING")
 399    if not macro:
 400        return self.__parse_having(skip_having_token=skip_having_token)  # type: ignore
 401
 402    macro.this.append("expressions", self.__parse_having(skip_having_token=True))  # type: ignore
 403    return macro
 404
 405
 406def _parse_order(
 407    self: Parser, this: t.Optional[exp.Expr] = None, skip_order_token: bool = False
 408) -> t.Optional[exp.Expr]:
 409    macro = _parse_matching_macro(self, "ORDER_BY")
 410    if not macro:
 411        return self.__parse_order(this, skip_order_token=skip_order_token)  # type: ignore
 412
 413    macro.this.append("expressions", self.__parse_order(this, skip_order_token=True))  # type: ignore
 414    return macro
 415
 416
 417def _parse_limit(
 418    self: Parser,
 419    this: t.Optional[exp.Expr] = None,
 420    top: bool = False,
 421    skip_limit_token: bool = False,
 422) -> t.Optional[exp.Expr]:
 423    macro = _parse_matching_macro(self, "TOP" if top else "LIMIT")
 424    if not macro:
 425        return self.__parse_limit(this, top=top, skip_limit_token=skip_limit_token)  # type: ignore
 426
 427    macro.this.append("expressions", self.__parse_limit(this, top=top, skip_limit_token=True))  # type: ignore
 428    return macro
 429
 430
 431def _parse_value(self: Parser, values: bool = True) -> t.Optional[exp.Expr]:
 432    wrapped = self._match(TokenType.L_PAREN, advance=False)
 433
 434    # The base _parse_value method always constructs a Tuple instance. This is problematic when
 435    # generating values with a macro function, because it's impossible to tell whether the user's
 436    # intention was to construct a row or a column with the VALUES expression. To avoid this, we
 437    # amend the AST such that the Tuple is replaced by the macro function call itself.
 438    expr = self.__parse_value()  # type: ignore
 439    if expr and not wrapped and isinstance(seq_get(expr.expressions, 0), MacroFunc):
 440        return expr.expressions[0]
 441
 442    return expr
 443
 444
 445def _parse_macro_or_clause(self: Parser, parser: t.Callable) -> t.Optional[exp.Expr]:
 446    return _parse_macro(self) if self._match(TokenType.PARAMETER) else parser()
 447
 448
 449def _parse_props(self: Parser) -> t.Optional[exp.Expr]:
 450    key = self._parse_id_var(any_token=True)
 451    if not key:
 452        return None
 453
 454    name = key.name.lower()
 455    if name == "time_data_type":
 456        # TODO: if we make *_data_type a convention to parse things into exp.DataType, we could make this more generic
 457        value = self._parse_types(schema=True)
 458    elif name == "when_matched":
 459        # Parentheses around the WHEN clauses can be used to disambiguate them from other properties
 460        value = self._parse_wrapped(
 461            lambda: _parse_macro_or_clause(self, self._parse_when_matched),
 462            optional=True,
 463        )
 464    elif name == "merge_filter":
 465        value = self._parse_conjunction()
 466    elif self._match(TokenType.L_PAREN):
 467        value = self.expression(exp.Tuple(expressions=self._parse_csv(self._parse_equality)))
 468        self._match_r_paren()
 469    else:
 470        value = self._parse_bracket(self._parse_field(any_token=True))
 471
 472    if name == "path" and value:
 473        # Make sure if we get a windows path that it is converted to posix
 474        value = exp.Literal.string(value.this.replace("\\", "/"))  # type: ignore
 475
 476    return self.expression(exp.Property(this=name, value=value))
 477
 478
 479def _parse_types(
 480    self: Parser,
 481    check_func: bool = False,
 482    schema: bool = False,
 483    allow_identifiers: bool = True,
 484    with_collation: bool = False,
 485) -> t.Optional[exp.Expr]:
 486    start = self._curr
 487    parsed_type = self.__parse_types(  # type: ignore
 488        check_func=check_func,
 489        schema=schema,
 490        allow_identifiers=allow_identifiers,
 491        with_collation=with_collation,
 492    )
 493
 494    if schema and parsed_type:
 495        parsed_type.meta["sql"] = self._find_sql(start, self._prev)
 496
 497    return parsed_type
 498
 499
 500# Only needed for Snowflake: its "staged file" syntax (@<path>) clashes with our macro
 501# var syntax. By converting the Var representation to a MacroVar, we should be able to
 502# handle both use cases: if there's no value in the MacroEvaluator's context for that
 503# MacroVar, it'll render into @<path>, so it won't break staged file path references.
 504#
 505# See: https://docs.snowflake.com/en/user-guide/querying-stage
 506def _parse_table_parts(
 507    self: Parser,
 508    schema: bool = False,
 509    is_db_reference: bool = False,
 510    wildcard: bool = False,
 511    fast: bool = False,
 512) -> exp.Table | StagedFilePath:
 513    index = self._index
 514    table = self.__parse_table_parts(  # type: ignore
 515        schema=schema, is_db_reference=is_db_reference, wildcard=wildcard, fast=fast
 516    )
 517
 518    if table is None:
 519        return table  # type: ignore[return-value]
 520
 521    table_arg = table.this
 522    name = table_arg.name if isinstance(table_arg, exp.Var) else ""
 523
 524    if name.startswith(SQLMESH_MACRO_PREFIX):
 525        # In these cases, we don't want to produce a `StagedFilePath` node:
 526        #
 527        # - @'...' needs to parsed as a string template
 528        # - @{foo}.bar needs to be parsed as a table with a macro var part
 529        # - @name(arg1 [, arg2 ...]) needs to be parsed as a macro function call
 530        #
 531        # These cases can unambiguously be parsed using the base `_parse_table_parts`, as there
 532        # is no overlap with staged files https://docs.snowflake.com/en/user-guide/querying-stage
 533        if (
 534            self._prev.token_type == TokenType.STRING
 535            or "{" in name
 536            or (
 537                self._curr
 538                and self._prev.token_type in (TokenType.L_PAREN, TokenType.R_PAREN)
 539                and self._curr.text.upper() not in ("FILE_FORMAT", "PATTERN")
 540                and not (table.args.get("format") or table.args.get("pattern"))
 541            )
 542        ):
 543            self._retreat(index)
 544            return Parser._parse_table_parts(
 545                self, schema=schema, is_db_reference=is_db_reference, fast=fast
 546            )  # type: ignore[return-value]
 547
 548        table_arg.replace(MacroVar(this=name[1:]))
 549        return StagedFilePath(**table.args)
 550
 551    return table
 552
 553
 554# Only needed for T-SQL: it spells a column's nullability right after its type, e.g.
 555# ALTER TABLE t ALTER COLUMN c INT NOT NULL. Without this the trailing clause is left
 556# over, so the whole statement falls back to a Command and any macros it contains (such as
 557# @this_model) are no longer resolved, which means they reach the engine verbatim.
 558#
 559# See: https://learn.microsoft.com/en-us/sql/t-sql/statements/alter-table-transact-sql
 560def _parse_alter_table_alter(self: Parser) -> t.Optional[exp.Expr]:
 561    alter_column = self.__parse_alter_table_alter()  # type: ignore
 562
 563    if isinstance(alter_column, exp.AlterColumn) and alter_column.args.get("dtype"):
 564        if self._match_pair(TokenType.NOT, TokenType.NULL):
 565            alter_column.set("allow_null", False)
 566        elif self._match(TokenType.NULL):
 567            alter_column.set("allow_null", True)
 568
 569    return alter_column
 570
 571
 572def altercolumn_sql(self: Generator, expression: exp.AlterColumn) -> str:
 573    sql = self._altercolumn_sql(expression)  # type: ignore
 574
 575    # sqlglot's generator returns as soon as it renders the type, so the nullability parsed
 576    # above has to be appended here
 577    allow_null = expression.args.get("allow_null")
 578    if expression.args.get("dtype") and allow_null is not None:
 579        sql = f"{sql} NULL" if allow_null else f"{sql} NOT NULL"
 580
 581    return sql
 582
 583
 584def _parse_if(self: Parser) -> t.Optional[exp.Expr]:
 585    # If we fail to parse an IF function with expressions as arguments, we then try
 586    # to parse a statement / command to support the macro @IF(condition, statement)
 587    index = self._index
 588    try:
 589        if self.dialect == "tsql":
 590            if not (self._index >= 2 and self._tokens[self._index - 2].text == "@"):
 591                return self.__parse_if()  # type: ignore
 592            return Parser.__parse_if(self)  # type: ignore
 593        return self.__parse_if()  # type: ignore
 594    except ParseError:
 595        self._retreat(index)
 596        self._match_l_paren()
 597
 598        cond = self._parse_conjunction()
 599        self._match(TokenType.COMMA)
 600
 601        # Try to parse a known statement, otherwise fall back to parsing a command
 602        # Since the trailing `)` token is not expected by the statement parsers, we
 603        # remove it from the token stream before trying to parse the statement.
 604        last_token = self._tokens[-1]
 605        if last_token.token_type == TokenType.R_PAREN:
 606            self._tokens[-2].comments.extend(last_token.comments)
 607            self._tokens.pop()
 608            if hasattr(self, "_tokens_size"):
 609                # keep _tokens_size in sync sqlglot 30.0.3 caches len(_tokens)
 610                # _advance() tries to read tokens[index + 1] past the new end
 611                self._tokens_size -= 1
 612        else:
 613            self.raise_error("Expecting )")
 614
 615        index = self._index
 616        stmt = self._parse_statement()
 617        if self._curr:
 618            self._retreat(index)
 619            stmt = self._parse_as_command(self._tokens[index])
 620
 621        return exp.Anonymous(this="IF", expressions=[cond, stmt])
 622
 623
 624def _create_parser(expression_type: t.Type[exp.Expr], table_keys: t.List[str]) -> t.Callable:
 625    def parse(self: Parser) -> t.Optional[exp.Expr]:
 626        from sqlmesh.core.model.kind import ModelKindName
 627
 628        expressions: t.List[exp.Expr] = []
 629
 630        while True:
 631            prev_property = seq_get(expressions, -1)
 632            if not self._match(TokenType.COMMA, expression=prev_property) and expressions:
 633                break
 634
 635            key_expression = self._parse_id_var(any_token=True)
 636            if not key_expression:
 637                break
 638
 639            # This allows macro functions that programmaticaly generate the property key-value pair
 640            if isinstance(key_expression, MacroFunc):
 641                expressions.append(key_expression)
 642                continue
 643
 644            key = key_expression.name.lower()
 645
 646            start = self._curr
 647            value: t.Optional[exp.Expr | str]
 648
 649            if key in table_keys:
 650                value = self._parse_table_parts()
 651                if value and self._prev.token_type == TokenType.STRING:
 652                    self.raise_error(
 653                        f"'{key}' property cannot be a string value: {value}. "
 654                        "Please use the identifier syntax instead, e.g. foo.bar instead of 'foo.bar'"
 655                    )
 656            elif key == "columns":
 657                value = self._parse_schema()
 658            elif key == "kind":
 659                field = _parse_macro_or_clause(self, lambda: self._parse_id_var(any_token=True))
 660
 661                if not field or isinstance(field, (MacroVar, MacroFunc)):
 662                    value = field
 663                else:
 664                    try:
 665                        kind = ModelKindName[field.name.upper()]
 666                    except KeyError:
 667                        raise SQLMeshError(
 668                            f"Model kind specified as '{field.name}', but that is not a valid model kind.\n\nPlease specify one of {', '.join(ModelKindName)}."
 669                        )
 670
 671                    if kind in (
 672                        ModelKindName.INCREMENTAL_BY_TIME_RANGE,
 673                        ModelKindName.INCREMENTAL_BY_UNIQUE_KEY,
 674                        ModelKindName.INCREMENTAL_BY_PARTITION,
 675                        ModelKindName.INCREMENTAL_UNMANAGED,
 676                        ModelKindName.SEED,
 677                        ModelKindName.VIEW,
 678                        ModelKindName.SCD_TYPE_2,
 679                        ModelKindName.SCD_TYPE_2_BY_TIME,
 680                        ModelKindName.SCD_TYPE_2_BY_COLUMN,
 681                        ModelKindName.CUSTOM,
 682                    ) and self._match(TokenType.L_PAREN, advance=False):
 683                        props = self._parse_wrapped_csv(functools.partial(_parse_props, self))
 684                    else:
 685                        props = None
 686
 687                    value = self.expression(ModelKind(this=kind.value, expressions=props))
 688            elif key == "expression":
 689                value = self._parse_conjunction()
 690            elif key == "partitioned_by":
 691                partitioned_by = self._parse_partitioned_by()
 692                if isinstance(partitioned_by.this, exp.Schema):
 693                    value = exp.tuple_(*partitioned_by.this.expressions)
 694                else:
 695                    value = partitioned_by.this
 696            elif key == "clustered_by":
 697                # Bare AUTO / NONE are Databricks liquid clustering keywords, not column refs.
 698                # Detect keywords by token type: unquoted bare identifiers arrive as VAR tokens.
 699                # Backtick-quoted identifiers (e.g. `auto`) have IDENTIFIER token type and are
 700                # treated as real column names.
 701                if (
 702                    self._curr is not None
 703                    and self._curr.token_type == TokenType.VAR
 704                    and self._curr.text.upper() in LIQUID_CLUSTERING_KEYWORDS
 705                ):
 706                    value = exp.Var(this=self._curr.text.upper())
 707                    self._advance()
 708                else:
 709                    parsed = self._parse_bracket(self._parse_field(any_token=True))
 710                    # Unwrap Paren wrapping a bare column to match partitioned_by normalisation:
 711                    # clustered_by (a) → stored as Column(a), not Paren(Column(a)).
 712                    # Preserve parens around function expressions: (TO_DATE(col)) stays as-is.
 713                    if isinstance(parsed, exp.Paren) and isinstance(parsed.this, exp.Column):
 714                        value = parsed.unnest()
 715                    else:
 716                        value = parsed
 717            else:
 718                value = self._parse_bracket(self._parse_field(any_token=True))
 719
 720            if isinstance(value, exp.Expr):
 721                value.meta["sql"] = self._find_sql(start, self._prev)
 722
 723            expressions.append(self.expression(exp.Property(this=key, value=value)))
 724
 725        return self.expression(expression_type(expressions=expressions))
 726
 727    return parse
 728
 729
 730PARSERS = {
 731    "MODEL": _create_parser(Model, ["name"]),
 732    "AUDIT": _create_parser(Audit, ["model"]),
 733    "METRIC": _create_parser(Metric, ["name"]),
 734}
 735
 736
 737_SQLMESH_META_DIALECT = "sqlmesh_meta_dialect"
 738
 739
 740def _holds_expression(annotation: t.Any, _visited: t.Optional[t.FrozenSet[t.Any]] = None) -> bool:
 741    """Whether a declared field type bottoms out in a SQLGlot expression.
 742
 743    Covers List[exp.Expr], Optional[Dict[str, exp.DataType]], Optional[exp.Tuple], the
 744    nested Tuple[str, Dict[str, exp.Expr]] shape used by audits/signals, and nested
 745    Pydantic models that themselves wrap an expression field, such as `TimeColumn`
 746    (IncrementalByTimeRangeKind.time_column).
 747
 748    Stops at `_ModelKind` subclasses without recursing into their fields: a `kind`
 749    property's own nested properties are independently dialect-tagged via the
 750    `ModelKind` expression node's own meta when `_props_sql` recurses into them, so
 751    treating the `kind` field itself as "holds an expression" -- true only because some
 752    other member of the `ModelKind` union has an expression field, e.g.
 753    `IncrementalByTimeRangeKind.time_column` -- would route its entire subtree,
 754    including scalar sibling properties like `forward_only`, through a dialect-specific
 755    generator and transpile them when they shouldn't be (tsql booleans becoming
 756    `(1 = 1)`, which silently reparses as `False`).
 757    """
 758    from sqlmesh.core.model.kind import _ModelKind
 759
 760    if isinstance(annotation, type):
 761        if issubclass(annotation, exp.Expr):
 762            return True
 763        if issubclass(annotation, _ModelKind):
 764            return False
 765        visited = _visited or frozenset()
 766        if annotation in visited:
 767            return False
 768        if hasattr(annotation, "model_fields"):
 769            visited = visited | {annotation}
 770            return any(
 771                _holds_expression(field.annotation, visited)
 772                for field in annotation.model_fields.values()
 773            )
 774        return False
 775    return any(_holds_expression(arg, _visited) for arg in t.get_args(annotation))
 776
 777
 778@functools.lru_cache(maxsize=1)
 779def _meta_render_policy() -> t.Dict[str, bool]:
 780    """Map header property name -> whether its value is warehouse SQL.
 781
 782    Derived from the field declarations themselves, so it stays correct as properties
 783    are added: expression-typed values (columns, audits, physical_properties, ...) are
 784    the user's warehouse SQL and must render in the model's dialect, while scalar-typed
 785    values (allow_partials, description, kind, ...) are SQLMesh's own semantics and must
 786    stay dialect-agnostic -- transpiling those is what corrupts `allow_partials TRUE`
 787    into tsql's unparseable `(1 = 1)`.
 788    """
 789    import inspect
 790
 791    from sqlmesh.core.audit.definition import ModelAudit
 792    from sqlmesh.core.metric.definition import MetricMeta
 793    from sqlmesh.core.model import kind as kind_module
 794    from sqlmesh.core.model.meta import ModelMeta
 795
 796    sources: t.List[t.Any] = [ModelMeta, ModelAudit, MetricMeta]
 797    sources.extend(
 798        obj
 799        for name, obj in vars(kind_module).items()
 800        if inspect.isclass(obj) and hasattr(obj, "model_fields") and name.endswith("Kind")
 801    )
 802
 803    policy: t.Dict[str, bool] = {}
 804    for source in sources:
 805        for name, field in source.model_fields.items():
 806            policy.setdefault((field.alias or name).lower(), _holds_expression(field.annotation))
 807
 808    # `ModelMeta._pre_root_validator` (sqlmesh/core/model/meta.py) renames these two
 809    # user-facing property names to their target field before Pydantic validation, so
 810    # they never surface as a `Field(alias=...)` for the reflection above to find. Give
 811    # each the render policy of the field it is renamed to.
 812    pre_validator_aliases = {
 813        "grain": "grains",
 814        "table_properties": "physical_properties",
 815    }
 816    for alias, target in pre_validator_aliases.items():
 817        if target in policy:
 818            policy[alias] = policy[target]
 819
 820    return policy
 821
 822
 823@functools.lru_cache(maxsize=None)
 824def _dialect_renders_array_as_brackets(dialect_name: t.Optional[str]) -> bool:
 825    """Whether `dialect_name`'s own generator spells an array literal as `[a, b]`.
 826
 827    Checked by actually rendering a sample `exp.Array` with that dialect, rather than
 828    inspecting `Dialect.ARRAY_SIZE_NAME` or similar generator flags, because the
 829    generator is the single source of truth for what a dialect's array syntax looks
 830    like and there is no single shared flag for it across dialects. This also covers
 831    dialects (tsql, sqlite, tableau, exasol, fabric) that reuse `[`/`]` for identifier
 832    quoting and therefore render arrays as `ARRAY(...)` instead: rewriting their
 833    `tags`/`ignored_rules` value to `[a, b]` would not be an array literal in their
 834    grammar at all, so it silently reparses as one bracket-quoted identifier and
 835    corrupts the value. An unrecognized dialect name renders with the generic
 836    generator, which itself does not use brackets, so it falls back to `False`.
 837    """
 838    try:
 839        sample = exp.Array(expressions=[exp.Literal.string("x")])
 840        return sample.sql(dialect=dialect_name).startswith("[")
 841    except Exception:
 842        return False
 843
 844
 845def _props_sql(self: Generator, expressions: t.List[exp.Expr]) -> str:
 846    props = []
 847    size = len(expressions)
 848
 849    for i, prop in enumerate(expressions):
 850        parent = prop.parent
 851        meta_dialect = parent.meta.get(_SQLMESH_META_DIALECT) if parent else None
 852
 853        def render_with_model_dialect(node: exp.Expr, **overrides: t.Any) -> str:
 854            opts: t.Dict[str, t.Any] = {
 855                "dialect": meta_dialect,
 856                "pretty": self.pretty,
 857                "identify": self.identify,
 858                "normalize": self.normalize,
 859                "pad": self.pad,
 860                "indent": self._indent,
 861                "normalize_functions": self.normalize_functions,
 862                "leading_comma": self.leading_comma,
 863                "max_text_width": self.max_text_width,
 864                "comments": self.comments,
 865            }
 866            opts.update(overrides)
 867
 868            # Keep boolean literals anywhere in the value (audit args, physical_properties,
 869            # merge_filter, ...) as `TRUE`/`FALSE`: tsql would otherwise emit `(1 = 1)`,
 870            # which reformats differently on the next pass. The value is transpiled with
 871            # the model dialect anyway when it is used, e.g. in the rendered audit query.
 872            def keep_boolean_literal(n: exp.Expr) -> exp.Expr:
 873                if not isinstance(n, exp.Boolean):
 874                    return n
 875                literal = exp.var("TRUE" if n.this else "FALSE")
 876                literal.comments = n.comments
 877                return literal
 878
 879            return node.transform(keep_boolean_literal).sql(**opts)
 880
 881        if isinstance(prop, MacroFunc):
 882            # A macro in property position wraps user-authored arguments, so it carries
 883            # warehouse SQL the same way `columns` or `audits` do. Clear the outer node's
 884            # own comments (not `.this`'s, which `_macro_func_sql` already attaches)
 885            # before rendering with the model dialect, mirroring what `comment=False`
 886            # does for the non-dialect path below -- passing `comments=False` here
 887            # instead would build a fresh Generator with comments globally disabled,
 888            # silently dropping every comment in the subtree rather than just the
 889            # redundant outer one.
 890            if meta_dialect:
 891                prop_for_render = prop.copy()
 892                prop_for_render.comments = None
 893                sql = self.indent(render_with_model_dialect(prop_for_render))
 894            else:
 895                sql = self.indent(self.sql(prop, comment=False))
 896        else:
 897            value = prop.args.get("value")
 898
 899            if (
 900                meta_dialect
 901                and isinstance(value, exp.Expr)
 902                and _meta_render_policy().get(prop.name.lower())
 903            ):
 904                value_sql = render_with_model_dialect(value)
 905            elif (
 906                meta_dialect
 907                and isinstance(value, exp.Array)
 908                and _dialect_renders_array_as_brackets(meta_dialect)
 909            ):
 910                # Dialect-agnostic properties (e.g. `tags`, `ignored_rules`) that hold a
 911                # list still go through the base (dialect=None) generator, which renders
 912                # an `exp.Array` as `ARRAY(...)`. On BigQuery `ARRAY(` is parsed as a
 913                # subquery constructor, so a multi-element `ARRAY('a', 'b')` fails to
 914                # reparse ("Required keyword: 'value' missing for Property"). Render it
 915                # as a bracketed list literal instead -- but only for dialects that
 916                # actually spell arrays that way; dialects that reuse `[`/`]` for
 917                # identifier quoting (tsql, sqlite, ...) keep the generic `ARRAY(...)`
 918                # form, which they parse back correctly. The elements themselves stay on
 919                # the dialect-agnostic path (`self.expressions`, not
 920                # `render_with_model_dialect`): these are SQLMesh's own scalar values
 921                # (tag/rule name strings), not user warehouse SQL, so they must not be
 922                # transpiled with the model dialect (e.g. tsql boolean literals turning
 923                # into `(1 = 1)`).
 924                value_sql = f"[{self.expressions(value, flat=True)}]"
 925            else:
 926                value_sql = self.sql(prop, "value")
 927
 928            sql = self.indent(f"{prop.name} {value_sql}")
 929
 930        if i < size - 1:
 931            sql += ","
 932
 933        props.append(self.maybe_comment(sql, expression=prop))
 934
 935    return "\n".join(props)
 936
 937
 938def _on_virtual_update_sql(self: Generator, expressions: t.List[exp.Expr]) -> str:
 939    statements = "\n".join(
 940        self.sql(expression)
 941        if isinstance(expression, JinjaStatement)
 942        else f"{self.sql(expression)};"
 943        for expression in expressions
 944    )
 945    return f"{ON_VIRTUAL_UPDATE_BEGIN};\n{statements}\n{ON_VIRTUAL_UPDATE_END};"
 946
 947
 948def _sqlmesh_ddl_sql(self: Generator, expression: Model | Audit | Metric, name: str) -> str:
 949    return "\n".join([f"{name} (", _props_sql(self, expression.expressions), ")"])
 950
 951
 952def _model_kind_sql(self: Generator, expression: ModelKind) -> str:
 953    props = _props_sql(self, expression.expressions)
 954    if props:
 955        return "\n".join([f"{expression.this} (", props, ")"])
 956    return expression.name.upper()
 957
 958
 959def _macro_keyword_func_sql(self: Generator, expression: exp.Expr) -> str:
 960    name = expression.name
 961    keyword = name.replace("_", " ")
 962    *args, clause = expression.expressions
 963    macro = f"@{name}({self.format_args(*args)})"
 964    return self.sql(clause).replace(keyword, macro, 1)
 965
 966
 967def _macro_func_sql(self: Generator, expression: MacroFunc) -> str:
 968    expression = expression.this
 969    name = expression.name
 970    if name in KEYWORD_MACROS:
 971        sql = _macro_keyword_func_sql(self, expression)
 972    else:
 973        sql = f"@{name}({self.format_args(*expression.expressions)})"
 974    return self.maybe_comment(sql, expression)
 975
 976
 977def _whens_sql(self: Generator, expression: exp.Whens) -> str:
 978    if isinstance(expression.parent, exp.Merge):
 979        return self.whens_sql(expression)
 980
 981    # If the `WHEN` clauses aren't part of a MERGE statement (e.g. they
 982    # appear in the `MODEL` DDL), then we will wrap them with parentheses.
 983    return self.wrap(self.expressions(expression, sep=" ", indent=False))
 984
 985
 986def _parse_interval_span(self: Parser, this: exp.Expr) -> exp.Interval:
 987    interval = self.__parse_interval_span(this)  # type: ignore
 988    # Without this, @unit in `INTERVAL @value @unit` is misread as an alias.
 989    if not interval.args.get("unit") and self._match(TokenType.PARAMETER):
 990        macro = _parse_macro(self)
 991        if macro is not None:
 992            interval.set("unit", macro)
 993    return interval
 994
 995
 996def _override(klass: t.Type[Tokenizer | Parser | Generator], func: t.Callable) -> None:
 997    name = func.__name__
 998    if getattr(klass, name, None) is func:
 999        # Already overridden. Re-applying would save the override itself as the
1000        # "original", making the wrapper call itself and recurse infinitely.
1001        return
1002    setattr(klass, f"_{name}", getattr(klass, name))
1003    setattr(klass, name, func)
1004
1005
1006def format_model_expressions(
1007    expressions: t.List[exp.Expr],
1008    dialect: t.Optional[str] = None,
1009    rewrite_casts: bool = True,
1010    normalize_functions: t.Union[str, bool, None] = False,
1011    **kwargs: t.Any,
1012) -> str:
1013    """Format a model's expressions into a standardized format.
1014
1015    Args:
1016        expressions: The model's expressions, must be at least model def + query.
1017        dialect: The dialect to render the expressions as.
1018        rewrite_casts: Whether to rewrite all casts to use the :: syntax.
1019        normalize_functions: How to normalize function name casing.
1020
1021            * ``False`` (default) — preserves the original spelling of custom and audit
1022              function names.  SQLGlot built-in functions may still canonicalize because
1023              the parser discards the original token.
1024            * ``"upper"`` — uppercases all function names including custom audit
1025              references.
1026            * ``"lower"`` — lowercases all function names including built-ins.
1027            * ``True`` — defers to SQLGlot's generator default (uppercase).
1028            * ``None`` — passes ``None`` directly to the SQLGlot generator, which
1029              defers to SQLGlot's own default (typically uppercase, but may vary by
1030              dialect).  Note: this is the **direct generator API** behaviour.  When
1031              called via ``FormatConfig``, ``None`` is excluded by Pydantic's
1032              ``exclude_none`` serialization and this function receives its own ``False``
1033              default instead — so the two paths are not equivalent.
1034        **kwargs: Additional keyword arguments to pass to the sql generator.
1035
1036    Returns:
1037        A string representing the formatted model.
1038    """
1039
1040    def tag_meta_dialect(expression: exp.Expr) -> exp.Expr:
1041        """Record the model dialect on meta nodes so `_props_sql` can render the
1042        warehouse-SQL properties (columns, audits, physical_properties, ...) with it
1043        while the SQLMesh-owned ones stay dialect-agnostic. Tags nested ModelKind
1044        nodes too, since kinds carry expression properties of their own such as
1045        `time_data_type` and `unique_key`."""
1046        if not dialect or not is_meta_expression(expression):
1047            return expression
1048
1049        expression = expression.copy()
1050        for node in expression.find_all(Model, Audit, Metric, ModelKind):
1051            node.meta[_SQLMESH_META_DIALECT] = dialect
1052        expression.meta[_SQLMESH_META_DIALECT] = dialect
1053        return expression
1054
1055    if len(expressions) == 1 and is_meta_expression(expressions[0]):
1056        # Meta expressions (MODEL/AUDIT/METRIC) are SQLMesh DDL, not standard SQL,
1057        # so they must never be transpiled to the target dialect (e.g. tsql would
1058        # rewrite a boolean property like `allow_partials TRUE` to `(1 = 1)`).
1059        # Individual properties whose values *are* warehouse SQL still render with
1060        # the model dialect -- see `_props_sql` / `_meta_render_policy`.
1061        return tag_meta_dialect(expressions[0]).sql(
1062            pretty=True, dialect=None, normalize_functions=normalize_functions
1063        )
1064
1065    if rewrite_casts:
1066
1067        def cast_to_colon(node: exp.Expr) -> exp.Expr:
1068            # Directly check type instead of isinstance to avoid rewriting subclasses of CAST, e.g. JSONCast
1069            if type(node) is exp.Cast and not any(
1070                # Only convert CAST into :: if it doesn't have additional args set, otherwise this
1071                # conversion could alter the semantics (eg. changing SAFE_CAST in BigQuery to CAST)
1072                arg
1073                for name, arg in node.args.items()
1074                if name not in ("this", "to")
1075            ):
1076                this = node.this
1077
1078                if not isinstance(this, (exp.Binary, exp.Unary)) or isinstance(this, exp.Paren):
1079                    cast = DColonCast(this=this, to=node.to)
1080                    cast.comments = node.comments
1081                    node = cast
1082
1083            exp.replace_children(node, cast_to_colon)
1084            return node
1085
1086        new_expressions = []
1087        for expression in expressions:
1088            expression = expression.copy()
1089            exp.replace_children(expression, cast_to_colon)
1090            new_expressions.append(expression)
1091
1092        expressions = new_expressions
1093
1094    return ";\n\n".join(
1095        # Meta expressions (MODEL/AUDIT/METRIC) are SQLMesh DDL and must stay
1096        # dialect-agnostic; only the actual query/statement expressions transpile.
1097        tag_meta_dialect(expression).sql(
1098            pretty=True,
1099            dialect=None if is_meta_expression(expression) else dialect,
1100            normalize_functions=normalize_functions,
1101            **kwargs,
1102        )
1103        for expression in expressions
1104    ).strip()
1105
1106
1107def text_diff(
1108    a: t.List[exp.Expr],
1109    b: t.List[exp.Expr],
1110    a_dialect: t.Optional[str] = None,
1111    b_dialect: t.Optional[str] = None,
1112) -> str:
1113    """Find the unified text diff between two expressions."""
1114    a_sql = [
1115        line
1116        for expr in a
1117        for line in expr.sql(pretty=True, comments=False, dialect=a_dialect).split("\n")
1118    ]
1119    b_sql = [
1120        line
1121        for expr in b
1122        for line in expr.sql(pretty=True, comments=False, dialect=b_dialect).split("\n")
1123    ]
1124    return "\n".join(unified_diff(a_sql, b_sql))
1125
1126
1127WS_OR_COMMENT = r"(?:\s|--[^\n]*\n|/\*.*?\*/)"
1128HEADER = r"\b(?:model|audit)\b(?=\s*\()"
1129KEY_BOUNDARY = r"(?:\(|,)"  # key is preceded by either '(' or ','
1130DIALECT_VALUE = r"['\"]?(?P<dialect>[a-z][a-z0-9]*)['\"]?"
1131VALUE_BOUNDARY = r"(?=,|\))"  # value is followed by comma or closing paren
1132
1133DIALECT_PATTERN = re.compile(
1134    rf"{HEADER}.*?{KEY_BOUNDARY}{WS_OR_COMMENT}*dialect{WS_OR_COMMENT}+{DIALECT_VALUE}{WS_OR_COMMENT}*{VALUE_BOUNDARY}",
1135    re.IGNORECASE | re.DOTALL,
1136)
1137
1138
1139def _is_command_statement(command: str, tokens: t.List[Token], pos: int) -> bool:
1140    try:
1141        return (
1142            tokens[pos].text.upper() == command.upper()
1143            and tokens[pos + 1].token_type == TokenType.SEMICOLON
1144        )
1145    except IndexError:
1146        return False
1147
1148
1149JINJA_QUERY_BEGIN = "JINJA_QUERY_BEGIN"
1150JINJA_STATEMENT_BEGIN = "JINJA_STATEMENT_BEGIN"
1151JINJA_END = "JINJA_END"
1152ON_VIRTUAL_UPDATE_BEGIN = "ON_VIRTUAL_UPDATE_BEGIN"
1153ON_VIRTUAL_UPDATE_END = "ON_VIRTUAL_UPDATE_END"
1154
1155
1156def _is_jinja_statement_begin(tokens: t.List[Token], pos: int) -> bool:
1157    return _is_command_statement(JINJA_STATEMENT_BEGIN, tokens, pos)
1158
1159
1160def _is_jinja_query_begin(tokens: t.List[Token], pos: int) -> bool:
1161    return _is_command_statement(JINJA_QUERY_BEGIN, tokens, pos)
1162
1163
1164def _is_jinja_end(tokens: t.List[Token], pos: int) -> bool:
1165    return _is_command_statement(JINJA_END, tokens, pos)
1166
1167
1168def jinja_query(query: str) -> JinjaQuery:
1169    return JinjaQuery(this=exp.Literal.string(query.strip()))
1170
1171
1172def jinja_statement(statement: str) -> JinjaStatement:
1173    return JinjaStatement(this=exp.Literal.string(statement.strip()))
1174
1175
1176def _is_virtual_statement_begin(tokens: t.List[Token], pos: int) -> bool:
1177    return _is_command_statement(ON_VIRTUAL_UPDATE_BEGIN, tokens, pos)
1178
1179
1180def _is_virtual_statement_end(tokens: t.List[Token], pos: int) -> bool:
1181    return _is_command_statement(ON_VIRTUAL_UPDATE_END, tokens, pos)
1182
1183
1184def virtual_statement(statements: t.List[exp.Expr]) -> VirtualUpdateStatement:
1185    return VirtualUpdateStatement(expressions=statements)
1186
1187
1188class ChunkType(Enum):
1189    JINJA_QUERY = auto()
1190    JINJA_STATEMENT = auto()
1191    SQL = auto()
1192    VIRTUAL_STATEMENT = auto()
1193    VIRTUAL_JINJA_STATEMENT = auto()
1194
1195
1196def parse_one(
1197    sql: str, dialect: t.Optional[str] = None, into: t.Optional[exp.IntoType] = None
1198) -> exp.Expr:
1199    expressions = parse(sql, default_dialect=dialect, match_dialect=False, into=into)
1200    if not expressions:
1201        raise SQLMeshError(f"No expressions found in '{sql}'")
1202    elif len(expressions) > 1:
1203        raise SQLMeshError(f"Multiple expressions found in '{sql}'")
1204    return expressions[0]
1205
1206
1207def parse(
1208    sql: str,
1209    default_dialect: t.Optional[str] = None,
1210    match_dialect: bool = True,
1211    into: t.Optional[exp.IntoType] = None,
1212) -> t.List[exp.Expr]:
1213    """Parse a sql string.
1214
1215    Supports parsing model definition.
1216    If a jinja block is detected, the query is stored as raw string in a Jinja node.
1217
1218    Args:
1219        sql: The sql based definition.
1220        default_dialect: The dialect to use if the model does not specify one.
1221
1222    Returns:
1223        A list of the parsed expressions: [Model, *Statements, Query, *Statements]
1224    """
1225    match = match_dialect and DIALECT_PATTERN.search(sql[:MAX_MODEL_DEFINITION_SIZE])
1226    dialect_str = match.group("dialect") if match else None
1227    dialect = Dialect.get_or_raise(dialect_str or default_dialect)
1228
1229    tokens = dialect.tokenize(sql)
1230    chunks: t.List[t.Tuple[t.List[Token], ChunkType]] = [([], ChunkType.SQL)]
1231    total = len(tokens)
1232
1233    pos = 0
1234    virtual = False
1235    while pos < total:
1236        token = tokens[pos]
1237        if _is_virtual_statement_end(tokens, pos):
1238            chunks[-1][0].append(token)
1239            virtual = False
1240            chunks.append(([], ChunkType.SQL))
1241            pos += 2
1242        elif _is_jinja_end(tokens, pos) or (
1243            chunks[-1][1] == ChunkType.SQL
1244            and token.token_type == TokenType.SEMICOLON
1245            and pos < total - 1
1246        ):
1247            if token.token_type == TokenType.SEMICOLON:
1248                pos += 1
1249            else:
1250                # Jinja end statement
1251                chunks[-1][0].append(token)
1252                pos += 2
1253            chunks.append(
1254                (
1255                    [],
1256                    ChunkType.VIRTUAL_STATEMENT
1257                    if virtual and tokens[pos] != ON_VIRTUAL_UPDATE_END
1258                    else ChunkType.SQL,
1259                )
1260            )
1261        elif _is_jinja_query_begin(tokens, pos):
1262            chunks.append(([token], ChunkType.JINJA_QUERY))
1263            pos += 2
1264        elif _is_jinja_statement_begin(tokens, pos):
1265            chunks.append(([token], ChunkType.JINJA_STATEMENT))
1266            pos += 2
1267        elif _is_virtual_statement_begin(tokens, pos):
1268            chunks.append(([token], ChunkType.VIRTUAL_STATEMENT))
1269            pos += 2
1270            virtual = True
1271        else:
1272            chunks[-1][0].append(token)
1273            pos += 1
1274
1275    parser = dialect.parser()
1276    expressions: t.List[exp.Expr] = []
1277
1278    def parse_sql_chunk(chunk: t.List[Token], meta_sql: bool = True) -> t.List[exp.Expr]:
1279        parsed_expressions: t.List[t.Optional[exp.Expr]] = (
1280            parser.parse(chunk, sql) if into is None else parser.parse_into(into, chunk, sql)
1281        )
1282        expressions = []
1283        for expression in parsed_expressions:
1284            if expression:
1285                if meta_sql:
1286                    expression.meta["sql"] = parser._find_sql(chunk[0], chunk[-1])
1287                expressions.append(expression)
1288        return expressions
1289
1290    def parse_jinja_chunk(chunk: t.List[Token], meta_sql: bool = True) -> exp.Expr:
1291        start, *_, end = chunk
1292        segment = sql[start.end + 2 : end.start - 1]
1293        factory = jinja_query if chunk_type == ChunkType.JINJA_QUERY else jinja_statement
1294        expression = factory(segment.strip())
1295        if meta_sql:
1296            expression.meta["sql"] = sql[start.start : end.end + 1]
1297        return expression
1298
1299    def parse_virtual_statement(
1300        chunks: t.List[t.Tuple[t.List[Token], ChunkType]], pos: int
1301    ) -> t.Tuple[t.List[exp.Expr], int]:
1302        # For virtual statements we need to handle both SQL and Jinja nested blocks within the chunk
1303        virtual_update_statements: t.List[exp.Expr] = []
1304        start = chunks[pos][0][0].start
1305
1306        while (
1307            chunks[pos - 1][0] == [] or chunks[pos - 1][0][-1].text.upper() != ON_VIRTUAL_UPDATE_END
1308        ):
1309            chunk, chunk_type = chunks[pos]
1310            if chunk_type == ChunkType.JINJA_STATEMENT:
1311                virtual_update_statements.append(parse_jinja_chunk(chunk, False))
1312            else:
1313                virtual_update_statements.extend(
1314                    parse_sql_chunk(
1315                        chunk[int(chunk[0].text.upper() == ON_VIRTUAL_UPDATE_BEGIN) : -1], False
1316                    ),
1317                )
1318            pos += 1
1319
1320        if virtual_update_statements:
1321            statements = virtual_statement(virtual_update_statements)
1322            end = chunk[-1].end + 1
1323            statements.meta["sql"] = sql[start:end]
1324            return [statements], pos
1325
1326        return [], pos
1327
1328    pos = 0
1329    total_chunks = len(chunks)
1330    while pos < total_chunks:
1331        chunk, chunk_type = chunks[pos]
1332        if chunk_type == ChunkType.VIRTUAL_STATEMENT:
1333            virtual_expression, pos = parse_virtual_statement(chunks, pos)
1334            expressions.extend(virtual_expression)
1335        elif chunk_type == ChunkType.SQL:
1336            expressions.extend(parse_sql_chunk(chunk))
1337        else:
1338            expressions.append(parse_jinja_chunk(chunk))
1339        pos += 1
1340
1341    return expressions
1342
1343
1344def extend_sqlglot() -> None:
1345    """Extend SQLGlot with SQLMesh's custom macro aware dialect."""
1346    tokenizers = {Tokenizer}
1347    parsers = {Parser}
1348    generators = {Generator}
1349
1350    for dialect in Dialect.classes.values():
1351        # Athena picks a different Tokenizer / Parser / Generator depending on the query
1352        # so this ensures that the extra ones it defines are also extended
1353        if dialect == athena.Athena:
1354            tokenizers.add(athena._TrinoTokenizer)
1355            parsers.add(AthenaTrinoParser)
1356            generators.add(athena_generators.AthenaTrinoGenerator)
1357            generators.add(athena_generators._HiveGenerator)
1358
1359        if hasattr(dialect, "Tokenizer"):
1360            tokenizers.add(dialect.Tokenizer)
1361        if hasattr(dialect, "Parser"):
1362            parsers.add(dialect.Parser)
1363        if hasattr(dialect, "Generator"):
1364            generators.add(dialect.Generator)
1365
1366    for tokenizer in tokenizers:
1367        tokenizer.VAR_SINGLE_TOKENS.update(SQLMESH_MACRO_PREFIX)
1368
1369    for parser in parsers:
1370        parser.FUNCTIONS.update({"JINJA": Jinja.from_arg_list, "METRIC": MetricAgg.from_arg_list})
1371        parser.PLACEHOLDER_PARSERS.update({TokenType.PARAMETER: _parse_macro})
1372        parser.QUERY_MODIFIER_PARSERS.update(
1373            {TokenType.PARAMETER: lambda self: _parse_body_macro(self)}
1374        )
1375
1376    for generator in generators:
1377        if MacroFunc not in generator.TRANSFORMS:
1378            generator.TRANSFORMS.update(
1379                {
1380                    Audit: lambda self, e: _sqlmesh_ddl_sql(self, e, "AUDIT"),
1381                    DColonCast: lambda self, e: f"{self.sql(e, 'this')}::{self.sql(e, 'to')}",
1382                    Jinja: lambda self, e: e.name,
1383                    JinjaQuery: lambda self, e: f"{JINJA_QUERY_BEGIN};\n{e.name}\n{JINJA_END};",
1384                    JinjaStatement: lambda self, e: (
1385                        f"{JINJA_STATEMENT_BEGIN};\n{e.name}\n{JINJA_END};"
1386                    ),
1387                    VirtualUpdateStatement: lambda self, e: _on_virtual_update_sql(self, e),
1388                    MacroDef: lambda self, e: f"@DEF({self.sql(e.this)}, {self.sql(e.expression)})",
1389                    MacroFunc: _macro_func_sql,
1390                    MacroStrReplace: lambda self, e: f"@{self.sql(e.this)}",
1391                    MacroSQL: lambda self, e: f"@SQL({self.sql(e.this)})",
1392                    MacroVar: lambda self, e: f"@{e.name}",
1393                    Metric: lambda self, e: _sqlmesh_ddl_sql(self, e, "METRIC"),
1394                    Model: lambda self, e: _sqlmesh_ddl_sql(self, e, "MODEL"),
1395                    ModelKind: _model_kind_sql,
1396                    PythonCode: lambda self, e: self.expressions(e, sep="\n", indent=False),
1397                    StagedFilePath: lambda self, e: self.table_sql(e),
1398                    exp.Whens: _whens_sql,
1399                }
1400            )
1401        if MacroDef not in generator.WITH_SEPARATED_COMMENTS:
1402            generator.WITH_SEPARATED_COMMENTS = (
1403                *generator.WITH_SEPARATED_COMMENTS,
1404                Model,
1405                MacroDef,
1406            )
1407
1408        if MacroVar not in generator.UNWRAPPED_INTERVAL_VALUES:
1409            generator.UNWRAPPED_INTERVAL_VALUES = (
1410                *generator.UNWRAPPED_INTERVAL_VALUES,
1411                MacroStrReplace,
1412                MacroVar,
1413            )
1414
1415    _override(Parser, _parse_select)
1416    _override(Parser, _parse_statement)
1417    _override(Parser, _parse_join)
1418    _override(Parser, _parse_order)
1419    _override(Parser, _parse_where)
1420    _override(Parser, _parse_group)
1421    _override(Parser, _parse_with)
1422    _override(Parser, _parse_having)
1423    _override(Parser, _parse_limit)
1424    _override(Parser, _parse_value)
1425    _override(Parser, _parse_lambda)
1426    _override(Parser, _parse_types)
1427    _override(Parser, _parse_if)
1428    _override(TSQL.Parser, Parser._parse_if)
1429    _override(Parser, _parse_id_var)
1430    _override(Parser, _parse_interval_span)
1431    _override(Parser, _warn_unsupported)
1432    _override(Snowflake.Parser, _parse_table_parts)
1433    _override(TSQL.Parser, _parse_alter_table_alter)
1434    _override(TSQL.Generator, altercolumn_sql)
1435
1436    # DuckDB's prefix absolute power operator `@` clashes with the macro syntax
1437    DuckDB.Parser.NO_PAREN_FUNCTION_PARSERS.pop("@", None)
1438
1439
1440def select_from_values(
1441    values: t.List[t.Tuple[t.Any, ...]],
1442    columns_to_types: t.Dict[str, exp.DataType],
1443    batch_size: int = 0,
1444    alias: str = "t",
1445) -> t.Iterator[exp.Select]:
1446    """Generate a VALUES expression that has a select wrapped around it to cast the values to their correct types.
1447
1448    Args:
1449        values: List of values to use for the VALUES expression.
1450        columns_to_types: Mapping of column names to types to assign to the values.
1451        batch_size: The maximum number of tuples per batches. Defaults to sys.maxsize if <= 0.
1452        alias: The alias to assign to the values expression. If not provided then will default to "t"
1453
1454    Returns:
1455        This method operates as a generator and yields a VALUES expression.
1456    """
1457    if batch_size <= 0:
1458        batch_size = sys.maxsize
1459    num_rows = len(values)
1460    for i in range(0, num_rows, batch_size):
1461        yield select_from_values_for_batch_range(
1462            values=values,
1463            target_columns_to_types=columns_to_types,
1464            batch_start=i,
1465            batch_end=min(i + batch_size, num_rows),
1466            alias=alias,
1467        )
1468
1469
1470def select_from_values_for_batch_range(
1471    values: t.List[t.Tuple[t.Any, ...]],
1472    target_columns_to_types: t.Dict[str, exp.DataType],
1473    batch_start: int,
1474    batch_end: int,
1475    alias: str = "t",
1476    source_columns: t.Optional[t.List[str]] = None,
1477) -> exp.Select:
1478    source_columns = source_columns or list(target_columns_to_types)
1479    source_columns_to_types = get_source_columns_to_types(target_columns_to_types, source_columns)
1480
1481    if not values:
1482        # Ensures we don't generate an empty VALUES clause & forces a zero-row output
1483        where = exp.false()
1484        expressions = [
1485            tuple(exp.cast(exp.null(), to=kind) for kind in source_columns_to_types.values())
1486        ]
1487    else:
1488        where = None
1489        expressions = [
1490            tuple(transform_values(v, source_columns_to_types))
1491            for v in values[batch_start:batch_end]
1492        ]
1493
1494    values_exp = exp.values(expressions, alias=alias, columns=source_columns_to_types)
1495    if values:
1496        # BigQuery crashes on `SELECT CAST(x AS TIMESTAMP) FROM UNNEST([NULL]) AS x`, but not
1497        # on `SELECT CAST(x AS TIMESTAMP) FROM UNNEST([CAST(NULL AS TIMESTAMP)]) AS x`. This
1498        # ensures nulls under the `Values` expression are cast to avoid similar issues.
1499        for value, kind in zip(
1500            values_exp.expressions[0].expressions, source_columns_to_types.values()
1501        ):
1502            if isinstance(value, exp.Null):
1503                value.replace(exp.cast(value, to=kind))
1504
1505    casted_columns = [
1506        exp.alias_(
1507            exp.cast(
1508                exp.column(column) if column in source_columns_to_types else exp.Null(), to=kind
1509            ),
1510            column,
1511            copy=False,
1512        )
1513        for column, kind in target_columns_to_types.items()
1514    ]
1515    return exp.select(*casted_columns).from_(values_exp, copy=False).where(where, copy=False)
1516
1517
1518def pandas_to_sql(
1519    df: pd.DataFrame,
1520    columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
1521    batch_size: int = 0,
1522    alias: str = "t",
1523) -> t.Iterator[exp.Select]:
1524    """Convert a pandas dataframe into a VALUES sql statement.
1525
1526    Args:
1527        df: A pandas dataframe to convert.
1528        columns_to_types: Mapping of column names to types to assign to the values.
1529        batch_size: The maximum number of tuples per batches. Defaults to sys.maxsize if <= 0.
1530        alias: The alias to assign to the values expression. If not provided then will default to "t"
1531
1532    Returns:
1533        This method operates as a generator and yields a VALUES expression.
1534    """
1535    yield from select_from_values(
1536        values=list(df.itertuples(index=False, name=None)),
1537        columns_to_types=columns_to_types or columns_to_types_from_df(df),
1538        batch_size=batch_size,
1539        alias=alias,
1540    )
1541
1542
1543def set_default_catalog(
1544    table: str | exp.Table,
1545    default_catalog: t.Optional[str],
1546) -> exp.Table:
1547    table = exp.to_table(table)
1548
1549    if default_catalog and not table.catalog and table.db:
1550        table.set("catalog", exp.parse_identifier(default_catalog))
1551
1552    return table
1553
1554
1555@lru_cache(maxsize=16384)
1556def normalize_model_name(
1557    table: str | exp.Table | exp.Column,
1558    default_catalog: t.Optional[str],
1559    dialect: DialectType = None,
1560) -> str:
1561    if isinstance(table, exp.Column):
1562        table = exp.table_(table.this, db=table.args.get("table"), catalog=table.args.get("db"))
1563    else:
1564        # We are relying on sqlglot's flexible parsing here to accept quotes from other dialects.
1565        # Ex: I have a a normalized name of '"my_table"' but the dialect is spark and therefore we should
1566        # expect spark quotes to be backticks ('`') instead of double quotes ('"'). sqlglot today is flexible
1567        # and will still parse this correctly and we rely on that.
1568        table = exp.to_table(table, dialect=dialect)
1569
1570    table = set_default_catalog(table, default_catalog)
1571    # An alternative way to do this is the following: exp.table_name(table, dialect=dialect, identify=True)
1572    # This though would result in the names being normalized to the target dialect AND the quotes while the below
1573    # approach just normalizes the names.
1574    # By just normalizing names and using sqlglot dialect for quotes this makes it easier for dialects that have
1575    # compatible normalization strategies but incompatible quoting to still work together without user hassle
1576    return exp.table_name(normalize_identifiers(table, dialect=dialect), identify=True)
1577
1578
1579def find_tables(
1580    expression: exp.Expr, default_catalog: t.Optional[str], dialect: DialectType = None
1581) -> t.Set[str]:
1582    """Find all tables referenced in a query.
1583
1584    Caches the result in the meta field 'tables'.
1585
1586    Args:
1587        expressions: The query to find the tables in.
1588        dialect: The dialect to use for normalization of table names.
1589
1590    Returns:
1591        A Set of all the table names.
1592    """
1593    if TABLES_META not in expression.meta:
1594        expression.meta[TABLES_META] = {
1595            normalize_model_name(table, default_catalog=default_catalog, dialect=dialect)
1596            for scope in traverse_scope(expression)
1597            for table in scope.tables
1598            if table.name and table.name not in scope.cte_sources
1599        }
1600    return expression.meta[TABLES_META]
1601
1602
1603def add_table(node: exp.Expr, table: str) -> exp.Expr:
1604    """Add a table to all columns in an expression."""
1605
1606    def _transform(node: exp.Expr) -> exp.Expr:
1607        if isinstance(node, exp.Column) and not node.table:
1608            return exp.column(node.this, table=table)
1609        if isinstance(node, exp.Identifier):
1610            return exp.column(node, table=table)
1611        return node
1612
1613    return node.transform(_transform)
1614
1615
1616def transform_values(
1617    values: t.Tuple[t.Any, ...], columns_to_types: t.Dict[str, exp.DataType]
1618) -> t.Iterator[t.Any]:
1619    """Perform transformations on values given columns_to_types."""
1620
1621    def _transform_value(value: t.Any, dtype: exp.DataType) -> t.Any:
1622        if (
1623            isinstance(value, list)
1624            and dtype.is_type(*exp.DataType.ARRAY_TYPES)
1625            and len(dtype.expressions) == 1
1626        ):
1627            element_type = dtype.expressions[0]
1628            return exp.convert([_transform_value(v, element_type) for v in value])
1629
1630        if (
1631            isinstance(value, dict)
1632            and dtype.is_type(*exp.DataType.STRUCT_TYPES)
1633            and len(value) == len(dtype.expressions)
1634        ):
1635            expressions = []
1636            for (field_name, field_value), field_type in zip(value.items(), dtype.expressions):
1637                if isinstance(field_type, exp.ColumnDef):
1638                    field_type = field_type.kind
1639                else:
1640                    field_type = exp.DataType.build(exp.DataType.Type.UNKNOWN)
1641
1642                expressions.append(
1643                    exp.PropertyEQ(
1644                        this=exp.to_identifier(field_name),
1645                        expression=_transform_value(field_value, field_type),
1646                    )
1647                )
1648
1649            return exp.Struct(expressions=expressions)
1650
1651        if dtype.is_type(exp.DataType.Type.JSON):
1652            return exp.func("PARSE_JSON", f"'{value}'")
1653
1654        return exp.convert(value)
1655
1656    for col_value, col_type in zip(values, columns_to_types.values()):
1657        yield _transform_value(col_value, col_type)
1658
1659
1660def to_schema(sql_path: str | exp.Table, dialect: DialectType = None) -> exp.Table:
1661    if isinstance(sql_path, exp.Table) and sql_path.this is None:
1662        return sql_path
1663    table = exp.to_table(
1664        sql_path.copy() if isinstance(sql_path, exp.Table) else sql_path, dialect=dialect
1665    )
1666    table.set("catalog", table.args.get("db"))
1667    table.set("db", table.args.get("this"))
1668    table.set("this", None)
1669    return table
1670
1671
1672def schema_(
1673    db: exp.Identifier | str,
1674    catalog: t.Optional[exp.Identifier | str] = None,
1675    quoted: t.Optional[bool] = None,
1676) -> exp.Table:
1677    """Build a Schema.
1678
1679    Args:
1680        db: Database name.
1681        catalog: Catalog name.
1682        quoted: Whether to force quotes on the schema's identifiers.
1683
1684    Returns:
1685        The new Schema instance.
1686    """
1687    return exp.Table(
1688        this=None,
1689        db=exp.to_identifier(db, quoted=quoted) if db else None,
1690        catalog=exp.to_identifier(catalog, quoted=quoted) if catalog else None,
1691    )
1692
1693
1694def normalize_mapping_schema(schema: t.Dict, dialect: DialectType) -> MappingSchema:
1695    return MappingSchema(_unquote_schema(schema), dialect=dialect, normalize=False)
1696
1697
1698def _unquote_schema(schema: t.Dict) -> t.Dict:
1699    """SQLGlot schema expects unquoted normalized keys."""
1700    return {
1701        k.strip('"'): _unquote_schema(v) if isinstance(v, dict) else v for k, v in schema.items()
1702    }
1703
1704
1705@contextmanager
1706def normalize_and_quote(
1707    query: E, dialect: DialectType, default_catalog: t.Optional[str], quote: bool = True
1708) -> t.Iterator[E]:
1709    qualify_tables(query, catalog=default_catalog, dialect=dialect)
1710    normalize_identifiers(query, dialect=dialect)
1711    yield query
1712    if quote:
1713        quote_identifiers(query, dialect=dialect)
1714
1715
1716def interpret_expression(e: exp.Expr) -> exp.Expr | str | int | float | bool:
1717    if e.is_int:
1718        return int(e.this)
1719    if e.is_number:
1720        return float(e.this)
1721    if isinstance(e, (exp.Literal, exp.Boolean)):
1722        return e.this
1723    return e
1724
1725
1726def interpret_key_value_pairs(
1727    e: exp.Tuple,
1728) -> t.Dict[str, exp.Expr | str | int | float | bool]:
1729    return {i.this.name: interpret_expression(i.expression) for i in e.expressions}
1730
1731
1732def extract_func_call(
1733    v: exp.Expr, allow_tuples: bool = False
1734) -> t.Tuple[str, t.Dict[str, exp.Expr]]:
1735    kwargs = {}
1736
1737    if isinstance(v, exp.Anonymous):
1738        func = v.name
1739        args = v.expressions
1740    elif isinstance(v, exp.Func):
1741        func = v.sql_name()
1742        args = list(v.args.values())
1743    elif isinstance(v, exp.Paren):
1744        func = ""
1745        args = [v.this]
1746    elif isinstance(v, exp.Tuple):  # airflow only
1747        if not allow_tuples:
1748            raise ConfigError("Audit name is missing (eg. MY_AUDIT())")
1749
1750        func = ""
1751        args = v.expressions
1752    else:
1753        return v.name.lower(), {}
1754
1755    for arg in args:
1756        if not isinstance(arg, (exp.PropertyEQ, exp.EQ)):
1757            raise ConfigError(
1758                f"Function '{func}' must be called with key-value arguments like {func}(arg := value)."
1759            )
1760        kwargs[arg.left.name.lower()] = arg.right
1761    return func.lower(), kwargs
1762
1763
1764def extract_function_calls(func_calls: t.Any, allow_tuples: bool = False) -> t.Any:
1765    """Used for extracting function calls for signals or audits."""
1766
1767    if isinstance(func_calls, (exp.Tuple, exp.Array)):
1768        return [extract_func_call(i, allow_tuples=allow_tuples) for i in func_calls.expressions]
1769    if isinstance(func_calls, exp.Paren):
1770        return [extract_func_call(func_calls.this, allow_tuples=allow_tuples)]
1771    if isinstance(func_calls, exp.Expr):
1772        return [extract_func_call(func_calls, allow_tuples=allow_tuples)]
1773    if isinstance(func_calls, list):
1774        function_calls = []
1775        for entry in func_calls:
1776            if isinstance(entry, dict):
1777                args = entry
1778                name = "" if allow_tuples else entry.pop("name")
1779            elif isinstance(entry, (tuple, list)):
1780                name, args = entry
1781            else:
1782                raise ConfigError(f"Audit must be a dictionary or named tuple. Got {entry}.")
1783
1784            function_calls.append(
1785                (
1786                    name.lower(),
1787                    {
1788                        key: parse_one(value) if isinstance(value, str) else value
1789                        for key, value in args.items()
1790                    },
1791                )
1792            )
1793
1794        return function_calls
1795
1796    return func_calls or []
1797
1798
1799def is_meta_expression(v: t.Any) -> bool:
1800    return isinstance(v, (Audit, Metric, Model))
1801
1802
1803def replace_merge_table_aliases(expression: exp.Expr, dialect: t.Optional[str] = None) -> exp.Expr:
1804    """
1805    Resolves references from the "source" and "target" tables (or their DBT equivalents)
1806    with the corresponding SQLMesh merge aliases (MERGE_SOURCE_ALIAS and MERGE_TARGET_ALIAS)
1807    """
1808    from sqlmesh.core.engine_adapter.base import MERGE_SOURCE_ALIAS, MERGE_TARGET_ALIAS
1809
1810    if isinstance(expression, exp.Column) and (first_part := expression.parts[0]):
1811        if first_part.this.lower() in ("target", "dbt_internal_dest", "__merge_target__"):
1812            first_part.replace(exp.to_identifier(MERGE_TARGET_ALIAS, quoted=True))
1813        elif first_part.this.lower() in ("source", "dbt_internal_source", "__merge_source__"):
1814            first_part.replace(exp.to_identifier(MERGE_SOURCE_ALIAS, quoted=True))
1815
1816    return expression