Concurrent writes reconciliation in Delta Lake

trinodb / trino

Official repository of Trino, the distributed SQL query engine for big data, formerly known as PrestoSQL (https://trino.io)

https://trino.io

Apache License 2.0

10.49k stars 3.02k forks source link

Concurrent writes reconciliation in Delta Lake #16985

Open findepi opened 1 year ago

findepi commented 1 year ago

Support transactions that e.g. insert data concurrently, or such that modify data within disjoint data sets (eg partitions).

Thelin90 commented 1 year ago

Hello.

I have seen this behaviour for a while now.

You don't see this behaviour in spark, delta-rs when you do concurrent appends.

The delta connector does not seem to lock in the correct metadata delta version like other engines do.

I have seen this behaviour for

appends (works fine in spark since it commits the delta log on the driver and writes on the workers, will collide if using multi cluster)
- with delta-rs you can use native dynamodb lock
deletes(works fine in spark since it commits the delta log on the driver and writes on the workers, will collide if using multi cluster)
- with delta-rs you can use native dynamodb lock
merge (works fine in spark since it commits the delta log on the driver and writes on the workers, will collide if using multi cluster)

findinpath commented 1 year ago

Scenario for realizing that there is a conflict and the query needs to be aborted (error key plays an important role here to notify the client accordingly):

start:

0001.json
table has file A with rows 1-10

User1: update row 5:
0002.json
  remove entry: delete A
  add entry: add file C (rows 1-4, 6-10 same, 5 modified)

User2: update row 2:
0003.json
  remove entry: delete A
  add entry: add file B (rows 1, 3-10 same, 2 modified)

realize conflict

When writing on different partitions of the table therefore the connector should be able to cope transparently for the user with concurrent inserts - there are no shared resources which are being modified by the queries running concurrently.

findinpath commented 1 year ago

Basic algorithm (shared from an answer of @findepi )

detect concurrent transaction (already there - see TransactionConflictException)
see what were the transactions committed in the meantime and whether transaction being committed still makes sense and can continue
- if yes, re-commit with higher transaction number
- rinse & repeat in case of another concurrent commit

findinpath commented 8 months ago

Overview of the PRs used to cover the concurrent reconciliation functionality:

[x] https://github.com/trinodb/trino/pull/18506
[ ] https://github.com/trinodb/trino/pull/20983
[ ] TODO Add concurrent writes reconciliation for Serializable INSERT in Delta Lake
[ ] https://github.com/trinodb/trino/pull/20492
[ ] https://github.com/trinodb/trino/pull/18521 concurrent reconciliation support on DELETE pushdown / TRUNCATE operations
[ ] https://github.com/trinodb/trino/pull/21727 concurrent reconciliation support on UPDATE/MERGE/DELETE operations
[ ] https://github.com/trinodb/trino/pull/22443 concurrent reconciliation support for the optimize procedure

findepi commented 5 months ago

@findinpath awesome progress!

TODO concurrent reconciliation support for the optimize procedure

by any chance, is in the works?

findinpath commented 5 months ago

by any chance, is in the works?

I plan to create a PR for handling optimize during this week.

https://github.com/trinodb/trino/pull/22443