CDC Multiple sequence

[junhonglau]

Currently the sequence by only able to accept just one column and we need to cater for multiple columns as streaming world might have more than one key to sort - event_timestamp, enqueue timestamp, and sequence id

[ravi-databricks]

@junhonglau Lakeflow Declarative Pipelines have create_auto_cdc_flow API which is implemented in DLT-META.
In case of multiple seq cols need to add this

[junhonglau]

Yes understand that LDP are now able to support multiple seq cols but dont think DLT META is able to support as the required input in a struct type. The CDCApplyChanges parse the sequence_by as a String so it wont be able to support.

Correct me if I'm wrong. If we already able to support now, please let me know how we can write the data flow spec.

[ravi-databricks]

@junhonglau yes we need to add struct for multiple cols in dlt-meta.

[ravi-databricks]

we might need following change in dataflow_pipeline.py allowing sequence by col to accept comma separated list of cols

        sequence_by = cdc_apply_changes.sequence_by
        if ',' in sequence_by:
            sequence_cols = [col.strip() for col in sequence_by.split(',')]
            sequence_by = struct(*sequence_cols)  # Use struct() from pyspark.sql.functions

        dlt.create_auto_cdc_flow(
            target=target_table,
            source=self.view_name,
            keys=cdc_apply_changes.keys,
            sequence_by=sequence_by,
            where=cdc_apply_changes.where,
            ignore_null_updates=cdc_apply_changes.ignore_null_updates,
            apply_as_deletes=apply_as_deletes,
            apply_as_truncates=apply_as_truncates,
            column_list=cdc_apply_changes.column_list,
            except_column_list=cdc_apply_changes.except_column_list,
            stored_as_scd_type=cdc_apply_changes.scd_type,
            track_history_column_list=cdc_apply_changes.track_history_column_list,
            track_history_except_column_list=cdc_apply_changes.track_history_except_column_list,
            flow_name=cdc_apply_changes.flow_name,
            once=cdc_apply_changes.once,
            ignore_null_updates_column_list=cdc_apply_changes.ignore_null_updates_column_list,
            ignore_null_updates_except_column_list=cdc_apply_changes.ignore_null_updates_except_column_list
        )

[ravi-databricks]

@junhonglau please check issue_223 branch and see if it works

[junhonglau]

Hi Ravi, yes it able to work. since it will be part of the version 0.0.10 when are we officially releasing this?