Add support for storing metadata to metastore in Delta Lake

Author: ebyhr

docs/src/main/sphinx/connector/delta-lake.md

@@ -161,6 +161,13 @@ values. Typical usage does not require you to configure them.
   - Maximum number of metastore data objects per transaction in the Hive
     metastore cache.
   - `1000`
+* - `delta.metastore.store-table-metadata`
+  - Store table comments and colum definitions in the metastore. The write
+    permission is required to update the metastore.
+  - `false`
+* - `delta.metastore.store-table-metadata-threads`
+  - Number of threads used for storing table metadata in metastore.
+  - `5`
 * - `delta.delete-schema-locations-fallback`
   - Whether schema locations are deleted when Trino can't determine whether they
     contain external files.

plugin/trino-delta-lake/pom.xml

@@ -130,6 +130,11 @@
             <artifactId>trino-plugin-toolkit</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>io.trino.hive</groupId>
+            <artifactId>hive-thrift</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>jakarta.annotation</groupId>
             <artifactId>jakarta.annotation-api</artifactId>
@@ -181,6 +186,21 @@
             <artifactId>jmxutils</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>aws-core</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>glue</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>utils</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-annotations</artifactId>
@@ -275,12 +295,6 @@
             <scope>runtime</scope>
         </dependency>
 
-        <dependency>
-            <groupId>software.amazon.awssdk</groupId>
-            <artifactId>glue</artifactId>
-            <scope>runtime</scope>
-        </dependency>
-
         <dependency>
             <groupId>com.github.docker-java</groupId>
             <artifactId>docker-java-api</artifactId>

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

@@ -74,6 +74,8 @@ public class DeltaLakeConfig
     private boolean collectExtendedStatisticsOnWrite = true;
     private HiveCompressionCodec compressionCodec = HiveCompressionCodec.SNAPPY;
     private long perTransactionMetastoreCacheMaximumSize = 1000;
+    private boolean storeTableMetadataEnabled;
+    private int storeTableMetadataThreads = 5;
     private boolean deleteSchemaLocationsFallback;
     private String parquetTimeZone = TimeZone.getDefault().getID();
     private DataSize targetMaxFileSize = DataSize.of(1, GIGABYTE);
@@ -377,6 +379,33 @@ public DeltaLakeConfig setPerTransactionMetastoreCacheMaximumSize(long perTransa
         return this;
     }
 
+    public boolean isStoreTableMetadataEnabled()
+    {
+        return storeTableMetadataEnabled;
+    }
+
+    @Config("delta.metastore.store-table-metadata")
+    @ConfigDescription("Store table metadata in metastore")
+    public DeltaLakeConfig setStoreTableMetadataEnabled(boolean storeTableMetadataEnabled)
+    {
+        this.storeTableMetadataEnabled = storeTableMetadataEnabled;
+        return this;
+    }
+
+    @Min(0) // Allow 0 to use the same thread for testing purpose
+    public int getStoreTableMetadataThreads()
+    {
+        return storeTableMetadataThreads;
+    }
+
+    @Config("delta.metastore.store-table-metadata-threads")
+    @ConfigDescription("Number of threads used for storing table metadata in metastore")
+    public DeltaLakeConfig setStoreTableMetadataThreads(int storeTableMetadataThreads)
+    {
+        this.storeTableMetadataThreads = storeTableMetadataThreads;
+        return this;
+    }
+
     public boolean isDeleteSchemaLocationsFallback()
     {
         return this.deleteSchemaLocationsFallback;

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

@@ -16,6 +16,7 @@
 import com.google.inject.Inject;
 import io.airlift.json.JsonCodec;
 import io.trino.filesystem.TrinoFileSystemFactory;
+import io.trino.plugin.deltalake.metastore.DeltaLakeTableMetadataScheduler;
 import io.trino.plugin.deltalake.metastore.HiveMetastoreBackedDeltaLakeMetastore;
 import io.trino.plugin.deltalake.statistics.CachingExtendedStatisticsAccess;
 import io.trino.plugin.deltalake.statistics.FileBasedTableStatisticsProvider;
@@ -56,6 +57,7 @@ public class DeltaLakeMetadataFactory
     private final long perTransactionMetastoreCacheMaximumSize;
     private final boolean deleteSchemaLocationsFallback;
     private final boolean useUniqueTableLocation;
+    private final DeltaLakeTableMetadataScheduler metadataScheduler;
 
     private final boolean allowManagedTableRename;
     private final String trinoVersion;
@@ -76,7 +78,8 @@ public DeltaLakeMetadataFactory(
             DeltaLakeRedirectionsProvider deltaLakeRedirectionsProvider,
             CachingExtendedStatisticsAccess statisticsAccess,
             @AllowDeltaLakeManagedTableRename boolean allowManagedTableRename,
-            NodeVersion nodeVersion)
+            NodeVersion nodeVersion,
+            DeltaLakeTableMetadataScheduler metadataScheduler)
     {
         this.hiveMetastoreFactory = requireNonNull(hiveMetastoreFactory, "hiveMetastore is null");
         this.fileSystemFactory = requireNonNull(fileSystemFactory, "fileSystemFactory is null");
@@ -98,6 +101,7 @@ public DeltaLakeMetadataFactory(
         this.useUniqueTableLocation = deltaLakeConfig.isUniqueTableLocation();
         this.allowManagedTableRename = allowManagedTableRename;
         this.trinoVersion = requireNonNull(nodeVersion, "nodeVersion is null").toString();
+        this.metadataScheduler = requireNonNull(metadataScheduler, "metadataScheduler is null");
     }
 
     public DeltaLakeMetadata create(ConnectorIdentity identity)
@@ -135,6 +139,7 @@ public DeltaLakeMetadata create(ConnectorIdentity identity)
                 deleteSchemaLocationsFallback,
                 deltaLakeRedirectionsProvider,
                 statisticsAccess,
+                metadataScheduler,
                 useUniqueTableLocation,
                 allowManagedTableRename);
     }

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

@@ -27,6 +27,7 @@
 import io.trino.plugin.deltalake.cache.DeltaLakeCacheKeyProvider;
 import io.trino.plugin.deltalake.functions.tablechanges.TableChangesFunctionProvider;
 import io.trino.plugin.deltalake.functions.tablechanges.TableChangesProcessorProvider;
+import io.trino.plugin.deltalake.metastore.DeltaLakeTableMetadataScheduler;
 import io.trino.plugin.deltalake.procedure.DropExtendedStatsProcedure;
 import io.trino.plugin.deltalake.procedure.FlushMetadataCacheProcedure;
 import io.trino.plugin.deltalake.procedure.OptimizeTableProcedure;
@@ -117,6 +118,9 @@ public void setup(Binder binder)
         binder.bind(TransactionLogAccess.class).in(Scopes.SINGLETON);
         newExporter(binder).export(TransactionLogAccess.class)
                 .as(generator -> generator.generatedNameOf(TransactionLogAccess.class, catalogName.get().toString()));
+        binder.bind(DeltaLakeTableMetadataScheduler.class).in(Scopes.SINGLETON);
+        newExporter(binder).export(DeltaLakeTableMetadataScheduler.class)
+                .as(generator -> generator.generatedNameOf(DeltaLakeTableMetadataScheduler.class, catalogName.get().toString()));
 
         binder.bind(TransactionLogWriterFactory.class).in(Scopes.SINGLETON);
         binder.bind(TransactionLogSynchronizerManager.class).in(Scopes.SINGLETON);

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

@@ -75,6 +75,7 @@ public final class DeltaLakeSessionProperties
     private static final String PROJECTION_PUSHDOWN_ENABLED = "projection_pushdown_enabled";
     private static final String QUERY_PARTITION_FILTER_REQUIRED = "query_partition_filter_required";
     private static final String CHECKPOINT_FILTERING_ENABLED = "checkpoint_filtering_enabled";
+    private static final String STORE_TABLE_METADATA = "store_table_metadata";
 
     private final List<PropertyMetadata<?>> sessionProperties;
 
@@ -230,7 +231,12 @@ public DeltaLakeSessionProperties(
                         CHECKPOINT_FILTERING_ENABLED,
                         "Use filter in checkpoint reader",
                         deltaLakeConfig.isCheckpointFilteringEnabled(),
-                        false));
+                        false),
+                booleanProperty(
+                        STORE_TABLE_METADATA,
+                        "Store table metadata in metastore",
+                        deltaLakeConfig.isStoreTableMetadataEnabled(),
+                        true));
     }
 
     @Override
@@ -348,4 +354,9 @@ public static boolean isCheckpointFilteringEnabled(ConnectorSession session)
     {
         return session.getProperty(CHECKPOINT_FILTERING_ENABLED, Boolean.class);
     }
+
+    public static boolean isStoreTableMetadataInMetastoreEnabled(ConnectorSession session)
+    {
+        return session.getProperty(STORE_TABLE_METADATA, Boolean.class);
+    }
 }

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

@@ -53,6 +53,11 @@ public void commit(ConnectorTransactionHandle transaction)
     {
         MemoizedMetadata deltaLakeMetadata = transactions.remove(transaction);
         checkArgument(deltaLakeMetadata != null, "no such transaction: %s", transaction);
+        deltaLakeMetadata.optionalGet().ifPresent(metadata -> {
+            try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(getClass().getClassLoader())) {
+                metadata.commit();
+            }
+        });
     }
 
     public void rollback(ConnectorTransactionHandle transaction)

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

@@ -0,0 +1,29 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.trino.plugin.deltalake;
+
+import com.google.inject.BindingAnnotation;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+import static java.lang.annotation.ElementType.FIELD;
+import static java.lang.annotation.ElementType.METHOD;
+import static java.lang.annotation.ElementType.PARAMETER;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+@Retention(RUNTIME)
+@Target({FIELD, PARAMETER, METHOD})
+@BindingAnnotation
+public @interface MaxTableParameterLength {}