Flink CDC streaming write support

[xx789633]

I'm trying to integrate Flink CDC with Lance, but I found the capabilities of Lance java api are quite limited.

It seems the fragment API https://github.com/lancedb/lance/blob/main/java/core/src/main/java/com/lancedb/lance/Fragment.java#L147 can only be used for appending data.

What I'm looking for is a delta writer to write both insert records and deletes, something like this: https://github.com/apache/iceberg/blob/main/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/BaseDeltaTaskWriter.java#L79.

@jackye1995 @majin1102 @westonpace Could you possibly give some advice? Thank you.

[jackye1995]

@cwang9208 what you linked is a bit different, because it is BaseEqualityDeltaWriter, which will write equality deletes. And this feature easily hits performance issues and people rarely use it for large scale production workloads. My understanding is that even Alibaba, who originally contributed this feature, have moved away to Paimon and abandoned this approach at this point, but please correct me if I am wrong here.

With what we have now in Lance rust, take a look at our merge-insert semantics: https://github.com/lancedb/lance/blob/main/rust/lance/src/dataset/write/merge_insert.rs#L248-L282. If this semantics is sufficient, let's expose that in Java, and then you can accumulate some batch and then write it with that API.

[majin1102]

When it comes to CDC, I think the first issue we need to consider is primary key.

If the table format doesn't have a primary key model. FlinkCDC writer would not know how to locate rows with the same primary key into one writer and reader(this is also called bucket index on primary key), then the runtime performance would be a mess. I think that's the problem as @jackye1995 mentioned: Iceberg is just an append-only model with partial primary key support in Flink(resolve the writer issue, leave the reader issue). There are also other gaps to talk. Let's just skip.

Primary key model could bring a better experiance out of box(like paimon and hudi). But I think this is really a complicated and big desicion to make when it comes to primary key issue, becase the engine runtime is totally different and hard to align across engines. For now, Iceberg and Delta still don't support primary key model as well as Lance.

The ususal solution of CDC on append-only format model like Iceberg and Delta is using merge into SQL(no need to intruduce primary key), which resolve the primary key match by batch shuffle totally in writing phase. Better efficient and perfermance with a little trade off on arrival latency. This is the same semantics with the merge-insert as metioned.

https://github.com/lancedb/lance/blob/main/rust/lance/src/dataset/write/merge_insert.rs

From my understanding, this implementation is a standalone execution right? @jackye1995 We may need to check the gap when using Flink or Spark to perform merge into operation. I think standalone execution by API may be a rare scenario in Java ecosystem

Back to CDC solution, I think a practical way is to split write view and read view:

1. FlinkCDC writes to an append-only table supporting incremental query(Paimon/Hudi/Iceberg, Lance maybe later?). This is a write view, would provide a good performace of writing with low arrival latency.
2. A merge into job scheduled periodically. this is the read view with better read performance and a bit high arrival latency

This approach has quite a few practice in CDC scenarios. We could also wrap the procedures above into one management system/platform like apache amoro, only for reference.

[jackye1995]

When it comes to CDC, I think the first issue we need to consider is primary key.

For primary key, I am working on that based on #3842, was trying to complete the full prototype before proceeding with further discussion, but maybe I can first publish the primary key format change.

Back to CDC solution, I think a practical way is to split write view and read view:

I don't think that is needed? Assuming we expose the merge-insert API in Java, you can have the writer side accumulating a batch and then use the merge-insert API to commit the batch at the checkpointing interval. That is also what the referenced Iceberg writer is doing.

In this architecture, the data is not available for reader before the checkpoint. But that is what the LSM upsert proposal is trying to achieve, I am expecting that prototype to be done this month, and we could reconcile the architecture at that time.

[majin1102]

In this architecture, the data is not available for reader before the checkpoint. But that is what the LSM upsert proposal is trying to achieve, I am expecting that prototype to be done this month, and we could reconcile the architecture at that time.

That's really quick and can't wait to see. 👍

I don't think that is needed? Assuming we expose the merge-insert API in Java, you can have the writer side accumulating a batch and then use the merge-insert API to commit the batch at the checkpointing interval. That is also what the referenced Iceberg writer is doing.

I don't know if I understand correctly. Do you mean flink writers could write merged data distrubuted and use merge-insert API to commit the writed fragments in the committer operator? If I did't get it wrong I think how to streamingly write merged data on an append-only format is not solved yet. For batch computing it could make a read-and-write plan while streaming writers is tricky to act like that?

Or do you mean Lance could solve this issue by providing merge-insert API and the new primary key model? That makes sense to me. I think Iceberg Flink writers could do that because equility-deleting protocal and the field-id act as the primary key model for Flink snenarios. The streaming write issue would be solved on your new primary key model. Otherwise I think spitting the streaming write and batch operation could be an optional solution for append-only formats(besides, the streaming data could server as some type of raw data).

This is really my big data views and thanks for correcting me if I am wrong

[majin1102]

but maybe I can first publish the primary key format change.

+1

[jackye1995]

Do you mean flink writers could write merged data distrubuted and use merge-insert API to commit the writed fragments in the committer operator?

no, I was more thinking about a more naive model that you directly have distributed (write + commit) in each partitioned writer. And this is possible in Lance because it can resolve conflicts at row level and auto-retry, this is more efficient than Iceberg. Based on our testing you can concurrently commit 128 merge-inserts without problem.

[xx789633]

Do you mean flink writers could write merged data distrubuted and use merge-insert API to commit the writed fragments in the committer operator?

Hopefully, I think a conflict-free approach using the fragment-level API should achieve better performance, but at the cost of development.

no, I was more thinking about a more naive model that you directly have distributed (write + commit) in each partitioned writer. And this is possible in Lance because it can resolve conflicts at row level and auto-retry, this is more efficient than Iceberg. Based on our testing you can concurrently commit 128 merge-inserts without problem.

If we adopt this approach, is there any advice on how to minimize the conflicts between writers? Like redistributing the data to the writers based on keys in Flink's SupportsPreWriteTopology. @jackye1995

[xx789633]

Besides, the merge-insert API is not enough for handling delete operations from Flink CDC.

[jackye1995]

If we adopt this approach, is there any advice on how to minimize the conflicts between writers? Like redistributing the data to the writers based on keys in Flink's SupportsPreWriteTopology

I think that makes sense. Today Lance does not have a partitioning concept, so the engine can just partition input stream on its own, hash distribution probably makes the most sense in that situation?

[jackye1995]

Besides, the merge-insert API is not enough for handling delete operations from Flink CDC.

yes. After #3842 I mentioned above the API will be updated to support deletes by having an additional delete marker column to indicate deletion. So if we can proceed in parallel that would be amazing.

[xx789633]

Great! Can't wait to to see the primary key dataset and the updated API. Let's work in parallel!

[xx789633]

FYI, I've completed the integration of Lance with Fluss - an Apache Incubator project that serves as a streaming storage solution purpose-built for real-time analytics. apache/fluss#1155