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

[findinpath]

Description

Follow-up from #18506

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

On tables with isolation level WriteSerializable, allow committing
non-blind append INSERT operations in a concurrent context by
placing these operations right after any other previously concurrently
completed write operations.

For disambiguation, INSERT operations that contain subqueries reading the same
table are considered non-blind inserts.

Disallow committing 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
• add files committed in the meantime should be read by
  the current insert operations
• remove files committed in the meantime conflict with the add files read
  by the current insert 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.

Consider the Delta Lake table t with the following definition:

CREATE TABLE t(data integer);
INSERT INTO t VALUES 0;

There will be always reconciliation when dealing with the following concurrent
INSERT statements:

INSERT INTO t SELECT count(*) FROM t; -- non-blind insert
INSERT INTO t VALUES 2; -- blind insert

The expected state of the table t will be independently of the order of the way
in which the operations were actually committed to the transaction log:

0
1
2

On the other hand, there can't be reconciliation when dealing with the following
concurrent INSERT statements:

INSERT INTO t SELECT count(*) FROM t; -- non-blind insert
INSERT INTO t SELECT count(*) FROM t; -- non-blind insert

The reason why the reconciliation will fail is because there can't be picked any
serializable sequence of statements to explain the state of the table if both
statements would commit successfully.
Just for the sake of demonstration, if both statements would complete successfully,
the table state would be:

0
1
1

However, independently how we would pick a serializable sequence of operations
performed on the table, the table state would look like:

0
1
2

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

Release notes

( ) This is not user-visible or is 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 non-blind append INSERT operation. ({issue}`issuenumber`)

[ebyhr]

/test-with-secrets sha=e452a09cf3bf6726cd1461b197f0797c784d75c7

[github-actions]

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

[pajaks]

Should it match source enforceDomains with source addAction. Or with target?

[findinpath]

The reconciliation is done AFAIU based on the enforced partition constraints of the source tables (for non-blind INSERT or UPDATE/MERGE/DELETE operations).
If the operations writes new data - lNSERT operations - , this is not relevant for the WriteSerializable isolation level.
This aspect becomes relevant when dealing with Serializable operations on the same partition of the table - out of scope for the current PR.

[pajaks]

Shouldn't name of this variable be more like conflictingAddAction? Right now unreadAddAction indicates that those add action are not conflicting because we don't read from them.

[findinpath]

The files are not necessarily conflicting.
It may very well be that the operations don't overlap at row level.
Here we're just aborting the operation if we're findining new active files that should have been taken into account

[pajaks]

Any reason why this test is not in BaseDeltaLakeConnectorSmokeTest as well?

[findinpath]

I'm reluctant that there is added value to test extensively on BaseDeltaLakeConnectorSmokeTest.
The class would not be a "smoke" test class anymore.

I trust that the local file store does indeed throw FileAlreadyExistsException when a concurrent operation creates a namesake transaction log file and that's why I'd be inclined to say TestDeltaLakeLocalConcurrentWritesTest is fit to hold an increased battery of tests for concurrent writes reconciliation functionality.

[ebyhr]

Almost good to me.

[alexjo2144]

Could you summarize the data in this summary more? Looks like you could throw out any add file commits that are blind appends entirely, and for other commits you need a set of the partitions but not the rest of the data.

[findinpath]

Keeping only:

• canonical partition keys for the added files
• a flag stating whether or not the commit contains remove files.

[alexjo2144]

From what I can tell the conflict handling here is still safe, just a few questions/comments.

[findinpath]

@ebyhr , @alexjo2144 AC.
Please make another round of review 🙏

[ebyhr]

/test-with-secrets sha=4d0024c45c16cde39d18dc160e6eef83abcd619e

[github-actions]

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

[ebyhr]

@findinpath Could you take a look at the following failure? It's happening on master:

Error:    TestDeltaLakeLocalConcurrentWritesTest.testConcurrentInsertsSelectingFromTheSameTable:177->testConcurrentInsertsSelectingFromTheSameTable:228->AbstractTestQueryFramework.assertQuery:350 For query 20240329_103745_00065_e4xfv: 
 SELECT * FROM test_concurrent_inserts_select_from_same_table_h28nk6v7lf
not equal
Actual rows (up to 100 of 1 extra rows shown, 3 rows in total):
    [1, 10]
Expected rows (up to 100 of 1 missing rows shown, 3 rows in total):
    [2, 10]

https://github.com/trinodb/trino/actions/runs/8480015101/job/23235027443