Fix flakiness of testConcurrentInsertsSelectingFromTheSameTable

[chenjian2664]

Description

The Delta Lake connector does not guarantee that queries are executed in a strict order. Queries may start concurrently, which makes count-based assertions in tests non-deterministic.

Additional context and related issues

Release notes

(x) 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.
( ) Release notes are required, with the following suggested text:

## Section
* Fix some things. ({issue}`issuenumber`)

[findinpath]

@chenjian2664 note that this PR relates to #27484

Queries may start concurrently, which makes count-based assertions in tests non-deterministic.

This is not true.
There seems to be indeed an issue when testing against the local filesystem which has not been reproduced against object storage providers.

[findinpath]

@chenjian2664 could you please add reference in the description to recorded CI failures against object storage providers?

[findinpath]

The flakiness is generated by testing against the local file system.

See #27484 (comment)

If the same issue would occur against an object storage, we would have a bug in the implementation.

[chenjian2664]

@findinpath just considering the reading phrase, when the queries arrive at the connector, it is possible it reads the same version of the table right?

[findinpath]

@findinpath just considering the reading phrase, when the queries arrive at the connector, it is possible it reads the same version of the table right?

yes, sure.

INSERT INTO " + tableName + " SELECT COUNT(*), 10 AS part FROM " + tableName

This is an operation which reads from the same table it inserts into.
This operation is therefore not a "blind append".

In case of dealing with a non-blind append, we will be checking whether other concurrent operations that are also non-blind inserts have inserted data and if so, fail the operation:

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

Lines 2452 to 2455 in 7269eba

	boolean readWholeTable = enforcedSourcePartitionConstraints.isAll();
	if (readWholeTable) {
	throw new TransactionFailedException("Conflicting concurrent writes found. Data files added in the modified table by concurrent write operation.");
	}

So, even if two operations are reading the same version of the table, only one of these operations will actually succeed in performing the commit successfully.

[chenjian2664]

@findinpath Thanks for guiding me to the place that committing the insert.
I think the problem is here

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

Line 2354 in b3c5bab

long currentVersion = getMandatoryCurrentVersion(fileSystem, handle.location(), readVersion.get());

, concurrent reads are possibly here and still reads the same readVersion - if it is, we won't step into checkForConcurrentTransactionConflicts at all.