Add support for storing metadata to metastore in Delta Lake

[ebyhr]

Description

This PR adds a new delta.metastore.store-table-metadata config property (true by default) and stores the last transaction number and schema string to table parameter on the metastore when the property is enabled.

Conditions we update the metastore:

• Add these table parameters during CREATE (OR REPLACE) TABLE (AS) statements. This is non-async operation because we have to create or replace metastore tables either.
• Repalce metastore table in getTableHandle asynchronously if the transaction number doesn't exist in metastore or the number is older than _delta_log directory
• Repalce metastore table in other DDL or DML asynchronously

The cached information is used when listing table comments or columns.
Test result with 100 tables having 7 transaction logs:

SELECT * FROM system.metadata.table_comments WHERE schema_name = 'xxx';
3:09 → 1:44 (55%)

SELECT * FROM information_schema.columns WHERE table_schema = 'xxx';
3:30 → 1:42 (48%)

We could improve more once we use Glue specific API call (AWSGlue#getTables). This can be handled in follow-up.

Release notes

(x) Release notes are required, with the following suggested text:

# Delta Lake
* Add support for caching table metadata to metastore. ({issue}`issuenumber`)

[findinpath]

Pls rebase to deal with code conflicts.

[findinpath]

Looks very promissing !!!

[findinpath]

We need compatibility tests with Delta Lake OSS where Spark writes to the table and Trino is constrained to update the metadata caching properties on any read/write operation.

[ebyhr]

Addressed comments partially. I will add more tests to TestDeltaLakeMetastoreAccessOperations.

[findinpath]

Have you taken into consideration base64(compress(schema))?

[findepi]

Good question.
Regardless of whether we compress or not, we should have a check on length.

trino/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/GlueIcebergUtil.java

Lines 90 to 92 in d60b4d0

	boolean canPersistComment = (comment == null || comment.length() <= GLUE_TABLE_PARAMETER_LENGTH_LIMIT);
	boolean canPersistColumnInfo = glueColumns.isPresent();
	boolean canPersistMetadata = canPersistComment && canPersistColumnInfo;

[findinpath]

In case of referencing multiple times the same table in a query, is this check supposed to avoid calling metastore table replace repeatedly for the same table?

I don't follow where we ensure that the call is made only once.

I'm thinking that this should be ensured with the help of a ConcurrentMap like in io.trino.plugin.deltalake.DeltaLakeMetadata#getSnapshot(io.trino.spi.connector.ConnectorSession, io.trino.spi.connector.SchemaTableName, java.lang.String, java.util.Optional<java.lang.Long>)

[findepi]

I don't follow where we ensure that the call is made only once.

good point. i don't think we do

we should maintain a sort of a queue of pending updates
and scheduling new update logically remove previous entry

the ordering of updates doesn't matter (we can deliver them out of order, as long as we deduplicate), and we want to deduplicate, so we only need a Map<SchemaTableName, /* update info */>

Moreover, the update should be propagated to the metastore after transaction is committed.

Thus

• DeltaLakeMetadata should collect stuff to update (there shouldn't be any duplicates here, as DeltaLakeMetadata should not observe two different versions of one table, unless AS OF is used0)
• on commit, it should move the collected stuff to update to update manager singleton, to perform actual update
• the update manager should deduplicate (updates coming from different transactions) and perform actual updates

[findepi]

(skimmed)

[findepi]

I don't follow where we ensure that the call is made only once.

good point. i don't think we do

we should maintain a sort of a queue of pending updates
and scheduling new update logically remove previous entry

the ordering of updates doesn't matter (we can deliver them out of order, as long as we deduplicate), and we want to deduplicate, so we only need a Map<SchemaTableName, /* update info */>

Moreover, the update should be propagated to the metastore after transaction is committed.

Thus

• DeltaLakeMetadata should collect stuff to update (there shouldn't be any duplicates here, as DeltaLakeMetadata should not observe two different versions of one table, unless AS OF is used0)
• on commit, it should move the collected stuff to update to update manager singleton, to perform actual update
• the update manager should deduplicate (updates coming from different transactions) and perform actual updates

[findepi]

This looks like efficient operation, but it's not good API at least for Glue. Getting table names is as expensive as getting all table information (because that's what's happening under the covers)
We should add something like HiveMetastore.streamTables that would return full table objects in an iterative manner. I happen to have something like that implemented, so I can share the code later. For now, we can just use listTables, without extracting new streamTables method.

[findepi]

And without cache, we do only GET_TABLE, right?

would it be possible to have "do cache metadata" as a boolean variable in the test
so that the expected can be expressed as a conditional expression based on "do cache metadata"? this would make the test tell the story more directly

[ebyhr]

Just rebased on master to resolve conflicts. (No change)

[ebyhr]

Rebased on master to resolve logical conflicts.

[ebyhr]

Rebased on master to resolve conflicts.

[SemionPar]

unrelated question: Why is ThreadContextClassLoader required for metadata.commit()?

[wendigo]

probably to ensure that metadata.commit() call will be executed in the context of the delta lake plugin classloader

[ebyhr]

@wendigo Please take another look.

[wendigo]

LGTM % fixups