Add supported drop cascade object indicators

Author: erindru

sqlmesh/core/engine_adapter/athena.py

@@ -45,6 +45,7 @@ class AthenaEngineAdapter(PandasNativeFetchDFSupportMixin, RowDiffMixin):
     # >>> self._execute('/* test */ DESCRIBE foo')
     #     pyathena.error.OperationalError: FAILED: ParseException line 1:0 cannot recognize input near '/' '*' 'test'
     ATTACH_CORRELATION_ID = False
+    SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["DATABASE", "SCHEMA"]
 
     def __init__(
         self, *args: t.Any, s3_warehouse_location: t.Optional[str] = None, **kwargs: t.Any

sqlmesh/core/engine_adapter/base.py

@@ -108,6 +108,7 @@ class EngineAdapter:
     SUPPORTS_CLONING = False
     SUPPORTS_MANAGED_MODELS = False
     SUPPORTS_CREATE_DROP_CATALOG = False
+    SUPPORTED_DROP_CASCADE_OBJECT_KINDS: t.List[str] = []
     SCHEMA_DIFFER = SchemaDiffer()
     SUPPORTS_TUPLE_IN = True
     HAS_VIEW_BINDING = False
@@ -1060,6 +1061,7 @@ def _drop_object(
         name: TableName | SchemaName,
         exists: bool = True,
         kind: str = "TABLE",
+        cascade: bool = False,
         **drop_args: t.Any,
     ) -> None:
         """Drops an object.
@@ -1070,8 +1072,13 @@ def _drop_object(
             name: The name of the table to drop.
             exists: If exists, defaults to True.
             kind: What kind of object to drop. Defaults to TABLE
+            cascade: Whether or not to DROP ... CASCADE.
+                Note that this is ignored for :kind's that are not present in self.SUPPORTED_DROP_CASCADE_OBJECT_KINDS
             **drop_args: Any extra arguments to set on the Drop expression
         """
+        if cascade and kind.upper() in self.SUPPORTED_DROP_CASCADE_OBJECT_KINDS:
+            drop_args["cascade"] = cascade
+
         self.execute(exp.Drop(this=exp.to_table(name), kind=kind, exists=exists, **drop_args))
 
     def get_alter_expressions(

sqlmesh/core/engine_adapter/base_postgres.py

@@ -24,6 +24,7 @@ class BasePostgresEngineAdapter(EngineAdapter):
     DEFAULT_BATCH_SIZE = 400
     COMMENT_CREATION_TABLE = CommentCreationTable.COMMENT_COMMAND_ONLY
     COMMENT_CREATION_VIEW = CommentCreationView.COMMENT_COMMAND_ONLY
+    SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["SCHEMA", "TABLE", "VIEW"]
 
     def columns(
         self, table_name: TableName, include_pseudo_columns: bool = False

sqlmesh/core/engine_adapter/bigquery.py

@@ -68,6 +68,7 @@ class BigQueryEngineAdapter(InsertOverwriteWithMergeMixin, ClusteredByMixin, Row
     SUPPORTS_CLONING = True
     MAX_TABLE_COMMENT_LENGTH = 1024
     MAX_COLUMN_COMMENT_LENGTH = 1024
+    SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["SCHEMA"]
 
     SCHEMA_DIFFER = SchemaDiffer(
         compatible_types={
@@ -1260,21 +1261,6 @@ def _native_df_to_pandas_df(
 
         return super()._native_df_to_pandas_df(query_or_df)
 
-    def _drop_object(
-        self,
-        name: TableName | SchemaName,
-        exists: bool = True,
-        kind: str = "TABLE",
-        **drop_args: t.Any,
-    ) -> None:
-        if kind.upper() == "TABLE" and "cascade" in drop_args:
-            # BigQuery doesnt support DROP CASCADE for tables
-            # ref: https://cloud.google.com/bigquery/docs/reference/standard-sql/data-definition-language#drop_table_statement
-            # so set it to False here so SQLGlot doesnt output a CASCADE argument
-            drop_args["cascade"] = False
-
-        super()._drop_object(name=name, exists=exists, kind=kind, **drop_args)
-
     @property
     def _query_data(self) -> t.Any:
         return self._connection_pool.get_attribute("query_data")

sqlmesh/core/engine_adapter/clickhouse.py

@@ -614,6 +614,7 @@ def _drop_object(
         name: TableName | SchemaName,
         exists: bool = True,
         kind: str = "TABLE",
+        cascade: bool = False,
         **drop_args: t.Any,
     ) -> None:
         """Drops an object.
@@ -626,7 +627,6 @@ def _drop_object(
             kind: What kind of object to drop. Defaults to TABLE
             **drop_args: Any extra arguments to set on the Drop expression
         """
-        drop_args.pop("cascade", None)
         self.execute(
             exp.Drop(
                 this=exp.to_table(name),

sqlmesh/core/engine_adapter/duckdb.py

@@ -37,6 +37,7 @@ class DuckDBEngineAdapter(LogicalMergeMixin, GetCurrentCatalogFromFunctionMixin,
     COMMENT_CREATION_TABLE = CommentCreationTable.COMMENT_COMMAND_ONLY
     COMMENT_CREATION_VIEW = CommentCreationView.COMMENT_COMMAND_ONLY
     SUPPORTS_CREATE_DROP_CATALOG = True
+    SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["SCHEMA", "TABLE", "VIEW"]
 
     @property
     def catalog_support(self) -> CatalogSupport:

sqlmesh/core/engine_adapter/snowflake.py

@@ -55,6 +55,7 @@ class SnowflakeEngineAdapter(GetCurrentCatalogFromFunctionMixin, ClusteredByMixi
     SUPPORTS_MANAGED_MODELS = True
     CURRENT_CATALOG_EXPRESSION = exp.func("current_database")
     SUPPORTS_CREATE_DROP_CATALOG = True
+    SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["DATABASE", "SCHEMA", "TABLE"]
     SCHEMA_DIFFER = SchemaDiffer(
         parameterized_type_defaults={
             exp.DataType.build("BINARY", dialect=DIALECT).this: [(8388608,)],

sqlmesh/core/engine_adapter/spark.py

@@ -57,6 +57,7 @@ class SparkEngineAdapter(
     # currently check for storage formats we say we don't support REPLACE TABLE
     SUPPORTS_REPLACE_TABLE = False
     QUOTE_IDENTIFIERS_IN_VIEWS = False
+    SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["DATABASE", "SCHEMA"]
 
     WAP_PREFIX = "wap_"
     BRANCH_PREFIX = "branch_"

sqlmesh/core/engine_adapter/trino.py

@@ -53,6 +53,7 @@ class TrinoEngineAdapter(
     COMMENT_CREATION_TABLE = CommentCreationTable.IN_SCHEMA_DEF_NO_CTAS
     COMMENT_CREATION_VIEW = CommentCreationView.COMMENT_COMMAND_ONLY
     SUPPORTS_REPLACE_TABLE = False
+    SUPPORTED_DROP_CASCADE_OBJECT_KINDS = ["SCHEMA"]
     DEFAULT_CATALOG_TYPE = "hive"
     QUOTE_IDENTIFIERS_IN_VIEWS = False
     SCHEMA_DIFFER = SchemaDiffer(