Add impersonation to HiveMetadataFactory

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

@@ -192,7 +192,7 @@ public HiveMetadataFactory(
     public TransactionalMetadata create(ConnectorIdentity identity, boolean autoCommit)
     {
         HiveMetastoreClosure hiveMetastoreClosure = new HiveMetastoreClosure(
-                memoizeMetastore(metastoreFactory.createMetastore(), perTransactionCacheMaximumSize)); // per-transaction cache
+                memoizeMetastore(metastoreFactory.createMetastore(Optional.of(identity)), perTransactionCacheMaximumSize)); // per-transaction cache
 
         SemiTransactionalHiveMetastore metastore = new SemiTransactionalHiveMetastore(
                 hdfsEnvironment,

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

@@ -41,6 +41,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.OptionalInt;
 import java.util.Set;
 
@@ -156,7 +157,7 @@ private ConnectorPageSink createPageSink(HiveWritableTableHandle handle, boolean
                 session.getQueryId(),
                 new HivePageSinkMetadataProvider(
                         handle.getPageSinkMetadata(),
-                        new HiveMetastoreClosure(memoizeMetastore(metastoreFactory.createMetastore(), perTransactionMetastoreCacheMaximumSize)),
+                        new HiveMetastoreClosure(memoizeMetastore(metastoreFactory.createMetastore(Optional.of(session.getIdentity())), perTransactionMetastoreCacheMaximumSize)),
                         new HiveIdentity(session)),
                 typeManager,
                 hdfsEnvironment,

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

@@ -25,6 +25,7 @@
 import io.trino.plugin.hive.metastore.procedure.FlushHiveMetastoreCacheProcedure;
 import io.trino.plugin.hive.metastore.recording.RecordingHiveMetastoreDecoratorModule;
 import io.trino.spi.procedure.Procedure;
+import io.trino.spi.security.ConnectorIdentity;
 
 import java.util.Comparator;
 import java.util.List;
@@ -85,9 +86,15 @@ public DecoratingHiveMetastoreFactory(HiveMetastoreFactory delegate, Set<HiveMet
         }
 
         @Override
-        public HiveMetastore createMetastore()
+        public boolean isImpersonationEnabled()
         {
-            HiveMetastore metastore = delegate.createMetastore();
+            return delegate.isImpersonationEnabled();
+        }
+
+        @Override
+        public HiveMetastore createMetastore(Optional<ConnectorIdentity> identity)
+        {
+            HiveMetastore metastore = delegate.createMetastore(identity);
             for (HiveMetastoreDecorator decorator : sortedDecorators) {
                 metastore = decorator.decorate(metastore);
             }

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

@@ -13,11 +13,22 @@
  */
 package io.trino.plugin.hive.metastore;
 
+import io.trino.spi.security.ConnectorIdentity;
+
+import java.util.Optional;
+
 import static java.util.Objects.requireNonNull;
 
 public interface HiveMetastoreFactory
 {
-    HiveMetastore createMetastore();
+    boolean isImpersonationEnabled();
+
+    /**
+     * Create a metastore instance for the identity. An empty identity will
+     * only be provided when impersonation is disabled, and global caching is
+     * enabled.
+     */
+    HiveMetastore createMetastore(Optional<ConnectorIdentity> identity);
 
     static HiveMetastoreFactory ofInstance(HiveMetastore metastore)
     {
@@ -35,7 +46,13 @@ private StaticHiveMetastoreFactory(HiveMetastore metastore)
         }
 
         @Override
-        public HiveMetastore createMetastore()
+        public boolean isImpersonationEnabled()
+        {
+            return false;
+        }
+
+        @Override
+        public HiveMetastore createMetastore(Optional<ConnectorIdentity> identity)
         {
             return metastore;
         }

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/alluxio/AlluxioHiveMetastoreFactory.java

@@ -17,9 +17,12 @@
 import io.trino.plugin.hive.metastore.HiveMetastore;
 import io.trino.plugin.hive.metastore.HiveMetastoreFactory;
 import io.trino.plugin.hive.metastore.MetastoreConfig;
+import io.trino.spi.security.ConnectorIdentity;
 
 import javax.inject.Inject;
 
+import java.util.Optional;
+
 public class AlluxioHiveMetastoreFactory
         implements HiveMetastoreFactory
 {
@@ -33,7 +36,13 @@ public AlluxioHiveMetastoreFactory(TableMasterClient client, MetastoreConfig met
     }
 
     @Override
-    public HiveMetastore createMetastore()
+    public boolean isImpersonationEnabled()
+    {
+        return false;
+    }
+
+    @Override
+    public HiveMetastore createMetastore(Optional<ConnectorIdentity> identity)
     {
         return metastore;
     }

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/cache/SharedHiveMetastoreCache.java

@@ -18,6 +18,7 @@
 import io.trino.plugin.hive.metastore.HiveMetastore;
 import io.trino.plugin.hive.metastore.HiveMetastoreFactory;
 import io.trino.spi.NodeManager;
+import io.trino.spi.security.ConnectorIdentity;
 import org.weakref.jmx.Flatten;
 import org.weakref.jmx.Nested;
 
@@ -96,11 +97,12 @@ public HiveMetastoreFactory createCachingHiveMetastoreFactory(HiveMetastoreFacto
             return metastoreFactory;
         }
 
+        // caching hive metastore currently handles caching for multiple users internally
         CachingHiveMetastore cachingHiveMetastore = cachingHiveMetastore(
-                metastoreFactory.createMetastore(),
                 // Loading of cache entry in CachingHiveMetastore might trigger loading of another cache entry for different object type
                 // In case there are no empty executor slots, such operation would deadlock. Therefore, a reentrant executor needs to be
                 // used.
+                metastoreFactory.createMetastore(Optional.empty()),
                 new ReentrantBoundedExecutor(executorService, maxMetastoreRefreshThreads),
                 metastoreCacheTtl,
                 metastoreRefreshInterval,
@@ -119,7 +121,13 @@ private CachingHiveMetastoreFactory(CachingHiveMetastore metastore)
         }
 
         @Override
-        public HiveMetastore createMetastore()
+        public boolean isImpersonationEnabled()
+        {
+            return metastore.isImpersonationEnabled();
+        }
+
+        @Override
+        public HiveMetastore createMetastore(Optional<ConnectorIdentity> identity)
         {
             return metastore;
         }

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

@@ -18,9 +18,12 @@
 import io.trino.plugin.hive.metastore.HiveMetastore;
 import io.trino.plugin.hive.metastore.HiveMetastoreFactory;
 import io.trino.plugin.hive.metastore.MetastoreConfig;
+import io.trino.spi.security.ConnectorIdentity;
 
 import javax.inject.Inject;
 
+import java.util.Optional;
+
 public class FileHiveMetastoreFactory
         implements HiveMetastoreFactory
 {
@@ -34,7 +37,13 @@ public FileHiveMetastoreFactory(NodeVersion nodeVersion, HdfsEnvironment hdfsEnv
     }
 
     @Override
-    public HiveMetastore createMetastore()
+    public boolean isImpersonationEnabled()
+    {
+        return false;
+    }
+
+    @Override
+    public HiveMetastore createMetastore(Optional<ConnectorIdentity> identity)
     {
         return metastore;
     }

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

@@ -104,8 +104,6 @@
 import io.trino.spi.statistics.ColumnStatisticType;
 import io.trino.spi.type.Type;
 import org.apache.hadoop.fs.Path;
-import org.weakref.jmx.Flatten;
-import org.weakref.jmx.Managed;
 
 import javax.annotation.Nullable;
 import javax.inject.Inject;
@@ -269,8 +267,6 @@ private static AWSCredentialsProvider getCustomAWSCredentialsProvider(String pro
         }
     }
 
-    @Managed
-    @Flatten
     public GlueMetastoreStats getStats()
     {
         return stats;

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

@@ -15,11 +15,14 @@
 
 import io.trino.plugin.hive.metastore.HiveMetastore;
 import io.trino.plugin.hive.metastore.HiveMetastoreFactory;
+import io.trino.spi.security.ConnectorIdentity;
 import org.weakref.jmx.Flatten;
 import org.weakref.jmx.Managed;
 
 import javax.inject.Inject;
 
+import java.util.Optional;
+
 import static java.util.Objects.requireNonNull;
 
 public class GlueHiveMetastoreFactory
@@ -42,7 +45,13 @@ public GlueHiveMetastore getMetastore()
     }
 
     @Override
-    public HiveMetastore createMetastore()
+    public boolean isImpersonationEnabled()
+    {
+        return false;
+    }
+
+    @Override
+    public HiveMetastore createMetastore(Optional<ConnectorIdentity> identity)
     {
         return metastore;
     }

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

@@ -15,9 +15,12 @@
 
 import io.trino.plugin.hive.metastore.HiveMetastore;
 import io.trino.plugin.hive.metastore.HiveMetastoreFactory;
+import io.trino.spi.security.ConnectorIdentity;
 
 import javax.inject.Inject;
 
+import java.util.Optional;
+
 import static java.util.Objects.requireNonNull;
 
 public class BridgingHiveMetastoreFactory
@@ -32,7 +35,13 @@ public BridgingHiveMetastoreFactory(ThriftMetastore thriftMetastore)
     }
 
     @Override
-    public HiveMetastore createMetastore()
+    public boolean isImpersonationEnabled()
+    {
+        return thriftMetastore.isImpersonationEnabled();
+    }
+
+    @Override
+    public HiveMetastore createMetastore(Optional<ConnectorIdentity> identity)
     {
         return new BridgingHiveMetastore(thriftMetastore);
     }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/TrinoCatalogFactory.java

@@ -25,6 +25,8 @@
 
 import javax.inject.Inject;
 
+import java.util.Optional;
+
 import static io.trino.plugin.hive.metastore.cache.CachingHiveMetastore.memoizeMetastore;
 import static io.trino.plugin.iceberg.IcebergSecurityConfig.IcebergSecurity.SYSTEM;
 import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
@@ -75,7 +77,7 @@ public TrinoCatalog create(ConnectorIdentity identity)
             case HIVE_METASTORE:
                 return new TrinoHiveCatalog(
                         catalogName,
-                        memoizeMetastore(metastoreFactory.createMetastore(), 1000),
+                        memoizeMetastore(metastoreFactory.createMetastore(Optional.of(identity)), 1000),
                         hdfsEnvironment,
                         typeManager,
                         tableOperationsProvider,