Add failing test for multiple upserts in same transaction

Author: koenvo

tests/table/test_upsert.py

@@ -732,3 +732,39 @@ def test_transaction(catalog: Catalog) -> None:
     df = table.scan().to_arrow()
 
     assert df_before_transaction == df
+
+
+def test_transaction_multiple_upserts(catalog: Catalog) -> None:
+    identifier = "default.test_multi_upsert"
+    _drop_table(catalog, identifier)
+
+    schema = Schema(
+        NestedField(1, "id", IntegerType(), required=True),
+        NestedField(2, "name", StringType(), required=True),
+        identifier_field_ids=[1],
+    )
+
+    tbl = catalog.create_table(identifier, schema=schema)
+
+    # Define exact schema: required int32 and required string
+    arrow_schema = pa.schema([
+        pa.field("id", pa.int32(), nullable=False),
+        pa.field("name", pa.string(), nullable=False),
+    ])
+
+    tbl.append(pa.Table.from_pylist([{"id": 1, "name": "Alice"}], schema=arrow_schema))
+
+    df = pa.Table.from_pylist([{"id": 2, "name": "Bob"}, {"id": 1, "name": "Alicia"}], schema=arrow_schema)
+
+    with tbl.transaction() as txn:
+        # This should read the uncommitted changes?
+        txn.upsert(df, join_cols=["id"])
+
+        txn.upsert(df, join_cols=["id"])
+
+    result = tbl.scan().to_arrow().to_pylist()
+    assert sorted(result, key=lambda x: x["id"]) == [
+        {"id": 1, "name": "Alicia"},
+        {"id": 2, "name": "Bob"},
+    ]
+