JeonDaehong opened a new issue, #18552:
URL: https://github.com/apache/iceberg/issues/18552
### Apache Iceberg version
1.10.1 (`apache/iceberg-rest-fixture:1.10.1`, i.e. `RESTCatalogServlet` with
`CatalogHandlers`)
### Query engine
None (REST API directly)
### Please describe the bug 🐞
`POST /v1/{prefix}/transactions/commit` does not isolate requirements on
tables that the transaction does not update.
Two transactions are sent at the same time. Each one changes one table and
asserts, with a requirement-only table change (`updates: []`), that the other
table is unchanged:
- T1: table `a` adds a schema and sets it current
(`assert-current-schema-id` on `a`); table `b` has no updates and
`assert-current-schema-id` = b's current id.
- T2: the same with `a` and `b` swapped.
Both transactions commit (HTTP 200) in **29 of 30** rounds (30 of 30 in
another run), although each one invalidates the other's assertion. With a
serializable implementation at most one of them commits per round; Lakekeeper
0.13.6 and Nessie 0.108.8 behave that way (0 of 30).
It looks like `CatalogHandlers.commitTransaction` validates the requirements
of every table change while building the per-table transactions, then commits
the tables one by one; a change without updates is never committed, so its
requirements are not checked again at commit time. (#16232 describes a related
lost update in `BaseTransaction`.)
Why it matters: a service that validates a write against several tables,
e.g. foreign keys between them, needs the catalog to fail a transaction when
one of the tables it read has changed. With the current behaviour a child
insert validated against a parent key and a delete of that parent key can both
commit.
We also observed, under sustained concurrent commits to the same tables,
transactions answered 204 whose snapshot then was not in the table; we have not
minimized that one yet.
Reproduction (Python 3, standard library only), against `docker run -p
8181:8181 apache/iceberg-rest-fixture:1.10.1`:
```python
"""Two concurrent multi-table commits, each changing one table and asserting
the other is
unchanged. A serializable catalog commits at most one of them per round.
python skew_repro.py http://localhost:8181 [prefix] [bearer-token]
"""
import json, sys, threading, urllib.request, urllib.error
base = sys.argv[1].rstrip("/") + "/v1/" + (sys.argv[2] + "/" if
len(sys.argv) > 2 and sys.argv[2] else "")
token = sys.argv[3] if len(sys.argv) > 3 else None
def call(method, path, body=None):
req = urllib.request.Request(base + path, method=method,
data=json.dumps(body).encode() if body is
not None else None,
headers={"Content-Type": "application/json",
**({"Authorization": f"Bearer
{token}"} if token else {})})
try:
with urllib.request.urlopen(req) as r:
return r.status, (json.loads(r.read() or b"null"))
except urllib.error.HTTPError as e:
return e.code, e.read().decode()
schema = {"type": "struct", "fields": [{"id": 1, "name": "id", "required":
True, "type": "long"}]}
call("POST", "namespaces", {"namespace": ["skew"]})
for t in ("a", "b"):
call("DELETE", f"namespaces/skew/tables/{t}")
call("POST", "namespaces/skew/tables", {"name": t, "schema": schema,
"stage-create": False})
def meta(t):
return call("GET", f"namespaces/skew/tables/{t}")[1]["metadata"]
def change(t, other, other_schema_id, n):
m = meta(t)
return {"identifier": {"namespace": ["skew"], "name": t},
"requirements": [{"type": "assert-current-schema-id",
"current-schema-id": m["current-schema-id"]}],
"updates": [{"action": "add-schema", "last-column-id": 100 + n,
"schema": {"type": "struct", "fields": [
{"id": 1, "name": "id", "required": True,
"type": "long"},
{"id": 100 + n, "name": f"c{n}", "required":
False, "type": "long"}]}},
{"action": "set-current-schema", "schema-id": -1}]},
{
"identifier": {"namespace": ["skew"], "name": other},
"requirements": [{"type": "assert-current-schema-id",
"current-schema-id": other_schema_id}],
"updates": []}
both = 0
codes = {}
for n in range(1, 31):
a, b = meta("a")["current-schema-id"], meta("b")["current-schema-id"]
bodies = [{"table-changes": list(change("a", "b", b, 2 * n))},
{"table-changes": list(change("b", "a", a, 2 * n + 1))}]
status = [None, None]
barrier = threading.Barrier(2)
def run(i):
barrier.wait()
status[i] = call("POST", "transactions/commit", bodies[i])[0]
threads = [threading.Thread(target=run, args=(i,)) for i in (0, 1)]
[t.start() for t in threads]
[t.join() for t in threads]
both += all(s in (200, 204) for s in status)
for s in status:
codes[s] = codes.get(s, 0) + 1
print(f"both transactions committed in {both} of 30 rounds; status codes
{codes}")
```
```
$ python skew_repro.py http://localhost:8181
both transactions committed in 29 of 30 rounds; status codes {200: 59, 409:
1}
```
Found while testing [Open Integrity
Plane](https://github.com/JeonDaehong/integrity-plane), an Iceberg REST proxy
that enforces keys across tables; the fuller probe is
`compat/multitable_probe.py` there.
### Willingness to contribute
- [ ] I can contribute a fix for this bug independently
- [ ] I would be willing to contribute a fix for this bug with guidance from
the Iceberg community
- [x] I cannot contribute a fix for this bug at this time
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
