MERGE INTO ignores an unenforced primary key: concurrent inserts of the same new keys duplicate rows

[Smarra]

Summary

When a Lance table has an unenforced primary key (ALTER TABLE ... SET UNENFORCED PRIMARY KEY), two concurrent MERGE INTO ... WHEN NOT MATCHED THEN INSERT statements that insert the same not-yet-present keys both commit, producing duplicate rows. lance-core's own merge_insert deduplicates the identical race using the declared key.

Environment

• lance-spark: reproduced on 0.7.1 (bundles lance-core 9.0.0) and 0.8.0-beta.1 (lance-core
  11.0.0-beta.10); Spark 3.5.2, Scala 2.12. Code inspection of main shows the same gap.

Steps to reproduce

Reproduced with the following self-contained ScalaTest spec (session config matches lance-spark's own test base). Two identical MERGEs of the same three not-yet-present keys race into an empty table that has an unenforced primary key declared.

import java.nio.file.Files
import java.util.UUID
import java.util.concurrent.{CountDownLatch, Executors, TimeUnit}

import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.types._
import org.scalatest.{FunSpec, Matchers}

class MergeUnenforcedPkReproSpec extends FunSpec with Matchers {

  private val root = Files.createTempDirectory("lance-merge-pk").toString

  private lazy val spark = SparkSession.builder()
    .master("local[4]")
    .config("spark.sql.extensions", "org.lance.spark.extensions.LanceSparkSessionExtensions")
    .config("spark.sql.catalog.lance_test", "org.lance.spark.LanceNamespaceSparkCatalog")
    .config("spark.sql.catalog.lance_test.impl", "dir")
    .config("spark.sql.catalog.lance_test.root", root)
    .config("spark.sql.catalog.lance_test.single_level_ns", "true")
    .getOrCreate()

  private val table = "lance_test.default.merge_pk_race"
  private val schema = StructType(Seq(
    StructField("id1", StringType, nullable = false),
    StructField("id2", StringType, nullable = false),
    StructField("payload", StringType, nullable = true)
  ))
  private val rows = Seq(Row("c1", "t1", "a"), Row("c1", "t2", "b"), Row("c2", "t1", "c"))

  // Two identical MERGEs on separate threads, released together to maximise overlap.
  private def raceTwoMerges(): Unit = {
    val start = new CountDownLatch(1)
    val pool = Executors.newFixedThreadPool(2)
    val task = new Runnable {
      override def run(): Unit = {
        val df = spark.createDataFrame(spark.sparkContext.parallelize(rows, 1), schema)
        val view = "src_" + UUID.randomUUID().toString.replace("-", "")
        df.createOrReplaceGlobalTempView(view)
        try {
          start.await()
          spark.sql(
            s"""MERGE INTO $table AS t USING global_temp.`$view` AS s
                ON t.id1 = s.id1 AND t.id2 = s.id2
                WHEN MATCHED THEN UPDATE SET *
                WHEN NOT MATCHED THEN INSERT *""")
        } finally spark.catalog.dropGlobalTempView(view)
      }
    }
    (1 to 2).foreach(_ => pool.submit(task))
    start.countDown()
    pool.shutdown()
    pool.awaitTermination(120, TimeUnit.SECONDS) shouldBe true
  }

  describe("lance-spark MERGE with an unenforced primary key") {
    it("two concurrent MERGEs of the same new keys should not duplicate") {
      spark.sql(
        s"CREATE TABLE $table (id1 STRING NOT NULL, id2 STRING NOT NULL, payload STRING) USING lance")
      spark.sql(s"ALTER TABLE $table SET UNENFORCED PRIMARY KEY (id1, id2)")

      raceTwoMerges()

      val count = spark.table(table).count()
      info(s"row count = $count (expected 3)")
      count shouldBe 3 // observed 6
    }
  }
}

Result: 6 rows. Expected: 3.

lance-core baseline (dedupes correctly)

The same race through lance-core's merge_insert (pylance) returns 3 rows once the key is declared, and 6 without it:

schema = pa.schema([
    pa.field("id1", pa.string(), nullable=False,
             metadata={"lance-schema:unenforced-primary-key": "true"}),
    pa.field("id2", pa.string(), nullable=False,
             metadata={"lance-schema:unenforced-primary-key": "true"}),
    pa.field("payload", pa.string()),
])
# two threads, released together:
#   lance.dataset(uri).merge_insert(["id1", "id2"]) \
#       .when_matched_update_all().when_not_matched_insert_all().execute(rows)
# -> 3 rows with the declared key (6 rows without it)

Root cause

lance-core deduplicates concurrent inserts inside merge_insert: it builds a KeyExistenceFilter from the inserted key values and attaches it to the transaction as inserted_rows_filter (rust/lance/src/dataset/write/merge_insert/exec/write.rs). The commit conflict resolver rejects a second transaction whose inserted keys intersect the first's (rust/lance/src/io/commit/conflict_resolver.rs, check_update_txn).

lance-spark's MERGE INTO does not go through merge_insert. It uses Spark's row-level framework (LanceRowLevelOperationBuilder -> SparkPositionDeltaWrite), which commits an org.lance.operation.Update with inserted_rows_filter = None. So concurrent inserts always take the no-filter path in the resolver and never conflict. SET UNENFORCED PRIMARY KEY (SetUnenforcedPrimaryKeyExec) only writes the schema metadata; nothing in the write path reads it. There are no references to KeyExistenceFilter or inserted_rows_filter in lance-spark's main source.