Repository: spetlr-org/spetlr (language: python)

Issue:
feat: delete schema on mismatch (delta and sqlserver)
Waits for #35  and #32 ....

In order to dynamically drop and recreate test tables, this function helps


Discussion / hints:

Fix patch:
diff --git a/docs/utilities/README.md b/docs/utilities/README.md
index bcfa4eb9..a813aed5 100644
--- a/docs/utilities/README.md
+++ b/docs/utilities/README.md
@@ -6,6 +6,7 @@ Utilities in spetlr:
 * [Test Utilities](#test-utilities)
 * [Git Hooks](#git-hooks)
 * [Cleanup Test Tables](#cleanup-test-tables)
+* [Delete Mismatched Schemas](#delete-mismatched-schemas)
 
 ## Api Auto Config
 
@@ -104,3 +105,52 @@ class ExampleTests(DataframeTestCase):
     def tearDownClass(cls) -> None:
         CleanupTestDatabases()
 ```
+
+## Delete Mismatched Schemas
+If the dataplatform needs an dynamic approach to automatically drop
+and recreate a table, if the table defined in code and the table
+that exists in production mismatches. Then, you can use `DeleteMismatchedSchemas`.
+
+*It is the responsibility of the developer to only add tables,
+where the code has the property that it can rebuild dropped tables.*
+
+Run this in your setup job:
+
+```python
+from spetlr.utils import DeleteMismatchedSchemas
+from tests.cluster.delta.SparkExecutor import SparkSqlExecutor
+
+# Replace the SparkSqlExecutor with your dataplatform executor
+DeleteMismatchedSchemas(spark_executor=SparkSqlExecutor())
+```
+
+If the tables must not be down for too long (e.g Sql Server tables used in reports).
+One could do the mismatch deletion in the ETL job:
+
+```python
+from spetlr.utils import DeleteMismatchedSchemas
+from tests.cluster.delta.SparkExecutor import SparkSqlExecutor
+
+# Replace the SparkSqlExecutor with your dataplatform executor
+DeleteMismatchedSchemas(table_ids_to_check=["SqlServerTableId"],
+                        spark_executor=SparkSqlExecutor())
+# Recreate the table
+SparkSqlExecutor().execute_sql_file("some-file")
+```
+
+To configure which tables to delete on mismatch either you define it in the SPETLR yml configuration:
+
+```yaml
+SparkTestTable1:
+  name: "{SparkTestDb}.tbl1"
+  path: "{SparkTestDb_path}/tbl1"
+  delete_on_delta_schema_mismatch: true
+```
+
+or, you give the ids as input:
+
+````python
+from spetlr.utils import DeleteMismatchedSchemas
+
+DeleteMismatchedSchemas(["TABLEID"])
+````
diff --git a/src/spetlr/utils/DeleteMismatchedSchemas.py b/src/spetlr/utils/DeleteMismatchedSchemas.py
new file mode 100644
index 00000000..570557f7
--- /dev/null
+++ b/src/spetlr/utils/DeleteMismatchedSchemas.py
@@ -0,0 +1,85 @@
+import json
+from textwrap import indent
+from typing import List
+
+from pyspark.sql.utils import AnalysisException
+
+from spetlr.configurator import Configurator
+from spetlr.delta import DeltaHandle
+from spetlr.delta.delta_handle import DeltaHandleInvalidName
+from spetlr.sql import SqlExecutor
+from spetlr.utils import CleanupTestDatabases
+
+
+def get_table_ids_to_check():
+    print("Remember to initialize the configurator first!")
+    c = Configurator()
+    table_ids_to_check = []
+    for key in c._raw_resource_details.keys():
+        if c.table_property(key, "delete_on_delta_schema_mismatch", False):
+            table_ids_to_check.append(key)
+    return table_ids_to_check
+
+
+def DeleteMismatchedSchemas(
+    table_ids_to_check: List[str] = None,
+    spark_executor: SqlExecutor = None,
+    sql_files_pattern: str = "*",
+):
+    """For the following tables, if the production schema does not match
+    the configured schema, delete the table.
+
+    It is the responsibility of the developer to only add tables here where the
+    code has the property that it can rebuild dropped tables.
+
+    Tables can be flagged by using ´delete_on_delta_schema_mismatch´
+    as Configurator property.
+
+    """
+
+    if table_ids_to_check is None:
+        table_ids_to_check = get_table_ids_to_check()
+
+    executor = spark_executor
+
+    # first gather the configured schemas from the test databases
+    print("Remember to initialize the configurator first!")
+    configurator = Configurator()
+    configurator.set_debug()
+    executor.execute_sql_file(sql_files_pattern)
+
+    schemas = {}
+    for tbl_id in table_ids_to_check:
+        try:
+            schemas[tbl_id] = DeltaHandle.from_tc(tbl_id).read().schema
+        except DeltaHandleInvalidName:
+            print(f"Not a valid delta handle {tbl_id}")
+            continue
+
+    # cleanup
+    CleanupTestDatabases()
+
+    # now get the production schemas from the production tables
+    affected_keys = []
+    configurator.set_prod()
+    for tbl_id in table_ids_to_check:
+        try:
+            prod_schema = DeltaHandle.from_tc(tbl_id).read().schema
+        except AnalysisException:
+            print(f"Exception in reading production version of table id {tbl_id}")
+            continue
+        except DeltaHandleInvalidName:
+            print(f"Not a valid delta handle {tbl_id}")
+            continue
+        if prod_schema != schemas[tbl_id]:
+            print(f"Schema mismatch detected for table id {tbl_id}.")
+            print(
+                "  Production table schema:",
+                indent(json.dumps(prod_schema.jsonValue(), indent=4), "  "),
+            )
+            print(
+                "  Configured table schema:",
+                indent(json.dumps(schemas[tbl_id].jsonValue(), indent=4), "  "),
+            )
+            DeltaHandle.from_tc(tbl_id).drop_and_delete()
+            affected_keys.append(tbl_id)
diff --git a/src/spetlr/utils/__init__.py b/src/spetlr/utils/__init__.py
index e79d63dc..3ae0fd49 100644
--- a/src/spetlr/utils/__init__.py
+++ b/src/spetlr/utils/__init__.py
@@ -1,5 +1,6 @@
 from .CleanupTestDatabases import CleanupTestDatabases
 from .DataframeCreator import DataframeCreator
+from .DeleteMismatchedSchemas import DeleteMismatchedSchemas
 from .DropOldestDuplicates import DropOldestDuplicates
 from .GetMergeStatement import GetMergeStatement
 from .MockExtractor import MockExtractor
@@ -14,4 +15,5 @@
     DropOldestDuplicates,
     SelectAndCastColumns,
     CleanupTestDatabases,
+    DeleteMismatchedSchemas,
 ]


Test patch:
diff --git a/tests/cluster/delta/extras/tablenames.yml b/tests/cluster/delta/extras/tablenames.yml
index 28f5290e..3c3073f6 100644
--- a/tests/cluster/delta/extras/tablenames.yml
+++ b/tests/cluster/delta/extras/tablenames.yml
@@ -32,3 +32,4 @@ DeleteDataLoaderDb:
 DeleteDataLoaderDummy:
   name: "{DeleteDataLoaderDb}.Dummy"
   path: "{DeleteDataLoaderDb_path}/dummy"
+
diff --git a/tests/cluster/utils/__init__.py b/tests/cluster/utils/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/cluster/utils/extras/__init__.py b/tests/cluster/utils/extras/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/cluster/utils/extras/changed_schema.sql b/tests/cluster/utils/extras/changed_schema.sql
new file mode 100644
index 00000000..bfcc407b
--- /dev/null
+++ b/tests/cluster/utils/extras/changed_schema.sql
@@ -0,0 +1,9 @@
+CREATE DATABASE IF NOT EXISTS {SparkTestDbMismatch}
+LOCATION "{SparkTestDbMismatch_path}";
+
+CREATE TABLE IF NOT EXISTS {MismatchSparkTestTable}(
+a int,
+b int -- Added a new column
+)
+USING DELTA
+LOCATION "{MismatchSparkTestTable_path}"
diff --git a/tests/cluster/utils/extras/initial_schema.sql b/tests/cluster/utils/extras/initial_schema.sql
new file mode 100644
index 00000000..e41a4a31
--- /dev/null
+++ b/tests/cluster/utils/extras/initial_schema.sql
@@ -0,0 +1,8 @@
+CREATE DATABASE IF NOT EXISTS {SparkTestDbMismatch}
+LOCATION "{SparkTestDbMismatch_path}";
+
+CREATE TABLE IF NOT EXISTS {MismatchSparkTestTable}(
+a int
+)
+USING DELTA
+LOCATION "{MismatchSparkTestTable_path}"
diff --git a/tests/cluster/utils/extras/tablenames.yml b/tests/cluster/utils/extras/tablenames.yml
new file mode 100644
index 00000000..ee5ebf98
--- /dev/null
+++ b/tests/cluster/utils/extras/tablenames.yml
@@ -0,0 +1,8 @@
+SparkTestDbMismatch:
+  name: my_db1{mismatchuuid}_mismatch{ID}
+  path: /tmp/foo/bar/my_db_mismatch{mismatchuuid}{ID}/
+
+MismatchSparkTestTable:
+  name: "{SparkTestDbMismatch}.tbl"
+  path: "{SparkTestDbMismatch_path}/tbl"
+  delete_on_delta_schema_mismatch: true
diff --git a/tests/cluster/utils/test_delete_schema_mismatch.py b/tests/cluster/utils/test_delete_schema_mismatch.py
new file mode 100644
index 00000000..bb07b28c
--- /dev/null
+++ b/tests/cluster/utils/test_delete_schema_mismatch.py
@@ -0,0 +1,96 @@
+import uuid as _uuid
+
+import pyspark.sql.types as T
+from spetlrtools.testing import DataframeTestCase
+
+from spetlr.configurator import Configurator
+from spetlr.delta import DbHandle, DeltaHandle
+from spetlr.sql import SqlExecutor
+from spetlr.utils import DeleteMismatchedSchemas
+from tests.cluster.utils import extras
+
+
+class DeleteSchemaMismatch(DataframeTestCase):
+
+    """
+    Test class for DeleteMismatchedSchemas.
+
+    This class contains tests for the `DeleteMismatchedSchemas` utility class.
+    It verifies the behavior of the utility when dealing with Delta tables
+    that have schema mismatches.
+
+    The class defines the initial and changed schema for testing purposes.
+    It also sets up the necessary configuration and resources for the tests.
+
+    NB: This test is special, since it expects production tables.
+        Do not use .set_prod() in tests.
+
+    """
+
+    initial_schema_delta = T.StructType(
+        [
+            T.StructField("a", T.IntegerType(), True),
+        ]
+    )
+
+    changed_schema_delta = T.StructType(
+        [
+            T.StructField("a", T.IntegerType(), True),
+            T.StructField("b", T.IntegerType(), True),
+        ]
+    )
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        Configurator().register("mismatchuuid", str(_uuid.uuid4().hex))
+        Configurator().add_resource_path(extras)
+
+        # This extra configuration is necessary
+        # since SPETLR tests runs parallel on multiple cluster
+        # but in same Databricks workspace
+        # The production tables will interfere with each other without an unique id
+
+        # This is a special case.
+        # Dont use set_prod in testcases
+        Configurator().set_prod()
+
+        cls.executor = SqlExecutor(base_module=extras)
+
+    def test_01_no_change_delta(self):
+        """Test scenario with no schema change in Delta table."""
+        dbhandle = DbHandle.from_tc("SparkTestDbMismatch")
+        dbhandle.drop_cascade()
+
+        # Setup production table
+
+        self.executor.execute_sql_file("initial_schema")
+
+        dh = DeltaHandle.from_tc("MismatchSparkTestTable")
+
+        self.assertEqual(dh.read().schema, self.initial_schema_delta)
+
+        DeleteMismatchedSchemas(
+            spark_executor=self.executor, sql_files_pattern="initial_schema"
+        )
+
+        self.assertEqual(dh.read().schema, self.initial_schema_delta)
+
+    def test_02_change_delta(self):
+        """Test scenario with schema change in Delta table."""
+        # Ensure that the database is dropped
+        dbhandle = DbHandle.from_tc("SparkTestDbMismatch")
+        dbhandle.drop_cascade()
+
+        # Setup production table
+        self.executor.execute_sql_file("initial_schema")
+
+        dh = DeltaHandle.from_tc("MismatchSparkTestTable")
+
+        self.assertEqual(dh.read().schema, self.initial_schema_delta)
+
+        DeleteMismatchedSchemas(
+            spark_executor=self.executor, sql_files_pattern="changed_schema"
+        )
+        self.executor.execute_sql_file("changed_schema")
+
+        self.assertEqual(dh.read().schema, self.changed_schema_delta)