Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
liamperritt merged 3 commits into from
Apr 7, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
573d74d
Fix FlowAppendView support for 'once' flag by forcing batch read
liamperritt Mar 1, 2026
06b7c6f
Add new sample dataflow for append_view_once_flow
liamperritt Apr 2, 2026
1d196d8
Improve flow name in append_view_once dataflow
liamperritt Apr 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 37 additions & 0 deletions .../bronze_sample/src/dataflows/feature_samples/dataflowspec/append_view_once_flow_main.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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"
}
}
}
}
}
}
]
}
7 changes: 5 additions & 2 deletions src/dataflow/flows/append_view.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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
Copy link
Copy Markdown
Collaborator

@haillew haillew Apr 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

what happens in streaming flows if you have this turned on, have we tested this? and can we please add a sample for this as well if possible

Copy link
Copy Markdown
Collaborator Author

@liamperritt liamperritt Apr 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've successfully tested this manually, but I'm happy to add a sample as well.

Copy link
Copy Markdown
Collaborator

@haillew haillew Apr 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

awesome, yeah that would be good if you could add one into the features samples under bronze, we use these as regression testing for now and will later turn into the tests for CICD pipeline so if we have a test for every feature that would help make sure we don't break anything on future developments

Copy link
Copy Markdown
Collaborator Author

@liamperritt liamperritt Apr 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, sounds good.

I just finished testing this change on new samples locally with these results:

  • When the source view has mode: batch, the flow succeeds (the batch view is inserted "once" into the target streaming table during the initial pipeline run, and then ignored in subsequent runs).
  • When the source view has mode: stream, the flow fails with View 'v_append_view_once_stream_flow' is a streaming view and must be referenced using readStream.

This behaviour is expected, as customers should only ever use a batch view as the source for an append once flow. As outlined in the append_flow documentation:

Using once=True changes the flow: the return value must be a batch DataFrame in this case, not a streaming DataFrame.

Copy link
Copy Markdown
Collaborator Author

@liamperritt liamperritt Apr 2, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here's a link to my test pipeline run: https://e2-demo-field-eng.cloud.databricks.com/pipelines/d55b8d12-d5c0-4e8e-85d9-234f3d99d594/updates/d2073a2f-c1ca-4094-9727-7b18212c67bd?o=1444828305810485

The relevant table is named append_view_once_flow

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([
Expand Down