Fix: Enable schema binding when creating views for queries with recursive CTEs in redshift (#4006)

Author: izeigerman

sqlmesh/core/engine_adapter/mixins.py

@@ -306,9 +306,7 @@ def _build_create_table_exp(
 
             temp_view_name = self._get_temp_table("ctas")
 
-            self.create_view(
-                temp_view_name, select_statement, replace=False, no_schema_binding=False
-            )
+            self.create_view(temp_view_name, select_statement, replace=False)
             try:
                 columns_to_types_from_view = self._default_precision_to_max(
                     self.columns(temp_view_name)

sqlmesh/core/engine_adapter/redshift.py

@@ -217,11 +217,13 @@ def create_view(
         underlying table without dropping the view first. This is a problem for us since we want to be able to
         swap tables out from under views. Therefore, we create the view as non-binding.
         """
-
-        if create_kwargs.pop("no_schema_binding", None) is False:
-            logger.warning(
-                "The 'no_schema_binding' attribute is deprecated. Views in Redshift are created as non-binding."
+        no_schema_binding = True
+        if isinstance(query_or_df, exp.Expression):
+            # We can't include NO SCHEMA BINDING if the query has a recursive CTE
+            has_recursive_cte = any(
+                w.args.get("recursive", False) for w in query_or_df.find_all(exp.With)
             )
+            no_schema_binding = not has_recursive_cte
 
         return super().create_view(
             view_name,
@@ -232,7 +234,7 @@ def create_view(
             materialized_properties,
             table_description=table_description,
             column_descriptions=column_descriptions,
-            no_schema_binding=True,
+            no_schema_binding=no_schema_binding,
             view_properties=view_properties,
             **create_kwargs,
         )

tests/core/engine_adapter/test_redshift.py

@@ -123,6 +123,46 @@ def test_create_table_from_query_exists_no_if_not_exists(
     columns_mock.assert_called_once_with(exp.table_("__temp_ctas_test_random_id", quoted=True))
 
 
+def test_create_table_recursive_cte(adapter: t.Callable, mocker: MockerFixture):
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.base.random_id",
+        return_value="test_random_id",
+    )
+
+    mocker.patch(
+        "sqlmesh.core.engine_adapter.redshift.RedshiftEngineAdapter.table_exists",
+        return_value=True,
+    )
+
+    columns_mock = mocker.patch(
+        "sqlmesh.core.engine_adapter.redshift.RedshiftEngineAdapter.columns",
+        return_value={
+            "a": exp.DataType.build("VARCHAR(MAX)", dialect="redshift"),
+            "b": exp.DataType.build("VARCHAR(60)", dialect="redshift"),
+            "c": exp.DataType.build("VARCHAR(MAX)", dialect="redshift"),
+            "d": exp.DataType.build("VARCHAR(MAX)", dialect="redshift"),
+            "e": exp.DataType.build("TIMESTAMP", dialect="redshift"),
+        },
+    )
+
+    adapter.ctas(
+        table_name="test_schema.test_table",
+        query_or_df=parse_one(
+            "WITH RECURSIVE cte AS (SELECT * FROM table WHERE FALSE LIMIT 0) SELECT a, b, x + 1 AS c, d AS d, e FROM cte WHERE d > 0 AND FALSE LIMIT 0",
+            dialect="redshift",
+        ),
+        exists=False,
+    )
+
+    assert to_sql_calls(adapter) == [
+        'CREATE VIEW "__temp_ctas_test_random_id" AS WITH RECURSIVE "cte" AS (SELECT * FROM "table") SELECT "a", "b", "x" + 1 AS "c", "d" AS "d", "e" FROM "cte"',
+        'DROP VIEW IF EXISTS "__temp_ctas_test_random_id" CASCADE',
+        'CREATE TABLE "test_schema"."test_table" ("a" VARCHAR(MAX), "b" VARCHAR(60), "c" VARCHAR(MAX), "d" VARCHAR(MAX), "e" TIMESTAMP)',
+    ]
+
+    columns_mock.assert_called_once_with(exp.table_("__temp_ctas_test_random_id", quoted=True))
+
+
 def test_create_table_from_query_exists_and_if_not_exists(
     adapter: t.Callable, mocker: MockerFixture
 ):