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

findinpath · 2023-08-02T20:55:45Z

Description

Allow committing blind append INSERT operations in a concurrent context by
placing these operations right after any other previously concurrently
completed write operations.

Disallow committing blind insert operations in any of the following cases:

table schema change has been committed in the meantime
table protocol change has been committed in the meantime

INSERT operations that contain subqueries reading the same table are
subject to concurrent write failures.

Relates to #16985

This change primarily affects storages which offer "put-If-Absent" consistency guarantees (ABFS, GCS, ).
In case of AWS S3 it is attempted to offer a "best effort" alternative to having a full-blown locking mechanism, by retrying to persist the transaction log file corresponding to the transaction with jitter.

https://docs.databricks.com/en/optimizations/isolation-level.html#isolation-levels

Additional context and related issues

https://docs.delta.io/latest/concurrency-control.html
https://docs.databricks.com/en/optimizations/isolation-level.html#isolation-levels

Delta Lake OSS Source Code

https://github.com/delta-io/delta/blob/28881f5710e7b0b0bf2d959e36a5b6ebd75c62ae/spark/src/main/scala/org/apache/spark/sql/delta/ConflictChecker.scala

Delta Lake OSS Code source explained

https://books.japila.pl/delta-lake-internals/OptimisticTransactionImpl/#checkForConflicts

This PR introduces the trino-spi method in ConnectorMetadata

    /**
     * Finish insert query
     */
    default Optional<ConnectorOutputMetadata> finishInsert(
            ConnectorSession session,
            ConnectorInsertTableHandle insertHandle,
            List<ConnectorTableHandle> sourceTableHandles,
            Collection<Slice> fragments,
            Collection<ComputedStatistics> computedStatistics)
    {
        // Delegate to deprecated SPI to not break existing connectors
        return finishInsert(session, insertHandle, fragments, computedStatistics);
    }

This is done in order to be able to figure out whether the INSERT is not a blind append operation

INSERT INTO tableA SELECT * FROM tableA WHERE partitionkey = xxx

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
* Concurrent write reconciliation for blind append INSERT operation. ({issue}`issuenumber`)

ebyhr · 2023-09-05T11:21:43Z

/test-with-secrets sha=bf19346564de038c46513dccb0ef1630bbe4f5f6

github-actions · 2023-09-05T13:38:25Z

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

findinpath · 2023-09-05T20:27:25Z

@ebyhr pls do run the PR again with secrets. I've discovered a little issue in identifying transaction conflicts in my code.

...rino-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeConnectorSmokeTest.java

pajaks · 2023-09-06T12:42:15Z

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

            writeCommitted = true;
-            writeCheckpointIfNeeded(session, handle.getTableName(), handle.getLocation(), checkpointInterval, commitVersion);
+            writeCheckpointIfNeeded(session, handle.getTableName(), handle.getLocation(), handle.getMetadataEntry().getCheckpointInterval(), commitVersion);

            if (isCollectExtendedStatisticsColumnStatisticsOnWrite(session) && !computedStatistics.isEmpty() && !dataFileInfos.isEmpty()) {


Current mechanism also prevents concurrent writes to extended statistics. With this new change there is possibility to race condition when writing them.

~~There should be only one checkpoint per version written - if the process manages to write the x.log transaction log, there's no other process trying to create a checkpoint for the same version.~~

Comment relates to #16088

This could probably be done on a mechanism similar to what we have on https://github.com/trinodb/trino/blob/675defb40e12734f5d067848a14857ff66c80019/plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/transactionlog/writer/TransactionLogSynchronizer.java implementations. Doesn't make sense because we currently update the same stats file, not always a new one.

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

alexjo2144

Looking good besides a few nits

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

alexjo2144 · 2024-02-02T16:07:10Z

...rino-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeConnectorSmokeTest.java

@@ -2263,6 +2269,63 @@ public void testPartitionFilterIncluded()
        }
    }

+    @Test


I'd run this a few times

alexjo2144 · 2024-02-02T16:09:15Z

...lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeMinioAndHmsConnectorSmokeTest.java

+    @Disabled
+    public void testConcurrentInsertsReconciliationForBlindInserts()
+    {
+        // testConcurrentInsertsReconciliation requires safe writes capability to avoid test flakiness


Since all the writes are happening in Trino, this should work. The writes on s3 are unsafe if you're also writing from Spark.

Since all the writes are happening in Trino, this should work.

It works most of the time, but sometimes it fails - it happened in the CI and this is why I've decided to disable the test.

IIRC it was linked to Transaction log locked... messages.

In the context of retrying to flush a transaction log file from multiple Trino queries it may happen that multiple locks will exist for the same transaction log file.

Add the possibility to perform analysis on the dependencies of the INSERT statement. Specifically one connector could potentially figure out whether concurrent INSERT operations which add data into the same table as the one from which data is being selected collide with each other.

findinpath · 2024-02-12T17:07:25Z

Rebasing on master to address conflicts.
The commits affected:

Add scaffolding for using LocalTransactionLogSynchronizer in tests commit
the test TestDeltaLakeLocalConcurrentWritesTest from Add concurrent writes reconciliation for blind append INSERT operations commit

alexjo2144 · 2024-02-13T22:06:52Z

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

+                optionalCommitInfo = Optional.of(transactionLogEntry.getCommitInfo());
+            }
+            else if (transactionLogEntry.getAdd() != null) {
+                addedFilesBuilder.add(transactionLogEntry.getAdd());


If you don't need these saved in memory yet, let's not keep them around. You'll need them for updates/deletes but we're not there yet. Same with remove entries. The ones that are singletons not lists are fine with me.

alexjo2144 · 2024-02-13T22:12:16Z

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

@@ -2446,6 +2543,26 @@ private CommitInfoEntry getCommitInfoEntry(
            long createdTime,
            String operation,
            long readVersion)
+    {


I would probably just inline this one, rather than have two helper methods which just call a constructor

alexjo2144 · 2024-02-13T22:13:01Z

Make sure to get a run with secrets done

ebyhr · 2024-02-13T22:14:18Z

/test-with-secrets sha=e3c91e5238fbe099145d20924dcb13804f645ac9

github-actions · 2024-02-13T22:43:13Z

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

Allow committing blind append INSERT operations in a concurrent context by placing these operations right after any other previously concurrently completed write operations. Disallow committing blind insert operations in any of the following cases: - table schema change has been committed in the meantime - table protocol change has been committed in the meantime INSERT operations that contain subqueries reading the same table are subject to concurrent write failures.

ebyhr · 2024-02-16T02:20:37Z

/test-with-secrets sha=efe610257a489242ee65cb62b54d8919adfa217f

github-actions · 2024-02-16T02:42:47Z

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

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

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch from 8caeeee to fa9b21d Compare August 2, 2023 20:57

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

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch 3 times, most recently from b3514ff to c8a13ef Compare August 3, 2023 06:43

findinpath marked this pull request as ready for review August 3, 2023 06:43

findinpath requested review from findepi, ebyhr and homar August 3, 2023 06:43

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch 3 times, most recently from b723285 to 882f911 Compare August 3, 2023 09:30

findinpath mentioned this pull request Aug 3, 2023

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

Merged

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch 2 times, most recently from 661f4ba to 0023308 Compare August 3, 2023 13:26

tomfutago mentioned this pull request Aug 22, 2023

adds celo to transfers duneanalytics/spellbook#4060

Merged

findinpath self-assigned this Sep 5, 2023

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch from 0023308 to bf19346 Compare September 5, 2023 11:17

findinpath requested a review from pajaks September 5, 2023 11:21

github-actions bot added the tests:hive label Sep 5, 2023

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch from bf19346 to 507bae0 Compare September 5, 2023 20:26

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch from 507bae0 to a4d1c72 Compare September 5, 2023 20:35

pajaks reviewed Sep 6, 2023

View reviewed changes

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch from a4d1c72 to c70f3a7 Compare September 6, 2023 13:32

pajaks reviewed Sep 7, 2023

View reviewed changes

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch from c70f3a7 to 3f4a218 Compare September 14, 2023 10:23

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch from 2e39b72 to f3d4706 Compare February 2, 2024 13:39

alexjo2144 approved these changes Feb 2, 2024

View reviewed changes

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch from f3d4706 to c3b4200 Compare February 12, 2024 17:04

findinpath added 7 commits February 12, 2024 18:05

Signal transaction conflict when the transaction log file already exists

8e48709

Throw TransactionConflictException for multiple live locks

f51caa2

In the context of retrying to flush a transaction log file from multiple Trino queries it may happen that multiple locks will exist for the same transaction log file.

Consider writing of table statistics for CTAS or INSERT as optional

31cdd9b

Add utility method for retrieving table isolation level

d2c81ad

Extract logic for appending transaction log file for INSERT

0c549d7

Add scaffolding for using LocalTransactionLogSynchronizer in tests

916af43

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch 2 times, most recently from 6231236 to e3c91e5 Compare February 12, 2024 17:25

findinpath requested a review from alexjo2144 February 12, 2024 17:25

alexjo2144 approved these changes Feb 13, 2024

View reviewed changes

ebyhr approved these changes Feb 14, 2024

View reviewed changes

findinpath force-pushed the findinpath/delta-concurrent-writes-reconciliation branch from e3c91e5 to efe6102 Compare February 15, 2024 13:11

ebyhr merged commit 63216e0 into trinodb:master Feb 16, 2024
98 of 99 checks passed

github-actions bot added this to the 440 milestone Feb 16, 2024

colebow mentioned this pull request Feb 21, 2024

Add Trino 440 release notes #20797

Merged

This was referenced Mar 7, 2024

Add concurrent writes reconciliation for non-blind INSERT in Delta Lake #20983

Merged

Concurrent writes reconciliation in Delta Lake #16985

Open

findinpath mentioned this pull request Mar 28, 2024

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

Merged

findinpath mentioned this pull request Apr 26, 2024

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

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

findinpath commented Aug 2, 2023 •

edited

Loading

ebyhr commented Sep 5, 2023

github-actions bot commented Sep 5, 2023

findinpath commented Sep 5, 2023

pajaks Sep 6, 2023

findinpath Sep 6, 2023 •

edited

Loading

alexjo2144 left a comment

alexjo2144 Feb 2, 2024

alexjo2144 Feb 2, 2024

findinpath Feb 12, 2024

findinpath commented Feb 12, 2024 •

edited

Loading

alexjo2144 Feb 13, 2024

alexjo2144 Feb 13, 2024

alexjo2144 commented Feb 13, 2024

ebyhr commented Feb 13, 2024

github-actions bot commented Feb 13, 2024 •

edited

Loading

ebyhr commented Feb 16, 2024

github-actions bot commented Feb 16, 2024 •

edited

Loading

@@ @@ -2263,6 +2269,63 @@ public void testPartitionFilterIncluded() @@
                       }
                   }
+                  @Test

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

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

Conversation

findinpath commented Aug 2, 2023 • edited Loading

Description

Additional context and related issues

Release notes

ebyhr commented Sep 5, 2023

github-actions bot commented Sep 5, 2023

findinpath commented Sep 5, 2023

pajaks Sep 6, 2023

Choose a reason for hiding this comment

findinpath Sep 6, 2023 • edited Loading

Choose a reason for hiding this comment

alexjo2144 left a comment

Choose a reason for hiding this comment

alexjo2144 Feb 2, 2024

Choose a reason for hiding this comment

alexjo2144 Feb 2, 2024

Choose a reason for hiding this comment

findinpath Feb 12, 2024

Choose a reason for hiding this comment

findinpath commented Feb 12, 2024 • edited Loading

alexjo2144 Feb 13, 2024

Choose a reason for hiding this comment

alexjo2144 Feb 13, 2024

Choose a reason for hiding this comment

alexjo2144 commented Feb 13, 2024

ebyhr commented Feb 13, 2024

github-actions bot commented Feb 13, 2024 • edited Loading

ebyhr commented Feb 16, 2024

github-actions bot commented Feb 16, 2024 • edited Loading

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

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

findinpath commented Aug 2, 2023 •

edited

Loading

findinpath Sep 6, 2023 •

edited

Loading

findinpath commented Feb 12, 2024 •

edited

Loading

github-actions bot commented Feb 13, 2024 •

edited

Loading

github-actions bot commented Feb 16, 2024 •

edited

Loading