Fix: #1435 introduced invalid sql for querying the state sync (#1479)

Author: tobymao

sqlmesh/core/state_sync/engine_adapter.py

@@ -215,7 +215,7 @@ def invalidate_environment(self, name: str) -> None:
         if name == c.PROD:
             raise SQLMeshError("Cannot invalidate the production environment.")
 
-        filter_expr = exp.to_column("name").eq(name)
+        filter_expr = exp.column("name").eq(name)
 
         self.engine_adapter.update_table(
             self.environments_table,
@@ -226,7 +226,7 @@ def invalidate_environment(self, name: str) -> None:
     def delete_expired_environments(self) -> t.List[Environment]:
         now_ts = now_timestamp()
         filter_expr = exp.LTE(
-            this=exp.to_column("expiration_ts"),
+            this=exp.column("expiration_ts"),
             expression=exp.Literal.number(now_ts),
         )
 
@@ -249,19 +249,16 @@ def delete_expired_environments(self) -> t.List[Environment]:
 
     def delete_snapshots(self, snapshot_ids: t.Iterable[SnapshotIdLike]) -> None:
         self.engine_adapter.delete_from(
-            self.snapshots_table, where=self._snapshot_id_filter(snapshot_ids, self.snapshots_table)
+            self.snapshots_table, where=self._snapshot_id_filter(snapshot_ids)
         )
 
     def snapshots_exist(self, snapshot_ids: t.Iterable[SnapshotIdLike]) -> t.Set[SnapshotId]:
         return {
             SnapshotId(name=name, identifier=identifier)
             for name, identifier in self.engine_adapter.fetchall(
-                exp.select(
-                    exp.to_column(f"{self.snapshots_table}.name"),
-                    exp.to_column(f"{self.snapshots_table}.identifier"),
-                )
+                exp.select("name", "identifier")
                 .from_(self.snapshots_table)
-                .where(self._snapshot_id_filter(snapshot_ids, self.snapshots_table)),
+                .where(self._snapshot_id_filter(snapshot_ids)),
                 quote_identifiers=True,
             )
         }
@@ -293,7 +290,7 @@ def _update_environment(self, environment: Environment) -> None:
         self.engine_adapter.delete_from(
             self.environments_table,
             where=exp.EQ(
-                this=exp.to_column("name"),
+                this=exp.column("name"),
                 expression=exp.Literal.string(environment.name),
             ),
         )
@@ -309,7 +306,7 @@ def _update_snapshot(self, snapshot: Snapshot) -> None:
         self.engine_adapter.update_table(
             self.snapshots_table,
             {"snapshot": snapshot.json()},
-            where=self._snapshot_id_filter([snapshot.snapshot_id], self.snapshots_table),
+            where=self._snapshot_id_filter([snapshot.snapshot_id]),
             contains_json=True,
         )
 
@@ -362,24 +359,18 @@ def _get_snapshots(
             A dictionary of snapshot ids to snapshots for ones that could be found.
         """
         query = (
-            exp.select(exp.to_column(f"{self.snapshots_table}.snapshot"))
-            .from_(self.snapshots_table)
-            .where(
-                self._snapshot_id_filter(snapshot_ids, self.snapshots_table)
-                if snapshot_ids
-                else None
-            )
+            exp.select(exp.column("snapshot", table="snapshots"))
+            .from_(exp.to_table(self.snapshots_table).as_("snapshots"))
+            .where(self._snapshot_id_filter(snapshot_ids, "snapshots") if snapshot_ids else None)
         )
         if hydrate_seeds:
-            query = query.select(exp.to_column(f"{self.seeds_table}.content")).join(
-                self.seeds_table,
+            query = query.select(exp.column("content", table="seeds")).join(
+                exp.to_table(self.seeds_table).as_("seeds"),
                 on=exp.and_(
-                    *[
-                        exp.to_column(f"{self.snapshots_table}.{col}").eq(
-                            exp.to_column(f"{self.seeds_table}.{col}")
-                        )
-                        for col in ["name", "identifier"]
-                    ]
+                    exp.column("name", table="snapshots").eq(exp.column("name", table="seeds")),
+                    exp.column("identifier", table="snapshots").eq(
+                        exp.column("identifier", table="seeds")
+                    ),
                 ),
                 join_type="left",
             )
@@ -436,9 +427,9 @@ def _get_snapshots_with_same_version(
             return []
 
         query = (
-            exp.select(exp.to_column(f"{self.snapshots_table}.snapshot"))
-            .from_(self.snapshots_table)
-            .where(self._snapshot_name_version_filter(snapshots, self.snapshots_table))
+            exp.select("snapshot")
+            .from_(exp.to_table(self.snapshots_table).as_("snapshots"))
+            .where(self._snapshot_name_version_filter(snapshots))
         )
         if lock_for_update:
             query = query.lock(copy=False)
@@ -477,7 +468,7 @@ def _get_environment(
         row = self.engine_adapter.fetchone(
             self._environments_query(
                 where=exp.EQ(
-                    this=exp.to_column("name"),
+                    this=exp.column("name"),
                     expression=exp.Literal.string(environment),
                 ),
                 lock_for_update=lock_for_update,
@@ -579,18 +570,18 @@ def _get_snapshot_intervals(
         query = (
             exp.select(
                 "id",
-                exp.to_column(f"{self.intervals_table}.name"),
-                exp.to_column(f"{self.intervals_table}.identifier"),
+                exp.column("name", table="intervals"),
+                exp.column("identifier", table="intervals"),
                 "version",
                 "start_ts",
                 "end_ts",
                 "is_dev",
                 "is_removed",
             )
-            .from_(self.intervals_table)
+            .from_(exp.to_table(self.intervals_table).as_("intervals"))
             .order_by(
-                exp.to_column(f"{self.intervals_table}.name"),
-                exp.to_column(f"{self.intervals_table}.identifier"),
+                exp.column("name", table="intervals"),
+                exp.column("identifier", table="intervals"),
                 "created_ts",
                 "is_removed",
             )
@@ -599,25 +590,23 @@ def _get_snapshot_intervals(
         if uncompacted_only:
             query.join(
                 exp.select("name", "identifier")
-                .from_(self.intervals_table)
+                .from_(exp.to_table(self.intervals_table).as_("intervals"))
                 .where(exp.column("is_compacted").not_())
                 .distinct()
                 .subquery(alias="uncompacted"),
                 on=exp.and_(
-                    *[
-                        exp.to_column(f"{self.intervals_table}.{col}").eq(
-                            exp.column(col, table="uncompacted")
-                        )
-                        for col in ["name", "identifier"]
-                    ]
+                    exp.column("name", table="intervals").eq(
+                        exp.column("name", table="uncompacted")
+                    ),
+                    exp.column("identifier", table="intervals").eq(
+                        exp.column("identifier", table="uncompacted")
+                    ),
                 ),
                 copy=False,
             )
 
         if snapshots:
-            query.where(
-                self._snapshot_name_version_filter(snapshots, self.intervals_table), copy=False
-            )
+            query.where(self._snapshot_name_version_filter(snapshots, "intervals"), copy=False)
         elif snapshots is not None:
             return (set(), [])
 
@@ -880,7 +869,7 @@ def map_data_versions(
             self.unpause_snapshots(updated_prod_environment.snapshots, now_timestamp())
 
     def _snapshot_id_filter(
-        self, snapshot_ids: t.Iterable[SnapshotIdLike], fq_table_name: str
+        self, snapshot_ids: t.Iterable[SnapshotIdLike], alias: t.Optional[str] = None
     ) -> t.Union[exp.In, exp.Boolean, exp.Condition]:
         if not snapshot_ids:
             return exp.false()
@@ -889,24 +878,24 @@ def _snapshot_id_filter(
                 exp.Tuple,
                 exp.convert(
                     (
-                        exp.to_column(f"{fq_table_name}.name"),
-                        exp.to_column(f"{fq_table_name}.identifier"),
+                        exp.column("name", table=alias),
+                        exp.column("identifier", table=alias),
                     )
                 ),
             ).isin(*[(snapshot_id.name, snapshot_id.identifier) for snapshot_id in snapshot_ids])
         else:
             return exp.or_(
                 *[
                     exp.and_(
-                        exp.to_column(f"{fq_table_name}.name").eq(snapshot_id.name),
-                        exp.to_column(f"{fq_table_name}.identifier").eq(snapshot_id.identifier),
+                        exp.column("name", table=alias).eq(snapshot_id.name),
+                        exp.column("identifier", table=alias).eq(snapshot_id.identifier),
                     )
                     for snapshot_id in snapshot_ids
                 ]
             )
 
     def _snapshot_name_version_filter(
-        self, snapshot_name_versions: t.Iterable[SnapshotNameVersionLike], fq_table_name: str
+        self, snapshot_name_versions: t.Iterable[SnapshotNameVersionLike], alias: str = "snapshots"
     ) -> t.Union[exp.In, exp.Boolean, exp.Condition]:
         if not snapshot_name_versions:
             return exp.false()
@@ -915,8 +904,8 @@ def _snapshot_name_version_filter(
                 exp.Tuple,
                 exp.convert(
                     (
-                        exp.to_column(f"{fq_table_name}.name"),
-                        exp.to_column(f"{fq_table_name}.version"),
+                        exp.column("name", table=alias),
+                        exp.column("version", table=alias),
                     )
                 ),
             ).isin(
@@ -929,8 +918,8 @@ def _snapshot_name_version_filter(
             return exp.or_(
                 *[
                     exp.and_(
-                        exp.to_column(f"{fq_table_name}.name").eq(snapshot_name_version.name),
-                        exp.to_column(f"{fq_table_name}.version").eq(snapshot_name_version.version),
+                        exp.column("name", table=alias).eq(snapshot_name_version.name),
+                        exp.column("version", table=alias).eq(snapshot_name_version.version),
                     )
                     for snapshot_name_version in snapshot_name_versions
                 ]