fix(flows): FlowAppendView support for 'once' flag by forcing batch read (#22)

* Fix FlowAppendView support for 'once' flag by forcing batch read

* Add new sample dataflow for append_view_once_flow

* Improve flow name in append_view_once dataflow

Author: liamperritt

samples/bronze_sample/src/dataflows/feature_samples/dataflowspec/append_view_once_flow_main.json

@@ -0,0 +1,37 @@
+{
+    "dataFlowId": "append_view_once_flow",
+    "dataFlowGroup": "feature_samples_general",
+    "dataFlowType": "flow",
+    "targetFormat": "delta",
+    "targetDetails": {
+        "table": "append_view_once_flow",
+        "tableProperties": {
+            "delta.enableChangeDataFeed": "true"
+        }
+    },
+    "flowGroups": [
+        {
+            "flowGroupId": "main",
+            "flows": {
+                "f_customer_append_view_once": {
+                    "flowType": "append_view",
+                    "flowDetails": {
+                        "targetTable": "append_view_once_flow",
+                        "sourceView": "v_append_view_once_flow",
+                        "once": true
+                    },
+                    "views": {
+                        "v_append_view_once_flow": {
+                            "mode": "batch",
+                            "sourceType": "delta",
+                            "sourceDetails": {
+                                "database": "{staging_schema}",
+                                "table": "customer"
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    ]
+}

src/dataflow/flows/append_view.py

@@ -31,7 +31,7 @@ def columnPrefixExceptions(self) -> List[str]:
 
     @property
     def once(self) -> bool:
-        """Get the once flag."""
+        """Get the once flag. Note: Setting 'once' requires a batch read."""
         return self.flowDetails.get("once", False)
 
     def create_flow(
@@ -55,14 +55,17 @@ def get_column_prefix_exceptions(flow_config: FlowConfig) -> List[str]:
             return column_prefix_exceptions
 
         spark = self.spark
+        spark_reader = spark.readStream
+        if self.once:
+            spark_reader = spark.read
         exclude_columns = flow_config.exclude_columns
         column_prefix_exceptions = get_column_prefix_exceptions(flow_config)
 
         source_view_name = f'live.{self.sourceView}'
 
         @dp.append_flow(name=self.flowName, target=self.targetTable, once=self.once)
         def flow_transform():
-            df = spark.readStream.table(source_view_name)
+            df = spark_reader.table(source_view_name)
             if "column_prefix" in self.flowDetails:
                 prefix = f"{self.columnPrefix.lower()}_"
                 df = df.select([