Fix: only drop LIMIT 0 in VarcharSizeWorkaroundMixin table creation

Author: georgesittas

sqlmesh/core/engine_adapter/mixins.py

@@ -297,11 +297,15 @@ def _build_create_table_exp(
             and statement.expression.args["limit"].expression.this == "0"
         ):
             assert not isinstance(table_name_or_schema, exp.Schema)
+
             # redshift and mssql have a bug where CTAS statements have non determistic types. if a limit
             # is applied to a ctas statement, VARCHAR types default to 1 in some instances.
             select_statement = statement.expression.copy()
             for select_or_union in select_statement.find_all(exp.Select, exp.SetOperation):
-                select_or_union.set("limit", None)
+                limit = select_or_union.args.get("limit")
+                if limit is not None and limit.expression.this == "0":
+                    limit.pop()
+
                 select_or_union.set("where", None)
 
             temp_view_name = self._get_temp_table("ctas")

tests/core/engine_adapter/test_mssql.py

@@ -730,11 +730,24 @@ def test_create_table_from_query(make_mocked_engine_adapter: t.Callable, mocker:
         ),
         exists=False,
     )
-
     assert to_sql_calls(adapter) == [
         "CREATE VIEW [__temp_ctas_test_random_id] AS SELECT [a], [b], [x] + 1 AS [c], [d] AS [d], [e] FROM (SELECT * FROM [table]);",
         "DROP VIEW IF EXISTS [__temp_ctas_test_random_id];",
         "CREATE TABLE [test_schema].[test_table] ([a] VARCHAR(MAX), [b] VARCHAR(60), [c] VARCHAR(MAX), [d] VARCHAR(MAX), [e] DATETIME2);",
     ]
 
     columns_mock.assert_called_once_with(exp.table_("__temp_ctas_test_random_id", quoted=True))
+
+    # We don't want to drop anything other than LIMIT 0
+    # See https://github.com/TobikoData/sqlmesh/issues/4048
+    adapter.ctas(
+        table_name="test_schema.test_table",
+        query_or_df=parse_one(
+            "SELECT * FROM (SELECT * FROM t WHERE FALSE LIMIT 1) WHERE FALSE LIMIT 0"
+        ),
+        exists=False,
+    )
+    assert (
+        "CREATE VIEW [__temp_ctas_test_random_id] AS SELECT * FROM (SELECT TOP 1 * FROM [t]);"
+        in to_sql_calls(adapter)
+    )