Fix: Dont normalize aliases in merge and when matched

Author: themisvaltinos

sqlmesh/core/dialect.py

@@ -1412,18 +1412,10 @@ def replace_merge_table_aliases(
     """
     from sqlmesh.core.engine_adapter.base import MERGE_SOURCE_ALIAS, MERGE_TARGET_ALIAS
 
-    normalized_merge_source_alias = quote_identifiers(
-        normalize_identifiers(exp.to_identifier(MERGE_SOURCE_ALIAS), dialect), dialect=dialect
-    )
-
-    normalized_merge_target_alias = quote_identifiers(
-        normalize_identifiers(exp.to_identifier(MERGE_TARGET_ALIAS), dialect), dialect=dialect
-    )
-
     if isinstance(expression, exp.Column) and (first_part := expression.parts[0]):
         if first_part.this.lower() in ("target", "dbt_internal_dest", "__merge_target__"):
-            first_part.replace(normalized_merge_target_alias)
+            first_part.replace(exp.to_identifier(MERGE_TARGET_ALIAS, quoted=True))
         elif first_part.this.lower() in ("source", "dbt_internal_source", "__merge_source__"):
-            first_part.replace(normalized_merge_source_alias)
+            first_part.replace(exp.to_identifier(MERGE_SOURCE_ALIAS, quoted=True))
 
     return expression

sqlmesh/core/model/kind.py

@@ -478,10 +478,9 @@ def _when_matched_validator(
                 v = v[1:-1]
 
             v = t.cast(exp.Whens, d.parse_one(v, into=exp.Whens, dialect=dialect))
-        else:
-            v = t.cast(exp.Whens, v.transform(d.replace_merge_table_aliases, dialect=dialect))
 
-        return validate_expression(v, dialect=dialect)
+        v = validate_expression(v, dialect=dialect)
+        return t.cast(exp.Whens, v.transform(d.replace_merge_table_aliases, dialect=dialect))
 
     @field_validator("merge_filter", mode="before")
     def _merge_filter_validator(
@@ -497,10 +496,9 @@ def _merge_filter_validator(
         if isinstance(v, str):
             v = v.strip()
             v = d.parse_one(v, dialect=dialect)
-        else:
-            v = v.transform(d.replace_merge_table_aliases, dialect=dialect)
 
-        return validate_expression(v, dialect=dialect)
+        v = validate_expression(v, dialect=dialect)
+        return v.transform(d.replace_merge_table_aliases, dialect=dialect)
 
     @property
     def data_hash_values(self) -> t.List[t.Optional[str]]:

tests/core/test_model.py

@@ -5370,7 +5370,7 @@ def test_when_matched():
     """
     )
 
-    expected_when_matched = "(WHEN MATCHED THEN UPDATE SET `__merge_target__`.`salary` = COALESCE(`__merge_source__`.`salary`, `__merge_target__`.`salary`))"
+    expected_when_matched = "(WHEN MATCHED THEN UPDATE SET `__MERGE_TARGET__`.`salary` = COALESCE(`__MERGE_SOURCE__`.`salary`, `__MERGE_TARGET__`.`salary`))"
 
     model = load_sql_based_model(expressions, dialect="hive")
     assert model.kind.when_matched.sql(dialect="hive") == expected_when_matched
@@ -5404,9 +5404,9 @@ def test_when_matched():
   kind INCREMENTAL_BY_UNIQUE_KEY (
     unique_key ("purchase_order_id"),
     when_matched (
-      WHEN MATCHED AND "__merge_source__"."_operation" = 1 THEN DELETE
-      WHEN MATCHED AND "__merge_source__"."_operation" <> 1 THEN UPDATE SET
-        "__merge_target__"."purchase_order_id" = 1
+      WHEN MATCHED AND "__MERGE_SOURCE__"."_operation" = 1 THEN DELETE
+      WHEN MATCHED AND "__MERGE_SOURCE__"."_operation" <> 1 THEN UPDATE SET
+        "__MERGE_TARGET__"."purchase_order_id" = 1
     ),
     batch_concurrency 1,
     forward_only FALSE,
@@ -5457,7 +5457,7 @@ def fingerprint_merge(
   kind INCREMENTAL_BY_UNIQUE_KEY (
     unique_key ("purchase_order_id"),
     when_matched (
-      WHEN MATCHED AND "__merge_source__"."salary" <> "__merge_target__"."salary" THEN UPDATE SET
+      WHEN MATCHED AND "__MERGE_SOURCE__"."salary" <> "__MERGE_TARGET__"."salary" THEN UPDATE SET
         ARRAY('target.update_datetime = source.update_datetime', 'target.salary = source.salary')
     ),
     batch_concurrency 1,
@@ -5491,8 +5491,8 @@ def test_when_matched_multiple():
     )
 
     expected_when_matched = [
-        "WHEN MATCHED AND `__merge_source__`.`x` = 1 THEN UPDATE SET `__merge_target__`.`salary` = COALESCE(`__merge_source__`.`salary`, `__merge_target__`.`salary`)",
-        "WHEN MATCHED THEN UPDATE SET `__merge_target__`.`salary` = COALESCE(`__merge_source__`.`salary`, `__merge_target__`.`salary`)",
+        "WHEN MATCHED AND `__MERGE_SOURCE__`.`x` = 1 THEN UPDATE SET `__MERGE_TARGET__`.`salary` = COALESCE(`__MERGE_SOURCE__`.`salary`, `__MERGE_TARGET__`.`salary`)",
+        "WHEN MATCHED THEN UPDATE SET `__MERGE_TARGET__`.`salary` = COALESCE(`__MERGE_SOURCE__`.`salary`, `__MERGE_TARGET__`.`salary`)",
     ]
 
     model = load_sql_based_model(expressions, dialect="hive", variables={"schema": "db"})
@@ -5533,13 +5533,13 @@ def test_when_matched_merge_filter_multi_part_columns():
     )
 
     expected_when_matched = [
-        "WHEN MATCHED AND `__merge_source__`.`record`.`nested_record`.`field` = 1 THEN UPDATE SET `__merge_target__`.`repeated_record`.`sub_repeated_record`.`sub_field` = COALESCE(`__merge_source__`.`repeated_record`.`sub_repeated_record`.`sub_field`, `__merge_target__`.`repeated_record`.`sub_repeated_record`.`sub_field`)",
-        "WHEN MATCHED THEN UPDATE SET `__merge_target__`.`repeated_record`.`sub_repeated_record`.`sub_field` = COALESCE(`__merge_source__`.`repeated_record`.`sub_repeated_record`.`sub_field`, `__merge_target__`.`repeated_record`.`sub_repeated_record`.`sub_field`)",
+        "WHEN MATCHED AND `__MERGE_SOURCE__`.`record`.`nested_record`.`field` = 1 THEN UPDATE SET `__MERGE_TARGET__`.`repeated_record`.`sub_repeated_record`.`sub_field` = COALESCE(`__MERGE_SOURCE__`.`repeated_record`.`sub_repeated_record`.`sub_field`, `__MERGE_TARGET__`.`repeated_record`.`sub_repeated_record`.`sub_field`)",
+        "WHEN MATCHED THEN UPDATE SET `__MERGE_TARGET__`.`repeated_record`.`sub_repeated_record`.`sub_field` = COALESCE(`__MERGE_SOURCE__`.`repeated_record`.`sub_repeated_record`.`sub_field`, `__MERGE_TARGET__`.`repeated_record`.`sub_repeated_record`.`sub_field`)",
     ]
 
     expected_merge_filter = (
-        "`__merge_source__`.`record`.`nested_record`.`field` < `__merge_target__`.`record`.`nested_record`.`field` AND "
-        "`__merge_target__`.`repeated_record`.`sub_repeated_record`.`sub_field` > `__merge_source__`.`repeated_record`.`sub_repeated_record`.`sub_field`"
+        "`__MERGE_SOURCE__`.`record`.`nested_record`.`field` < `__MERGE_TARGET__`.`record`.`nested_record`.`field` AND "
+        "`__MERGE_TARGET__`.`repeated_record`.`sub_repeated_record`.`sub_field` > `__MERGE_SOURCE__`.`repeated_record`.`sub_repeated_record`.`sub_field`"
     )
 
     model = load_sql_based_model(expressions, dialect="bigquery", variables={"schema": "db"})
@@ -5568,7 +5568,7 @@ def test_when_matched_normalization() -> None:
             when_matched (
                 WHEN MATCHED THEN UPDATE SET
                     target.key_a = source.key_a,
-                    target.key_b = source.key_b,            
+                    target.key_b = source.key_b,
             )
           )
         );
@@ -6569,7 +6569,7 @@ def test_unrendered_macros_sql_model(mocker: MockerFixture) -> None:
     assert model.unique_key[0] == exp.column("a", quoted=True)
     assert (
         t.cast(exp.Expression, model.merge_filter).sql()
-        == '"__merge_source__"."id" > 0 AND "__merge_target__"."updated_at" < @end_ds AND "__merge_source__"."updated_at" > @start_ds AND @merge_filter_var'
+        == '"__MERGE_SOURCE__"."id" > 0 AND "__MERGE_TARGET__"."updated_at" < @end_ds AND "__MERGE_SOURCE__"."updated_at" > @start_ds AND @merge_filter_var'
     )
 
 
@@ -6665,7 +6665,7 @@ def model_with_macros(evaluator, **kwargs):
     assert python_sql_model.unique_key[0] == exp.column("a", quoted=True)
     assert (
         python_sql_model.merge_filter.sql()
-        == '"__merge_source__"."id" > 0 AND "__merge_target__"."updated_at" < @end_ds AND "__merge_source__"."updated_at" > @start_ds AND @merge_filter_var'
+        == '"__MERGE_SOURCE__"."id" > 0 AND "__MERGE_TARGET__"."updated_at" < @end_ds AND "__MERGE_SOURCE__"."updated_at" > @start_ds AND @merge_filter_var'
     )
 
 
@@ -7752,7 +7752,7 @@ def test_model_kind_to_expression():
         .sql()
         == """INCREMENTAL_BY_UNIQUE_KEY (
 unique_key ("a"),
-when_matched (WHEN MATCHED THEN UPDATE SET "__merge_target__"."b" = COALESCE("__merge_source__"."b", "__merge_target__"."b")),
+when_matched (WHEN MATCHED THEN UPDATE SET "__MERGE_TARGET__"."b" = COALESCE("__MERGE_SOURCE__"."b", "__MERGE_TARGET__"."b")),
 batch_concurrency 1,
 forward_only FALSE,
 disable_restatement FALSE,
@@ -7780,7 +7780,7 @@ def test_model_kind_to_expression():
         .sql()
         == """INCREMENTAL_BY_UNIQUE_KEY (
 unique_key ("a"),
-when_matched (WHEN MATCHED AND "__merge_source__"."x" = 1 THEN UPDATE SET "__merge_target__"."b" = COALESCE("__merge_source__"."b", "__merge_target__"."b") WHEN MATCHED THEN UPDATE SET "__merge_target__"."b" = COALESCE("__merge_source__"."b", "__merge_target__"."b")),
+when_matched (WHEN MATCHED AND "__MERGE_SOURCE__"."x" = 1 THEN UPDATE SET "__MERGE_TARGET__"."b" = COALESCE("__MERGE_SOURCE__"."b", "__MERGE_TARGET__"."b") WHEN MATCHED THEN UPDATE SET "__MERGE_TARGET__"."b" = COALESCE("__MERGE_SOURCE__"."b", "__MERGE_TARGET__"."b")),
 batch_concurrency 1,
 forward_only FALSE,
 disable_restatement FALSE,
@@ -8041,7 +8041,7 @@ def test_merge_filter():
     """
     )
 
-    expected_incremental_predicate = f"`{MERGE_SOURCE_ALIAS.lower()}`.`salary` > 0"
+    expected_incremental_predicate = f"`{MERGE_SOURCE_ALIAS}`.`salary` > 0"
 
     model = load_sql_based_model(expressions, dialect="hive")
     assert model.kind.merge_filter.sql(dialect="hive") == expected_incremental_predicate
@@ -8084,19 +8084,19 @@ def test_merge_filter():
   kind INCREMENTAL_BY_UNIQUE_KEY (
     unique_key ("purchase_order_id"),
     when_matched (
-      WHEN MATCHED AND "{MERGE_SOURCE_ALIAS.lower()}"."_operation" = 1 THEN DELETE
-      WHEN MATCHED AND "{MERGE_SOURCE_ALIAS.lower()}"."_operation" <> 1 THEN UPDATE SET
-        "{MERGE_TARGET_ALIAS.lower()}"."purchase_order_id" = 1
+      WHEN MATCHED AND "{MERGE_SOURCE_ALIAS}"."_operation" = 1 THEN DELETE
+      WHEN MATCHED AND "{MERGE_SOURCE_ALIAS}"."_operation" <> 1 THEN UPDATE SET
+        "{MERGE_TARGET_ALIAS}"."purchase_order_id" = 1
     ),
     merge_filter (
-      "{MERGE_SOURCE_ALIAS.lower()}"."ds" > (
+      "{MERGE_SOURCE_ALIAS}"."ds" > (
         SELECT
           MAX("ds")
         FROM "db"."test"
       )
-      AND "{MERGE_SOURCE_ALIAS.lower()}"."ds" > @start_ds
-      AND "{MERGE_SOURCE_ALIAS.lower()}"."_operation" <> 1
-      AND "{MERGE_TARGET_ALIAS.lower()}"."start_date" > CURRENT_DATE + INTERVAL '7' DAY
+      AND "{MERGE_SOURCE_ALIAS}"."ds" > @start_ds
+      AND "{MERGE_SOURCE_ALIAS}"."_operation" <> 1
+      AND "{MERGE_TARGET_ALIAS}"."start_date" > CURRENT_DATE + INTERVAL '7' DAY
     ),
     batch_concurrency 1,
     forward_only FALSE,
@@ -8114,7 +8114,7 @@ def test_merge_filter():
     rendered_merge_filters = model.render_merge_filter(start="2023-01-01", end="2023-01-02")
     assert (
         rendered_merge_filters.sql(dialect="hive")
-        == "(`__merge_source__`.`ds` > (SELECT MAX(`ds`) FROM `db`.`test`) AND `__merge_source__`.`ds` > '2023-01-01' AND `__merge_source__`.`_operation` <> 1 AND `__merge_target__`.`start_date` > CURRENT_DATE + INTERVAL '7' DAY)"
+        == "(`__MERGE_SOURCE__`.`ds` > (SELECT MAX(`ds`) FROM `db`.`test`) AND `__MERGE_SOURCE__`.`ds` > '2023-01-01' AND `__MERGE_SOURCE__`.`_operation` <> 1 AND `__MERGE_TARGET__`.`start_date` > CURRENT_DATE + INTERVAL '7' DAY)"
     )
 
 

tests/core/test_snapshot_evaluator.py

@@ -2200,18 +2200,14 @@ def test_create_incremental_by_unique_key_updated_at_exp(adapter_mock, make_snap
                     source=False,
                     then=exp.Update(
                         expressions=[
-                            exp.column("name", MERGE_TARGET_ALIAS.lower(), quoted=True).eq(
-                                exp.column("name", MERGE_SOURCE_ALIAS.lower(), quoted=True)
+                            exp.column("name", MERGE_TARGET_ALIAS, quoted=True).eq(
+                                exp.column("name", MERGE_SOURCE_ALIAS, quoted=True)
                             ),
-                            exp.column("updated_at", MERGE_TARGET_ALIAS.lower(), quoted=True).eq(
+                            exp.column("updated_at", MERGE_TARGET_ALIAS, quoted=True).eq(
                                 exp.Coalesce(
-                                    this=exp.column(
-                                        "updated_at", MERGE_SOURCE_ALIAS.lower(), quoted=True
-                                    ),
+                                    this=exp.column("updated_at", MERGE_SOURCE_ALIAS, quoted=True),
                                     expressions=[
-                                        exp.column(
-                                            "updated_at", MERGE_TARGET_ALIAS.lower(), quoted=True
-                                        )
+                                        exp.column("updated_at", MERGE_TARGET_ALIAS, quoted=True)
                                     ],
                                 )
                             ),
@@ -2269,23 +2265,19 @@ def test_create_incremental_by_unique_key_multiple_updated_at_exp(adapter_mock,
             expressions=[
                 exp.When(
                     matched=True,
-                    condition=exp.column("id", MERGE_SOURCE_ALIAS.lower(), quoted=True).eq(
+                    condition=exp.column("id", MERGE_SOURCE_ALIAS, quoted=True).eq(
                         exp.Literal.number(1)
                     ),
                     then=exp.Update(
                         expressions=[
-                            exp.column("name", MERGE_TARGET_ALIAS.lower(), quoted=True).eq(
-                                exp.column("name", MERGE_SOURCE_ALIAS.lower(), quoted=True)
+                            exp.column("name", MERGE_TARGET_ALIAS, quoted=True).eq(
+                                exp.column("name", MERGE_SOURCE_ALIAS, quoted=True)
                             ),
-                            exp.column("updated_at", MERGE_TARGET_ALIAS.lower(), quoted=True).eq(
+                            exp.column("updated_at", MERGE_TARGET_ALIAS, quoted=True).eq(
                                 exp.Coalesce(
-                                    this=exp.column(
-                                        "updated_at", MERGE_SOURCE_ALIAS.lower(), quoted=True
-                                    ),
+                                    this=exp.column("updated_at", MERGE_SOURCE_ALIAS, quoted=True),
                                     expressions=[
-                                        exp.column(
-                                            "updated_at", MERGE_TARGET_ALIAS.lower(), quoted=True
-                                        )
+                                        exp.column("updated_at", MERGE_TARGET_ALIAS, quoted=True)
                                     ],
                                 )
                             ),
@@ -2297,18 +2289,14 @@ def test_create_incremental_by_unique_key_multiple_updated_at_exp(adapter_mock,
                     source=False,
                     then=exp.Update(
                         expressions=[
-                            exp.column("name", MERGE_TARGET_ALIAS.lower(), quoted=True).eq(
-                                exp.column("name", MERGE_SOURCE_ALIAS.lower(), quoted=True)
+                            exp.column("name", MERGE_TARGET_ALIAS, quoted=True).eq(
+                                exp.column("name", MERGE_SOURCE_ALIAS, quoted=True)
                             ),
-                            exp.column("updated_at", MERGE_TARGET_ALIAS.lower(), quoted=True).eq(
+                            exp.column("updated_at", MERGE_TARGET_ALIAS, quoted=True).eq(
                                 exp.Coalesce(
-                                    this=exp.column(
-                                        "updated_at", MERGE_SOURCE_ALIAS.lower(), quoted=True
-                                    ),
+                                    this=exp.column("updated_at", MERGE_SOURCE_ALIAS, quoted=True),
                                     expressions=[
-                                        exp.column(
-                                            "updated_at", MERGE_TARGET_ALIAS.lower(), quoted=True
-                                        )
+                                        exp.column("updated_at", MERGE_TARGET_ALIAS, quoted=True)
                                     ],
                                 )
                             ),
@@ -2395,16 +2383,16 @@ def test_create_incremental_by_unique_key_merge_filter(adapter_mock, make_snapsh
     assert model.merge_filter == exp.And(
         this=exp.And(
             this=exp.GT(
-                this=exp.column("id", MERGE_SOURCE_ALIAS.lower(), quoted=True),
+                this=exp.column("id", MERGE_SOURCE_ALIAS, quoted=True),
                 expression=exp.Literal(this="0", is_string=False),
             ),
             expression=exp.LT(
-                this=exp.column("updated_at", MERGE_TARGET_ALIAS.lower(), quoted=True),
+                this=exp.column("updated_at", MERGE_TARGET_ALIAS, quoted=True),
                 expression=d.MacroVar(this="end_ds"),
             ),
         ),
         expression=exp.GT(
-            this=exp.column("updated_at", MERGE_SOURCE_ALIAS.lower(), quoted=True),
+            this=exp.column("updated_at", MERGE_SOURCE_ALIAS, quoted=True),
             expression=d.MacroVar(this="start_ds"),
         ),
     )
@@ -2436,15 +2424,11 @@ def test_create_incremental_by_unique_key_merge_filter(adapter_mock, make_snapsh
                     matched=True,
                     then=exp.Update(
                         expressions=[
-                            exp.column("updated_at", MERGE_TARGET_ALIAS.lower(), quoted=True).eq(
+                            exp.column("updated_at", MERGE_TARGET_ALIAS, quoted=True).eq(
                                 exp.Coalesce(
-                                    this=exp.column(
-                                        "updated_at", MERGE_SOURCE_ALIAS.lower(), quoted=True
-                                    ),
+                                    this=exp.column("updated_at", MERGE_SOURCE_ALIAS, quoted=True),
                                     expressions=[
-                                        exp.column(
-                                            "updated_at", MERGE_TARGET_ALIAS.lower(), quoted=True
-                                        )
+                                        exp.column("updated_at", MERGE_TARGET_ALIAS, quoted=True)
                                     ],
                                 )
                             ),
@@ -2456,16 +2440,16 @@ def test_create_incremental_by_unique_key_merge_filter(adapter_mock, make_snapsh
         merge_filter=exp.And(
             this=exp.And(
                 this=exp.GT(
-                    this=exp.column("id", MERGE_SOURCE_ALIAS.lower(), quoted=True),
+                    this=exp.column("id", MERGE_SOURCE_ALIAS, quoted=True),
                     expression=exp.Literal(this="0", is_string=False),
                 ),
                 expression=exp.LT(
-                    this=exp.column("updated_at", MERGE_TARGET_ALIAS.lower(), quoted=True),
+                    this=exp.column("updated_at", MERGE_TARGET_ALIAS, quoted=True),
                     expression=exp.Literal(this="2020-01-02", is_string=True),
                 ),
             ),
             expression=exp.GT(
-                this=exp.column("updated_at", MERGE_SOURCE_ALIAS.lower(), quoted=True),
+                this=exp.column("updated_at", MERGE_SOURCE_ALIAS, quoted=True),
                 expression=exp.Literal(this="2020-01-01", is_string=True),
             ),
         ),

tests/dbt/test_config.py

@@ -135,7 +135,7 @@ def test_model_to_sqlmesh_fields():
     assert kind.on_destructive_change == OnDestructiveChange.ALLOW
     assert (
         kind.merge_filter.sql(dialect=model.dialect)
-        == """55 > "__merge_source__"."b" AND "__merge_target__"."session_start" > CURRENT_DATE + INTERVAL '7' DAY"""
+        == """55 > "__MERGE_SOURCE__"."b" AND "__MERGE_TARGET__"."session_start" > CURRENT_DATE + INTERVAL '7' DAY"""
     )
 
     model = model_config.update_with({"dialect": "snowflake"}).to_sqlmesh(context)