Pass allowFailure flag to statistics access

[chenjian2664]

Description

Concurrent access to metadata statistics may lead to errors:

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]

as the current implementation does not provide concurrency guarantees.
Since statistics access is typically non-essential to query execution, this PR introduces pass a flag to define the failure semantics, enabling best-effort handling where metadata statistics access failures do not interrupt the main execution path.

Fixes #21725
Fixes #22455

Additional context and related issues

• 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.
( ) 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`)

[findepi]

Concurrent access to metadata statistics may lead to errors:

We can fix concurrent access/modification quite easily.
E.g. we can have a backup file a reader would use.

How will we fix concurrent writes?
I think it might be time to fix them?

Since statistics access is typically non-essential to query execution, this PR introduces a flag to define the failure semantics

This is not great

• it hides problem rather than fix it
• the assumption is wrong. A bad query plan can lead to query execution being 1000x more expensive or simply impossible to execute.

[chenjian2664]

E.g. we can have a backup file a reader would use.

What's would that be like? there will be still a conflict when we are (concurrent) baking up, or still a chance reading an updating backup.

but If you've see an easy way I can have a try on it

How will we fix concurrent writes?

We don't fix the concurrent writes here, like previously we are passing the ignoreFailure in here
But we didn't notice that reading also could fail - I pass the ignoreFailure into for reading, this is major semantic I want to update in this pr.

Could you clarify what kind of "concurrent writes" you have in mind -- are we talking about within a single cluster or across multiple clusters?
If it's just within a single cluster, we could likely handle it by adding synchronization around the read/write methods.
For cross cluster cases, I don't think we currently have a reliable way to support all environments, similar to the log synchronizer challenges discussed in #28092

But it wouldn't matter if it is just a optimization we could just ignore it.

it hides problem rather than fix it

I might be missing something, but why do we have an ignoreFailure flag when writing statistics? My understanding is that they're mainly for optimizing metadata access, so I assumed failures wouldn't be critical.

[findinpath]

This contribution is in line with #18506

While the approach of "ignoring failures" is certainly not ideal, pragmatically this contribution serves at least in avoiding test flakiness.

Let's use the opportunity and come up with a rough action plan on how to deal with Delta Lake table extended statistics before we go forward and merge and forget about the underlying issue (the fact that multiple readers & writers are trying to read & write from the very same extended stats file sometimes concurrently).

[chenjian2664]

temporary solution #29170