Store extended statistics filename in Delta Lake CommitInfoEntry

[chenjian2664]

Description

This change records the extended statistics filename in the commitInfo, when updating extended stats, we first write the statistics file, then leverage the commit log mechanism to handle concurrent conflicts - in other words it will allow the concurrent read/write if the log synchronizer support it.

This contribution ensures that there are unique stats files for each transaction log file (when the writing of the extended stats is enabled).

It is a fix for the existing flaky tests:

io.trino.testing.QueryFailedException: Failed to write Delta Lake transaction log entry
	at io.trino.testing.AbstractTestingTrinoClient.execute(AbstractTestingTrinoClient.java:138)
	at io.trino.testing.DistributedQueryRunner.executeInternal(DistributedQueryRunner.java:664)
	at io.trino.testing.DistributedQueryRunner.execute(DistributedQueryRunner.java:640)
	at io.trino.plugin.deltalake.TestDeltaLakeLocalConcurrentWritesTest.lambda$testConcurrentInsertsSelectingFromTheSameVersionedTable$2(TestDeltaLakeLocalConcurrentWritesTest.java:240)
	Suppressed: java.lang.Exception: SQL: INSERT INTO test_concurrent_inserts_select_from_same_versioned_table_jlty6w4qop SELECT 3, 'd' AS part FROM test_concurrent_inserts_select_from_same_versioned_table_jlty6w4qop FOR VERSION AS OF 0
Caused by: io.trino.spi.TrinoException: Failed to write Delta Lake transaction log entry
	at io.trino.plugin.deltalake.DeltaLakeMetadata.finishInsert(DeltaLakeMetadata.java:2696)
Caused by: io.trino.spi.TrinoException: Error reading statistics from cache
	at io.trino.plugin.deltalake.statistics.CachingExtendedStatisticsAccess.readExtendedStatistics(CachingExtendedStatisticsAccess.java:68)
	at io.trino.plugin.deltalake.DeltaLakeMetadata.updateTableStatistics(DeltaLakeMetadata.java:4818)
Caused by: java.lang.RuntimeException: Failed to decode JSON
	at io.trino.plugin.deltalake.statistics.MetaDirStatisticsAccess.decodeAndRethrowIfNotFound(MetaDirStatisticsAccess.java:142)
	at io.trino.plugin.deltalake.statistics.MetaDirStatisticsAccess.readExtendedStatistics(MetaDirStatisticsAccess.java:83)
Caused by: java.lang.IllegalArgumentException: Invalid JSON bytes for [simple type, class io.trino.plugin.deltalake.statistics.ExtendedStatistics]
	at io.airlift.json.JsonCodec.fromJson(JsonCodec.java:242)
Caused by: com.fasterxml.jackson.databind.exc.MismatchedInputException: No content to map due to end-of-input
 at [Source: REDACTED (StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION disabled); line: 1]

This is an alternative to #29006, but requires extensive modifications to the current implementation.

Design

According to the Delta Lake protocol specification:
https://github.com/delta-io/delta/blob/master/PROTOCOL.md#commit-provenance-information

Implementations are free to store any valid JSON-formatted data via the commitInfo action.

The commitInfo action is therefore the most appropriate place to persist additional metadata during a commit.
In Trino, the CommitInfoEntry is written for every data modification operation. This makes it a natural integration point for recording extended statistics. In contrast, MetadataEntry and ProtocolEntry are not suitable for this purpose, as they are intended for schema or protocol changes and should not be rewritten on every commit solely to update statistics.

The write flow for extended statistics follows the same pattern as data file writes:

• First, write the statistics file.
• Then, include the generated file path in the CommitInfoEntry.
• • Finally, rely on the commit mechanism to handle conflict resolution in concurrent scenarios.

Review

For reviewers:

• Commits 1–7 primarily focus on code cleanup. They also expose CommitInfo via DeltaLakeMetadata#getTableHandle and cache it within TableSnapshot.
• Commit 8 introduces the core change that persisting the extended statistics file path in CommitInfo.

Limitation

Concurrency guarantees

Conflict resolution is bounded by the capabilities of the underlying log synchronizer. In multi-cluster (e.g., cloud) deployments, this support is limited and depends on the guarantees provided by the underlying storage system.

Engine-specific visibility

Extended statistics are Trino-specific. When a table is written by other engines, these statistics will not be updated or recognized, which may lead to stale statistics being used by Trino.

Fixes:

• Flaky testConcurrentInsertsSelectingFromTheSameVersionedTable and testConcurrentInsertsSelectingFromTheSameTemporalVersionedTable in TestDeltaLakeLocalConcurrentWritesTest #21725
• TestDeltaLakeLocalConcurrentWritesTest.testConcurrentInsertsReconciliationForMixedInserts flaky: Failed to decode JSON from statistics access #22455
• DeltaLake rework the way cacheMetadataEntries are processed #12032
• Add extended statistics synchronisation when version conflict for INSERT is resolved in delta lake #16088
• Improve extended stats management in Delta Lake #29064

Fixes #21725
Fixes #22455
Fixes #12032
Fixes #16088
Fixes #29064

Additional context and related issues

• Pass allowFailure flag to statistics access #29006
• Synchronize Delta log writing in tests too #28092
• Make transaction log publish atomic in local file system #27904 (comment)
• Add concurrent writes reconciliation for blind append INSERT in Delta Lake #18506 the flag was introduced

Release notes

(x) This is not user-visible or is docs only, and no release notes are required.

[findinpath]

i'm missing integration test coverage for this new development

we should cover situations where a commit does extended stats writing and does not write extended stats, but they are retrieved nevertheless.

also vacuuming doesn't seem to be covered in the integration tests.

[chenjian2664]

Could you please elaborate more about "commit extended stats but not write"?

[chenjian2664]

@findepi Would you mind to take a look when you are available :)