fix(fabric): Correct catalog.schema parsing

Author: mattiasthalen

sqlmesh/core/engine_adapter/fabric.py

@@ -390,31 +390,49 @@ def drop_schema(
         """
         logger.debug(f"drop_schema called with: {schema_name} (type: {type(schema_name)})")
 
-        # Parse schema_name into an exp.Table to properly handle both string and Table cases
-        table = exp.to_table(schema_name)
-
-        if table.catalog:
-            # 3-part name detected (catalog.db.table) - this shouldn't happen for schema operations
-            raise SQLMeshError(
-                f"Invalid schema name format: {schema_name}. Expected 'schema' or 'catalog.schema'"
-            )
-        elif table.db:
-            # Catalog-qualified schema: catalog.schema
-            catalog_name = table.db
-            schema_only = table.name
-            logger.debug(
-                f"Detected catalog.schema format: catalog='{catalog_name}', schema='{schema_only}'"
-            )
-
-            # Switch to the catalog first
-            self.set_current_catalog(catalog_name)
-
-            # Use just the schema name
-            super().drop_schema(schema_only, ignore_if_not_exists, cascade, **drop_args)
+        # Handle Table objects created by schema_() function
+        if isinstance(schema_name, exp.Table) and not schema_name.name:
+            # This is a schema Table object - check for catalog qualification
+            if schema_name.catalog:
+                # Catalog-qualified schema: catalog.schema
+                catalog_name = schema_name.catalog
+                schema_only = schema_name.db
+                logger.debug(
+                    f"Detected catalog-qualified schema: catalog='{catalog_name}', schema='{schema_only}'"
+                )
+                # Switch to the catalog first
+                self.set_current_catalog(catalog_name)
+                # Use just the schema name
+                super().drop_schema(schema_only, ignore_if_not_exists, cascade, **drop_args)
+            else:
+                # Schema only, no catalog
+                schema_only = schema_name.db
+                logger.debug(f"Detected schema-only: schema='{schema_only}'")
+                super().drop_schema(schema_only, ignore_if_not_exists, cascade, **drop_args)
         else:
-            # No catalog qualification, use as-is
-            logger.debug(f"No catalog detected, using original: {schema_name}")
-            super().drop_schema(schema_name, ignore_if_not_exists, cascade, **drop_args)
+            # Handle string or table name inputs by parsing as table
+            table = exp.to_table(schema_name)
+
+            if table.catalog:
+                # 3-part name detected (catalog.db.table) - this shouldn't happen for schema operations
+                raise SQLMeshError(
+                    f"Invalid schema name format: {schema_name}. Expected 'schema' or 'catalog.schema', got 3-part name"
+                )
+            elif table.db:
+                # Catalog-qualified schema: catalog.schema
+                catalog_name = table.db
+                schema_only = table.name
+                logger.debug(
+                    f"Detected catalog.schema format: catalog='{catalog_name}', schema='{schema_only}'"
+                )
+                # Switch to the catalog first
+                self.set_current_catalog(catalog_name)
+                # Use just the schema name
+                super().drop_schema(schema_only, ignore_if_not_exists, cascade, **drop_args)
+            else:
+                # No catalog qualification, use as-is
+                logger.debug(f"No catalog detected, using original: {schema_name}")
+                super().drop_schema(schema_name, ignore_if_not_exists, cascade, **drop_args)
 
     def create_schema(
         self,
@@ -428,31 +446,49 @@ def create_schema(
         """
         logger.debug(f"create_schema called with: {schema_name} (type: {type(schema_name)})")
 
-        # Parse schema_name into an exp.Table to properly handle both string and Table cases
-        table = exp.to_table(schema_name)
-
-        if table.catalog:
-            # 3-part name detected (catalog.db.table) - this shouldn't happen for schema operations
-            raise SQLMeshError(
-                f"Invalid schema name format: {schema_name}. Expected 'schema' or 'catalog.schema'"
-            )
-        elif table.db:
-            # Catalog-qualified schema: catalog.schema
-            catalog_name = table.db
-            schema_only = table.name
-            logger.debug(
-                f"Detected catalog.schema format: catalog='{catalog_name}', schema='{schema_only}'"
-            )
-
-            # Switch to the catalog first
-            self.set_current_catalog(catalog_name)
-
-            # Use just the schema name
-            super().create_schema(schema_only, ignore_if_exists, **kwargs)
+        # Handle Table objects created by schema_() function
+        if isinstance(schema_name, exp.Table) and not schema_name.name:
+            # This is a schema Table object - check for catalog qualification
+            if schema_name.catalog:
+                # Catalog-qualified schema: catalog.schema
+                catalog_name = schema_name.catalog
+                schema_only = schema_name.db
+                logger.debug(
+                    f"Detected catalog-qualified schema: catalog='{catalog_name}', schema='{schema_only}'"
+                )
+                # Switch to the catalog first
+                self.set_current_catalog(catalog_name)
+                # Use just the schema name
+                super().create_schema(schema_only, ignore_if_exists, **kwargs)
+            else:
+                # Schema only, no catalog
+                schema_only = schema_name.db
+                logger.debug(f"Detected schema-only: schema='{schema_only}'")
+                super().create_schema(schema_only, ignore_if_exists, **kwargs)
         else:
-            # No catalog qualification, use as-is
-            logger.debug(f"No catalog detected, using original: {schema_name}")
-            super().create_schema(schema_name, ignore_if_exists, **kwargs)
+            # Handle string or table name inputs by parsing as table
+            table = exp.to_table(schema_name)
+
+            if table.catalog:
+                # 3-part name detected (catalog.db.table) - this shouldn't happen for schema operations
+                raise SQLMeshError(
+                    f"Invalid schema name format: {schema_name}. Expected 'schema' or 'catalog.schema', got 3-part name"
+                )
+            elif table.db:
+                # Catalog-qualified schema: catalog.schema
+                catalog_name = table.db
+                schema_only = table.name
+                logger.debug(
+                    f"Detected catalog.schema format: catalog='{catalog_name}', schema='{schema_only}'"
+                )
+                # Switch to the catalog first
+                self.set_current_catalog(catalog_name)
+                # Use just the schema name
+                super().create_schema(schema_only, ignore_if_exists, **kwargs)
+            else:
+                # No catalog qualification, use as-is
+                logger.debug(f"No catalog detected, using original: {schema_name}")
+                super().create_schema(schema_name, ignore_if_exists, **kwargs)
 
     def _ensure_schema_exists(self, table_name: TableName) -> None:
         """