Add support for lock-free commit in Iceberg HMS catalog

docs/src/main/sphinx/object-storage/metastores.md

@@ -208,6 +208,39 @@ properties:
   -
 :::
 
+(iceberg-hive-catalog)=
+### Iceberg-specific Hive catalog configuration properties
+
+When using the Hive catalog, the Iceberg connector supports the same
+{ref}`general Thrift metastore configuration properties <hive-thrift-metastore>` 
+as previously described with the following additional property:
+
+:::{list-table} Iceberg Hive catalog configuration property
+:widths: 35, 50, 15
+:header-rows: 1
+
+* - Property name
+  - Description
+  - Default
+* - `iceberg.hive-catalog.locking-enabled`
+  - Commit to tables using Hive locks.
+  - `true`
+:::
+
+:::{warning}
+Setting `iceberg.hive-catalog.locking-enabled=false` will cause the catalog to
+commit to tables without using Hive locks. This should only be set to false if all
+following conditions are met:
+
+* [HIVE-26882](https://issues.apache.org/jira/browse/HIVE-26882) is available on
+  the Hive metastore server. Requires version 2.3.10, 4.0.0-beta-1 or later.
+* [HIVE-28121](https://issues.apache.org/jira/browse/HIVE-28121) is available on
+  the Hive metastore server, if it is backed by MySQL or MariaDB. Requires version
+  2.3.10, 4.1.0, 4.0.1 or later.
+* All other catalogs committing to tables that this catalogs commits to are also
+  on Iceberg 1.3 or later, and disabled Hive locks on commit.
+:::
+
 (hive-thrift-metastore-authentication)=
 ### Thrift metastore authentication
 

lib/trino-metastore/src/main/java/io/trino/metastore/HiveMetastore.java

@@ -90,7 +90,7 @@ default boolean useSparkTableStatistics()
      * alter one field of a table object previously acquired from getTable is
      * probably not what you want.
      */
-    void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges);
+    void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges, Map<String, String> environmentContext);
 
     void renameTable(String databaseName, String tableName, String newDatabaseName, String newTableName);
 

lib/trino-metastore/src/main/java/io/trino/metastore/cache/CachingHiveMetastore.java

@@ -651,10 +651,10 @@ public void dropTable(String databaseName, String tableName, boolean deleteData)
     }
 
     @Override
-    public void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges)
+    public void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges, Map<String, String> environmentContext)
     {
         try {
-            delegate.replaceTable(databaseName, tableName, newTable, principalPrivileges);
+            delegate.replaceTable(databaseName, tableName, newTable, principalPrivileges, environmentContext);
         }
         finally {
             invalidateTable(databaseName, tableName);

lib/trino-metastore/src/main/java/io/trino/metastore/tracing/TracingHiveMetastore.java

@@ -236,13 +236,13 @@ public void dropTable(String databaseName, String tableName, boolean deleteData)
     }
 
     @Override
-    public void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges)
+    public void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges, Map<String, String> environmentContext)
     {
         Span span = tracer.spanBuilder("HiveMetastore.replaceTable")
                 .setAttribute(SCHEMA, databaseName)
                 .setAttribute(TABLE, tableName)
                 .startSpan();
-        withTracing(span, () -> delegate.replaceTable(databaseName, tableName, newTable, principalPrivileges));
+        withTracing(span, () -> delegate.replaceTable(databaseName, tableName, newTable, principalPrivileges, environmentContext));
     }
 
     @Override

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

@@ -13,6 +13,7 @@
  */
 package io.trino.plugin.deltalake.metastore;
 
+import com.google.common.collect.ImmutableMap;
 import io.trino.metastore.Database;
 import io.trino.metastore.HiveMetastore;
 import io.trino.metastore.PrincipalPrivileges;
@@ -108,7 +109,7 @@ public void createTable(Table table, PrincipalPrivileges principalPrivileges)
     @Override
     public void replaceTable(Table table, PrincipalPrivileges principalPrivileges)
     {
-        delegate.replaceTable(table.getDatabaseName(), table.getTableName(), table, principalPrivileges);
+        delegate.replaceTable(table.getDatabaseName(), table.getTableName(), table, principalPrivileges, ImmutableMap.of());
     }
 
     @Override

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/metastore/file/DeltaLakeFileMetastoreTableOperations.java

@@ -47,6 +47,6 @@ public void commitToExistingTable(SchemaTableName schemaTableName, long version,
                 .putAll(tableMetadataParameters(version, schemaString, tableComment))
                 .buildKeepingLast();
         Table updatedTable = currentTable.withParameters(parameters);
-        metastore.replaceTable(currentTable.getDatabaseName(), currentTable.getTableName(), updatedTable, buildInitialPrivilegeSet(currentTable.getOwner().orElseThrow()));
+        metastore.replaceTable(currentTable.getDatabaseName(), currentTable.getTableName(), updatedTable, buildInitialPrivilegeSet(currentTable.getOwner().orElseThrow()), ImmutableMap.of());
     }
 }

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/metastore/thrift/DeltaLakeThriftMetastoreTableOperations.java

@@ -66,7 +66,7 @@ public void commitToExistingTable(SchemaTableName schemaTableName, long version,
                     .buildKeepingLast();
             Table updatedTable = currentTable.withParameters(parameters);
 
-            metastore.replaceTable(currentTable.getDatabaseName(), currentTable.getTableName(), updatedTable, buildInitialPrivilegeSet(currentTable.getOwner().orElseThrow()));
+            metastore.replaceTable(currentTable.getDatabaseName(), currentTable.getTableName(), updatedTable, buildInitialPrivilegeSet(currentTable.getOwner().orElseThrow()), ImmutableMap.of());
         }
         finally {
             thriftMetastore.releaseTableLock(lockId);

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeFileOperations.java

@@ -14,6 +14,7 @@
 package io.trino.plugin.deltalake;
 
 import com.google.common.collect.HashMultiset;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableMultiset;
 import com.google.common.collect.Multiset;
 import com.google.common.io.Resources;
@@ -1265,6 +1266,6 @@ private void removeLastTransactionVersionFromMetastore(String schemaName, String
         Table newMetastoreTable = Table.builder(table)
                 .setParameters(filterKeys(table.getParameters(), key -> !key.equals("trino_last_transaction_version")))
                 .build();
-        metastore.replaceTable(table.getDatabaseName(), table.getTableName(), newMetastoreTable, buildInitialPrivilegeSet(table.getOwner().orElseThrow()));
+        metastore.replaceTable(table.getDatabaseName(), table.getTableName(), newMetastoreTable, buildInitialPrivilegeSet(table.getOwner().orElseThrow()), ImmutableMap.of());
     }
 }

plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/metastore/TestDeltaLakeMetastoreAccessOperations.java

@@ -13,6 +13,7 @@
  */
 package io.trino.plugin.deltalake.metastore;
 
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableMultiset;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Multiset;
@@ -574,7 +575,7 @@ private void removeMetadataCachingPropertiesFromMetastore(String tableName)
         Table newMetastoreTable = Table.builder(table)
                 .setParameters(Maps.filterKeys(table.getParameters(), key -> !key.equals("trino_last_transaction_version")))
                 .build();
-        metastore.replaceTable(table.getDatabaseName(), table.getTableName(), newMetastoreTable, buildInitialPrivilegeSet(table.getOwner().orElseThrow()));
+        metastore.replaceTable(table.getDatabaseName(), table.getTableName(), newMetastoreTable, buildInitialPrivilegeSet(table.getOwner().orElseThrow()), ImmutableMap.of());
     }
 
     private Session sessionWithStoreTableMetadata(boolean storeTableMetadata)

plugin/trino-hive/src/main/java/io/trino/plugin/hive/TrinoViewHiveMetastore.java

@@ -92,7 +92,7 @@ public void createView(ConnectorSession session, SchemaTableName schemaViewName,
                 throw new ViewAlreadyExistsException(schemaViewName);
             }
 
-            metastore.replaceTable(schemaViewName.getSchemaName(), schemaViewName.getTableName(), table, principalPrivileges);
+            metastore.replaceTable(schemaViewName.getSchemaName(), schemaViewName.getTableName(), table, principalPrivileges, ImmutableMap.of());
             return;
         }
 
@@ -225,6 +225,6 @@ private void replaceView(ConnectorSession session, SchemaTableName viewName, Tab
 
         PrincipalPrivileges principalPrivileges = isUsingSystemSecurity ? NO_PRIVILEGES : buildInitialPrivilegeSet(session.getUser());
 
-        metastore.replaceTable(viewName.getSchemaName(), viewName.getTableName(), viewBuilder.build(), principalPrivileges);
+        metastore.replaceTable(viewName.getSchemaName(), viewName.getTableName(), viewBuilder.build(), principalPrivileges, ImmutableMap.of());
     }
 }

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/SemiTransactionalHiveMetastore.java

@@ -594,7 +594,7 @@ public synchronized void dropTable(ConnectorSession session, String databaseName
 
     public synchronized void replaceTable(String databaseName, String tableName, Table table, PrincipalPrivileges principalPrivileges)
     {
-        setExclusive(delegate -> delegate.replaceTable(databaseName, tableName, table, principalPrivileges));
+        setExclusive(delegate -> delegate.replaceTable(databaseName, tableName, table, principalPrivileges, ImmutableMap.of()));
     }
 
     public synchronized void renameTable(String databaseName, String tableName, String newDatabaseName, String newTableName)
@@ -3063,7 +3063,7 @@ public void run(HiveMetastore metastore, AcidTransaction transaction)
                 metastore.alterTransactionalTable(newTable, transaction.getAcidTransactionId(), transaction.getWriteId(), principalPrivileges);
             }
             else {
-                metastore.replaceTable(newTable.getDatabaseName(), newTable.getTableName(), newTable, principalPrivileges);
+                metastore.replaceTable(newTable.getDatabaseName(), newTable.getTableName(), newTable, principalPrivileges, ImmutableMap.of());
             }
         }
 
@@ -3077,7 +3077,7 @@ public void undo(HiveMetastore metastore, AcidTransaction transaction)
                 metastore.alterTransactionalTable(oldTable, transaction.getAcidTransactionId(), transaction.getWriteId(), principalPrivileges);
             }
             else {
-                metastore.replaceTable(oldTable.getDatabaseName(), oldTable.getTableName(), oldTable, principalPrivileges);
+                metastore.replaceTable(oldTable.getDatabaseName(), oldTable.getTableName(), oldTable, principalPrivileges, ImmutableMap.of());
             }
         }
     }

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/file/FileHiveMetastore.java

@@ -586,7 +586,7 @@ public synchronized void dropTable(String databaseName, String tableName, boolea
     }
 
     @Override
-    public synchronized void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges)
+    public synchronized void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges, Map<String, String> environmentContext)
     {
         Table table = getRequiredTable(databaseName, tableName);
         if (!table.getDatabaseName().equals(databaseName) || !table.getTableName().equals(tableName)) {

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/glue/GlueHiveMetastore.java

@@ -606,7 +606,7 @@ private void deleteDir(Location path)
     }
 
     @Override
-    public void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges)
+    public void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges, Map<String, String> environmentContext)
     {
         if (!tableName.equals(newTable.getTableName()) || !databaseName.equals(newTable.getDatabaseName())) {
             throw new TrinoException(NOT_SUPPORTED, "Table rename is not yet supported by Glue service");

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/thrift/BridgingHiveMetastore.java

@@ -247,9 +247,9 @@ public void dropTable(String databaseName, String tableName, boolean deleteData)
     }
 
     @Override
-    public void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges)
+    public void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges, Map<String, String> environmentContext)
     {
-        alterTable(databaseName, tableName, toMetastoreApiTable(newTable, principalPrivileges));
+        alterTable(databaseName, tableName, toMetastoreApiTable(newTable, principalPrivileges), environmentContext);
     }
 
     @Override
@@ -292,7 +292,7 @@ public void setTableOwner(String databaseName, String tableName, HivePrincipal p
                 .setOwner(Optional.of(principal.getName()))
                 .build();
 
-        delegate.alterTable(databaseName, tableName, toMetastoreApiTable(newTable));
+        delegate.alterTable(databaseName, tableName, toMetastoreApiTable(newTable), ImmutableMap.of());
     }
 
     @Override
@@ -360,7 +360,12 @@ public void dropColumn(String databaseName, String tableName, String columnName)
 
     private void alterTable(String databaseName, String tableName, io.trino.hive.thrift.metastore.Table table)
     {
-        delegate.alterTable(databaseName, tableName, table);
+        delegate.alterTable(databaseName, tableName, table, ImmutableMap.of());
+    }
+
+    private void alterTable(String databaseName, String tableName, io.trino.hive.thrift.metastore.Table table, Map<String, String> context)
+    {
+        delegate.alterTable(databaseName, tableName, table, context);
     }
 
     @Override

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/thrift/ThriftHiveMetastore.java

@@ -434,7 +434,7 @@ public void updateTableStatistics(String databaseName, String tableName, Optiona
         if (acidWriteId.isPresent()) {
             modifiedTable.setWriteId(acidWriteId.getAsLong());
         }
-        alterTable(databaseName, tableName, modifiedTable);
+        alterTable(databaseName, tableName, modifiedTable, ImmutableMap.of());
 
         io.trino.metastore.Table table = fromMetastoreApiTable(modifiedTable);
         List<ColumnStatisticsObj> metastoreColumnStatistics = updatedStatistics.columnStatistics().entrySet().stream()
@@ -962,7 +962,7 @@ private static boolean isManagedTable(Table table)
     }
 
     @Override
-    public void alterTable(String databaseName, String tableName, Table table)
+    public void alterTable(String databaseName, String tableName, Table table, Map<String, String> environmentContext)
     {
         if (!Objects.equals(databaseName, table.getDbName())) {
             validateObjectName(table.getDbName());
@@ -980,7 +980,10 @@ public void alterTable(String databaseName, String tableName, Table table)
                             // This prevents Hive 3.x from collecting basic table stats at table creation time.
                             // These stats are not useful by themselves and can take a very long time to collect when creating an
                             // external table over a large data set.
-                            context.setProperties(ImmutableMap.of("DO_NOT_UPDATE_STATS", "true"));
+                            context.setProperties(ImmutableMap.<String, String>builder()
+                                    .put("DO_NOT_UPDATE_STATS", "true")
+                                    .putAll(environmentContext)
+                                    .buildOrThrow());
                             client.alterTableWithEnvironmentContext(databaseName, tableName, table, context);
                         }
                         return null;

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/thrift/ThriftMetastore.java

@@ -57,7 +57,7 @@ public sealed interface ThriftMetastore
 
     void dropTable(String databaseName, String tableName, boolean deleteData);
 
-    void alterTable(String databaseName, String tableName, Table table);
+    void alterTable(String databaseName, String tableName, Table table, Map<String, String> environmentContext);
 
     void alterTransactionalTable(Table table, long transactionId, long writeId);
 

plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveMetadataListing.java

@@ -340,7 +340,7 @@ public void dropTable(String databaseName, String tableName, boolean deleteData)
         }
 
         @Override
-        public void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges)
+        public void replaceTable(String databaseName, String tableName, Table newTable, PrincipalPrivileges principalPrivileges, Map<String, String> environmentContext)
         {
             throw new UnsupportedOperationException();
         }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/file/FileMetastoreTableOperations.java

@@ -13,6 +13,7 @@
  */
 package io.trino.plugin.iceberg.catalog.file;
 
+import com.google.common.collect.ImmutableMap;
 import io.trino.annotation.NotThreadSafe;
 import io.trino.metastore.PrincipalPrivileges;
 import io.trino.metastore.Table;
@@ -87,7 +88,7 @@ private void commitTableUpdate(Table table, TableMetadata metadata, BiFunction<T
         PrincipalPrivileges privileges = table.getOwner().map(MetastoreUtil::buildInitialPrivilegeSet).orElse(NO_PRIVILEGES);
 
         try {
-            metastore.replaceTable(database, table.getTableName(), updatedTable, privileges);
+            metastore.replaceTable(database, table.getTableName(), updatedTable, privileges, ImmutableMap.of());
         }
         catch (RuntimeException e) {
             if (e instanceof TrinoException trinoException &&