Merge remote-tracking branch 'upstream/main' into local-lint-option

Author: evieira

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"

docs/integrations/engines/redshift.md

@@ -29,6 +29,7 @@ pip install "sqlmesh[redshift]"
 | `region`                | The AWS region of the Amazon Redshift cluster                                                               | string |    N     |
 | `cluster_identifier`    | The cluster identifier of the Amazon Redshift cluster                                                       | string |    N     |
 | `iam`                   | If IAM authentication is enabled. IAM must be True when authenticating using an IdP                         |  dict  |    N     |
+| `db_user`               | The database user to authenticate as. Required when using IAM authentication                                | string |    N     |
 | `is_serverless`         | If the Amazon Redshift cluster is serverless (Default: `False`)                                             |  bool  |    N     |
 | `serverless_acct_id`    | The account ID of the serverless cluster                                                                    | string |    N     |
 | `serverless_work_group` | The name of work group for serverless end point                                                             | string |    N     |

docs/prerequisites.md

@@ -4,7 +4,7 @@ This page describes the system prerequisites needed to run SQLMesh and provides
 
 ## SQLMesh prerequisites
 
-You'll need Python 3.8 or higher to use SQLMesh. You can check your python version by running the following command:
+You'll need Python 3.9 or higher to use SQLMesh. You can check your python version by running the following command:
 ```bash
 python3 --version
 ```

pyproject.toml

@@ -7,7 +7,6 @@ authors = [{ name = "SQLMesh Contributors" }]
 license = { file = "LICENSE" }
 requires-python = ">= 3.9"
 dependencies = [
-    "astor",
     "click",
     "croniter",
     "duckdb>=0.10.0,!=0.10.3",
@@ -202,7 +201,6 @@ disable_error_code = "annotation-unchecked"
 [[tool.mypy.overrides]]
 module = [
     "api.*",
-    "astor.*",
     "IPython.*",
     "hyperscript.*",
     "py.*",

sqlmesh/core/config/connection.py

@@ -13,8 +13,8 @@
 from sys import version_info
 
 import pydantic
+from pydantic import Field, computed_field
 from packaging import version
-from pydantic import Field
 from pydantic_core import from_json
 from sqlglot import exp
 from sqlglot.errors import ParseError
@@ -110,7 +110,14 @@ class ConnectionConfig(abc.ABC, BaseConfig):
     catalog_type_overrides: t.Optional[t.Dict[str, str]] = None
 
     # Whether to share a  single connection across threads or create a new connection per thread.
-    shared_connection: t.ClassVar[bool] = False
+    #
+    # MyPy throws a "Decorators on top of @property are not supported" error despite this being a
+    # valid decoration, and Pydantic recommend disabling the MyPy hint for this reason - see:
+    # https://pydantic.dev/docs/validation/2.0/usage/computed_fields/
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def shared_connection(self) -> bool:
+        return False
 
     @property
     @abc.abstractmethod
@@ -311,7 +318,10 @@ class BaseDuckDBConnectionConfig(ConnectionConfig):
 
     token: t.Optional[str] = None
 
-    shared_connection: t.ClassVar[bool] = True
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def shared_connection(self) -> bool:
+        return True
 
     _data_file_to_adapter: t.ClassVar[t.Dict[str, EngineAdapter]] = {}
 
@@ -820,11 +830,15 @@ class DatabricksConnectionConfig(ConnectionConfig):
     DISPLAY_NAME: t.ClassVar[t.Literal["Databricks"]] = "Databricks"
     DISPLAY_ORDER: t.ClassVar[t.Literal[3]] = 3
 
-    shared_connection: t.ClassVar[bool] = True
-
     _concurrent_tasks_validator = concurrent_tasks_validator
     _http_headers_validator = http_headers_validator
 
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def shared_connection(self) -> bool:
+        """The connection should only be shared if U2M OAuth is being used"""
+        return self.auth_type is not None and self.oauth_client_secret is None
+
     @model_validator(mode="before")
     def _databricks_connect_validator(cls, data: t.Any) -> t.Any:
         # SQLQueryContextLogger will output any error SQL queries even if they are in a try/except block.
@@ -1321,6 +1335,7 @@ class RedshiftConnectionConfig(ConnectionConfig):
         region: The AWS region where the Amazon Redshift cluster is located.
         cluster_identifier: The cluster identifier of the Amazon Redshift cluster.
         iam: If IAM authentication is enabled. Default value is False. IAM must be True when authenticating using an IdP.
+        db_user: The database user to authenticate as. Required when using IAM authentication.
         is_serverless: Redshift end-point is serverless or provisional. Default value false.
         serverless_acct_id: The account ID of the serverless. Default value None
         serverless_work_group: The name of work group for serverless end point. Default value None.
@@ -1346,6 +1361,7 @@ class RedshiftConnectionConfig(ConnectionConfig):
     region: t.Optional[str] = None
     cluster_identifier: t.Optional[str] = None
     iam: t.Optional[bool] = None
+    db_user: t.Optional[str] = None
     is_serverless: t.Optional[bool] = None
     serverless_acct_id: t.Optional[str] = None
     serverless_work_group: t.Optional[str] = None
@@ -1383,6 +1399,7 @@ def _connection_kwargs_keys(self) -> t.Set[str]:
             "region",
             "cluster_identifier",
             "iam",
+            "db_user",
             "is_serverless",
             "serverless_acct_id",
             "serverless_work_group",

sqlmesh/core/config/loader.py

@@ -272,4 +272,4 @@ def convert_config_type(
     config_obj: Config,
     config_type: t.Type[C],
 ) -> C:
-    return config_type.parse_obj(config_obj.dict())
+    return config_type.parse_obj(config_obj.dict(exclude_computed_fields=True))

sqlmesh/core/engine_adapter/databricks.py

@@ -154,7 +154,7 @@ def _set_spark_engine_adapter_if_needed(self) -> None:
             host=self._extra_config["databricks_connect_server_hostname"],
             token=self._extra_config.get("databricks_connect_access_token"),
         )
-        if "databricks_connect_use_serverless" in self._extra_config:
+        if self._extra_config.get("databricks_connect_use_serverless"):
             connect_kwargs["serverless"] = True
         else:
             connect_kwargs["cluster_id"] = self._extra_config["databricks_connect_cluster_id"]

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]]