Merge remote-tracking branch 'origin/main' into fix/surrogate-key-sha2-hex

Author: Pawansingh3889

Makefile

@@ -63,7 +63,8 @@ install-dev-dbt-%:
 	fi; \
 	if [ "$$version" = "1.3.0" ]; then \
 		echo "Applying overrides for dbt $$version - upgrading google-cloud-bigquery"; \
-		$(PIP) install 'google-cloud-bigquery>=3.0.0' --upgrade; \
+		$(PIP) install 'google-cloud-bigquery>=3.0.0' \
+			'pyOpenSSL>=24.0.0' --upgrade; \
 	fi; \
 	mv pyproject.toml.backup pyproject.toml; \
 	echo "Restored original pyproject.toml"

sqlmesh/core/engine_adapter/fabric.py

@@ -58,6 +58,33 @@ def _target_catalog(self) -> t.Optional[str]:
     def _target_catalog(self, value: t.Optional[str]) -> None:
         self._connection_pool.set_attribute("target_catalog", value)
 
+    @property
+    def _connected_catalog(self) -> t.Optional[str]:
+        """Catalog the currently-open thread-local connection is actually using."""
+        return self._connection_pool.get_attribute("connected_catalog")
+
+    @_connected_catalog.setter
+    def _connected_catalog(self, value: t.Optional[str]) -> None:
+        self._connection_pool.set_attribute("connected_catalog", value)
+
+    def _normalize_catalog(self, catalog_name: t.Optional[str]) -> t.Optional[str]:
+        if not catalog_name:
+            return None
+
+        default_catalog = self._default_catalog or self._extra_config.get("database")
+        if default_catalog and catalog_name == default_catalog:
+            return None
+
+        return catalog_name
+
+    def _catalog_state_label(self, catalog_name: t.Optional[str]) -> str:
+        return (
+            catalog_name
+            or self._default_catalog
+            or self._extra_config.get("database")
+            or "<default>"
+        )
+
     @property
     def api_client(self) -> FabricHttpClient:
         # the requests Session is not guaranteed to be threadsafe
@@ -101,20 +128,28 @@ def _create_catalog(self, catalog_name: exp.Identifier) -> None:
     def _drop_catalog(self, catalog_name: exp.Identifier) -> None:
         """Drop a catalog (warehouse) in Microsoft Fabric via REST API."""
         warehouse_name = catalog_name.sql(dialect=self.dialect, identify=False)
-        current_catalog = self.get_current_catalog()
 
         logger.info(f"Deleting Fabric warehouse: {warehouse_name}")
         self.api_client.delete_warehouse(warehouse_name)
 
-        if warehouse_name == current_catalog:
-            # Somewhere around 2025-09-08, Fabric started validating the "Database=" connection argument and throwing 'Authentication failed' if the database doesnt exist
-            # In addition, set_current_catalog() is implemented using a threadlocal variable "target_catalog"
-            # So, when we drop a warehouse, and there are still threads with "target_catalog" set to reference it, any operations on those threads
-            # that use an either use an existing connection pointing to this warehouse or trigger a new connection
-            # will fail with an 'Authentication Failed' error unless we close all connections here, which also clears all the threadlocal data
+        # Close all connections if any thread may be using the dropped warehouse.
+        # We must check both the logical target and the physical connection catalog
+        # (falling back to the configured default when either is neutral) because
+        # Fabric validates the DATABASE= connection argument and raises
+        # 'Authentication Failed' when it points at a non-existent warehouse.
+        default_db = self._extra_config.get("database")
+        in_use = {
+            self.get_current_catalog() or default_db,
+            self._normalize_catalog(self._connected_catalog) or default_db,
+        }
+        if warehouse_name in in_use:
             self.close()
 
-    def set_current_catalog(self, catalog_name: str) -> None:
+    def get_current_catalog(self) -> t.Optional[str]:
+        """Return the explicit Fabric catalog target for the current thread."""
+        return self._normalize_catalog(self._target_catalog)
+
+    def set_current_catalog(self, catalog_name: t.Optional[str]) -> None:
         """
         Set the current catalog for Microsoft Fabric connections.
 
@@ -123,7 +158,8 @@ def set_current_catalog(self, catalog_name: str) -> None:
         recreate them with the new catalog in the connection configuration.
 
         Args:
-            catalog_name: The name of the catalog (warehouse) to switch to
+            catalog_name: The name of the catalog (warehouse) to switch to.
+                The configured default catalog is treated as the neutral state.
 
         Note:
             Fabric doesn't support catalog switching via USE statements because each
@@ -133,33 +169,60 @@ def set_current_catalog(self, catalog_name: str) -> None:
         See:
             https://learn.microsoft.com/en-us/fabric/data-warehouse/sql-query-editor#limitations
         """
-        current_catalog = self.get_current_catalog()
-
-        # If already using the requested catalog, do nothing
-        if current_catalog and current_catalog == catalog_name:
-            logger.debug(f"Already using catalog '{catalog_name}', no action needed")
+        target_catalog = self._normalize_catalog(catalog_name)
+        explicit_default_catalog = catalog_name is not None and target_catalog is None
+        connected_catalog = self._normalize_catalog(self._connected_catalog)
+
+        # An explicit request for the default catalog must also match the catalog
+        # used by the open connection. A lazy restore with None only updates the
+        # logical target and intentionally leaves that connection in place.
+        if self.get_current_catalog() == target_catalog and (
+            not explicit_default_catalog or connected_catalog is None
+        ):
+            logger.debug("Already using requested Fabric catalog state, no action needed")
             return
 
-        logger.info(f"Switching from catalog '{current_catalog}' to '{catalog_name}'")
-
-        # commit the transaction before closing the connection to help prevent errors like:
-        # > Snapshot isolation transaction failed in database because the object accessed by the statement has been modified by a
-        # > DDL statement in another concurrent transaction since the start of this transaction
-        # on subsequent queries in the new connection
-        self._connection_pool.commit()
-
-        # note: we call close() on the connection pool instead of self.close() because self.close() calls close_all()
-        # on the connection pool but we just want to close the connection for this thread
-        self._connection_pool.close()
-        self._target_catalog = catalog_name  # new connections will use this catalog
-
-        catalog_after_switch = self.get_current_catalog()
+        # Decide whether the open connection needs to be replaced.
+        #
+        # The set_catalog decorator restores the previous catalog (often None)
+        # after every catalog-scoped call.  For Fabric, a connection close +
+        # reopen is expensive because each new connection goes through ODBC and
+        # the Fabric gateway.  We therefore apply lazy connection management:
+        #
+        #  * When restoring to neutral (target=None): just update _target_catalog.
+        #    The existing connection stays alive and will be reused or replaced
+        #    on the next real switch, avoiding a pointless bounce through the
+        #    default catalog.
+        #
+        #  * When switching to a non-neutral catalog: only close/reopen if the
+        #    open connection is already on a different catalog.  If a previous
+        #    restore-to-neutral left the connection on the right catalog, we
+        #    skip the close entirely.
+        needs_reconnect = (target_catalog is not None or explicit_default_catalog) and (
+            connected_catalog != target_catalog
+        )
 
-        if catalog_after_switch != catalog_name:
-            # We need to raise an error if the catalog switch failed to prevent the operation that needed the catalog switch from being run against the wrong catalog
-            raise SQLMeshError(
-                f"Unable to switch catalog to {catalog_name}, catalog ended up as {catalog_after_switch}"
+        if needs_reconnect:
+            logger.info(
+                "Switching connection from catalog '%s' to '%s'",
+                self._catalog_state_label(connected_catalog),
+                self._catalog_state_label(target_catalog),
             )
+            # Commit before closing to avoid snapshot-isolation errors on
+            # subsequent queries in the new connection.
+            self._connection_pool.commit()
+            # note: close() on the pool (not self.close()) to only affect this
+            # thread's connection rather than all threads.
+            self._connection_pool.close()
+            self._connected_catalog = target_catalog
+        else:
+            logger.debug(
+                "Updating catalog target to '%s' (connection remains on '%s')",
+                self._catalog_state_label(target_catalog),
+                self._catalog_state_label(connected_catalog),
+            )
+
+        self._target_catalog = target_catalog
 
     def alter_table(
         self, alter_expressions: t.Union[t.List[exp.Alter], t.List[TableAlterOperation]]

sqlmesh/core/engine_adapter/mysql.py

@@ -2,7 +2,6 @@
 
 import logging
 import typing as t
-
 from sqlglot import exp, parse_one
 
 from sqlmesh.core.dialect import to_schema
@@ -22,6 +21,7 @@
 
 if t.TYPE_CHECKING:
     from sqlmesh.core._typing import SchemaName, TableName
+    from sqlmesh.core.engine_adapter._typing import QueryOrDF
 
 logger = logging.getLogger(__name__)
 
@@ -186,5 +186,88 @@ def _create_table_like(
             )
         )
 
+    def _replace_by_key(
+        self,
+        target_table: TableName,
+        source_table: QueryOrDF,
+        target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]],
+        key: t.Sequence[exp.Expr],
+        is_unique_key: bool,
+        source_columns: t.Optional[t.List[str]] = None,
+    ) -> None:
+        if len(key) <= 1:
+            return super()._replace_by_key(
+                target_table,
+                source_table,
+                target_columns_to_types,
+                key,
+                is_unique_key,
+                source_columns,
+            )
+
+        if target_columns_to_types is None:
+            target_columns_to_types = self.columns(target_table)
+
+        temp_table = self._get_temp_table(target_table)
+        column_names = list(target_columns_to_types or [])
+
+        target_alias = "_target"
+        temp_alias = "_temp"
+
+        with self.transaction():
+            self.ctas(
+                temp_table,
+                source_table,
+                target_columns_to_types=target_columns_to_types,
+                exists=False,
+                source_columns=source_columns,
+            )
+
+            try:
+                # Build a JOIN-based DELETE instead of using CONCAT_WS.
+                # CONCAT_WS prevents MySQL/MariaDB from using indexes, causing full table scans.
+                on_condition = exp.and_(
+                    *[
+                        self._qualify_columns(k, target_alias).eq(
+                            self._qualify_columns(k, temp_alias)
+                        )
+                        for k in key
+                    ]
+                )
+
+                target_table_aliased = exp.to_table(target_table).as_(target_alias, quoted=True)
+                temp_table_aliased = exp.to_table(temp_table).as_(temp_alias, quoted=True)
+
+                join = exp.Join(this=temp_table_aliased, kind="INNER", on=on_condition)
+                target_table_aliased.append("joins", join)
+
+                delete_stmt = exp.Delete(
+                    tables=[exp.to_table(target_alias)],
+                    this=target_table_aliased,
+                )
+                self.execute(delete_stmt)
+
+                insert_query = self._select_columns(target_columns_to_types).from_(temp_table)
+                if is_unique_key:
+                    insert_query = insert_query.distinct(*key)
+
+                insert_statement = exp.insert(
+                    insert_query,
+                    target_table,
+                    columns=column_names,
+                )
+                self.execute(insert_statement, track_rows_processed=True)
+            finally:
+                self.drop_table(temp_table)
+
+    @staticmethod
+    def _qualify_columns(expr: exp.Expr, table_alias: str) -> exp.Expr:
+        """Qualify unqualified column references in an expression with a table alias."""
+        expr = expr.copy()
+        for col in expr.find_all(exp.Column):
+            if not col.table:
+                col.set("table", exp.to_identifier(table_alias, quoted=True))
+        return expr
+
     def ping(self) -> None:
         self._connection_pool.get().ping(reconnect=False)

sqlmesh/core/model/definition.py

@@ -1712,6 +1712,7 @@ def render(
 
     def render_seed(self) -> t.Iterator[QueryOrDF]:
         import numpy as np
+        import pandas as pd
 
         self._ensure_hydrated()
 
@@ -1752,8 +1753,6 @@ def render_seed(self) -> t.Iterator[QueryOrDF]:
 
             # convert all date/time types to native pandas timestamp
             for column in [*date_columns, *datetime_columns]:
-                import pandas as pd
-
                 df[column] = pd.to_datetime(df[column], infer_datetime_format=True, errors="ignore")  # type: ignore
 
             # extract datetime.date from pandas timestamp for DATE columns
@@ -1769,7 +1768,7 @@ def render_seed(self) -> t.Iterator[QueryOrDF]:
                     )
 
             for column in bool_columns:
-                df[column] = df[column].apply(lambda i: str_to_bool(str(i)))
+                df[column] = df[column].apply(lambda i: None if pd.isna(i) else str_to_bool(str(i)))
 
             df.loc[:, string_columns] = df[string_columns].mask(
                 cond=lambda x: x.notna(),  # type: ignore

sqlmesh/integrations/dlt.py

@@ -63,10 +63,7 @@ def generate_dlt_models_and_settings(
     if db_type == "filesystem":
         connection_config = None
     else:
-        if dlt.__version__ >= "1.10.0":
-            client = pipeline.destination_client()
-        else:
-            client = pipeline._sql_job_client(schema)  # type: ignore
+        client = pipeline.destination_client()
         config = client.config
         credentials = config.credentials
         configs = {

sqlmesh/utils/date.py

@@ -344,6 +344,8 @@ def make_exclusive(time: TimeLike) -> datetime:
 
 
 def make_ts_exclusive(time: TimeLike, dialect: DialectType) -> datetime:
+    import pandas as pd
+
     ts = to_datetime(time)
     if dialect == "tsql":
         return to_utc_timestamp(ts) - pd.Timedelta(1, unit="ns")