feat: implement Bloom Filter concurrent conflict detection for merge insert operations

protos/join_key.proto

@@ -0,0 +1,52 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The Lance Authors
+
+syntax = "proto3";
+
+package lance.table;
+
+// Value of the join key representation (reserved for future use)
+// Currently filters operate on hashed values for compactness.
+message JoinKeyValue {
+  oneof value {
+    string string_value = 1;
+    int64 int64_value = 2;
+    uint64 uint64_value = 3;
+    bytes binary_value = 4;
+    CompositeKey composite = 5;
+  }
+}
+
+message CompositeKey {
+  repeated JoinKeyValue parts = 1;
+}
+
+// Exact set of join keys, represented by their 64-bit hashes.
+message ExactSet {
+  repeated uint64 key_hashes = 1;
+}
+
+// Bloom filter data for join key set membership tests.
+message BloomFilterData {
+  // Bitset backing the bloom filter.
+  bytes bitmap = 1;
+  // Number of hash functions used.
+  uint32 num_hashes = 2;
+  // Total number of bits in the bitmap.
+  uint32 bitmap_bits = 3;
+  // Reserved for future fields to avoid reuse.
+  reserved 4, 5;
+  reserved "hash_seed", "hash_algo";
+}
+
+// Join key metadata attached to a Transaction for conflict detection.
+message JoinKeyMetadata {
+  // Names of columns participating in the join key.
+  repeated string columns = 1;
+  oneof filter {
+    ExactSet exact_set = 2;
+    BloomFilterData bloom = 3;
+  }
+  // Reserved to allow schema evolution.
+  reserved 4, 5;
+}

protos/transaction.proto

@@ -5,6 +5,7 @@ syntax = "proto3";
 
 import "file.proto";
 import "table.proto";
+import "join_key.proto";
 import "google/protobuf/any.proto";
 
 package lance.table;
@@ -33,6 +34,8 @@ message Transaction {
   // __lance_commit_message is a reserved key
   map<string, string> transaction_properties = 4;
 
+  // Join key metadata using typed protobuf message. This is the sole carrier.
+  optional JoinKeyMetadata join_key_metadata = 6;
   // Add new rows to the dataset.
   message Append {
     // The new fragments to append.
@@ -299,6 +302,7 @@ message Transaction {
   }
 
   // Fields 200/202 (`blob_append` / `blob_overwrite`) previously represented blob dataset ops.
+  reserved 5;
   reserved 200, 202;
   reserved "blob_append", "blob_overwrite";
 }

python/src/transaction.rs

@@ -551,6 +551,7 @@ impl FromPyObject<'_> for PyLance<Transaction> {
             operation,
             tag: None,
             transaction_properties,
+            join_key_metadata: None,
         }))
     }
 }

rust/lance-index/src/scalar/bloomfilter.rs

@@ -18,7 +18,7 @@ use crate::scalar::{
 use crate::{pb, Any};
 use arrow_array::{Array, UInt64Array};
 mod as_bytes;
-mod sbbf;
+pub mod sbbf;
 use arrow_schema::{DataType, Field};
 use serde::{Deserialize, Serialize};
 

rust/lance-table/build.rs

@@ -19,6 +19,7 @@ fn main() -> Result<()> {
             "./protos/table.proto",
             "./protos/transaction.proto",
             "./protos/rowids.proto",
+            "./protos/join_key.proto",
         ],
         &["./protos"],
     )?;

rust/lance/src/dataset.rs

@@ -66,6 +66,7 @@ pub(crate) mod blob;
 mod branch_location;
 pub mod builder;
 pub mod cleanup;
+pub mod conflict_detection;
 pub mod delta;
 pub mod fragment;
 mod hash_joiner;