issue #112 fix for Silver Transformations - Silver Dataflowspec Table…

src/onboard_dataflowspec.py

@@ -197,6 +197,7 @@ def onboard_silver_dataflow_spec(self):
                 StructField(
                     "target_partition_cols", ArrayType(StringType(), True), True
                 ),
+                StructField("database", StringType(), True),
                 StructField("target_table", StringType(), True),
                 StructField("where_clause", ArrayType(StringType(), True), True),
             ]
@@ -221,11 +222,18 @@ def onboard_silver_dataflow_spec(self):
 
         logger.info(silver_transformation_json_file)
 
+        silver_transformation_json_df.show()
+        silver_data_flow_spec_df.show()
+
         silver_data_flow_spec_df = silver_transformation_json_df.join(
             silver_data_flow_spec_df,
-            silver_transformation_json_df.target_table
-            == silver_data_flow_spec_df.targetDetails["table"],
+            (silver_transformation_json_df.target_table == silver_data_flow_spec_df.targetDetails["table"]) &
+            ((silver_transformation_json_df.database.isNull()) | (
+                        silver_transformation_json_df.database == silver_data_flow_spec_df.targetDetails["database"]))
         )
+        if silver_data_flow_spec_df.count() != silver_transformation_json_df.count():
+            raise Exception("The database name specified in the silver transformation JSON does not match the one in the onboarding template's silver table. Please verify!")
+
         silver_dataflow_spec_df = (
             silver_data_flow_spec_df.drop("target_table")  # .drop("path")
             .drop("target_partition_cols")

tests/resources/onboarding_silver_spec_with_similar_table_names.json

@@ -0,0 +1,76 @@
+[
+    {
+        "data_flow_id": "100",
+        "data_flow_group": "A1",
+        "source_system": "MYSQL",
+        "source_format": "cloudFiles",
+        "source_details": {
+            "source_database": "APP",
+            "source_table": "CUSTOMERS",
+            "source_path_dev": "tests/resources/data/customers",
+            "source_schema_path": "tests/resources/schema/customer_schema.ddl",
+            "source_metadata": {
+                "include_autoloader_metadata_column": "True",
+                "autoloader_metadata_col_name": "source_metadata",
+                "select_metadata_cols": {
+                    "input_file_name": "_metadata.file_name",
+                    "input_file_path": "_metadata.file_path"
+                }
+            }
+        },
+        "bronze_database_dev": "bronze",
+        "bronze_database_staging": "bronze",
+        "bronze_database_prd": "bronze",
+        "bronze_table": "customers_cdc",
+        "bronze_reader_options": {
+            "cloudFiles.format": "json",
+            "cloudFiles.inferColumnTypes": "true",
+            "cloudFiles.rescuedDataColumn": "_rescued_data"
+        },
+        "bronze_table_path_dev": "tests/resources/delta/customers",
+        "bronze_cdc_apply_changes": {
+            "keys": [
+                "id"
+            ],
+            "sequence_by": "operation_date",
+            "scd_type": "1",
+            "apply_as_deletes": "operation = 'DELETE'",
+            "except_column_list": [
+                "operation",
+                "operation_date",
+                "_rescued_data"
+            ]
+        },
+        "bronze_table_properties": {
+            "pipelines.autoOptimize.managed": "false",
+            "pipelines.reset.allowed": "false"
+        },
+        "bronze_data_quality_expectations_json_dev": "tests/resources/dqe/customers/bronze_data_quality_expectations.json",
+        "silver_database_dev": "silver",
+        "silver_database_staging": "silver1",
+        "silver_database_prd": "silver",
+        "silver_table": "customers",
+        "silver_cdc_apply_changes": {
+            "keys": [
+                "id"
+            ],
+            "sequence_by": "operation_date",
+            "scd_type": "1",
+            "apply_as_deletes": "operation = 'DELETE'",
+            "except_column_list": [
+                "operation",
+                "operation_date",
+                "_rescued_data"
+            ]
+        },
+        "silver_table_path_dev": "tests/resources/data/silver/customers",
+        "silver_table_properties": {
+            "pipelines.autoOptimize.managed": "false",
+            "pipelines.reset.allowed": "false",
+            "pipelines.autoOptimize.zOrderCols": "id,email"
+        },
+        "silver_transformation_json_dev": "tests/resources/silver_transformations_with_similar_tables.json",
+        "silver_transformation_json_staging": "tests/resources/silver_transformations_with_similar_tables.json",
+        "silver_data_quality_expectations_json_dev": "tests/resources/dqe/customers/silver_data_quality_expectations.json"
+    }
+]

tests/resources/silver_transformations_with_similar_tables.json

@@ -0,0 +1,37 @@
+[
+  {
+    "target_table": "customers",
+    "select_exp": [
+      "address",
+      "email",
+      "firstname",
+      "id",
+      "lastname",
+      "operation_date",
+      "operation",
+      "_rescued_data"
+    ],
+    "where_clause": [
+      "id IS NOT NULL",
+      "email is not NULL"
+    ]
+  },
+  {
+    "target_table": "customers",
+    "database": "silver",
+    "select_exp": [
+      "address",
+      "email",
+      "firstname",
+      "id",
+      "lastname",
+      "operation_date",
+      "operation",
+      "_rescued_data"
+    ],
+    "where_clause": [
+      "id IS NOT NULL",
+      "email is not NULL"
+    ]
+  }
+]

tests/test_onboard_dataflowspec.py

@@ -384,3 +384,26 @@ def test_onboard_apply_changes_from_snapshot_negative(self):
         onboardDataFlowSpecs = OnboardDataflowspec(self.spark, onboarding_params_map, uc_enabled=True)
         with self.assertRaises(Exception):
             onboardDataFlowSpecs.onboard_bronze_dataflow_spec()
+
+    def test_onboard_silver_spec_positive(self):
+        onboarding_params_map = copy.deepcopy(self.onboarding_bronze_silver_params_map)
+        onboarding_params_map['env'] = 'dev'
+        onboarding_params_map["onboarding_file_path"] = self.onboarding_silver_spec_with_similar_table_names
+        onboardDataFlowSpecs = OnboardDataflowspec(self.spark, onboarding_params_map, uc_enabled=True)
+        onboardDataFlowSpecs.onboard_silver_dataflow_spec()
+
+        silver_dataflowSpec_df = self.read_dataflowspec(
+            self.onboarding_bronze_silver_params_map['database'],
+            self.onboarding_bronze_silver_params_map['silver_dataflowspec_table'])
+        silver_dataflowSpec_df.show(truncate=False)
+
+        self.assertEqual(silver_dataflowSpec_df.count(), 2)
+
+    def test_onboard_silver_spec_negative(self):
+        onboarding_params_map = copy.deepcopy(self.onboarding_bronze_silver_params_map)
+        onboarding_params_map['env'] = 'staging'
+        onboarding_params_map["onboarding_file_path"] = self.onboarding_silver_spec_with_similar_table_names
+        onboardDataFlowSpecs = OnboardDataflowspec(self.spark, onboarding_params_map, uc_enabled=True)
+        with self.assertRaises(Exception):
+            onboardDataFlowSpecs.onboard_silver_dataflow_spec()
+

tests/utils.py

@@ -76,6 +76,8 @@ def setUp(self):
             "uc_enabled": "True"
         }
 
+        self.onboarding_silver_spec_with_similar_table_names = "tests/resources/onboarding_silver_spec_with_similar_table_names.json"
+
     def tearDown(self):
         """Tear down."""
         self.deltaPipelinesMetaStoreOps.drop_database("ravi_dlt_demo")