[iceberg] Reduce redundant getTable calls in IcebergHiveMetadata

Author: imjalpreet

presto-iceberg/src/main/java/com/facebook/presto/iceberg/IcebergHiveMetadata.java

@@ -85,6 +85,8 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.TimeZone;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.stream.Stream;
 
 import static com.facebook.presto.hive.HiveStatisticsUtil.createPartitionStatistics;
@@ -152,6 +154,7 @@ public class IcebergHiveMetadata
     private final HdfsEnvironment hdfsEnvironment;
     private final DateTimeZone timeZone = DateTimeZone.forTimeZone(TimeZone.getTimeZone(ZoneId.of(TimeZone.getDefault().getID())));
     private final IcebergHiveTableOperationsConfig hiveTableOeprationsConfig;
+    private final ConcurrentMap<SchemaTableName, Optional<Table>> tablesCache = new ConcurrentHashMap<>();
 
     public IcebergHiveMetadata(
             ExtendedHiveMetastore metastore,
@@ -191,8 +194,7 @@ protected View getIcebergView(ConnectorSession session, SchemaTableName schemaTa
     @Override
     protected boolean tableExists(ConnectorSession session, SchemaTableName schemaTableName)
     {
-        IcebergTableName name = IcebergTableName.from(schemaTableName.getTableName());
-        Optional<Table> hiveTable = metastore.getTable(getMetastoreContext(session), schemaTableName.getSchemaName(), name.getTableName());
+        Optional<Table> hiveTable = getHiveTable(session, schemaTableName);
         if (!hiveTable.isPresent()) {
             return false;
         }
@@ -202,6 +204,14 @@ protected boolean tableExists(ConnectorSession session, SchemaTableName schemaTa
         return true;
     }
 
+    private Optional<Table> getHiveTable(ConnectorSession session, SchemaTableName schemaTableName)
+    {
+        IcebergTableName name = IcebergTableName.from(schemaTableName.getTableName());
+        return tablesCache.computeIfAbsent(
+                schemaTableName,
+                ignored -> metastore.getTable(getMetastoreContext(session), schemaTableName.getSchemaName(), name.getTableName()));
+    }
+
     @Override
     public List<String> listSchemaNames(ConnectorSession session)
     {
@@ -369,7 +379,7 @@ public void createView(ConnectorSession session, ConnectorTableMetadata viewMeta
                 encodeViewData(viewData));
         PrincipalPrivileges principalPrivileges = buildInitialPrivilegeSet(session.getUser());
 
-        Optional<Table> existing = metastore.getTable(metastoreContext, viewName.getSchemaName(), viewName.getTableName());
+        Optional<Table> existing = getHiveTable(session, viewName);
         if (existing.isPresent()) {
             if (!replace || !isPrestoView(existing.get())) {
                 throw new ViewAlreadyExistsException(viewName);
@@ -413,7 +423,7 @@ public Map<SchemaTableName, ConnectorViewDefinition> getViews(ConnectorSession s
         }
         MetastoreContext metastoreContext = getMetastoreContext(session);
         for (SchemaTableName schemaTableName : tableNames) {
-            Optional<Table> table = metastore.getTable(metastoreContext, schemaTableName.getSchemaName(), schemaTableName.getTableName());
+            Optional<Table> table = getHiveTable(session, schemaTableName);
             if (table.isPresent() && isPrestoView(table.get())) {
                 verifyAndPopulateViews(table.get(), schemaTableName, decodeViewData(table.get().getViewOriginalText().get()), views);
             }
@@ -527,7 +537,7 @@ public void finishStatisticsCollection(ConnectorSession session, ConnectorTableH
     {
         IcebergTableHandle icebergTableHandle = (IcebergTableHandle) tableHandle;
         MetastoreContext metastoreContext = getMetastoreContext(session);
-        Table table = metastore.getTable(metastoreContext, icebergTableHandle.getSchemaTableName().getSchemaName(), icebergTableHandle.getSchemaTableName().getTableName())
+        Table table = getHiveTable(session, icebergTableHandle.getSchemaTableName())
                 .orElseThrow(() -> new TableNotFoundException(icebergTableHandle.getSchemaTableName()));
 
         List<Column> partitionColumns = table.getPartitionColumns();