Fragment-Level Merge/Update columns interface for Java

[steFaiz]

Distributed column merging is a crucial feature that enables flexible schema evolution for users. In big data engine scenarios, we need to update columns in a distributed manner, where specific concurrent tasks operate only on specific fragments.

This is a complex functionality. At the computation engine level, the most fundamental requirement is the ability to perform merge and update operations on columns at the fragment granularity.

The purpose of this issue is to add column-level merge and update APIs to the Java module. This will serve as the cornerstone for the next implementation of distributed column merging/updating in Spark and Flink.

This issue mainly contains two interfaces:

1. MergeColumn interface: The jave side fragment#merge_columns method. This method will merge new columns into target fragment. A new fragment with the same FragId as well as the merged Schema will be returned.
2. UpdateColumn interface:
   In a distributed scenario, modifying the schema for each fragment individually is a risky operation. For this reason, we introduce an update column interface, which is semantically similar to Paimon's PartialUpdate mechanism. It builds upon the existing merge_column logic with the following key characteristics:
   a. The new batch updates an existing column rather than adding a new one.
   b. A join is still performed based on the values of a specified key column. For each matched row, the new value is determined by comparing the incoming value (value_new) with the current value (value_old):
   1. If value_new is present and not null, it is used as the updated value.
   2. If value_new is not present or null, but value_old is not null, the original value_old is kept.
   3. Otherwise (if both are null), the new value is null.

The UpdateColumn interface is based on the observation that in big data scenarios, users rarely add a complete set of new columns to a table all at once. Instead, they tend to add columns incrementally over the course of multiple jobs, as illustrated in the diagram below:

Without UpdateColumn interface, the result might be:

While with UpdateColumn, we can get the ideal result:

[Xuanwo]

Hi, thank you for working on this. However, I didn't expect any complex functionality to be implemented inside the Java binding. Ideally, all logic should reside in our Rust core, and the Java binding should simply expose it.

Is this issue intended to implement something new, or is it primarily serving as a tracking item for Java-side work?

[steFaiz]

Hi, thank you for working on this. However, I didn't expect any complex functionality to be implemented inside the Java binding. Ideally, all logic should reside in our Rust core, and the Java binding should simply expose it.

Is this issue intended to implement something new, or is it primarily serving as a tracking item for Java-side work?

@Xuanwo
My apologies if my initial description was not clear enough. Let me clarify the scope.
To be clear, we will not be adding any complex logic on the Java side. The Java layer will act solely as a thin interface (JNI wrapper) that directly calls the underlying Rust implementation. The title might be slightly misleading because our primary use case is integrating with big data engines, which are predominantly implemented in Java.

Actually, the work can be broken down into the following parts:

1. Implement the fragment update_column interface on the Rust side.
2. Implement the fragment merge_column interface on the Rust side (which is already natively supported).
3. Expose the fragment update_column interface on the Java side (via JNI).
4. Expose the fragment merge_column interface on the Java side (via JNI).

I've submitted the first PR 4649, which just exposes the rust-layer fragment-level merge_column method to java. The rust-layer update_column is also implemented, and the related PR will be created soon.

[majin1102]

Thanks for raising this issue.

This is a complex functionality. At the computation engine level, the most fundamental requirement is the ability to perform merge and update operations on columns at the fragment granularity.

The purpose of this issue is to add column-level merge and update APIs to the Java module. This will serve as the cornerstone for the next implementation of distributed column merging/updating in Spark and Flink.

If I understand right, the distributed merging columns scenario should use transaction and the merge operation interface(If I understand wrong, please correct me). I looked into #4649 and I think the api is logically right. But I don't figure out a case that we should partially merge one fragment as one transaction(I think the main usage of Dataset and Fragment operation API is for standalone and python cases). Can we talk more about the API you plan to use for partial fragment merge or update scenarios to make it solid? cc @jackye1995

For update_column, I did't get the the limitation of current solution. Could you elaberate more on this. Eventually I think we might need to consider or re-consider to extend the transaction and operation framework for your case.

[steFaiz]

Thank you for your attention and feedback!

1. Regarding your first question, the primary scenario we aim to cover is in the distributed computing domain. Whether using Ray (mainly Python) or Flink/Spark (Java/Scala), I believe the workflow is consistent: upstream nodes write individual fragments, and a downstream node performs a unified commit to finalize the transaction, as illustrated in the diagram below:

2. To elaborate on the second point, our proposal is driven by practical observations of how users work with very large-scale datasets (tens or even hundreds of billions of rows, a scale I believe Lance also aims to support). Users often don't add a complete new column in a single operation. Instead, they add columns incrementally through multiple Flink/Spark jobs, with each job covering only a subset of the rows.
   This behavior arises from several real-world constraints:

   • Decoupled Systems: Training and storage systems are typically separate. Data selected from the Lance lake format often passes through multiple systems before it's used for training, and we cannot force users to perform a full column update in a single step.
   • Conditional Feature Engineering: Users might filter a subset of data based on specific conditions and then generate new columns (i.e., features) only for that subset.

   All these scenarios result in partial column updates. To avoid losing previous updates from other jobs, we need to introduce this kind of partial-update mechanism.

We have already implemented distributed column addition and updates on top of Spark and Flink. We will describe these implementation details more thoroughly in subsequent issues and PRs.

[majin1102]

We have already implemented distributed column addition and updates on top of Spark and Flink. We will describe these implementation details more thoroughly in subsequent issues and PRs.

That sounds good.

But you didn't tell why don't use merge operation to do the job you metioned, especially for merge_column scenario. I believe this operation is born for this distributed scenario and it's already readfy for rust, python and java.

For the partial column update, I think what you want to do is manipulating part of columns as well as rows. Am I right? then I think the current merge operation should also support this situation(maybe needs a little test). merge operation is mainly used for columns operation, but anyway we need to write the total number of fragments into manifest in committing phase. Only difference is in or out of the operation. So I think we need to give a look to the transaction and operation framework and reconsider your cases.

Back to the first issue, anyway I don't think the fragment operation should be used in your distributed scenario because each operation is a transaction underlying. As you said, the topology the engine used is to have one single committer. We should use one single transaction to handle one commit instead of one transaction per fragement. Of course the api was right because rust side provided it.

[steFaiz]

We have already implemented distributed column addition and updates on top of Spark and Flink. We will describe these implementation details more thoroughly in subsequent issues and PRs.

That sounds good.

But you didn't tell why don't use merge operation to do the job you metioned, especially for merge_column scenario. I believe this operation is born for this distributed scenario and it's already readfy for rust, python and java.

For the partial column update, I think what you want to do is manipulating part of columns as well as rows. Am I right? then I think the current merge operation should also support this situation(maybe needs a little test). merge operation is mainly used for columns operation, but anyway we need to write the total number of fragments into manifest in committing phase. Only difference is in or out of the operation. So I think we need to give a look to the transaction and operation framework and reconsider your cases.

Back to the first issue, anyway I don't think the fragment operation should be used in your distributed scenario because each operation is a transaction underlying. As you said, the topology the engine used is to have one single committer. We should use one single transaction to handle one commit instead of one transaction per fragement. Of course the api was right because rust side provided it.

@majin1102 Thank you for the detailed explanation. It seems I may not have been clear enough about my implementation, or I might have misunderstood the existing Merge Operation you're referring to. You mentioned the merge operation, does this mean the merge method of lance::Dataset? We are pleasant to use any existing code, but it seems that the dataset-level merge operation can't be executed in a distributed manner. We hope that the merge process can be executed parallelly among different workers to maximize throughput.

It's indeed a widely adopted production practice for different workers to process exclusive data subsets, which aligns with the concept of fragments in Lance. In my implementation, after all fragments are processed by their respective workers, I do use a single Merge Operation as the content of one transaction to commit all the changes together.

Perhaps I'm misunderstanding the specific Merge Operation you mentioned. Could you please elaborate on which one it is? For instance, providing a code entry point would be very helpful. If Lance already has a merge operation designed for distributed execution as you described, I would be more than happy to reuse it.

Back to the first issue, anyway I don't think the fragment operation should be used in your distributed scenario because each operation is a transaction underlying. As you said, the topology the engine used is to have one single committer. We should use one single transaction to handle one commit instead of one transaction per fragement. Of course the api was right because rust side provided it.

Actually, the fragment-level Merge or Update interface is not a transaction. It will only create new data files and return the written filemetas. The dataset will not be modified until someone gathers all fragment metas and commit them, which is the committer operator in Flink/Spark.

[westonpace]

Ok. I think there is a some unfortunate naming confusion in the current code base which is adding to confusion on this issue. I think there are three operations that are being discussed by everyone.

A. Given an existing fragment, and new data, do a hash-join to align the new data with the existing fragment, and add new columns
B. Given an existing fragment, and new data, add new columns to the existing fragment without a join, by assuming the new data is already aligned
C. Given an existing fragment, and new data, do a hash-join to align the new data with the existing fragment, and update values in existing columns where the new value is not null

A exists already and is called merge in python and merge_columns in rust. The 4649 PR added support for this to Java.
B exists already and is called add_columns in rust but also called merge_columns in python in one spot. This feature is not being requested or modified in this issue.
C does not exist already, and this issue proposes we add it to rust and then java

I agree that C can be useful. C feels very similar to the dataset level merge_insert operation. It seems we do not have merge_insert at the fragment level today. I think the main difference between merge_insert and what you are proposing is that:

• In merge_insert, if you do not want to update a row, you don't specify a row with that key
• In this proposal, if you do not want to update a row, you specify a row with that key but null for all value columns

Would a fragment-level merge_insert work here? That would maybe help keep symmetry with the dataset level API?

[majin1102]

Actually, the fragment-level Merge or Update interface is not a transaction. It will only create new data files and return the written filemetas. The dataset will not be modified until someone gathers all fragment metas and commit them, which is the committer operator in Flink/Spark.

Sorry, I took it for granted. You are right about the fragment.merge_columns. Then I think we've been talking about two different things.

The merge operation. is https://github.com/lancedb/lance/blob/main/java/core/src/main/java/com/lancedb/lance/operation/Merge.java. You could find other operation type and the usage of them in the tests. They should be used in commit phases. For you cases, you could use the added API to generate new fragements and then use merge operation to commit them.

Would a fragment-level merge_insert work here? That would maybe help keep symmetry with the dataset level API?

Agree with @westonpace , I think a merge_insert operation on fragments would help you generate the updated fragments and you could still use merge operation to commit them.

[steFaiz]

Thank you so much for the excellent summary!

I agree that C can be useful. C feels very similar to the dataset level merge_insert operation. It seems we do not have merge_insert at the fragment level today. I think the main difference between merge_insert and what you are proposing is that:

• In merge_insert, if you do not want to update a row, you don't specify a row with that key
• In this proposal, if you do not want to update a row, you specify a row with that key but null for all value columns

Would a fragment-level merge_insert work here? That would maybe help keep symmetry with the dataset level API?

You're right, my proposal is essentially a fragment-level merge insert. I'm currently finalizing the Rust-side implementation and will submit a new PR for it shortly.

[jackye1995]

Do we have any work remaining for this? I think we can close this one now?