wangzhigang1999 opened a new issue, #9800:
URL: https://github.com/apache/paimon/issues/9800
### 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
