AnishMahto commented on code in PR #60086: URL: https://github.com/apache/spark/pull/60086#discussion_r4471309214
########## docs/declarative-pipelines-programming-guide.md: ########## @@ -620,13 +620,15 @@ dp.create_auto_cdc_flow( | `apply_as_deletes` | No | A boolean expression identifying events that represent deletes. Matching rows are removed from the target. A SQL expression string or a `Column`. | | `column_list` | No | The columns to include in the target. Mutually exclusive with `except_column_list`. | | `except_column_list` | No | The columns to exclude from the target; all other columns are included. Mutually exclusive with `column_list`. Commonly used to drop operation/metadata columns such as `op`. | -| `stored_as_scd_type` | No | The SCD type of the target. Pass `1` (or `"1"`) for the Type 1 behavior described here. Type 2 is also accepted; see [SPARK-58570](https://issues.apache.org/jira/browse/SPARK-58570). | +| `stored_as_scd_type` | No | The SCD type of the target: `1` or `2` (or `"1"` or `"2"`). Defaults to Type 1. See [SCD Type 2: Keeping History](#scd-type-2-keeping-history). | +| `track_history_column_list` | No | SCD Type 2 only. The columns whose changes open a new version. Keyword-only. Mutually exclusive with `track_history_except_column_list`. | +| `track_history_except_column_list` | No | SCD Type 2 only. The columns whose changes do not open a new version; changes to any other non-key column in the target do. Keyword-only. Mutually exclusive with `track_history_column_list`. | | `name` | No | The name of the flow. Defaults to the target table name. | Review Comment: nit: keep name above `track_history_column_list` but below `stored_as_scd_type`. `name` is a positional API argument, along with `stored_as_scd_type` and everything above. `track_history_column_list` is a key-word only argument (and everything below it should be too). ########## docs/declarative-pipelines-programming-guide.md: ########## @@ -807,6 +810,198 @@ spark-pipelines run The table is unchanged. This time the stale event is alone in its batch, so there is no newer event to supersede it within the batch; instead Auto CDC compares its version against the state already recorded for `id 1` and discards it, because version 1 is below the committed version 2. Applied in arrival order, this event would have wrongly reverted `id 1` to `alice`. +### SCD Type 2: Keeping History + +With SCD Type 2, the target keeps every version of a row instead of overwriting it. Auto CDC adds two columns to the target that record when each version was current: + +- `__START_AT` - the sequence value at which the version became current. +- `__END_AT` - the sequence value at which the version stopped being current, or `NULL` for the current version. + +Both columns have the type of the sequencing expression. Auto CDC adds them to the target schema; the source does not supply them. Type 2 has the same [Requirements](#requirements) as Type 1. + +The current version of each key is the row whose `__END_AT` is `NULL`: + +```sql +SELECT * FROM customers WHERE __END_AT IS NULL; +``` + +The version that was current at a given sequence value is the row whose interval contains that value. For `version` 1: + +```sql +SELECT * FROM customers +WHERE __START_AT <= 1 AND (__END_AT > 1 OR __END_AT IS NULL); +``` + +#### How Type 2 Applies Each Event + +- **Inserts and updates** - The first event for a key opens its first version. A later event that changes a tracked column closes the key's current version at the event's sequence value and opens a new version. See [Choosing Which Columns Open a New Version](#choosing-which-columns-open-a-new-version). +- **Deletes** - A delete closes the key's current version at the delete's sequence value and opens no new one, so the key has no row with a `NULL` `__END_AT`. The key's earlier versions stay in the target. A later insert for the key opens a new version. +- **Out-of-order events** - A late event is placed in the history at the position its sequence value gives it. An event older than all of a key's history adds an earlier version, and an event that falls inside an existing version splits that version in two. A late delete closes the version it falls inside at the delete's sequence value. The history is the same as if the events had arrived in order. +- **Duplicate events** - A re-delivered event adds no version. + +For example, a first run receives two events for `id 1`: + +| id | name | version | +|----|--------|---------| +| 1 | alice | 10 | +| 1 | alicia | 30 | + +The target holds two versions. The tables here omit the `__spark_autocdc_metadata` column (see [Auto CDC Considerations](#auto-cdc-considerations)): + +| id | name | version | `__START_AT` | `__END_AT` | +|----|--------|---------|--------------|------------| +| 1 | alice | 10 | 10 | 30 | +| 1 | alicia | 30 | 30 | NULL | + +A later run receives two late events, one inside the history and one before it: + +| id | name | version | +|----|--------|---------| +| 1 | alyssa | 20 | +| 1 | ally | 5 | + +`alyssa` splits the `alice` version at 20, and `ally` adds a version before `alice`: + +| id | name | version | `__START_AT` | `__END_AT` | +|----|--------|---------|--------------|------------| +| 1 | ally | 5 | 5 | 10 | +| 1 | alice | 10 | 10 | 20 | +| 1 | alyssa | 20 | 20 | 30 | +| 1 | alicia | 30 | 30 | NULL | + +#### Defining a Type 2 Flow + +In Python, pass `stored_as_scd_type=2` to `create_auto_cdc_flow`: + +```python +from pyspark import pipelines as dp +from pyspark.sql import DataFrame + [email protected] +def cdc_events() -> DataFrame: + return spark.readStream.table("cdc_source") + +dp.create_streaming_table("customers") + +dp.create_auto_cdc_flow( + target="customers", + source="cdc_events", + keys=["id"], + sequence_by="version", + apply_as_deletes="op = 'DELETE'", + except_column_list=["op"], + stored_as_scd_type=2, +) +``` + +In SQL, add `STORED AS SCD TYPE 2`: + +```sql +CREATE STREAMING TABLE customers +FLOW AUTO CDC +FROM STREAM(cdc_events) +KEYS (id) +APPLY AS DELETE WHEN op = 'DELETE' +SEQUENCE BY version +COLUMNS * EXCEPT (op) +STORED AS SCD TYPE 2; +``` + +#### Choosing Which Columns Open a New Version + +By default, every column written to the target except the keys is tracked, including the sequencing column. Because every event for a key has its own sequence value, each insert or update then opens a new version. + +Consecutive inserts and updates for a key that agree on every tracked column are coalesced into one version. That version keeps the `__START_AT` of the first event and takes the other column values of the latest one. Review Comment: optional: "that agree" might be too ambiguous. Maybe just spell out that they have equal values for all tracked columns. ########## docs/declarative-pipelines-programming-guide.md: ########## @@ -807,6 +810,198 @@ spark-pipelines run The table is unchanged. This time the stale event is alone in its batch, so there is no newer event to supersede it within the batch; instead Auto CDC compares its version against the state already recorded for `id 1` and discards it, because version 1 is below the committed version 2. Applied in arrival order, this event would have wrongly reverted `id 1` to `alice`. +### SCD Type 2: Keeping History + +With SCD Type 2, the target keeps every version of a row instead of overwriting it. Auto CDC adds two columns to the target that record when each version was current: + +- `__START_AT` - the sequence value at which the version became current. +- `__END_AT` - the sequence value at which the version stopped being current, or `NULL` for the current version. + +Both columns have the type of the sequencing expression. Auto CDC adds them to the target schema; the source does not supply them. Type 2 has the same [Requirements](#requirements) as Type 1. + +The current version of each key is the row whose `__END_AT` is `NULL`: + +```sql +SELECT * FROM customers WHERE __END_AT IS NULL; +``` + +The version that was current at a given sequence value is the row whose interval contains that value. For `version` 1: + +```sql +SELECT * FROM customers +WHERE __START_AT <= 1 AND (__END_AT > 1 OR __END_AT IS NULL); +``` + +#### How Type 2 Applies Each Event + +- **Inserts and updates** - The first event for a key opens its first version. A later event that changes a tracked column closes the key's current version at the event's sequence value and opens a new version. See [Choosing Which Columns Open a New Version](#choosing-which-columns-open-a-new-version). +- **Deletes** - A delete closes the key's current version at the delete's sequence value and opens no new one, so the key has no row with a `NULL` `__END_AT`. The key's earlier versions stay in the target. A later insert for the key opens a new version. +- **Out-of-order events** - A late event is placed in the history at the position its sequence value gives it. An event older than all of a key's history adds an earlier version, and an event that falls inside an existing version splits that version in two. A late delete closes the version it falls inside at the delete's sequence value. The history is the same as if the events had arrived in order. Review Comment: ```suggestion - **Out-of-order events** - A late event is placed in the history at the position its sequence value gives it, producing the same history as if the events had arrived in order. A late insert or update that changes a tracked column adds an earlier version if it precedes all of the key's history, or closes the version it falls inside at its sequence value and opens a new one; a late delete closes the version it falls inside at the delete's sequence value. ``` ########## docs/declarative-pipelines-programming-guide.md: ########## @@ -620,13 +620,15 @@ dp.create_auto_cdc_flow( | `apply_as_deletes` | No | A boolean expression identifying events that represent deletes. Matching rows are removed from the target. A SQL expression string or a `Column`. | Review Comment: > Matching rows are removed from the target We should update this description to express the difference between SCD1 and SCD2; in SCD2, the delete will attempt to close the matching row, not delete it. ########## docs/declarative-pipelines-programming-guide.md: ########## @@ -620,13 +620,15 @@ dp.create_auto_cdc_flow( | `apply_as_deletes` | No | A boolean expression identifying events that represent deletes. Matching rows are removed from the target. A SQL expression string or a `Column`. | | `column_list` | No | The columns to include in the target. Mutually exclusive with `except_column_list`. | | `except_column_list` | No | The columns to exclude from the target; all other columns are included. Mutually exclusive with `column_list`. Commonly used to drop operation/metadata columns such as `op`. | -| `stored_as_scd_type` | No | The SCD type of the target. Pass `1` (or `"1"`) for the Type 1 behavior described here. Type 2 is also accepted; see [SPARK-58570](https://issues.apache.org/jira/browse/SPARK-58570). | +| `stored_as_scd_type` | No | The SCD type of the target: `1` or `2` (or `"1"` or `"2"`). Defaults to Type 1. See [SCD Type 2: Keeping History](#scd-type-2-keeping-history). | +| `track_history_column_list` | No | SCD Type 2 only. The columns whose changes open a new version. Keyword-only. Mutually exclusive with `track_history_except_column_list`. | +| `track_history_except_column_list` | No | SCD Type 2 only. The columns whose changes do not open a new version; changes to any other non-key column in the target do. Keyword-only. Mutually exclusive with `track_history_column_list`. | | `name` | No | The name of the flow. Defaults to the target table name. | | `spark_conf` | No | Spark configuration to apply while the flow runs. Overrides configuration set on the destination, the pipeline, or the cluster. | Review Comment: `spark_conf` is keyword-only as well, so if we're gonna state that for `track_history_column_list`/``track_history_except_column_list`, we should state it for `spark_conf` too. ########## docs/declarative-pipelines-programming-guide.md: ########## @@ -807,6 +810,198 @@ spark-pipelines run The table is unchanged. This time the stale event is alone in its batch, so there is no newer event to supersede it within the batch; instead Auto CDC compares its version against the state already recorded for `id 1` and discards it, because version 1 is below the committed version 2. Applied in arrival order, this event would have wrongly reverted `id 1` to `alice`. +### SCD Type 2: Keeping History + +With SCD Type 2, the target keeps every version of a row instead of overwriting it. Auto CDC adds two columns to the target that record when each version was current: + +- `__START_AT` - the sequence value at which the version became current. +- `__END_AT` - the sequence value at which the version stopped being current, or `NULL` for the current version. + +Both columns have the type of the sequencing expression. Auto CDC adds them to the target schema; the source does not supply them. Type 2 has the same [Requirements](#requirements) as Type 1. + +The current version of each key is the row whose `__END_AT` is `NULL`: + +```sql +SELECT * FROM customers WHERE __END_AT IS NULL; +``` + +The version that was current at a given sequence value is the row whose interval contains that value. For `version` 1: + +```sql +SELECT * FROM customers +WHERE __START_AT <= 1 AND (__END_AT > 1 OR __END_AT IS NULL); +``` + +#### How Type 2 Applies Each Event + +- **Inserts and updates** - The first event for a key opens its first version. A later event that changes a tracked column closes the key's current version at the event's sequence value and opens a new version. See [Choosing Which Columns Open a New Version](#choosing-which-columns-open-a-new-version). +- **Deletes** - A delete closes the key's current version at the delete's sequence value and opens no new one, so the key has no row with a `NULL` `__END_AT`. The key's earlier versions stay in the target. A later insert for the key opens a new version. +- **Out-of-order events** - A late event is placed in the history at the position its sequence value gives it. An event older than all of a key's history adds an earlier version, and an event that falls inside an existing version splits that version in two. A late delete closes the version it falls inside at the delete's sequence value. The history is the same as if the events had arrived in order. Review Comment: optional: Maybe this is getting into too much detail, but the earlier version added and/or split will only occur if the rows are unique, as per the configured track history selection. ########## docs/declarative-pipelines-programming-guide.md: ########## Review Comment: Stale description that only applies to SCD1, we should update for both SCD1 and SCD2. ########## docs/declarative-pipelines-programming-guide.md: ########## @@ -807,6 +810,198 @@ spark-pipelines run The table is unchanged. This time the stale event is alone in its batch, so there is no newer event to supersede it within the batch; instead Auto CDC compares its version against the state already recorded for `id 1` and discards it, because version 1 is below the committed version 2. Applied in arrival order, this event would have wrongly reverted `id 1` to `alice`. +### SCD Type 2: Keeping History + +With SCD Type 2, the target keeps every version of a row instead of overwriting it. Auto CDC adds two columns to the target that record when each version was current: + +- `__START_AT` - the sequence value at which the version became current. +- `__END_AT` - the sequence value at which the version stopped being current, or `NULL` for the current version. + +Both columns have the type of the sequencing expression. Auto CDC adds them to the target schema; the source does not supply them. Type 2 has the same [Requirements](#requirements) as Type 1. + +The current version of each key is the row whose `__END_AT` is `NULL`: + +```sql +SELECT * FROM customers WHERE __END_AT IS NULL; +``` + +The version that was current at a given sequence value is the row whose interval contains that value. For `version` 1: + +```sql +SELECT * FROM customers +WHERE __START_AT <= 1 AND (__END_AT > 1 OR __END_AT IS NULL); +``` + +#### How Type 2 Applies Each Event + +- **Inserts and updates** - The first event for a key opens its first version. A later event that changes a tracked column closes the key's current version at the event's sequence value and opens a new version. See [Choosing Which Columns Open a New Version](#choosing-which-columns-open-a-new-version). +- **Deletes** - A delete closes the key's current version at the delete's sequence value and opens no new one, so the key has no row with a `NULL` `__END_AT`. The key's earlier versions stay in the target. A later insert for the key opens a new version. +- **Out-of-order events** - A late event is placed in the history at the position its sequence value gives it. An event older than all of a key's history adds an earlier version, and an event that falls inside an existing version splits that version in two. A late delete closes the version it falls inside at the delete's sequence value. The history is the same as if the events had arrived in order. +- **Duplicate events** - A re-delivered event adds no version. + +For example, a first run receives two events for `id 1`: + +| id | name | version | +|----|--------|---------| +| 1 | alice | 10 | +| 1 | alicia | 30 | Review Comment: keep an `op` column in this example too for consistency with other examples, and to make it clear to readers that both events are considered upserts. ########## docs/declarative-pipelines-programming-guide.md: ########## @@ -807,6 +810,198 @@ spark-pipelines run The table is unchanged. This time the stale event is alone in its batch, so there is no newer event to supersede it within the batch; instead Auto CDC compares its version against the state already recorded for `id 1` and discards it, because version 1 is below the committed version 2. Applied in arrival order, this event would have wrongly reverted `id 1` to `alice`. +### SCD Type 2: Keeping History + +With SCD Type 2, the target keeps every version of a row instead of overwriting it. Auto CDC adds two columns to the target that record when each version was current: + +- `__START_AT` - the sequence value at which the version became current. +- `__END_AT` - the sequence value at which the version stopped being current, or `NULL` for the current version. + +Both columns have the type of the sequencing expression. Auto CDC adds them to the target schema; the source does not supply them. Type 2 has the same [Requirements](#requirements) as Type 1. + +The current version of each key is the row whose `__END_AT` is `NULL`: + +```sql +SELECT * FROM customers WHERE __END_AT IS NULL; +``` + +The version that was current at a given sequence value is the row whose interval contains that value. For `version` 1: + +```sql +SELECT * FROM customers +WHERE __START_AT <= 1 AND (__END_AT > 1 OR __END_AT IS NULL); +``` + +#### How Type 2 Applies Each Event + +- **Inserts and updates** - The first event for a key opens its first version. A later event that changes a tracked column closes the key's current version at the event's sequence value and opens a new version. See [Choosing Which Columns Open a New Version](#choosing-which-columns-open-a-new-version). +- **Deletes** - A delete closes the key's current version at the delete's sequence value and opens no new one, so the key has no row with a `NULL` `__END_AT`. The key's earlier versions stay in the target. A later insert for the key opens a new version. +- **Out-of-order events** - A late event is placed in the history at the position its sequence value gives it. An event older than all of a key's history adds an earlier version, and an event that falls inside an existing version splits that version in two. A late delete closes the version it falls inside at the delete's sequence value. The history is the same as if the events had arrived in order. +- **Duplicate events** - A re-delivered event adds no version. + +For example, a first run receives two events for `id 1`: + +| id | name | version | +|----|--------|---------| +| 1 | alice | 10 | +| 1 | alicia | 30 | + +The target holds two versions. The tables here omit the `__spark_autocdc_metadata` column (see [Auto CDC Considerations](#auto-cdc-considerations)): + +| id | name | version | `__START_AT` | `__END_AT` | +|----|--------|---------|--------------|------------| +| 1 | alice | 10 | 10 | 30 | +| 1 | alicia | 30 | 30 | NULL | + +A later run receives two late events, one inside the history and one before it: + +| id | name | version | +|----|--------|---------| +| 1 | alyssa | 20 | +| 1 | ally | 5 | + +`alyssa` splits the `alice` version at 20, and `ally` adds a version before `alice`: + +| id | name | version | `__START_AT` | `__END_AT` | +|----|--------|---------|--------------|------------| +| 1 | ally | 5 | 5 | 10 | +| 1 | alice | 10 | 10 | 20 | +| 1 | alyssa | 20 | 20 | 30 | +| 1 | alicia | 30 | 30 | NULL | + +#### Defining a Type 2 Flow + +In Python, pass `stored_as_scd_type=2` to `create_auto_cdc_flow`: + +```python +from pyspark import pipelines as dp +from pyspark.sql import DataFrame + [email protected] +def cdc_events() -> DataFrame: + return spark.readStream.table("cdc_source") + +dp.create_streaming_table("customers") + +dp.create_auto_cdc_flow( + target="customers", + source="cdc_events", + keys=["id"], + sequence_by="version", + apply_as_deletes="op = 'DELETE'", + except_column_list=["op"], + stored_as_scd_type=2, +) +``` + +In SQL, add `STORED AS SCD TYPE 2`: + +```sql +CREATE STREAMING TABLE customers +FLOW AUTO CDC +FROM STREAM(cdc_events) +KEYS (id) +APPLY AS DELETE WHEN op = 'DELETE' +SEQUENCE BY version +COLUMNS * EXCEPT (op) +STORED AS SCD TYPE 2; +``` + +#### Choosing Which Columns Open a New Version + +By default, every column written to the target except the keys is tracked, including the sequencing column. Because every event for a key has its own sequence value, each insert or update then opens a new version. + +Consecutive inserts and updates for a key that agree on every tracked column are coalesced into one version. That version keeps the `__START_AT` of the first event and takes the other column values of the latest one. Review Comment: > and takes the other column values of the latest one. nit: I think we should be more precise here too. Ex. "untracked columns hold the latest event values received for/across the whole version." ########## docs/declarative-pipelines-programming-guide.md: ########## @@ -807,6 +810,198 @@ spark-pipelines run The table is unchanged. This time the stale event is alone in its batch, so there is no newer event to supersede it within the batch; instead Auto CDC compares its version against the state already recorded for `id 1` and discards it, because version 1 is below the committed version 2. Applied in arrival order, this event would have wrongly reverted `id 1` to `alice`. +### SCD Type 2: Keeping History + +With SCD Type 2, the target keeps every version of a row instead of overwriting it. Auto CDC adds two columns to the target that record when each version was current: + +- `__START_AT` - the sequence value at which the version became current. +- `__END_AT` - the sequence value at which the version stopped being current, or `NULL` for the current version. + +Both columns have the type of the sequencing expression. Auto CDC adds them to the target schema; the source does not supply them. Type 2 has the same [Requirements](#requirements) as Type 1. + +The current version of each key is the row whose `__END_AT` is `NULL`: + +```sql +SELECT * FROM customers WHERE __END_AT IS NULL; +``` + +The version that was current at a given sequence value is the row whose interval contains that value. For `version` 1: + +```sql +SELECT * FROM customers +WHERE __START_AT <= 1 AND (__END_AT > 1 OR __END_AT IS NULL); +``` + +#### How Type 2 Applies Each Event + +- **Inserts and updates** - The first event for a key opens its first version. A later event that changes a tracked column closes the key's current version at the event's sequence value and opens a new version. See [Choosing Which Columns Open a New Version](#choosing-which-columns-open-a-new-version). +- **Deletes** - A delete closes the key's current version at the delete's sequence value and opens no new one, so the key has no row with a `NULL` `__END_AT`. The key's earlier versions stay in the target. A later insert for the key opens a new version. +- **Out-of-order events** - A late event is placed in the history at the position its sequence value gives it. An event older than all of a key's history adds an earlier version, and an event that falls inside an existing version splits that version in two. A late delete closes the version it falls inside at the delete's sequence value. The history is the same as if the events had arrived in order. Review Comment: optional: Also add a note that an early-arriving delete that doesn't match against any row version is tombstoned and reconciled later. ########## docs/declarative-pipelines-programming-guide.md: ########## @@ -807,6 +810,198 @@ spark-pipelines run The table is unchanged. This time the stale event is alone in its batch, so there is no newer event to supersede it within the batch; instead Auto CDC compares its version against the state already recorded for `id 1` and discards it, because version 1 is below the committed version 2. Applied in arrival order, this event would have wrongly reverted `id 1` to `alice`. +### SCD Type 2: Keeping History + +With SCD Type 2, the target keeps every version of a row instead of overwriting it. Auto CDC adds two columns to the target that record when each version was current: + +- `__START_AT` - the sequence value at which the version became current. +- `__END_AT` - the sequence value at which the version stopped being current, or `NULL` for the current version. + +Both columns have the type of the sequencing expression. Auto CDC adds them to the target schema; the source does not supply them. Type 2 has the same [Requirements](#requirements) as Type 1. + +The current version of each key is the row whose `__END_AT` is `NULL`: + +```sql +SELECT * FROM customers WHERE __END_AT IS NULL; +``` + +The version that was current at a given sequence value is the row whose interval contains that value. For `version` 1: + +```sql +SELECT * FROM customers +WHERE __START_AT <= 1 AND (__END_AT > 1 OR __END_AT IS NULL); +``` + +#### How Type 2 Applies Each Event + +- **Inserts and updates** - The first event for a key opens its first version. A later event that changes a tracked column closes the key's current version at the event's sequence value and opens a new version. See [Choosing Which Columns Open a New Version](#choosing-which-columns-open-a-new-version). +- **Deletes** - A delete closes the key's current version at the delete's sequence value and opens no new one, so the key has no row with a `NULL` `__END_AT`. The key's earlier versions stay in the target. A later insert for the key opens a new version. +- **Out-of-order events** - A late event is placed in the history at the position its sequence value gives it. An event older than all of a key's history adds an earlier version, and an event that falls inside an existing version splits that version in two. A late delete closes the version it falls inside at the delete's sequence value. The history is the same as if the events had arrived in order. +- **Duplicate events** - A re-delivered event adds no version. + +For example, a first run receives two events for `id 1`: + +| id | name | version | +|----|--------|---------| +| 1 | alice | 10 | +| 1 | alicia | 30 | + +The target holds two versions. The tables here omit the `__spark_autocdc_metadata` column (see [Auto CDC Considerations](#auto-cdc-considerations)): + +| id | name | version | `__START_AT` | `__END_AT` | +|----|--------|---------|--------------|------------| +| 1 | alice | 10 | 10 | 30 | +| 1 | alicia | 30 | 30 | NULL | + +A later run receives two late events, one inside the history and one before it: + +| id | name | version | +|----|--------|---------| +| 1 | alyssa | 20 | +| 1 | ally | 5 | + +`alyssa` splits the `alice` version at 20, and `ally` adds a version before `alice`: + +| id | name | version | `__START_AT` | `__END_AT` | +|----|--------|---------|--------------|------------| +| 1 | ally | 5 | 5 | 10 | +| 1 | alice | 10 | 10 | 20 | +| 1 | alyssa | 20 | 20 | 30 | +| 1 | alicia | 30 | 30 | NULL | + +#### Defining a Type 2 Flow + +In Python, pass `stored_as_scd_type=2` to `create_auto_cdc_flow`: + +```python +from pyspark import pipelines as dp +from pyspark.sql import DataFrame + [email protected] +def cdc_events() -> DataFrame: + return spark.readStream.table("cdc_source") + +dp.create_streaming_table("customers") + +dp.create_auto_cdc_flow( + target="customers", + source="cdc_events", + keys=["id"], + sequence_by="version", + apply_as_deletes="op = 'DELETE'", + except_column_list=["op"], + stored_as_scd_type=2, +) +``` + +In SQL, add `STORED AS SCD TYPE 2`: + +```sql +CREATE STREAMING TABLE customers +FLOW AUTO CDC +FROM STREAM(cdc_events) +KEYS (id) +APPLY AS DELETE WHEN op = 'DELETE' +SEQUENCE BY version +COLUMNS * EXCEPT (op) +STORED AS SCD TYPE 2; +``` + +#### Choosing Which Columns Open a New Version + +By default, every column written to the target except the keys is tracked, including the sequencing column. Because every event for a key has its own sequence value, each insert or update then opens a new version. + +Consecutive inserts and updates for a key that agree on every tracked column are coalesced into one version. That version keeps the `__START_AT` of the first event and takes the other column values of the latest one. + +To choose the tracked columns, name them with `track_history_column_list` (Python) or `TRACK HISTORY ON (...)` (SQL), or exclude columns with `track_history_except_column_list` or `TRACK HISTORY ON * EXCEPT (...)`. The two forms are mutually exclusive, and both require SCD Type 2. Tracked columns must be columns written to the target and must not be keys; any other column fails with `AUTOCDC_COLUMNS_NOT_FOUND_IN_SCHEMA`. + +For example, to open a new version only when `name` changes: + +```python +dp.create_auto_cdc_flow( + target="customers", + source="cdc_events", + keys=["id"], + sequence_by="version", + apply_as_deletes="op = 'DELETE'", + except_column_list=["op"], + stored_as_scd_type=2, + track_history_column_list=["name"], +) +``` + +```sql +CREATE STREAMING TABLE customers +FLOW AUTO CDC +FROM STREAM(cdc_events) +KEYS (id) +APPLY AS DELETE WHEN op = 'DELETE' +SEQUENCE BY version +COLUMNS * EXCEPT (op) +STORED AS SCD TYPE 2 +TRACK HISTORY ON (name); +``` + +Given these events: + +| id | name | status | version | op | +|----|--------|----------|---------|--------| +| 1 | alice | active | 5 | UPSERT | +| 1 | alice | inactive | 10 | UPSERT | +| 1 | alicia | inactive | 15 | UPSERT | + +the first two events agree on `name` and are coalesced; the third changes `name` and opens a new version: + +| id | name | status | version | `__START_AT` | `__END_AT` | +|----|--------|----------|---------|--------------|------------| +| 1 | alice | inactive | 10 | 5 | 15 | +| 1 | alicia | inactive | 15 | 15 | NULL | + +With the default tracking, the same events produce three versions. + +#### Changing the SCD Type or Tracked Columns + +Like the key set (see [Changing the Key Set](#changing-the-key-set)), the SCD type and the set of tracked columns are part of the flow's persisted state. Changing the SCD type across incremental runs fails with `AUTOCDC_INVALID_STATE.SCD_TYPE_DRIFT`, and changing the tracked columns fails with `AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT`. With the default tracking or `track_history_except_column_list`, the tracked columns follow the columns written to the target, so adding or removing a column there also changes them. To make either change, fully refresh the target table: Review Comment: optional: might be worth mentioning there's an escape here without fully refreshing the table, if the target table was additively schema evolved (by far the more common case); the `AUTOCDC_INVALID_STATE.TRACK_HISTORY_DRIFT` error will give the exact expected track history column configuration, users should be able to directly paste this into their AutoCDC API callsite and switch from include-all/exclude to an explicit include list. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
