diff --git a/sqlmesh/core/dialect.py b/sqlmesh/core/dialect.py
index ed904cc4b3..6f5e3745fd 100644
--- a/sqlmesh/core/dialect.py
+++ b/sqlmesh/core/dialect.py
@@ -174,6 +174,7 @@ def _parse_id_var(
 
     while (
         identifier
+        and not identifier.args.get("quoted")
         and self._is_connected()
         and (
             self._match_texts(("{", SQLMESH_MACRO_PREFIX))
diff --git a/sqlmesh/core/macros.py b/sqlmesh/core/macros.py
index 9e7df5d111..554638aec7 100644
--- a/sqlmesh/core/macros.py
+++ b/sqlmesh/core/macros.py
@@ -1128,7 +1128,7 @@ def haversine_distance(
 def pivot(
     evaluator: MacroEvaluator,
     column: SQL,
-    values: t.List[SQL],
+    values: t.List[exp.Expression],
     alias: bool = True,
     agg: exp.Expression = exp.Literal.string("SUM"),
     cmp: exp.Expression = exp.Literal.string("="),
@@ -1146,10 +1146,10 @@ def pivot(
         >>> from sqlmesh.core.macros import MacroEvaluator
         >>> sql = "SELECT date_day, @PIVOT(status, ['cancelled', 'completed']) FROM rides GROUP BY 1"
         >>> MacroEvaluator().transform(parse_one(sql)).sql()
-        'SELECT date_day, SUM(CASE WHEN status = \\'cancelled\\' THEN 1 ELSE 0 END) AS "\\'cancelled\\'", SUM(CASE WHEN status = \\'completed\\' THEN 1 ELSE 0 END) AS "\\'completed\\'" FROM rides GROUP BY 1'
+        'SELECT date_day, SUM(CASE WHEN status = \\'cancelled\\' THEN 1 ELSE 0 END) AS "cancelled", SUM(CASE WHEN status = \\'completed\\' THEN 1 ELSE 0 END) AS "completed" FROM rides GROUP BY 1'
         >>> sql = "SELECT @PIVOT(a, ['v'], then_value := tv, suffix := '_sfx', quote := FALSE)"
         >>> MacroEvaluator(dialect="bigquery").transform(parse_one(sql)).sql("bigquery")
-        "SELECT SUM(CASE WHEN a = 'v' THEN tv ELSE 0 END) AS `v_sfx`"
+        "SELECT SUM(CASE WHEN a = 'v' THEN tv ELSE 0 END) AS v_sfx"
     """
     aggregates: t.List[exp.Expression] = []
     for value in values:
@@ -1157,12 +1157,12 @@ def pivot(
         if distinct:
             proj += "DISTINCT "
 
-        proj += f"CASE WHEN {column} {cmp.name} {value} THEN {then_value} ELSE {else_value} END) "
+        proj += f"CASE WHEN {column} {cmp.name} {value.sql(evaluator.dialect)} THEN {then_value} ELSE {else_value} END) "
         node = evaluator.parse_one(proj)
 
         if alias:
             node = node.as_(
-                f"{prefix.name}{value}{suffix.name}",
+                f"{prefix.name}{value.name}{suffix.name}",
                 quoted=quote,
                 copy=False,
                 dialect=evaluator.dialect,
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
index 961062fe45..8528dd4d1c 100644
--- a/sqlmesh/core/snapshot/evaluator.py
+++ b/sqlmesh/core/snapshot/evaluator.py
@@ -1110,7 +1110,10 @@ def _migrate_target_table(
     ) -> None:
         adapter = self.get_adapter(snapshot.model.gateway)
 
-        tmp_table_name = f"{target_table_name}_schema_tmp"
+        target_table = exp.to_table(target_table_name)
+        target_table.this.set("this", f"{target_table.name}_schema_tmp")
+
+        tmp_table_name = target_table.sql()
         if snapshot.is_materialized:
             self._execute_create(
                 snapshot=snapshot,
diff --git a/tests/core/engine_adapter/test_clickhouse.py b/tests/core/engine_adapter/test_clickhouse.py
index 39e317c7fa..188ae7f394 100644
--- a/tests/core/engine_adapter/test_clickhouse.py
+++ b/tests/core/engine_adapter/test_clickhouse.py
@@ -640,7 +640,7 @@ def test_scd_type_2_by_time(
     "test_valid_from",
     "test_valid_to",
     TRUE AS "_exists"
-  FROM ""__temp_target_efgh""
+  FROM "__temp_target_efgh"
   WHERE
     NOT "test_valid_to" IS NULL
 ), "latest" AS (
@@ -652,7 +652,7 @@ def test_scd_type_2_by_time(
     "test_valid_from",
     "test_valid_to",
     TRUE AS "_exists"
-  FROM ""__temp_target_efgh""
+  FROM "__temp_target_efgh"
   WHERE
     "test_valid_to" IS NULL
 ), "deleted" AS (
diff --git a/tests/core/test_dialect.py b/tests/core/test_dialect.py
index 11ffec3720..58e372c634 100644
--- a/tests/core/test_dialect.py
+++ b/tests/core/test_dialect.py
@@ -717,3 +717,8 @@ def test_sqlglot_extended_correctly(dialect: str) -> None:
     assert isinstance(value, exp.Table)
     assert value.sql() == "foo"
     assert ast.sql(dialect=dialect) == "MODEL (\nname foo\n)"
+
+
+def test_connected_identifier():
+    ast = d.parse_one("""SELECT ("x"at time zone 'utc')::timestamp as x""", "redshift")
+    assert ast.sql("redshift") == """SELECT CAST(("x" AT TIME ZONE 'utc') AS TIMESTAMP) AS x"""
diff --git a/tests/core/test_format.py b/tests/core/test_format.py
index 9b51220a9f..7d544eadf0 100644
--- a/tests/core/test_format.py
+++ b/tests/core/test_format.py
@@ -28,7 +28,7 @@ def test_format_files(tmp_path: pathlib.Path, mocker: MockerFixture):
     f3 = create_temp_file(
         tmp_path,
         pathlib.Path(audits_dir, "audit_1.sql"),
-        "AUDIT(name assert_positive_id, dialect 'duckdb'); SELECT  * FROM @this_model WHERE  \"CaseSensitive\"_item_id < 0;",
+        "AUDIT(name assert_positive_id, dialect 'duckdb'); SELECT  * FROM @this_model WHERE  \"CaseSensitive_item_id\" < 0;",
     )
     f4 = create_temp_file(
         tmp_path,