@@ -457,6 +457,14 @@ def get_current_catalog(self) -> t.Optional[str]:
         """Returns the catalog name of the current connection."""
         raise NotImplementedError()
 
+    def _get_current_catalog_for_set_catalog(self) -> t.Optional[str]:
+        """Catalog token @set_catalog saves and restores.
+
+        Override when get_current_catalog() reports a default catalog name but
+        the restore must keep an internal sentinel so the connection is not reopened.
+        """
+        return self.get_current_catalog()
+
     def set_current_catalog(self, catalog: str) -> None:
         """Sets the catalog name of the current connection."""
         raise NotImplementedError()
 
@@ -87,12 +87,13 @@ def _catalog_state_label(self, catalog_name: t.Optional[str]) -> str:
         )
 
     def _resolved_catalog(self) -> t.Optional[str]:
-        return (
-            self.get_current_catalog()
-            or self._normalize_catalog(self._connected_catalog)
-            or self._default_catalog
-            or self._extra_config.get("database")
-        )
+        # A lazy restore sets the logical target back to the default but leaves
+        # the open connection on the previous warehouse. Name that connection,
+        # because that is the warehouse the query hits.
+        connected_catalog = self._normalize_catalog(self._connected_catalog)
+        if self._normalize_catalog(self._target_catalog) is None and connected_catalog:
+            return connected_catalog
+        return self.get_current_catalog()
 
     @property
     def api_client(self) -> FabricHttpClient:
@@ -154,10 +155,29 @@ def _drop_catalog(self, catalog_name: exp.Identifier) -> None:
         if warehouse_name in in_use:
             self.close()
 
-    def get_current_catalog(self) -> t.Optional[str]:
-        """Return the explicit Fabric catalog target for the current thread."""
+    def _get_current_catalog_for_set_catalog(self) -> t.Optional[str]:
+        """Override of EngineAdapter._get_current_catalog_for_set_catalog.
+
+        The base method returns get_current_catalog(). @set_catalog saves that
+        value and writes it back with set_current_catalog. Fabric's public method
+        returns the configured warehouse name, so restoring it would reconnect.
+        This returns the normalized internal target instead. None is that default,
+        and restoring None skips the reconnect.
+        """
         return self._normalize_catalog(self._target_catalog)
 
+    def get_current_catalog(self) -> t.Optional[str]:
+        """Return the warehouse name for the current thread.
+
+        The configured default stays None internally so set_current_catalog can
+        skip a reconnect. Callers still receive that warehouse's name. None is
+        returned only when no warehouse is configured.
+        """
+        explicit_catalog = self._normalize_catalog(self._target_catalog)
+        if explicit_catalog:
+            return explicit_catalog
+        return self._default_catalog or self._extra_config.get("database")
+
     def set_current_catalog(self, catalog_name: t.Optional[str]) -> None:
         """
         Set the current catalog for Microsoft Fabric connections.
@@ -182,10 +202,12 @@ def set_current_catalog(self, catalog_name: t.Optional[str]) -> None:
         explicit_default_catalog = catalog_name is not None and target_catalog is None
         connected_catalog = self._normalize_catalog(self._connected_catalog)
 
+        # Compare the internal sentinel, not get_current_catalog(). That method
+        # returns the configured warehouse name when this sentinel is None.
         # 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 (
+        if self._normalize_catalog(self._target_catalog) == target_catalog and (
             not explicit_default_catalog or connected_catalog is None
         ):
             logger.debug("Already using requested Fabric catalog state, no action needed")
@@ -233,16 +255,40 @@ def set_current_catalog(self, catalog_name: t.Optional[str]) -> None:
 
         self._target_catalog = target_catalog
 
+    def _get_data_objects(
+        self, schema_name: t.Union[str, exp.Table], object_names: t.Optional[t.Set[str]] = None
+    ) -> t.List[DataObject]:
+        objects = super()._get_data_objects(schema_name, object_names)
+        # Runs before EngineAdapter.get_data_objects writes the cache. @set_catalog
+        # has already stripped the catalog, so the lookup key is schema.table.
+        # MSSQL stamps get_current_catalog() here, which is now the default warehouse
+        # name, and that would store catalog.schema.table instead. Blank that name so the
+        # stored key matches. get_data_objects puts it back on the objects it returns.
+        default_catalog = self._default_catalog or self._extra_config.get("database")
+        if not default_catalog:
+            return objects
+        return [
+            DataObject(
+                catalog=None if obj.catalog == default_catalog else obj.catalog,
+                schema=obj.schema_name,
+                name=obj.name,
+                type=obj.type,
+                clustering_key=obj.clustering_key,
+            )
+            for obj in objects
+        ]
+
     def get_data_objects(
         self,
         schema_name: t.Union[str, exp.Table],
         object_names: t.Optional[t.Set[str]] = None,
         safe_to_cache: bool = False,
     ) -> t.List[DataObject]:
-        # Fabric uses None as "default catalog" so we skip reconnects. Other engines
-        # return a real warehouse name here.
-        # Cache on schema.table due to @set_catalog stripping the catalog. Then put a
-        # warehouse name on the returned objects so listing matches other engines.
+        # Runs after the cache read. _get_data_objects blanked the default warehouse
+        # so the cached key stays schema.table. Callers still need that name on the
+        # returned objects. A schema that already names a warehouse wins; otherwise
+        # use the connection actually queried, which may still be the previous
+        # warehouse after a lazy restore.
         objects = super().get_data_objects(schema_name, object_names, safe_to_cache)
         catalog = to_schema(schema_name).catalog or self._resolved_catalog()
         if not catalog:
 
@@ -348,8 +348,11 @@ def internal_wrapper(*args: t.Any, **kwargs: t.Any) -> t.Any:
                         f"{engine_adapter.dialect} requires that all catalog operations be against a single catalog: {engine_adapter._default_catalog}. Provided catalog: {catalog_name}"
                     )
                 return func(*list_args, **kwargs)
-            # Set the catalog name on the engine adapter if needed
-            current_catalog = engine_adapter.get_current_catalog()
+            # Set the catalog name on the engine adapter if needed.
+            # See EngineAdapter._get_current_catalog_for_set_catalog. Fabric reports
+            # the configured warehouse name from get_current_catalog() and keeps None
+            # internally, so a restore of the default warehouse can skip a reconnect.
+            current_catalog = engine_adapter._get_current_catalog_for_set_catalog()
             if catalog_name != current_catalog:
                 engine_adapter.set_current_catalog(catalog_name)
                 resp = func(*list_args, **kwargs)
 
@@ -40,6 +40,7 @@ def _conflict_response(mocker: MockerFixture, error_code: str) -> t.Any:
     resp.raise_for_status.side_effect = requests.HTTPError(response=resp)
     return resp
 
+
 def _record_catalogs_at_execute(adapter: FabricEngineAdapter) -> t.List[t.Optional[str]]:
     catalogs: t.List[t.Optional[str]] = []
     real_execute = adapter.execute
@@ -52,15 +53,18 @@ def execute(*args: t.Any, **kwargs: t.Any) -> None:
     return catalogs
 
 
-def test_get_current_catalog_uses_only_explicit_target_catalog(
+def test_get_current_catalog_returns_configured_database_when_target_is_default(
     make_mocked_engine_adapter: t.Callable,
 ):
     adapter = make_mocked_engine_adapter(
         FabricEngineAdapter,
         database="default_catalog",
     )
 
-    assert adapter.get_current_catalog() is None
+    current_catalog = adapter.get_current_catalog()
+    assert current_catalog is not None
+    assert current_catalog.lower() == "default_catalog"
+    assert adapter._target_catalog is None
 
     adapter._target_catalog = "switched_catalog"
 
@@ -69,10 +73,31 @@ def test_get_current_catalog_uses_only_explicit_target_catalog(
     adapter._connection_pool.close()
 
     assert adapter._connection_pool.get_attribute("target_catalog") is None
-    assert adapter.get_current_catalog() is None
+    assert adapter.get_current_catalog() == "default_catalog"
     adapter.cursor.execute.assert_not_called()
 
 
+def test_shared_catalog_operations_can_lowercase_the_default_warehouse(
+    make_mocked_engine_adapter: t.Callable,
+):
+    """test_catalog_operations lowercases the current catalog, then restores it."""
+    adapter = make_mocked_engine_adapter(
+        FabricEngineAdapter,
+        database="ci_abc",
+    )
+
+    current_catalog = adapter.get_current_catalog().lower()
+    assert current_catalog == "ci_abc"
+    assert adapter._target_catalog is None
+
+    adapter.set_current_catalog("testing")
+    assert adapter.get_current_catalog().lower() == "testing"
+
+    adapter.set_current_catalog(current_catalog)
+    assert adapter.get_current_catalog().lower() == current_catalog
+    assert adapter._target_catalog is None
+
+
 def test_get_current_catalog_returns_none_without_target_or_database(
     make_mocked_engine_adapter: t.Callable,
 ):
@@ -108,7 +133,8 @@ def test_set_current_catalog_to_default_clears_explicit_target(
     adapter.set_current_catalog("planning")
     adapter.set_current_catalog("core")
 
-    assert adapter.get_current_catalog() is None
+    assert adapter._target_catalog is None
+    assert adapter.get_current_catalog() == "core"
     adapter.cursor.execute.assert_not_called()
 
 
@@ -130,8 +156,9 @@ def test_catalog_scoped_call_restores_to_neutral_without_close(
     # Decorator calls set_current_catalog("planning") then set_current_catalog(None).
     # Only the first call (None→planning) should trigger a connection close.
     assert close_spy.call_count == 1
+    assert adapter._target_catalog is None
     assert adapter._connected_catalog == "planning"
-    assert adapter.get_current_catalog() is None
+    assert adapter.get_current_catalog() == "core"
 
 
 def test_default_catalog_after_non_default_catalog_reconnects(
@@ -150,8 +177,9 @@ def test_default_catalog_after_non_default_catalog_reconnects(
     adapter.table_exists("core.db.table")
 
     assert close_spy.call_count == 2
+    assert adapter._target_catalog is None
     assert adapter._connected_catalog is None
-    assert adapter.get_current_catalog() is None
+    assert adapter.get_current_catalog() == "core"
 
 
 def test_repeated_same_catalog_reuses_connection(
@@ -507,7 +535,7 @@ def test_create_warehouse_unrelated_conflict_still_raises(
         fabric_http_client.create_warehouse("my_warehouse")
 
 
-def test_get_data_objects_uses_default_catalog_when_current_is_none(
+def test_get_data_objects_uses_configured_default_catalog(
     make_mocked_engine_adapter: t.Callable,
     mocker: MockerFixture,
 ) -> None:
@@ -517,7 +545,8 @@ def test_get_data_objects_uses_default_catalog_when_current_is_none(
         database="ci_abc",
         patch_get_data_objects=False,
     )
-    assert adapter.get_current_catalog() is None
+    assert adapter._target_catalog is None
+    assert adapter.get_current_catalog() == "ci_abc"
 
     mocker.patch.object(
         adapter,
@@ -564,9 +593,11 @@ def test_get_data_objects_labels_connected_warehouse_after_lazy_restore(
     make_mocked_engine_adapter: t.Callable,
     mocker: MockerFixture,
 ) -> None:
-    """Logical catalog is None; the connection is still on planning.
+    """Logical target is the default; the connection is still on planning.
 
-    Unqualified list must be planning. ci_abc.dbo must still switch.
+    The public catalog is the configured warehouse. The unqualified list must
+    still be planning, because that is the warehouse the query hits.
+    ci_abc.dbo must still switch.
     """
     adapter = make_mocked_engine_adapter(
         FabricEngineAdapter,
@@ -576,7 +607,9 @@ def test_get_data_objects_labels_connected_warehouse_after_lazy_restore(
     )
     adapter.set_current_catalog("planning")
     adapter.set_current_catalog(None)
-    assert adapter.get_current_catalog() is None
+    assert adapter._target_catalog is None
+    assert adapter._connected_catalog == "planning"
+    assert adapter.get_current_catalog() == "ci_abc"
 
     fetchdf = mocker.patch.object(
         adapter,