apache / apache/paimon

[Bug] [python] Duplicate rows when composite primary-key order differs from column order

Closed
#9,800 0 comments 0 reactions 0 assignees View on GitHub
Dominant language
Java
Stars
3.4k
Forks
1.4k
Avg merge
1d 11h
Merged PRs (30d)
396

Description

### Search before asking

- [x] I searched the issues and found no matching report.

### Paimon version

Apache master at `c081ccd0d7003ceae2517c6b592a947e824acc25`.

### Compute Engine

PyPaimon batch write/read API, Python 3.12.6, PyArrow 19.0.1, macOS arm64.

### Minimal reproduce step

Create a table with columns `(customer_id, order_id, amount)` and declare `PRIMARY KEY (order_id, customer_id)`. Write four orders, commit, then update one order in a second commit. The example uses a temporary local warehouse and one bucket.

Run with `paimon-python` from the above revision on `PYTHONPATH`:

```python
import tempfile

import pyarrow as pa
from pypaimon import CatalogFactory, Schema

def write(table, data):
builder = table.new_batch_write_builder()
writer, committer = builder.new_write(), builder.new_commit()
try:
writer.write_arrow(data)
committer.commit(writer.prepare_commit())
finally:
writer.close()
committer.close()

with tempfile.TemporaryDirectory() as warehouse:
catalog = CatalogFactory.create({"warehouse": warehouse})
catalog.create_database("test", False)
schema = pa.schema([
("customer_id", pa.int64()),
("order_id", pa.int64()),
("amount", pa.int64()),
])
catalog.create_table("test.orders", Schema.from_pyarrow_schema(
schema, primary_keys=["order_id", "customer_id"],
options={"bucket": "1"}), False)
table = catalog.get_table("test.orders")
write(table, pa.table({
"customer_id": [1, 2, 1, 2],
"order_id": [10, 10, 20, 20],
"amount": [100, 200, 300, 400],
}, schema=schema))
write(table, pa.table({
"customer_id": [1], "order_id": [20], "amount": [350],
}, schema=schema))

builder = table.new_read_builder()
actual = builder.new_read().to_arrow(builder.new_scan().plan().splits())
rows = actual.sort_by([
("customer_id", "ascending"), ("order_id", "ascending"),
]).to_pylist()
print(rows)
assert rows == [
{"customer_id": 1, "order_id": 10, "amount": 100},
{"customer_id": 1, "order_id": 20, "amount": 350},
{"customer_id": 2, "order_id": 10, "amount": 200},
{"customer_id": 2, "order_id": 20, "amount": 400},
]
```

### What doesn't meet your expectations?

I expect four rows, with the updated order `(customer_id=1, order_id=20)` retaining only `amount=350`. Instead, the reader returns five rows and the assertion fails:

```text
customer_id order_id amount
1 10 100
1 20 350
1 20 300
2 10 200
2 20 400
```

Changing only the primary-key declaration to `["customer_id", "order_id"]` makes the example pass.

### Anything else?

The writer sorts records by the declared primary-key order. In `paimon-python/pypaimon/read/split_read.py`, `SplitRead._create_key_value_fields` constructs keys by iterating over table columns instead. When the orders differ, the merge reader receives a stream that is no longer sorted by its key positions, so it can emit an updated row before reaching the old row for the same key.

Using the table's ordered `trimmed_primary_keys_fields` to construct read keys fixes this reproduction in local testing. I have also reproduced the duplicate-row behavior when the primary key includes a partition column, and after changing a primary-key column's position through schema evolution. I have not assessed Java, Flink or Spark readers.

### Are you willing to submit a PR?

Yes. A local fix and regression tests are ready.

Contributor guide

No contributing guide indexed for this repository

Research direction

Start in paimon-python/pypaimon/read/split_read.py, focusing on SplitRead._create_key_value_fields and the ordered primary-key fields used by the merge reader. Run the provided PyPaimon reproduction and inspect the ready regression tests; done means composite keys declared in a different order from table columns return four rows with the update applied once.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
data-engineering, databases
Issue type
Bug
Difficulty
3/5
Estimated time
1-2 days
Activity status
Active
Clarity
Clearly specified
Newbie friendliness
86/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.