Incorrect transaction isolation when inserting into Delta table while selecting from it

[findepi]

As found by CI https://github.com/trinodb/trino/actions/runs/19737237040/job/56552097779?pr=27480

Error:  Failures: 
Error:    TestDeltaLakeLocalConcurrentWritesTest.testConcurrentInsertsSelectingFromTheSameTable:136->testConcurrentInsertsSelectingFromTheSameTable:186->AbstractTestQueryFramework.assertQuery:356 For query 20251127_132020_00206_h9jyp: 
 SELECT * FROM test_concurrent_inserts_select_from_same_table_6htfdgywiq
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]

cc @findinpath @ebyhr

[findinpath]

I have explored several avenues on when this kind of unusual exceptions may happen:

• LocalOutputFile not failing when creating a namesake transaction file

OutputStream stream = Files.newOutputStream(path, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);

Unlikely. Even if so, the SELECT statement from the table would rather return 2 entries (given that one transaction log was swallowed) instead of 3

• DeltaLakeMetadata unobvious bug

io.trino.plugin.deltalake.DeltaLakeMetadata#checkForConcurrentTransactionConflicts
When inserting while reading from the same table the operation should fail:

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

Lines 2453 to 2456 in 1e3b669

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

One option that I can think of is that the INSERT operation is thought as blind, because in such a situation we don't evaluate for conflicts with other INSERT operations.

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

Line 2409 in 1e3b669

if (!enforcedSourcePartitionConstraints.isEmpty()) {

In such a situation we would end up with the tuples and this would match the failure reported above :

(0,10)
(1,10)
(1,10)

instead of having

(0,10)
(1,10)
(2,10)

Maybe there is a non-obvious issue in retrieving the table handles io.trino.sql.planner.optimizations.BeginTableWrite.Rewriter#findSourceTableHandles

trino/core/trino-main/src/main/java/io/trino/sql/planner/optimizations/BeginTableWrite.java

Lines 276 to 285 in 1e3b669

	private static List<TableHandle> findSourceTableHandles(PlanNode startNode)
	{
	return PlanNodeSearcher.searchFrom(startNode)
	.where(TableScanNode.class::isInstance)
	.findAll()
	.stream()
	.map(TableScanNode.class::cast)
	.map(TableScanNode::getTable)
	.collect(toImmutableList());
	}

[findepi]

• LocalOutputFile not failing when creating a namesake transaction file

Local FS doesn't support createExclusive. The creation via LocalTransactionLogSynchronizer is still exclusive, but not atomic. Another thread can see partially written empty file and -- not sure what it would do then.

Maybe it's worth trying to repro the issue with an object storage storage?
Partially written files are a local filesystem artifact, generally not present in scenarios we care most about.

[findinpath]

Additional failure in CI https://github.com/trinodb/trino/actions/runs/20140355697/job/57805861444?pr=27621