[Feature Request] Support for change data capture

[NOD507]

A change data feed option like the one on delta lake is very usefull for incremental syncs between tables or systems.

CDC is one feature thats not on open source delta lake or iceberg. I dont think theres a format that supports CDC on spark open source at this moment.

Hidden partitioning is currently a plus for iceberg vs delta lake, and CDC could be another one.

https://docs.databricks.com/delta/delta-change-data-feed.html

[flyrain]

I'm working on a Iceberg CDC design doc. It will be like the CDF(change data feed) in the proprietary Delta. Will keep you posted. It'd be helpful if you can share your CDC use case.

[SreeramGarlapati]

@flyrain - we started this design doc, which scopes the problem down to the tables that are written with MoR, a while back - after a community sync up. pl. see if you wanna adapt it or even change the same doc.

[flyrain]

Thanks for sharing, @SreeramGarlapati! That is a good start and I'm glad you guys have started working on it. Can we have a discuss when my design doc is ready?

[mohitgargk]

+1

@flyrain We are looking into two possible approaches here.

1. Materialised where, like DeltaLake Change Data Feed that is generated in the write path.
2. Change Data Feed is evaluated at the read path by consumers.

At this point of time, we are grokking and spiking on approach #2. Would love to know your early thoughts on this.

[flyrain]

Hi @mohitgargk, we had multiple discussions internally and are working on a doc. Will share it later. Generally speaking, CDC on MOR is in a better position, we should target it first. CDC on COW is almost impossible without additional metadata column or extra logging. cc @aokolnychyi @RussellSpitzer

[mohitgargk]

@flyrain Agree on that. With CoW, without any row-identifier, it is not possible to establish link with the new state and old state of the records without ambiguities.

For the CDF record structure, we grokked a bit on the design and came up with this.

Above proposal is based on our use case where we are interested in change of a certain column but also want the values of other columns. If the whole state is part of CDF record, the consumer can avoid query/lookup/join with the table to get rest of the state.

cc - @SreeramGarlapati @rawataaryan9 @dharamshoo

[NOD507]

I'm working on a Iceberg CDC design doc. It will be like the CDF(change data feed) in the proprietary Delta. Will keep you posted. It'd be helpful if you can share your CDC use case.

We are interested in CDC to be able to incrementally sync changes from a datalake to a datawarehouse. To be able to do ETL in the datalake and use the dwh for end user consumption.

[NOD507]

@flyrain Agree on that. With CoW, without any row-identifier, it is not possible to establish link with the new state and old state of the records without ambiguities.

For the CDF record structure, we grokked a bit on the design and came up with this.

Above proposal is based on our use case where we are interested in change of a certain column but also want the values of other columns. If the whole state is part of CDF record, the consumer can avoid query/lookup/join with the table to get rest of the state.

cc - @SreeramGarlapati @rawataaryan9 @dharamshoo

This design would work great for our use case, is exactly what we need. Thanks for sharing

[rajarshisarkar]

Hi @flyrain Please let us know once the design doc is ready.

[flyrain]

We've discussed the solution we are working on in today's community sync-up. The solution can support both COW and MOR. We think it is the good first step to approach CDC. Will post the design doc soon.

[flyrain]

The design doc is here, https://docs.google.com/document/d/1bN6rdLNcYOHnT3xVBfB33BoiPO06aKBo56SZmuU9pnY/edit?usp=sharing.

[Reo-LEI]

@flyrain We are looking into two possible approaches here.

1. Materialised where, like DeltaLake Change Data Feed that is generated in the write path.
2. Change Data Feed is evaluated at the read path by consumers.

I drafted a design document for case 1 base on write-time logging.

But as @flyrain mentioned in email, the approach based on write-time logging would require changing the table spec and changing all engines to fit the new table spec. Therefore, I would tend to take the approach of @flyrain and try to integrate it into flink.

Just paste my approach here so we can find it later when we need it.

[donatobarone]

+1 we have started to investigate iceberg internally and we realised that the CDC available today was just for append operations, which is definitely not ideal as processes will have to be built around the point in time functionalities to be able to get the changes.
Great that the community is looking into this

[aokolnychyi]

Based on Yufei’s design doc and what we discussed during the sync, I’d like to share my thoughts on what can be efficiently supported right now.

Assumptions

• Changes are consumed per snapshot, rows that are added and removed in the same snapshot can be skipped

Iceberg does not distinguish the order of records within a snapshot. Users may manually deduce this using a data column. However, there is nothing in the metadata to support this at the moment. I believe it is sufficient to have a summary of what changed per snapshot for most use cases (except the audit use case). Specifically, if we add a record and immediately delete it in the same snapshot (Flink CDC), it is okay to skip this record and not report it in the CDC log.

• Output only delete and insert record types

From what I see, just knowing what records to remove and to add should be enough to support most use cases. I have an example below even if the target system is not Iceberg. In the future, we can add pre/post images but computing that with the current table metadata would require a join and would be expensive.

• Table must have identity columns defined

Whenever we apply a delta log to a target table, most CDC use cases rely on a primary key or a set of identity columns. I think it is reasonable to assume Iceberg tables should have identity columns defined to support generation of CDC records.

• Delete records may only include values for identity columns

It is not required to output the entire deleted record if other columns are not stored in equality delete files. If values for other columns are not persisted, this would require an expensive join to reconstruct the entire row.

Unsupported

• Audit use cases as they require a seq number per row
• Pre/post update image (can be added in the future)

The CDC record can include:

_record_type, _commit_snapshot_id, _commit_timestamp, _commit_order, _identity_columns, col1, col2, ...

Algorithm (per snapshot)

• Build insert records
  • Read all data files written in this snapshot, applying any matching position delete files produced in the same snapshot. Equality deletes produced in this snapshot cannot apply.
• Build delete records
  • Read all data files marked as deleted in this snapshot (they always contain values for identity columns).
  • Read all equality deletes (they always contain values for identity columns)
  • Read position deletes, find the list of affected data files from previous snapshots, read those data files, join on _file, _pos with deletes, project identity columns.
• Append metadata such as _commit_snapshot_id or _record_type
• Union insert and delete records

The algorithm would only require one join to find identity columns for position deletes. However, it should be fairly efficient as position deletes are scoped to partitions and we know the exact file names to read. We don’t have to read any other files or do a full table scan.

I think it should be sufficient to have only delete/insert record types in most cases. A generic MERGE statement can be used to apply changes for a single snapshot to both Iceberg and non-Iceberg tables.

MERGE INTO target t USING source s
ON t.id_col = s.id_col AND s._record_type = 'DELETE'
WHEN MATCHED
  DELETE
WHEN NOT MATCHED
  INSERT …

If the destination table is another Iceberg table, we may skip the MERGE statement and write equality deletes and data files directly without querying the destination table.

Here are a few examples (inspired by what Yufei has put together) I went through to see if this approach is going to work (double check it, please).

[dharamsahoo]

Hi @aokolnychyi , thanks for the well illustrated writeup. I have a few comments related to the assumptions.

Output only delete and insert record types

Some consumers might want to directly consume the CDC records and generate events/notifications. It would be useful for such consumers to have CDC records relate to the actual operation on Iceberg table record. So having an Update record type would help to segregate it from Insert and Delete records in more convenient way.

Delete records may only include values for identity columns

For consumers wanting to generate notifications which include old & new values like sending email for customer preference update OR generate events when an attribute value crosses a threshold, the inclusion of non-identity column(s) in the Delete record (related to an Update operation on Iceberg table) will be very useful.

Also, I noticed that in example 2 above, CDC records are generated for unchanged records (id=106). For Copy-On-Write tables, would this be the behaviour of CDC? If so, for snapshots where a small fraction of records are added/modified, there would be too many CDC records.