1.PR Commnet: refactored dataflow where clause

2.PR Comment: removed --- from docs
3.PR Comment: Corrected delta read condition

Author: ravi-db

docs/content/getting_started/additionals.md

@@ -23,7 +23,7 @@ export DATABRICKS_TOKEN=<DATABRICKS TOKEN> # Account needs permission to create
 5. Run integration test against cloudfile or eventhub or kafka using below options:
     5a. Run the command for cloudfiles ```python integration-tests/run-integration-test.py  --cloud_provider_name=aws --dbr_version=11.3.x-scala2.12 --source=cloudfiles --dbfs_path=dbfs:/tmp/DLT-META/```
 
-    5b. Run the command for eventhub ```python integration-tests/run-integration-test.py --cloud_provider_name=azure --dbr_version=11.3.x-scala2.12 --source=eventhub --dbfs_path=dbfs:/tmp/DLT-META/ --eventhub_name=iot --eventhub_secrets_scope_name=eventhubs_creds --eventhub_namespace=int_test-standard --eventhub_port=9093 --eventhub_producer_accesskey_name=producer ----eventhub_consumer_accesskey_name=consumer```
+    5b. Run the command for eventhub ```python integration-tests/run-integration-test.py --cloud_provider_name=azure --dbr_version=11.3.x-scala2.12 --source=eventhub --dbfs_path=dbfs:/tmp/DLT-META/ --eventhub_name=iot --eventhub_secrets_scope_name=eventhubs_creds --eventhub_namespace=int_test-standard --eventhub_port=9093 --eventhub_producer_accesskey_name=producer --eventhub_consumer_accesskey_name=consumer```
 
     For eventhub integration tests, the following are the prerequisites:
     1. Needs eventhub instance running
@@ -34,7 +34,7 @@ export DATABRICKS_TOKEN=<DATABRICKS TOKEN> # Account needs permission to create
     1. Provide your eventhub topic : --eventhub_name
     2. Provide eventhub namespace : --eventhub_namespace
     3. Provide eventhub port : --eventhub_port
-    4. Provide databricks secret scope name : ----eventhub_secrets_scope_name
+    4. Provide databricks secret scope name : --eventhub_secrets_scope_name
     5. Provide eventhub producer access key name : --eventhub_producer_accesskey_name
     6. Provide eventhub access key name : --eventhub_consumer_accesskey_name
 

src/dataflow_pipeline.py

@@ -96,9 +96,9 @@ def read_bronze(self) -> DataFrame:
         """Read Bronze Table."""
         logger.info("In read_bronze func")
         bronze_dataflow_spec: BronzeDataflowSpec = self.dataflowSpec
-        if bronze_dataflow_spec.sourceFormat == "cloudFiles" or bronze_dataflow_spec.sourceFormat == "delta":
+        if bronze_dataflow_spec.sourceFormat == "cloudFiles":
             return PipelineReaders.read_dlt_cloud_files(self.spark, bronze_dataflow_spec, self.schema_json)
-        if bronze_dataflow_spec.sourceFormat == "delta":
+        elif bronze_dataflow_spec.sourceFormat == "delta":
             return PipelineReaders.read_dlt_delta(self.spark, bronze_dataflow_spec)
         elif bronze_dataflow_spec.sourceFormat == "eventhub" or bronze_dataflow_spec.sourceFormat == "kafka":
             return PipelineReaders.read_kafka(self.spark, bronze_dataflow_spec, self.schema_json)
@@ -117,12 +117,25 @@ def get_silver_schema(self):
             format="delta"
             # #f"{source_database}.{source_table}"
         ).selectExpr(*select_exp)
+        raw_delta_table_stream = self.__apply_where_clause(where_clause, raw_delta_table_stream)
+        return raw_delta_table_stream.schema
+    
+    def __apply_where_clause(self, where_clause, raw_delta_table_stream):
+        """This method apply where clause provided in silver transformations
+
+        Args:
+            where_clause (_type_): _description_
+            raw_delta_table_stream (_type_): _description_
+
+        Returns:
+            _type_: _description_
+        """
         if where_clause:
             where_clause_str = " ".join(where_clause)
             if len(where_clause_str.strip()) > 0:
                 for where_clause in where_clause:
                     raw_delta_table_stream = raw_delta_table_stream.where(where_clause)
-        return raw_delta_table_stream.schema
+        return raw_delta_table_stream
 
     def read_silver(self) -> DataFrame:
         """Read Silver tables."""

src/pipeline_readers.py

@@ -53,8 +53,6 @@ def read_dlt_delta(spark, bronze_dataflow_spec) -> DataFrame:
         Args:
             spark (_type_): _description_
             bronze_dataflow_spec (_type_): _description_
-            schema_json (_type_): _description_
-
         Returns:
             DataFrame: _description_
         """