Add concurrent writes reconciliation for `DELETE` pushdown/`TRUNCATE` in Delta Lake #18521

findinpath · 2023-08-03T12:08:22Z

Description

Allow committing operations based on delete pushdown / truncate
in a concurrent context by placing these operations right after
any other previously concurrently completed write operations.

Disallow committing the operation in any of the following cases:

table schema change has been committed in the meantime
table protocol change has been committed in the meantime
add files committed in the meantime should be read by
the current operation
remove files committed in the meantime conflict with the
add files read by the current operation

The current changes also take into consideration the delta.isolationLevel
table property of the Delta Lake table for DELETE/TRUNCATE operations.

Relevant example taken from Databricks documentation in regards to the
distinction between WriteSerializable and Serializable isolation levels:

For example, consider txn1, a long running delete and txn2,
which inserts blindly data into the table.
txn2 and txn1 complete and they are recorded in the order
txn2, txn1
into the history of the table.
According to the history, the data inserted in txn2 should not exist
in the table. For Serializable level, a reader would never see data
inserted by txn2. However, for the WriteSerializable level, a reader
could at some point see the data inserted by txn2.

A few words about WriteSerializable isolation level taken from delta.io javadocs:

This isolation level will ensure snapshot isolation consistency guarantee
between write operations only.
In other words, if only the write operations are considered, then
there exists a serializable sequence between them that would produce the same
result as seen in the table.

Additional context and related issues

INSERT scaffolding PRs:

Depends on the fix from #21330

Release notes

( ) This is not user-visible or docs only and no release notes are required.
( ) Release notes are required, please propose a release note for me.
(x) Release notes are required, with the following suggested text:

# Delta Lake
* Add support for concurrent `DELETE` pushdown / `TRUNCATE` queries. ({issue}`issuenumber`)

pajaks · 2024-04-08T11:00:23Z

core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java

-
-        return metadata.executeDelete(connectorSession, table.getConnectorHandle());
+        List<ConnectorTableHandle> sourceConnectorHandles = sourceTableHandles.stream()
+                .filter(handle -> handle.getCatalogHandle().equals(catalogHandle))


Any reason why we wan't only handles from the same catalog?

This is intentional in order to ensure that schema and table name are pointing to the very same table.

If we'd have catalog1.schema1.table1 and catalog2.schema1.table1 we wouldn't be able to tell whether these tables are one and the same (because they may point to completely different tables even though their schema and table name are the same).

pajaks · 2024-04-08T11:06:41Z

...in/src/main/java/io/trino/sql/planner/iterative/rule/PushMergeWriterDeleteIntoConnector.java

@@ -64,7 +65,8 @@ public Result apply(TableFinishNode node, Captures captures, Context context)
                .map(newHandle -> new TableDeleteNode(
                        context.getIdAllocator().getNextId(),
                        newHandle,
-                        getOnlyElement(node.getOutputSymbols())))
+                        getOnlyElement(node.getOutputSymbols()),
+                        ImmutableList.of(tableScan.getTable())))


Isn't the `tableScan.getTable() target table?

The scanned table is what is being read.
See also a few lines above which indicates as well that this is the input table.

metadata.applyDelete(context.getSession(), tableScan.getTable())

pajaks · 2024-04-08T11:14:15Z

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeCommitSummary.java

@@ -59,15 +61,16 @@ else if (transactionLogEntry.getAdd() != null) {
                addedFilesCanonicalPartitionValuesBuilder.add(transactionLogEntry.getAdd().getCanonicalPartitionValues());
            }
            else if (transactionLogEntry.getRemove() != null) {
-                removedFilesFound = true;
+                Map<String, String> partitionValues = transactionLogEntry.getRemove().partitionValues();
+                removedFilesCanonicalPartitionValuesBuilder.add(partitionValues == null ? ImmutableMap.of() : canonicalizePartitionValues(partitionValues));


Why do we add empty map to set?

Good catch.
In the unlikely case that the winning commit does not contain information about the partition values of the remove file, we should fail the operation

Should we also fix addFiles in the same way?

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

...lta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeLocalConcurrentWritesTest.java

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

...lta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeLocalConcurrentWritesTest.java

findinpath · 2024-04-10T19:22:08Z

@ebyhr pls run the PR with secrets

ebyhr · 2024-04-10T21:05:48Z

/test-with-secrets sha=e4de37b413e10dd0a9d255c95a868ccb1b4e4f01

github-actions · 2024-04-10T21:06:53Z

The CI workflow run with tests that require additional secrets finished as failure: https://github.com/trinodb/trino/actions/runs/8637812363

ebyhr · 2024-04-12T03:44:56Z

/test-with-secrets sha=cda867bfcfe862af396c91026220e33692f5e4ce

github-actions · 2024-04-12T03:56:32Z

The CI workflow run with tests that require additional secrets finished as failure: https://github.com/trinodb/trino/actions/runs/8656675392

core/trino-main/src/main/java/io/trino/sql/planner/plan/TableDeleteNode.java

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

…e tables" This reverts commit 9d0b9f4.

In case of performing commit retries, start seeking the latest version of the table from the version of the table read during the latest attempt to commit the insert operation. In the context of concurrent operations, this strategy can spare some unnecessary HEAD operations to iterate incrementally starting from the version read when the table handle was instantiated.

…operations

ebyhr · 2024-04-25T03:42:40Z

/test-with-secrets sha=10b2d559dfce2926e352613677558fec0bc68e45

github-actions · 2024-04-25T03:43:48Z

The CI workflow run with tests that require additional secrets has been started: https://github.com/trinodb/trino/actions/runs/8826682787

...lta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeLocalConcurrentWritesTest.java

Allow committing pushdown DELETE operations in a concurrent context by placing these operations right after any other previously concurrently completed write operations. Disallow committing the operation in any of the following cases: - table schema change has been committed in the meantime - table protocol change has been committed in the meantime - add files committed in the meantime should be read by the current operation - remove files committed in the meantime conflict with the add files read by the current operation The current changes also take into consideration the `delta.isolationLevel` table property of the Delta Lake table for DELETE operations. Relevant example taken from Databricks documentation in regards to the distinction between `WriteSerializable` and `Serializable` isolation levels: > For example, consider `txn1`, a long running delete and `txn2`, > which inserts blindly data into the table. > `txn2` and `txn1` complete and they are recorded in the order > `txn2, txn1` > into the history of the table. > According to the history, the data inserted in `txn2` should not exist > in the table. For `Serializable` level, a reader would never see data > inserted by `txn2`. However, for the `WriteSerializable` level, a reader > could at some point see the data inserted by `txn2`. A few words about WriteSerializable isolation level taken from delta.io javadocs: > This isolation level will ensure snapshot isolation consistency guarantee > between write operations only. > In other words, if only the write operations are considered, then > there exists a serializable sequence between them that would produce the same > result as seen in the table.

cla-bot bot added the cla-signed label Aug 3, 2023

github-actions bot added the delta-lake Delta Lake connector label Aug 3, 2023

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch from 258382d to b262cb7 Compare August 3, 2023 13:13

findinpath self-assigned this Sep 5, 2023

findinpath mentioned this pull request Sep 6, 2023

Add concurrent writes reconciliation for blind append INSERT in Delta Lake #18506

Merged

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch from b262cb7 to b05405f Compare January 28, 2024 22:23

github-actions bot added the tests:hive label Jan 28, 2024

findinpath mentioned this pull request Mar 7, 2024

Concurrent writes reconciliation in Delta Lake #16985

Open

findinpath mentioned this pull request Mar 28, 2024

FINDINPATH Delta Lake concurrent update merge delete reconciliation findinpath/trino#20

Closed

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch from b05405f to 01347f4 Compare March 28, 2024 20:42

findinpath changed the title ~~WiP Add concurrent writes reconciliation for UPDATE/MERGE/DELETE in Delta Lake~~ Add concurrent writes reconciliation for UPDATE/MERGE/DELETE in Delta Lake Mar 28, 2024

findinpath mentioned this pull request Mar 28, 2024

Disregard nodes which are update targets when selecting source tables #21315

Merged

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch 7 times, most recently from 2bbb70e to 6b94cfe Compare April 1, 2024 07:04

findinpath marked this pull request as ready for review April 1, 2024 07:04

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch from 6b94cfe to 39f1647 Compare April 1, 2024 08:35

findinpath requested review from alexjo2144 and ebyhr and removed request for alexjo2144 April 3, 2024 13:49

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch from 39f1647 to 74ab9dc Compare April 3, 2024 13:50

findinpath mentioned this pull request Apr 3, 2024

Add Trino 444 release notes #21212

Merged

findinpath requested a review from pajaks April 4, 2024 10:52

pajaks reviewed Apr 8, 2024

View reviewed changes

pajaks reviewed Apr 9, 2024

View reviewed changes

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch from ead4b33 to e4de37b Compare April 10, 2024 08:44

pajaks approved these changes Apr 10, 2024

View reviewed changes

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch 2 times, most recently from e29ae61 to cda867b Compare April 11, 2024 20:59

ebyhr reviewed Apr 12, 2024

View reviewed changes

core/trino-main/src/main/java/io/trino/sql/planner/plan/TableDeleteNode.java Outdated Show resolved Hide resolved

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java Show resolved Hide resolved

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch from cda867b to 1979c9c Compare April 12, 2024 20:30

findepi removed the tests:hive label Apr 18, 2024

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch from 1979c9c to bb6c2e3 Compare April 19, 2024 10:33

findinpath added 4 commits April 23, 2024 17:07

Revert "Disregard nodes which are update targets when selecting sourc…

69792bb

…e tables" This reverts commit 9d0b9f4.

Extract logic used to check for concurrent transaction conflicts

1bde88d

Extract logic for appending transaction log file for DELETE pushdown …

4cf2856

…operations

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch from bb6c2e3 to 10b2d55 Compare April 23, 2024 15:10

findinpath mentioned this pull request Apr 26, 2024

Add concurrent writes reconciliation for UPDATE/MERGE/DELETE in Delta Lake #21727

Merged

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch from 10b2d55 to 31db21d Compare April 26, 2024 10:51

findinpath changed the title ~~Add concurrent writes reconciliation for UPDATE/MERGE/DELETE in Delta Lake~~ Add concurrent writes reconciliation for DELETE pushdown/TRUNCATE in Delta Lake Apr 26, 2024

ebyhr approved these changes Apr 29, 2024

View reviewed changes

...lta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeLocalConcurrentWritesTest.java Outdated Show resolved Hide resolved

findinpath force-pushed the findinpath/delta-concurrent-update-merge-delete-reconciliation branch from 31db21d to f7829ae Compare April 29, 2024 09:39

empty

6f8b8d0

ebyhr merged commit e5f536e into trinodb:master May 8, 2024
99 checks passed

github-actions bot added this to the 447 milestone May 8, 2024

colebow mentioned this pull request May 8, 2024

Add Trino 447 release notes #21873

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add concurrent writes reconciliation for `DELETE` pushdown/`TRUNCATE` in Delta Lake #18521

Add concurrent writes reconciliation for `DELETE` pushdown/`TRUNCATE` in Delta Lake #18521

findinpath commented Aug 3, 2023 •

edited

pajaks Apr 8, 2024

findinpath Apr 9, 2024

pajaks Apr 8, 2024

findinpath Apr 9, 2024

pajaks Apr 8, 2024

findinpath Apr 9, 2024

pajaks Apr 10, 2024

findinpath commented Apr 10, 2024

ebyhr commented Apr 10, 2024

github-actions bot commented Apr 10, 2024 •

edited

ebyhr commented Apr 12, 2024

github-actions bot commented Apr 12, 2024 •

edited

ebyhr commented Apr 25, 2024

github-actions bot commented Apr 25, 2024

Add concurrent writes reconciliation for DELETE pushdown/TRUNCATE in Delta Lake #18521

Add concurrent writes reconciliation for DELETE pushdown/TRUNCATE in Delta Lake #18521

Conversation

findinpath commented Aug 3, 2023 • edited

Description

Additional context and related issues

Release notes

pajaks Apr 8, 2024

Choose a reason for hiding this comment

findinpath Apr 9, 2024

Choose a reason for hiding this comment

pajaks Apr 8, 2024

Choose a reason for hiding this comment

findinpath Apr 9, 2024

Choose a reason for hiding this comment

pajaks Apr 8, 2024

Choose a reason for hiding this comment

findinpath Apr 9, 2024

Choose a reason for hiding this comment

pajaks Apr 10, 2024

Choose a reason for hiding this comment

findinpath commented Apr 10, 2024

ebyhr commented Apr 10, 2024

github-actions bot commented Apr 10, 2024 • edited

ebyhr commented Apr 12, 2024

github-actions bot commented Apr 12, 2024 • edited

ebyhr commented Apr 25, 2024

github-actions bot commented Apr 25, 2024

Add concurrent writes reconciliation for `DELETE` pushdown/`TRUNCATE` in Delta Lake #18521

Add concurrent writes reconciliation for `DELETE` pushdown/`TRUNCATE` in Delta Lake #18521

findinpath commented Aug 3, 2023 •

edited

github-actions bot commented Apr 10, 2024 •

edited

github-actions bot commented Apr 12, 2024 •

edited