feat(fabric): Override create view

Author: mattiasthalen

sqlmesh/core/engine_adapter/fabric.py

@@ -433,3 +433,101 @@ def create_schema(
             # 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 create_view(
+        self,
+        view_name: t.Union[str, exp.Table],
+        query_or_df: t.Any,
+        columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
+        replace: bool = True,
+        materialized: bool = False,
+        materialized_properties: t.Optional[t.Dict[str, t.Any]] = None,
+        table_description: t.Optional[str] = None,
+        column_descriptions: t.Optional[t.Dict[str, str]] = None,
+        view_properties: t.Optional[t.Dict[str, exp.Expression]] = None,
+        **create_kwargs: t.Any,
+    ) -> None:
+        """
+        Override create_view to handle catalog-qualified view names.
+        Fabric doesn't support 'CREATE VIEW [catalog].[schema].[view]' syntax.
+        """
+        logger.debug(f"create_view called with: {view_name} (type: {type(view_name)})")
+
+        # Handle exp.Table objects that might be catalog-qualified
+        if isinstance(view_name, exp.Table):
+            if view_name.catalog:
+                # Has catalog qualification - switch to catalog and use schema.table
+                catalog_name = view_name.catalog
+                schema_name = view_name.db or ""
+                table_name = view_name.name
+
+                logger.debug(
+                    f"Detected exp.Table with catalog: catalog='{catalog_name}', schema='{schema_name}', table='{table_name}'"
+                )
+
+                # Switch to the catalog first
+                self.set_current_catalog(catalog_name)
+
+                # Create new Table expression without catalog
+                unqualified_view = exp.Table(this=table_name, db=schema_name)
+
+                super().create_view(
+                    unqualified_view,
+                    query_or_df,
+                    columns_to_types,
+                    replace,
+                    materialized,
+                    materialized_properties,
+                    table_description,
+                    column_descriptions,
+                    view_properties,
+                    **create_kwargs,
+                )
+                return
+
+        # Handle string view names that might be catalog-qualified
+        elif isinstance(view_name, str):
+            # Check if it's in catalog.schema.view format
+            parts = view_name.split(".")
+            if len(parts) == 3:
+                # catalog.schema.view format
+                catalog_name = parts[0].strip('"[]')
+                schema_name = parts[1].strip('"[]')
+                view_only = parts[2].strip('"[]')
+                unqualified_view_str = f"{schema_name}.{view_only}"
+                logger.debug(
+                    f"Detected catalog.schema.view format: catalog='{catalog_name}', unqualified='{unqualified_view_str}'"
+                )
+
+                # Switch to the catalog first
+                self.set_current_catalog(catalog_name)
+
+                # Use just the schema.view name
+                super().create_view(
+                    unqualified_view_str,
+                    query_or_df,
+                    columns_to_types,
+                    replace,
+                    materialized,
+                    materialized_properties,
+                    table_description,
+                    column_descriptions,
+                    view_properties,
+                    **create_kwargs,
+                )
+                return
+
+        # No catalog qualification, use as-is
+        logger.debug(f"No catalog detected, using original: {view_name}")
+        super().create_view(
+            view_name,
+            query_or_df,
+            columns_to_types,
+            replace,
+            materialized,
+            materialized_properties,
+            table_description,
+            column_descriptions,
+            view_properties,
+            **create_kwargs,
+        )