Fix table listing in Iceberg to skip non-Iceberg tables

presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/HiveMetastore.java

@@ -45,6 +45,8 @@ public interface HiveMetastore
 
     List<String> getAllTables(String databaseName);
 
+    List<String> getTablesWithParameter(String databaseName, String parameterKey, String parameterValue);
+
     List<String> getAllViews(String databaseName);
 
     void createDatabase(Database database);

presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/RecordingHiveMetastore.java

@@ -73,6 +73,7 @@ public class RecordingHiveMetastore
     private final Cache<HiveTableName, PartitionStatistics> tableStatisticsCache;
     private final Cache<Set<HivePartitionName>, Map<String, PartitionStatistics>> partitionStatisticsCache;
     private final Cache<String, List<String>> allTablesCache;
+    private final Cache<TablesWithParameterCacheKey, List<String>> tablesWithParameterCache;
     private final Cache<String, List<String>> allViewsCache;
     private final Cache<HivePartitionName, Optional<Partition>> partitionCache;
     private final Cache<HiveTableName, Optional<List<String>>> partitionNamesCache;
@@ -96,6 +97,7 @@ public RecordingHiveMetastore(@ForRecordingHiveMetastore HiveMetastore delegate,
         tableStatisticsCache = createCache(hiveConfig);
         partitionStatisticsCache = createCache(hiveConfig);
         allTablesCache = createCache(hiveConfig);
+        tablesWithParameterCache = createCache(hiveConfig);
         allViewsCache = createCache(hiveConfig);
         partitionCache = createCache(hiveConfig);
         partitionNamesCache = createCache(hiveConfig);
@@ -123,6 +125,7 @@ void loadRecording()
         tableStatisticsCache.putAll(toMap(recording.getTableStatistics()));
         partitionStatisticsCache.putAll(toMap(recording.getPartitionStatistics()));
         allTablesCache.putAll(toMap(recording.getAllTables()));
+        tablesWithParameterCache.putAll(toMap(recording.getTablesWithParameter()));
         allViewsCache.putAll(toMap(recording.getAllViews()));
         partitionCache.putAll(toMap(recording.getPartitions()));
         partitionNamesCache.putAll(toMap(recording.getPartitionNames()));
@@ -161,6 +164,7 @@ public void writeRecording()
                 toPairs(tableStatisticsCache),
                 toPairs(partitionStatisticsCache),
                 toPairs(allTablesCache),
+                toPairs(tablesWithParameterCache),
                 toPairs(allViewsCache),
                 toPairs(partitionCache),
                 toPairs(partitionNamesCache),
@@ -253,6 +257,13 @@ public List<String> getAllTables(String databaseName)
         return loadValue(allTablesCache, databaseName, () -> delegate.getAllTables(databaseName));
     }
 
+    @Override
+    public List<String> getTablesWithParameter(String databaseName, String parameterKey, String parameterValue)
+    {
+        TablesWithParameterCacheKey key = new TablesWithParameterCacheKey(databaseName, parameterKey, parameterValue);
+        return loadValue(tablesWithParameterCache, key, () -> delegate.getTablesWithParameter(databaseName, parameterKey, parameterValue));
+    }
+
     @Override
     public List<String> getAllViews(String databaseName)
     {
@@ -502,6 +513,7 @@ public static class Recording
         private final List<Pair<HiveTableName, PartitionStatistics>> tableStatistics;
         private final List<Pair<Set<HivePartitionName>, Map<String, PartitionStatistics>>> partitionStatistics;
         private final List<Pair<String, List<String>>> allTables;
+        private final List<Pair<TablesWithParameterCacheKey, List<String>>> tablesWithParameter;
         private final List<Pair<String, List<String>>> allViews;
         private final List<Pair<HivePartitionName, Optional<Partition>>> partitions;
         private final List<Pair<HiveTableName, Optional<List<String>>>> partitionNames;
@@ -520,6 +532,7 @@ public Recording(
                 @JsonProperty("tableStatistics") List<Pair<HiveTableName, PartitionStatistics>> tableStatistics,
                 @JsonProperty("partitionStatistics") List<Pair<Set<HivePartitionName>, Map<String, PartitionStatistics>>> partitionStatistics,
                 @JsonProperty("allTables") List<Pair<String, List<String>>> allTables,
+                @JsonProperty("tablesWithParameter") List<Pair<TablesWithParameterCacheKey, List<String>>> tablesWithParameter,
                 @JsonProperty("allViews") List<Pair<String, List<String>>> allViews,
                 @JsonProperty("partitions") List<Pair<HivePartitionName, Optional<Partition>>> partitions,
                 @JsonProperty("partitionNames") List<Pair<HiveTableName, Optional<List<String>>>> partitionNames,
@@ -536,6 +549,7 @@ public Recording(
             this.tableStatistics = tableStatistics;
             this.partitionStatistics = partitionStatistics;
             this.allTables = allTables;
+            this.tablesWithParameter = tablesWithParameter;
             this.allViews = allViews;
             this.partitions = partitions;
             this.partitionNames = partitionNames;
@@ -569,6 +583,12 @@ public List<Pair<HiveTableName, Optional<Table>>> getTables()
             return tables;
         }
 
+        @JsonProperty
+        public List<Pair<TablesWithParameterCacheKey, List<String>>> getTablesWithParameter()
+        {
+            return tablesWithParameter;
+        }
+
         @JsonProperty
         public List<Pair<String, Set<ColumnStatisticType>>> getSupportedColumnStatistics()
         {

presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/TablesWithParameterCacheKey.java

@@ -0,0 +1,80 @@
+/*
+ * 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.prestosql.plugin.hive.metastore;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import javax.annotation.concurrent.Immutable;
+
+import java.util.Objects;
+
+@Immutable
+public class TablesWithParameterCacheKey
+{
+    private final String databaseName;
+    private final String parameterKey;
+    private final String parameterValue;
+
+    @JsonCreator
+    public TablesWithParameterCacheKey(
+            @JsonProperty("databaseName") String databaseName,
+            @JsonProperty("parameterKey") String parameterKey,
+            @JsonProperty("parameterValue") String parameterValue)
+    {
+        this.databaseName = databaseName;
+        this.parameterKey = parameterKey;
+        this.parameterValue = parameterValue;
+    }
+
+    @JsonProperty
+    public String getDatabaseName()
+    {
+        return databaseName;
+    }
+
+    @JsonProperty
+    public String getParameterKey()
+    {
+        return parameterKey;
+    }
+
+    @JsonProperty
+    public String getParameterValue()
+    {
+        return parameterValue;
+    }
+
+    @Override
+    public boolean equals(Object o)
+    {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+
+        TablesWithParameterCacheKey other = (TablesWithParameterCacheKey) o;
+        return Objects.equals(databaseName, other.databaseName) &&
+                Objects.equals(parameterKey, other.parameterKey) &&
+                Objects.equals(parameterValue, other.parameterValue);
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return Objects.hash(databaseName, parameterKey, parameterValue);
+    }
+}

presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/cache/CachingHiveMetastore.java

@@ -35,6 +35,7 @@
 import io.prestosql.plugin.hive.metastore.PartitionWithStatistics;
 import io.prestosql.plugin.hive.metastore.PrincipalPrivileges;
 import io.prestosql.plugin.hive.metastore.Table;
+import io.prestosql.plugin.hive.metastore.TablesWithParameterCacheKey;
 import io.prestosql.plugin.hive.metastore.UserTableKey;
 import io.prestosql.spi.PrestoException;
 import io.prestosql.spi.security.RoleGrant;
@@ -86,6 +87,7 @@ public class CachingHiveMetastore
     private final LoadingCache<String, List<String>> databaseNamesCache;
     private final LoadingCache<HiveTableName, Optional<Table>> tableCache;
     private final LoadingCache<String, List<String>> tableNamesCache;
+    private final LoadingCache<TablesWithParameterCacheKey, List<String>> tablesWithParameterCache;
     private final LoadingCache<HiveTableName, PartitionStatistics> tableStatisticsCache;
     private final LoadingCache<HivePartitionName, PartitionStatistics> partitionStatisticsCache;
     private final LoadingCache<String, List<String>> viewNamesCache;
@@ -141,6 +143,9 @@ private CachingHiveMetastore(HiveMetastore delegate, Executor executor, Optional
         tableNamesCache = newCacheBuilder(expiresAfterWriteMillis, refreshMills, maximumSize)
                 .build(asyncReloading(CacheLoader.from(this::loadAllTables), executor));
 
+        tablesWithParameterCache = newCacheBuilder(expiresAfterWriteMillis, refreshMills, maximumSize)
+                .build(asyncReloading(CacheLoader.from(this::loadTablesMatchingParameter), executor));
+
         tableStatisticsCache = newCacheBuilder(expiresAfterWriteMillis, refreshMills, maximumSize)
                 .build(asyncReloading(new CacheLoader<HiveTableName, PartitionStatistics>()
                 {
@@ -373,6 +378,18 @@ private List<String> loadAllTables(String databaseName)
         return delegate.getAllTables(databaseName);
     }
 
+    @Override
+    public List<String> getTablesWithParameter(String databaseName, String parameterKey, String parameterValue)
+    {
+        TablesWithParameterCacheKey key = new TablesWithParameterCacheKey(databaseName, parameterKey, parameterValue);
+        return get(tablesWithParameterCache, key);
+    }
+
+    private List<String> loadTablesMatchingParameter(TablesWithParameterCacheKey key)
+    {
+        return delegate.getTablesWithParameter(key.getDatabaseName(), key.getParameterKey(), key.getParameterValue());
+    }
+
     @Override
     public List<String> getAllViews(String databaseName)
     {

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

@@ -90,6 +90,7 @@
 import static io.prestosql.plugin.hive.HiveErrorCode.HIVE_PARTITION_DROPPED_DURING_QUERY;
 import static io.prestosql.plugin.hive.HiveMetadata.TABLE_COMMENT;
 import static io.prestosql.plugin.hive.HivePartitionManager.extractPartitionValues;
+import static io.prestosql.plugin.hive.HiveUtil.PRESTO_VIEW_FLAG;
 import static io.prestosql.plugin.hive.HiveUtil.toPartitionValues;
 import static io.prestosql.plugin.hive.metastore.HivePrivilegeInfo.HivePrivilege.OWNERSHIP;
 import static io.prestosql.plugin.hive.metastore.MetastoreUtil.makePartName;
@@ -398,19 +399,26 @@ public synchronized List<String> getAllTables(String databaseName)
     }
 
     @Override
-    public synchronized List<String> getAllViews(String databaseName)
+    public synchronized List<String> getTablesWithParameter(String databaseName, String parameterKey, String parameterValue)
     {
+        requireNonNull(parameterKey, "parameterKey is null");
+        requireNonNull(parameterValue, "parameterValue is null");
+
         List<String> tables = getAllTables(databaseName);
 
-        List<String> views = tables.stream()
+        return tables.stream()
                 .map(tableName -> getTable(databaseName, tableName))
                 .filter(Optional::isPresent)
                 .map(Optional::get)
-                .filter(table -> table.getTableType().equals(VIRTUAL_VIEW.name()))
+                .filter(table -> parameterValue.equals(table.getParameters().get(parameterKey)))
                 .map(Table::getTableName)
                 .collect(toImmutableList());
+    }
 
-        return views;
+    @Override
+    public synchronized List<String> getAllViews(String databaseName)
+    {
+        return getTablesWithParameter(databaseName, PRESTO_VIEW_FLAG, "true");
     }
 
     @Override

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

@@ -365,6 +365,13 @@ public List<String> getAllTables(String databaseName)
         }
     }
 
+    @Override
+    public synchronized List<String> getTablesWithParameter(String databaseName, String parameterKey, String parameterValue)
+    {
+        // TODO
+        throw new UnsupportedOperationException("getTablesWithParameter for GlueHiveMetastore is not implemented");
+    }
+
     @Override
     public List<String> getAllViews(String databaseName)
     {

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

@@ -125,6 +125,12 @@ public List<String> getAllTables(String databaseName)
         return delegate.getAllTables(databaseName);
     }
 
+    @Override
+    public List<String> getTablesWithParameter(String databaseName, String parameterKey, String parameterValue)
+    {
+        return delegate.getTablesWithParameter(databaseName, parameterKey, parameterValue);
+    }
+
     @Override
     public List<String> getAllViews(String databaseName)
     {

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

@@ -73,6 +73,7 @@
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Function;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 
 import static com.google.common.base.Preconditions.checkArgument;
@@ -122,6 +123,9 @@ public class ThriftHiveMetastore
     private final AtomicInteger chosenGetTableAlternative = new AtomicInteger(Integer.MAX_VALUE);
     private final AtomicInteger chosenTableParamAlternative = new AtomicInteger(Integer.MAX_VALUE);
 
+    private static final Pattern TABLE_PARAMETER_SAFE_KEY_PATTERN = Pattern.compile("^[a-zA-Z_]+$");
+    private static final Pattern TABLE_PARAMETER_SAFE_VALUE_PATTERN = Pattern.compile("^[a-zA-Z0-9]*$");
+
     @Inject
     public ThriftHiveMetastore(MetastoreLocator metastoreLocator, ThriftHiveMetastoreConfig thriftConfig)
     {
@@ -208,6 +212,27 @@ public List<String> getAllTables(String databaseName)
         }
     }
 
+    @Override
+    public List<String> getTablesWithParameter(String databaseName, String parameterKey, String parameterValue)
+    {
+        try {
+            return retry()
+                    .stopOn(UnknownDBException.class)
+                    .stopOnIllegalExceptions()
+                    .run("getTablesWithParameter", stats.getGetTablesWithParameter().wrap(
+                            () -> doGetTablesWithParameter(databaseName, parameterKey, parameterValue)));
+        }
+        catch (UnknownDBException e) {
+            return ImmutableList.of();
+        }
+        catch (TException e) {
+            throw new PrestoException(HIVE_METASTORE_ERROR, e);
+        }
+        catch (Exception e) {
+            throw propagate(e);
+        }
+    }
+
     @Override
     public Optional<Table> getTable(String databaseName, String tableName)
     {
@@ -718,7 +743,7 @@ public List<String> getAllViews(String databaseName)
                     .stopOn(UnknownDBException.class)
                     .stopOnIllegalExceptions()
                     .run("getAllViews", stats.getGetAllViews().wrap(
-                            () -> getPrestoViews(databaseName)));
+                            () -> doGetTablesWithParameter(databaseName, PRESTO_VIEW_FLAG, "true")));
         }
         catch (UnknownDBException e) {
             return ImmutableList.of();
@@ -731,16 +756,25 @@ public List<String> getAllViews(String databaseName)
         }
     }
 
-    private List<String> getPrestoViews(String databaseName)
+    private List<String> doGetTablesWithParameter(String databaseName, String parameterKey, String parameterValue)
             throws TException
     {
+        checkArgument(TABLE_PARAMETER_SAFE_KEY_PATTERN.matcher(parameterKey).matches(), "Parameter key contains invalid characters: '%s'", parameterKey);
+        /*
+         * The parameter value is restricted to have only alphanumeric characters so that it's safe
+         * to be used against HMS. When using with a LIKE operator, the HMS may want the parameter
+         * value to follow a Java regex pattern or a SQL pattern. And it's hard to predict the
+         * HMS's behavior from outside. Also, by restricting parameter values, we avoid the problem
+         * of how to quote them when passing within the filter string.
+         */
+        checkArgument(TABLE_PARAMETER_SAFE_VALUE_PATTERN.matcher(parameterValue).matches(), "Parameter value contains invalid characters: '%s'", parameterValue);
         /*
          * Thrift call `get_table_names_by_filter` may be translated by Metastore to a SQL query against Metastore database.
          * Hive 2.3 on some databases uses CLOB for table parameter value column and some databases disallow `=` predicate over
          * CLOB values. At the same time, they allow `LIKE` predicates over them.
          */
-        String filterWithEquals = HIVE_FILTER_FIELD_PARAMS + PRESTO_VIEW_FLAG + " = \"true\"";
-        String filterWithLike = HIVE_FILTER_FIELD_PARAMS + PRESTO_VIEW_FLAG + " LIKE \"true\"";
+        String filterWithEquals = HIVE_FILTER_FIELD_PARAMS + parameterKey + " = \"" + parameterValue + "\"";
+        String filterWithLike = HIVE_FILTER_FIELD_PARAMS + parameterKey + " LIKE \"" + parameterValue + "\"";
 
         return alternativeCall(
                 chosenTableParamAlternative,

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

@@ -54,6 +54,8 @@ public interface ThriftMetastore
 
     List<String> getAllTables(String databaseName);
 
+    List<String> getTablesWithParameter(String databaseName, String parameterKey, String parameterValue);
+
     List<String> getAllViews(String databaseName);
 
     Optional<Database> getDatabase(String databaseName);

presto-hive/src/main/java/io/prestosql/plugin/hive/metastore/thrift/ThriftMetastoreStats.java

@@ -21,6 +21,7 @@ public class ThriftMetastoreStats
     private final ThriftMetastoreApiStats getAllDatabases = new ThriftMetastoreApiStats();
     private final ThriftMetastoreApiStats getDatabase = new ThriftMetastoreApiStats();
     private final ThriftMetastoreApiStats getAllTables = new ThriftMetastoreApiStats();
+    private final ThriftMetastoreApiStats getTablesWithParameter = new ThriftMetastoreApiStats();
     private final ThriftMetastoreApiStats getAllViews = new ThriftMetastoreApiStats();
     private final ThriftMetastoreApiStats getTable = new ThriftMetastoreApiStats();
     private final ThriftMetastoreApiStats getFields = new ThriftMetastoreApiStats();
@@ -70,6 +71,13 @@ public ThriftMetastoreApiStats getGetAllTables()
         return getAllTables;
     }
 
+    @Managed
+    @Nested
+    public ThriftMetastoreApiStats getGetTablesWithParameter()
+    {
+        return getTablesWithParameter;
+    }
+
     @Managed
     @Nested
     public ThriftMetastoreApiStats getGetAllViews()