sqlmesh.core.engine_adapter.postgres

112    def merge(
113        self,
114        target_table: TableName,
115        source_table: QueryOrDF,
116        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]],
117        unique_key: t.Sequence[exp.Expr],
118        when_matched: t.Optional[exp.Whens] = None,
119        merge_filter: t.Optional[exp.Expr] = None,
120        source_columns: t.Optional[t.List[str]] = None,
121        **kwargs: t.Any,
122    ) -> None:
123        # Merge isn't supported until Postgres 15
124        major, minor = self.server_version
125        merge_impl = super().merge if major >= 15 else partial(logical_merge, self)
126        merge_impl(  # type: ignore
127            target_table,
128            source_table,
129            target_columns_to_types,
130            unique_key,
131            when_matched=when_matched,
132            merge_filter=merge_filter,
133            source_columns=source_columns,
134        )

136    @cached_property
137    def server_version(self) -> t.Tuple[int, int]:
138        """Lazily fetch and cache major and minor server version"""
139        if result := self.fetchone("SHOW server_version"):
140            server_version, *_ = result
141            match = re.search(r"(\d+)\.(\d+)", server_version)
142            if match:
143                return int(match.group(1)), int(match.group(2))
144        return 0, 0

35    def columns(
36        self, table_name: TableName, include_pseudo_columns: bool = False
37    ) -> t.Dict[str, exp.DataType]:
38        """Fetches column names and types for the target table."""
39        table = exp.to_table(table_name)
40
41        sql = (
42            exp.select(
43                "attname AS column_name",
44                "pg_catalog.format_type(atttypid, atttypmod) AS data_type",
45            )
46            .from_("pg_catalog.pg_attribute")
47            .join("pg_catalog.pg_class", on="pg_class.oid = attrelid")
48            .join("pg_catalog.pg_namespace", on="pg_namespace.oid = relnamespace")
49            .where(
50                exp.and_(
51                    "attnum > 0",
52                    "NOT attisdropped",
53                    exp.column("relname").eq(table.alias_or_name),
54                )
55            )
56        )
57        if table.args.get("db"):
58            sql = sql.where(exp.column("nspname").eq(table.args["db"].name))
59
60        self.execute(sql)
61        resp = self.cursor.fetchall()
62        if not resp:
63            raise SQLMeshError(
64                f"Could not get columns for table '{table.sql(dialect=self.dialect)}'. Table not found."
65            )
66
67        return {
68            column_name: exp.DataType.build(data_type, dialect=self.dialect, udt=True)
69            for column_name, data_type in resp
70        }

 76    def table_exists(self, table_name: TableName) -> bool:
 77        """
 78        Postgres doesn't support describe so I'm using what the redshift cursor does to check if a table
 79        exists. We don't use this directly in order for this to work as a base class for other postgres
 80
 81        Reference: https://github.com/aws/amazon-redshift-python-driver/blob/master/redshift_connector/cursor.py#L528-L553
 82        """
 83        table = exp.to_table(table_name)
 84        data_object_cache_key = _get_data_object_cache_key(table.catalog, table.db, table.name)
 85        if data_object_cache_key in self._data_object_cache:
 86            logger.debug("Table existence cache hit: %s", data_object_cache_key)
 87            return self._data_object_cache[data_object_cache_key] is not None
 88
 89        sql = (
 90            exp.select("1")
 91            .from_("information_schema.tables")
 92            .where(f"table_name = '{table.alias_or_name}'")
 93        )
 94        database_name = table.db
 95        if database_name:
 96            sql = sql.where(f"table_schema = '{database_name}'")
 97
 98        self.execute(sql)
 99
100        result = self.cursor.fetchone()
101
102        return result[0] == 1 if result is not None else False

104    def create_view(
105        self,
106        view_name: TableName,
107        query_or_df: QueryOrDF,
108        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
109        replace: bool = True,
110        materialized: bool = False,
111        materialized_properties: t.Optional[t.Dict[str, t.Any]] = None,
112        table_description: t.Optional[str] = None,
113        column_descriptions: t.Optional[t.Dict[str, str]] = None,
114        view_properties: t.Optional[t.Dict[str, exp.Expr]] = None,
115        source_columns: t.Optional[t.List[str]] = None,
116        **create_kwargs: t.Any,
117    ) -> None:
118        """
119        Postgres has very strict rules around view replacement. For example the new query must generate an identical setFormatter
120        of columns, using the same column names and data types as the old one. We have to delete the old view instead of replacing it
121        to work around these constraints.
122
123        Reference: https://www.postgresql.org/docs/current/sql-createview.html
124        """
125        with self.transaction():
126            if replace:
127                self.drop_view(view_name, materialized=materialized)
128            super().create_view(
129                view_name,
130                query_or_df,
131                target_columns_to_types=target_columns_to_types,
132                replace=False,
133                materialized=materialized,
134                materialized_properties=materialized_properties,
135                table_description=table_description,
136                column_descriptions=column_descriptions,
137                view_properties=view_properties,
138                source_columns=source_columns,
139                **create_kwargs,
140            )

142    def drop_view(
143        self,
144        view_name: TableName,
145        ignore_if_not_exists: bool = True,
146        materialized: bool = False,
147        **kwargs: t.Any,
148    ) -> None:
149        kwargs["cascade"] = kwargs.get("cascade", True)
150        return super().drop_view(
151            view_name,
152            ignore_if_not_exists=ignore_if_not_exists,
153            materialized=materialized,
154            **kwargs,
155        )