MergeInsert produces duplicated rows

[jackye1995]

When concurrent processes both write new rows of the same primary key through merge-insert, currently we do not have a way to detect it as conflict because it does not touch any existing fragments.

I think this is "technically" the right behavior under snapshot isolation, given we do not guarantee exactly one row per on column. But from the perspective of the people using merge insert as a way to ensure one row per primary key, this basically breaks the assumption.

I am able to verify with this test:

    #[tokio::test]
    async fn test_concurrent_merge_insert_duplicate_new_rows() {
        // This test verifies that when two writers both do merge insert to add a new row
        // with the same key, the result dataset will have duplicated rows because
        // new rows don't have addresses yet so conflict detection doesn't catch this.
        
        let test_dir = tempdir().unwrap();
        let test_uri = test_dir.path().to_str().unwrap();
        
        // Create initial dataset with some existing data
        let schema = Arc::new(Schema::new(vec![
            Field::new("key", DataType::Int32, false),
            Field::new("value", DataType::Utf8, false),
            Field::new("writer", DataType::Utf8, false),
        ]));
        
        let initial_batch = RecordBatch::try_new(
            schema.clone(),
            vec![
                Arc::new(Int32Array::from(vec![1, 2, 3])),
                Arc::new(StringArray::from(vec!["value1", "value2", "value3"])),
                Arc::new(StringArray::from(vec!["initial", "initial", "initial"])),
            ],
        )
        .unwrap();
        
        let reader = Box::new(RecordBatchIterator::new(
            vec![Ok(initial_batch)],
            schema.clone(),
        ));
        
        let dataset = Dataset::write(reader, test_uri, Default::default())
            .await
            .unwrap();
        let initial_version = dataset.version().version;
        let dataset = Arc::new(dataset);
        
        // First writer: operates on current dataset version
        let writer1_batch = RecordBatch::try_new(
            schema.clone(),
            vec![
                Arc::new(Int32Array::from(vec![4])),
                Arc::new(StringArray::from(vec!["value_from_writer1"])),
                Arc::new(StringArray::from(vec!["writer1"])),
            ],
        )
        .unwrap();
        
        let job1 = MergeInsertBuilder::try_new(dataset.clone(), vec!["key".to_string()])
            .unwrap()
            .when_matched(WhenMatched::UpdateAll)
            .when_not_matched(WhenNotMatched::InsertAll)
            .try_build()
            .unwrap();
        
        let reader1 = Box::new(RecordBatchIterator::new(
            vec![Ok(writer1_batch)],
            schema.clone(),
        ));
        let stream1 = reader_to_stream(reader1);
        
        // Execute first merge insert
        let (dataset_after_writer1, stats1) = job1.execute(stream1).await.unwrap();
        assert_eq!(stats1.num_inserted_rows, 1, "Writer 1 should insert 1 row");
        
        // Second writer: simulates a concurrent writer by checking out the initial version
        // This mimics the scenario where writer2 started before writer1 committed
        let dataset_for_writer2 = dataset.checkout_version(initial_version).await.unwrap();
        
        let writer2_batch = RecordBatch::try_new(
            schema.clone(),
            vec![
                Arc::new(Int32Array::from(vec![4])),
                Arc::new(StringArray::from(vec!["value_from_writer2"])),
                Arc::new(StringArray::from(vec!["writer2"])),
            ],
        )
        .unwrap();
        
        let job2 = MergeInsertBuilder::try_new(Arc::new(dataset_for_writer2), vec!["key".to_string()])
            .unwrap()
            .when_matched(WhenMatched::UpdateAll)
            .when_not_matched(WhenNotMatched::InsertAll)
            .try_build()
            .unwrap();
        
        let reader2 = Box::new(RecordBatchIterator::new(
            vec![Ok(writer2_batch)],
            schema.clone(),
        ));
        let stream2 = reader_to_stream(reader2);
        
        // Execute second merge insert - this should succeed despite writer1 already adding key=4
        // because new rows don't have addresses for conflict detection
        let result2 = job2.execute(stream2).await;
        
        match result2 {
            Ok((dataset_after_writer2, stats2)) => {
                assert_eq!(stats2.num_inserted_rows, 1, "Writer 2 should also insert 1 row");
                
                // Use the most recent dataset version
                let final_dataset = if dataset_after_writer1.version().version > dataset_after_writer2.version().version {
                    dataset_after_writer1
                } else {
                    dataset_after_writer2
                };
                
                // Check if we have duplicate rows with key=4
                let result = final_dataset
                    .scan()
                    .filter("key = 4")
                    .unwrap()
                    .try_into_batch()
                    .await
                    .unwrap();
                
                let num_rows = result.num_rows();
                
                // This assertion demonstrates the bug: we expect 2 duplicate rows
                // because both writers succeeded in inserting the same key
                assert_eq!(
                    num_rows, 2,
                    "Expected 2 duplicate rows with key=4 due to concurrent insert of new rows \
                     (conflict detection doesn't work for new rows without addresses)"
                );
                
                // Verify both values are present
                let writer_col = result
                    .column_by_name("writer")
                    .unwrap()
                    .as_any()
                    .downcast_ref::<StringArray>()
                    .unwrap();
                
                let writers: Vec<&str> = (0..num_rows)
                    .map(|i| writer_col.value(i))
                    .collect();
                
                assert!(
                    writers.contains(&"writer1") && writers.contains(&"writer2"),
                    "Both writers should have successfully inserted their rows"
                );
            }
            Err(e) => {
                // If writer2 fails due to conflict detection, that would be the desired behavior
                // but currently this doesn't happen for new rows
                panic!(
                    "Writer 2 failed with error: {:?} - this shouldn't happen since new rows \
                     don't have addresses for conflict detection",
                    e
                );
            }
        }
    }

[jackye1995]

cc @wjones127 @yanghua

[yanghua]

I think this is "technically" the right behavior under snapshot isolation, given we do not guarantee exactly one row per on column.

I think this depends on whether the "column" is defined to be a primary key. If it is, then we need to make sure it's globally unique. If not, it may be reasonable.

[jackye1995]

the "column" is defined to be a primary key

Even when it is defined as a primary key, we clearly say it is unenforced, which means it is okay to have duplicates. Although I understand this is not an ideal situation...

[wjones127]

Yeah this is a known issue:

• Unique constraints on columns #4087
• bug: concurrent upsert can generate duplicates lancedb/lancedb#2463

I think the right approach will be to:

1. Allow user to define a unique constraint on a column
2. When attempting to commit the operation, pass in an array of new values so they can be checked for duplicates, even during the retry loop.
   • We'll probably want to integrate this with cache / some index for better performance.

[yanghua]

I think the right approach will be to:

1. Allow user to define a unique constraint on a column

2. When attempting to commit the operation, pass in an array of new values so they can be checked for duplicates, even during the retry loop.

   • We'll probably want to integrate this with cache / some index for better performance.

+1 for introducing the unique constraint on a column(primary key).

I discussed with @jackye1995 offline. We have a scenario that strongly depends on this.

[jackye1995]

introducing the unique constraint on a column(primary key).

@wjones127 how are you thinking about enforcing it? I guess one way is that we can collect and store the list of new keys as a part of the transaction so that a concurrent merge-insert can know if there is any conflict for the new key. But that could inflate the transaction file size a lot and also slows down the conflict check. Is there any better way in your mind?

[jackye1995]

One way I can think of to reduce the transaction size is to store a compact representation of the keys, e.g. a list of hashes of the primary key instead of the primary key itself.

We can also instead track a bloom filter of the keys, so when we do conflict resolution we might fail a concurrent transaction with false positive, but the storage remains constant.

[yanghua]

We can also instead track a bloom filter of the keys, so when we do conflict resolution we might fail a concurrent transaction with false positive, but the storage remains constant.

What about bloom filter + some second-level exact index structure(e.g. Trie)? We can use its compression-optimized version (for example, Radix Tree).

We have a case, the primary key is generated by several common prefixes like a-b-c-d-f, and the prefixes are highly duplicated.

[wjones127]

how are you thinking about enforcing it? I guess one way is that we can collect and store the list of new keys as a part of the transaction so that a concurrent merge-insert can know if there is any conflict for the new key.

Yeah that's basically what I was thinking. Like you both suggest, there are ways we can optimize it further.

[jackye1995]

@yanghua would you be interested in helping this?

[yanghua]

@yanghua would you be interested in helping this?

Yes, I am thinking about the design. Will try to figure out and share more.

[yanghua]

Hi @jackye1995 let's align some information.

Actually, I want to extend the design scope beyond this issue. I want to figure out a way to support "Global uniqueness of primary keys". Both for Append and MergeInsert operations. That means, if there are new records that want to be inserted into the Lance table, we must check that the primary key is globally unique.

So we need to fix two problems in the Lance format:

• How to judge if the primary key is globally unique(step 1);
• If there is a primary key conflict, how to handle the conflict(step 2);

The design for writing new records(append and merge insert) looks like this:

To do the judgment (step 1), we need an effective index structure to do fast lookup and compact storage.

The index structure should cover the global key space for the current snapshot.

It may be split and bound to the fragment level, and it may be inlined and stored in the transaction file.

We may need to align the above information first.

In the bi-weekly dev meeting, I said that the features of the HFile may match the requirement that I described about the index structure above. But that does not mean we should use it, and we can not use it. It may cause you to misunderstand.

Any thoughts?

[jackye1995]

Both for Append and MergeInsert operations. That means, if there are new records that want to be inserted into the Lance table, we must check that the primary key is globally unique.

Personally I am not a fan of this, you cannot really truly enforce it if someone just appends a full fragment to the manifest and don't check anything. If you want uniqueness, you just use merge-insert, that feels very straightforward to me.

Even if we want to achieve this, we can just add a syntax sugar to make Append a MergeInsert if there is enforced primary key, so it reduces back to this current problem?

[wjones127]

Personally I am not a fan of this, you cannot really truly enforce it if someone just appends a full fragment to the manifest and don't check anything. If you want uniqueness, you just use merge-insert, that feels very straightforward to me.

Oh I was thinking that the uniqueness check would happen in the Commit code. Same part where we do conflict resolution. We could error if the user passes a fragment but doesn't supply the set of ids they are trying to insert for verification. And if we do find a duplicate created by the insert, it should return an error and prevent the commit.

[yanghua]

you cannot really truly enforce it if someone just appends a full fragment to the manifest and don't check anything.

@jackye1995 You mean, use the fragment api to append data? Correct me if I am wrong. But If yes, I think we can check it. Just like @wjones127 said(and in my design picture), the check logic happens in the commit phase.

If you want uniqueness, you just use merge-insert, that feels very straightforward to me.

There is one key question:

• The MergeInsert operation does a join on the on column; we do not enforce it to be a primary key, right? If the on column is not the primary key and if invoking when_not_match_then_insert_all, how to check the duplicated primary keys?