IMPALA-12609: Implement SHOW METADATA TABLES IN statement to list Iceberg Metadata tables

After this change, the new SHOW METADATA TABLES IN statement can be used
to list all the available metadata tables of an Iceberg table.

Note that in contrast to querying the contents of Iceberg metadata tables,
this does not require fully qualified paths, e.g. both
  SHOW METADATA TABLES IN functional_parquet.iceberg_query_metadata;
and
  USE functional_parquet;
  SHOW METADATA TABLES IN iceberg_query_metadata;
work.

The available metadata tables for all Iceberg tables are the same,
corresponding to the values of the enum
"org.apache.iceberg.MetadataTableType", so there is actually no need to
pass the name of the regular table for which the metadata table list is
requested through Thrift. This change, however, does send the table name
because this way
 - if we add support for metadata tables for other table formats, the
   table name/path will be necessary to determine the correct list of
   metadata tables
 - we could later add support for different authorisation policies for
   individual tables
 - we can check also at the point of generating the list of metadata
   tables that the table is an Iceberg table

Testing:
 - added and updated tests in ParserTest, AnalyzeDDLTest, ToSqlTest and
   AuthorizationStmtTest
 - added a custom cluster test in test_authorization.py
 - added functional tests in iceberg-metadata-tables.test

Change-Id: Ide10ccf10fc0abf5c270119ba7092c67e712ec49
Reviewed-on: http://gerrit.cloudera.org:8080/21026
Tested-by: Impala Public Jenkins <impala-public-jenkins@cloudera.com>
Reviewed-by: Zoltan Borok-Nagy <boroknagyz@cloudera.com>

be/src/service/client-request-state.cc

@@ -366,15 +366,30 @@ Status ClientRequestState::ExecLocalCatalogOp(
       // A NULL pattern means match all tables of the specified table types. However,
       // Thrift string types can't be NULL in C++, so we have to test if it's set rather
       // than just blindly using the value.
-      const string* table_name =
+      const string* table_name_pattern =
-          params->__isset.show_pattern ? &(params->show_pattern) : NULL;
+          params->__isset.show_pattern ? &(params->show_pattern) : nullptr;
       TGetTablesResult table_names;
       const set<TImpalaTableType::type>& table_types = params->table_types;
-      RETURN_IF_ERROR(frontend_->GetTableNames(params->db, table_name,
+      RETURN_IF_ERROR(frontend_->GetTableNames(params->db, table_name_pattern,
           &query_ctx_.session, table_types, &table_names));
       SetResultSet(table_names.tables);
       return Status::OK();
     }
+    case TCatalogOpType::SHOW_METADATA_TABLES: {
+      const TShowTablesParams* params = &catalog_op.show_tables_params;
+      // A NULL pattern means match all tables of the specified table types. However,
+      // Thrift string types can't be NULL in C++, so we have to test if it's set rather
+      // than just blindly using the value.
+      const string* metadata_table_name_pattern =
+          params->__isset.show_pattern ? &(params->show_pattern) : nullptr;
+      DCHECK(params->__isset.tbl);
+      const string& table_name = params->tbl;
+      TGetTablesResult table_names;
+      RETURN_IF_ERROR(frontend_->GetMetadataTableNames(params->db, table_name,
+          metadata_table_name_pattern, &query_ctx_.session, &table_names));
+      SetResultSet(table_names.tables);
+      return Status::OK();
+    }
     case TCatalogOpType::SHOW_DBS: {
       const TShowDbsParams* params = &catalog_op.show_dbs_params;
       TGetDbsResult dbs;

be/src/service/frontend.cc

@@ -110,6 +110,7 @@ Frontend::Frontend() {
     {"updateExecutorMembership", "([B)V", &update_membership_id_},
     {"getCatalogMetrics", "()[B", &get_catalog_metrics_id_},
     {"getTableNames", "([B)[B", &get_table_names_id_},
+    {"getMetadataTableNames", "([B)[B", &get_metadata_table_names_id_},
     {"describeDb", "([B)[B", &describe_db_id_},
     {"describeTable", "([B)[B", &describe_table_id_},
     {"showCreateTable", "([B)Ljava/lang/String;", &show_create_table_id_},
@@ -217,11 +218,23 @@ Status Frontend::GetTableNames(const string& db, const string* pattern,
   TGetTablesParams params;
   params.__set_db(db);
   params.__set_table_types(table_types);
-  if (pattern != NULL) params.__set_pattern(*pattern);
+  if (pattern != nullptr) params.__set_pattern(*pattern);
-  if (session != NULL) params.__set_session(*session);
+  if (session != nullptr) params.__set_session(*session);
   return JniUtil::CallJniMethod(fe_, get_table_names_id_, params, table_names);
 }
 
+Status Frontend::GetMetadataTableNames(const string& db, const string& table_name,
+    const string* pattern, const TSessionState* session,
+    TGetTablesResult* metadata_table_names) {
+  TGetMetadataTablesParams params;
+  params.__set_db(db);
+  params.__set_tbl(table_name);
+  if (pattern != nullptr) params.__set_pattern(*pattern);
+  if (session != nullptr) params.__set_session(*session);
+  return JniUtil::CallJniMethod(fe_, get_metadata_table_names_id_, params,
+      metadata_table_names);
+}
+
 Status Frontend::GetDbs(const string* pattern, const TSessionState* session,
     TGetDbsResult* dbs) {
   TGetDbsParams params;

be/src/service/frontend.h

@@ -60,16 +60,16 @@ class Frontend {
   /// Get the metrics from the catalog used by this frontend.
   Status GetCatalogMetrics(TGetCatalogMetricsResult* resp);
 
-  /// Get all matching table names, per Hive's "SHOW TABLES <pattern>" regardless of the
+  /// Returns all matching table names, per Hive's "SHOW TABLES <pattern>" regardless of
-  /// table type.
+  /// the table type.
   Status GetTableNames(const std::string& db, const std::string* pattern,
       const TSessionState* session, TGetTablesResult* table_names);
 
   /// Returns all matching table names, per Hive's "SHOW TABLES <pattern>" such that each
   /// corresponds to a table whose type is in table_types for a non-empty table_types.
-  /// Each table name returned is unqualified.
+  /// Each table name returned is unqualified. If table_types is empty, then all types of
-  /// If table_types is empty, then all types of tables will be considered when their
+  /// tables will be considered when their table names are matched against the pattern.
-  /// table names are matched against the pattern.
+  ///
   /// If pattern is NULL, match all tables otherwise match only those tables that
   /// match the pattern string. Patterns are "p1|p2|p3" where | denotes choice,
   /// and each pN may contain wildcards denoted by '*' which match all strings.
@@ -82,6 +82,13 @@ class Frontend {
       const TSessionState* session, const std::set<TImpalaTableType::type>& table_types,
       TGetTablesResult* table_names);
 
+  /// Returns the list of metadata tables for the given table that match the pattern.
+  /// 'pattern' and 'session' are used as in GetTableNames(). Currently only Iceberg
+  /// metadata tables are supported.
+  Status GetMetadataTableNames(const string& db, const string& table_name,
+      const string* pattern, const TSessionState* session,
+      TGetTablesResult* metadata_table_names);
+
   /// Return all databases matching the optional argument 'pattern'.
   /// If pattern is NULL, match all databases otherwise match only those databases that
   /// match the pattern string. Patterns are "p1|p2|p3" where | denotes choice,
@@ -251,6 +258,7 @@ class Frontend {
   jmethodID update_membership_id_; // JniFrontend.updateExecutorMembership()
   jmethodID get_catalog_metrics_id_; // JniFrontend.getCatalogMetrics()
   jmethodID get_table_names_id_; // JniFrontend.getTableNames
+  jmethodID get_metadata_table_names_id_; // JniFrontend.getMetadataTableNames
   jmethodID describe_db_id_; // JniFrontend.describeDb
   jmethodID describe_table_id_; // JniFrontend.describeTable
   jmethodID show_create_table_id_; // JniFrontend.showCreateTable

be/src/service/impala-http-handler.cc

@@ -962,7 +962,7 @@ void ImpalaHttpHandler::CatalogHandler(const Webserver::WebRequest& req,
 
     TGetTablesResult get_table_results;
     status = server_->exec_env_->frontend()->GetTableNames(
-        db.db_name, NULL, NULL, &get_table_results);
+        db.db_name, nullptr, nullptr, &get_table_results);
     if (!status.ok()) {
       Value error(status.GetDetail().c_str(), document->GetAllocator());
       database.AddMember("error", error, document->GetAllocator());

common/thrift/Frontend.thrift

@@ -83,6 +83,22 @@ struct TGetTablesParams {
   4: optional set<CatalogService.TImpalaTableType> table_types = []
 }
 
+// Arguments to getMetadataTableNames, which returns the list of metadata tables of the
+// specified table.
+struct TGetMetadataTablesParams {
+  1: required string db
+
+  2: required string tbl
+
+  // If not set, match every table
+  3: optional string pattern
+
+  // Session state for the user who initiated this request. If authorization is
+  // enabled, only the tables this user has access to will be returned. If not
+  // set, access checks will be skipped (used for internal Impala requests)
+  4: optional Query.TSessionState session
+}
+
 // getTableNames returns a list of unqualified table names
 struct TGetTablesResult {
   1: list<string> tables
@@ -253,18 +269,21 @@ struct TShowFunctionsParams {
   3: optional string show_pattern
 }
 
-// Parameters for SHOW TABLES and SHOW VIEWS commands
+// Parameters for SHOW TABLES, SHOW METADATA TABLES and SHOW VIEWS commands
 struct TShowTablesParams {
   // Database to use for SHOW TABLES
   1: optional string db
 
+  // Set for querying the metadata tables of the given table.
+  2: optional string tbl
+
   // Optional pattern to match tables names. If not set, all tables from the given
   // database are returned.
-  2: optional string show_pattern
+  3: optional string show_pattern
 
   // This specifies the types of tables that should be returned. If not set, all types of
   // tables are considered when their names are matched against pattern.
-  3: optional set<CatalogService.TImpalaTableType> table_types = []
+  4: optional set<CatalogService.TImpalaTableType> table_types = []
 }
 
 // Parameters for SHOW FILES commands
@@ -441,6 +460,7 @@ enum TCatalogOpType {
   SHOW_CREATE_FUNCTION = 14
   DESCRIBE_HISTORY = 15
   SHOW_VIEWS = 16
+  SHOW_METADATA_TABLES = 17
 }
 
 // TODO: Combine SHOW requests with a single struct that contains a field

fe/src/main/cup/sql-parser.cup

@@ -381,6 +381,7 @@ nonterminal UseStmt use_stmt;
 nonterminal SetStmt set_stmt;
 nonterminal SetStmt unset_stmt;
 nonterminal ShowTablesStmt show_tables_stmt;
+nonterminal ShowMetadataTablesStmt show_metadata_tables_stmt;
 nonterminal ShowViewsStmt show_views_stmt;
 nonterminal ShowDbsStmt show_dbs_stmt;
 nonterminal ShowStatsStmt show_stats_stmt, show_partitions_stmt,
@@ -681,6 +682,8 @@ stmt ::=
   {: RESULT = use; :}
   | show_tables_stmt:show_tables
   {: RESULT = show_tables; :}
+  | show_metadata_tables_stmt:show_metadata_tables
+  {: RESULT = show_metadata_tables; :}
   | show_views_stmt:show_views
   {: RESULT = show_views; :}
   | show_dbs_stmt:show_dbs
@@ -2938,6 +2941,15 @@ show_tables_stmt ::=
   {: RESULT = new ShowTablesStmt(db, showPattern); :}
   ;
 
+show_metadata_tables_stmt ::=
+  KW_SHOW KW_METADATA KW_TABLES KW_IN table_name:tbl_name
+  {: RESULT = new ShowMetadataTablesStmt(tbl_name.getDb(), tbl_name.getTbl(), null); :}
+  | KW_SHOW KW_METADATA KW_TABLES KW_IN table_name:tbl_name
+      show_pattern:showPattern
+  {: RESULT = new ShowMetadataTablesStmt(
+      tbl_name.getDb(), tbl_name.getTbl(), showPattern); :}
+  ;
+
 show_views_stmt ::=
   KW_SHOW KW_VIEWS
   {: RESULT = new ShowViewsStmt(); :}

fe/src/main/java/org/apache/impala/analysis/AnalysisContext.java

@@ -119,6 +119,9 @@ public class AnalysisContext {
     public boolean isUseStmt() { return stmt_ instanceof UseStmt; }
     public boolean isSetStmt() { return stmt_ instanceof SetStmt; }
     public boolean isShowTablesStmt() { return stmt_ instanceof ShowTablesStmt; }
+    public boolean isShowMetadataTablesStmt() {
+      return stmt_ instanceof ShowMetadataTablesStmt;
+    }
     public boolean isShowViewsStmt() { return stmt_ instanceof ShowViewsStmt; }
     public boolean isDescribeHistoryStmt() {
       return stmt_ instanceof DescribeHistoryStmt;
@@ -177,9 +180,9 @@ public class AnalysisContext {
     }
 
     private boolean isViewMetadataStmt() {
-      return isShowFilesStmt() || isShowTablesStmt() || isShowViewsStmt() ||
+      return isShowFilesStmt() || isShowTablesStmt() || isShowMetadataTablesStmt() ||
-          isShowDbsStmt() || isShowFunctionsStmt() || isShowRolesStmt() ||
+          isShowViewsStmt() || isShowDbsStmt() || isShowFunctionsStmt() ||
-          isShowGrantPrincipalStmt() || isShowCreateTableStmt() ||
+          isShowRolesStmt() || isShowGrantPrincipalStmt() || isShowCreateTableStmt() ||
           isShowDataSrcsStmt() || isShowStatsStmt() || isDescribeTableStmt() ||
           isDescribeDbStmt() || isShowCreateFunctionStmt() || isDescribeHistoryStmt();
     }
@@ -211,8 +214,8 @@ public class AnalysisContext {
      */
     public boolean isSingleColumnPrivStmt() {
       return isDescribeTableStmt() || isResetMetadataStmt() || isUseStmt()
-          || isShowTablesStmt() || isShowViewsStmt() || isAlterTableStmt()
+          || isShowTablesStmt() || isShowMetadataTablesStmt() || isShowViewsStmt()
-          || isShowFunctionsStmt();
+          || isAlterTableStmt() || isShowFunctionsStmt();
     }
 
     public boolean isConvertTableToIcebergStmt() {
@@ -328,6 +331,11 @@ public class AnalysisContext {
       return (ShowTablesStmt) stmt_;
     }
 
+    public ShowMetadataTablesStmt getShowMetadataTablesStmt() {
+      Preconditions.checkState(isShowMetadataTablesStmt());
+      return (ShowMetadataTablesStmt) stmt_;
+    }
+
     public ShowViewsStmt getShowViewsStmt() {
       Preconditions.checkState(isShowViewsStmt());
       return (ShowViewsStmt) stmt_;

fe/src/main/java/org/apache/impala/analysis/ShowMetadataTablesStmt.java

@@ -0,0 +1,119 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 org.apache.impala.analysis;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.impala.analysis.Path.PathType;
+import org.apache.impala.catalog.FeTable;
+import org.apache.impala.catalog.FeIcebergTable;
+import org.apache.impala.catalog.TableLoadingException;
+import org.apache.impala.common.AnalysisException;
+import org.apache.impala.thrift.TShowTablesParams;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Representation of a SHOW METADATA TABLES [pattern] statement.
+ *
+ * This statement queries the list of metadata tables available for a table. This is
+ * currently only supported for Iceberg tables:
+ *
+ * Acceptable syntax:
+ *
+ * SHOW METADATA TABLES IN [database.]table
+ * SHOW METADATA TABLES IN [database.]table "pattern"
+ * SHOW METADATA TABLES IN [database.]table LIKE "pattern"
+ *
+ * As in Hive, the 'LIKE' is optional. In Hive, also SHOW TABLES unquotedpattern is
+ * accepted by the parser but returns no results. We don't support that syntax.
+ */
+public class ShowMetadataTablesStmt extends ShowTablesOrViewsStmt {
+  private final String tbl_;
+
+  public ShowMetadataTablesStmt(String database, String tbl, String pattern) {
+    super(database, pattern);
+    Preconditions.checkNotNull(tbl);
+    tbl_ = tbl;
+  }
+
+  @Override
+  public void collectTableRefs(List<TableRef> tblRefs) {
+    List<String> rawPath = createRawPath();
+    tblRefs.add(new TableRef(rawPath, null));
+  }
+
+  @Override
+  public void analyze(Analyzer analyzer) throws AnalysisException {
+    Preconditions.checkNotNull(tbl_);
+    super.analyze(analyzer);
+
+    String db = getDb();
+    analyzer.registerPrivReq(builder ->
+        builder.onTableUnknownOwner(db, tbl_)
+        .any()
+        .build());
+
+    Path resolvedPath;
+    try {
+      final List<String> rawPath = createRawPath();
+      resolvedPath = analyzer.resolvePath(rawPath, PathType.ANY);
+    } catch (TableLoadingException tle) {
+      throw new AnalysisException(tle.getMessage(), tle);
+    }
+
+    FeTable table = resolvedPath.getRootTable();
+    if (!(table instanceof FeIcebergTable)) {
+      throw new AnalysisException(
+          "The SHOW METADATA TABLES statement is only valid for Iceberg tables: '" + db +
+          "." + tbl_ + "' is not an Iceberg table.");
+    }
+  }
+
+  @Override
+  public TShowTablesParams toThrift() {
+    TShowTablesParams params = super.toThrift();
+    params.setTbl(tbl_);
+    return params;
+  }
+
+  @Override
+  public String toSql(ToSqlOptions options) {
+    StringBuilder sb = new StringBuilder("SHOW METADATA TABLES IN ");
+
+    String parsedDb = getParsedDb();
+    if (parsedDb != null) sb.append(parsedDb).append(".");
+
+    sb.append(tbl_);
+
+    String pattern = getPattern();
+    if (pattern != null) sb.append(" LIKE '").append(pattern).append("'");
+
+    return sb.toString();
+  }
+
+  private List<String> createRawPath() {
+    List<String> res = new ArrayList<>();
+    final String dbName = isAnalyzed() ? getDb() : getParsedDb();
+    if (dbName != null) res.add(dbName);
+    res.add(tbl_);
+    return res;
+  }
+}

fe/src/main/java/org/apache/impala/analysis/ShowTablesStmt.java

@@ -28,8 +28,8 @@ package org.apache.impala.analysis;
  * SHOW TABLES IN database "pattern"
  * SHOW TABLES IN database LIKE "pattern"
  *
- * In Hive, the 'LIKE' is optional. Also SHOW TABLES unquotedpattern is accepted
+ * As in Hive, the 'LIKE' is optional. In Hive, also SHOW TABLES unquotedpattern is
- * by the parser but returns no results. We don't support that syntax.
+ * accepted by the parser but returns no results. We don't support that syntax.
  */
 public class ShowTablesStmt extends ShowTablesOrViewsStmt {
   public ShowTablesStmt() { super(null, null); }
@@ -40,18 +40,18 @@ public class ShowTablesStmt extends ShowTablesOrViewsStmt {
 
   @Override
   public String toSql(ToSqlOptions options) {
-    if (getPattern() == null) {
+    StringBuilder sb = new StringBuilder("SHOW TABLES");
-      if (getParsedDb() == null) {
+
-        return "SHOW TABLES";
+    String parsedDb = getParsedDb();
-      } else {
+    if (parsedDb != null) {
-        return "SHOW TABLES IN " + getParsedDb();
+      sb.append(" IN ").append(parsedDb);
-      }
-    } else {
-      if (getParsedDb() == null) {
-        return "SHOW TABLES LIKE '" + getPattern() + "'";
-      } else {
-        return "SHOW TABLES IN " + getParsedDb() + " LIKE '" + getPattern() + "'";
-      }
     }
+
+    String pattern = getPattern();
+    if (pattern != null) {
+      sb.append(" LIKE '").append(pattern).append("'");
+    }
+
+    return sb.toString();
   }
 }

fe/src/main/java/org/apache/impala/service/Frontend.java

@@ -63,6 +63,7 @@ import org.apache.hadoop.hive.metastore.api.LockType;
 import org.apache.hadoop.hive.metastore.api.SQLForeignKey;
 import org.apache.hadoop.hive.metastore.api.SQLPrimaryKey;
 import org.apache.iceberg.HistoryEntry;
+import org.apache.iceberg.MetadataTableType;
 import org.apache.iceberg.Snapshot;
 import org.apache.iceberg.Table;
 import org.apache.impala.analysis.AlterDbStmt;
@@ -588,6 +589,11 @@ public class Frontend {
       ddl.setShow_tables_params(analysis.getShowTablesStmt().toThrift());
       metadata.setColumns(Arrays.asList(
           new TColumn("name", Type.STRING.toThrift())));
+    } else if (analysis.isShowMetadataTablesStmt()) {
+      ddl.op_type = TCatalogOpType.SHOW_METADATA_TABLES;
+      ddl.setShow_tables_params(analysis.getShowMetadataTablesStmt().toThrift());
+      metadata.setColumns(Arrays.asList(
+          new TColumn("name", Type.STRING.toThrift())));
     } else if (analysis.isShowViewsStmt()) {
       ddl.op_type = TCatalogOpType.SHOW_VIEWS;
       ddl.setShow_tables_params(analysis.getShowViewsStmt().toThrift());
@@ -1148,17 +1154,34 @@ public class Frontend {
     return getTableNames(dbName, matcher, user, /*tableTypes*/ Collections.emptySet());
   }
 
-  /**
+  /** Returns the names of the tables of types specified in 'tableTypes' in database
-   * Returns tables of types specified in 'tableTypes' in database 'dbName' that
+   * 'dbName' that are accessible to 'user'. Only tables that match the pattern of
-   * match the pattern of 'matcher' and are accessible to 'user'.
+   * 'matcher' are returned.
    */
-  public List<String> getTableNames(String dbName, PatternMatcher matcher,
+  public List<String> getTableNames(String dbName, PatternMatcher matcher, User user,
-      User user, Set<TImpalaTableType> tableTypes) throws ImpalaException {
+      Set<TImpalaTableType> tableTypes) throws ImpalaException {
     RetryTracker retries = new RetryTracker(
         String.format("fetching %s table names", dbName));
     while (true) {
       try {
-        return doGetTableNames(dbName, matcher, user, tableTypes);
+        return doGetCatalogTableNames(dbName, matcher, user, tableTypes);
+      } catch(InconsistentMetadataFetchException e) {
+        retries.handleRetryOrThrow(e);
+      }
+    }
+  }
+
+  /** Returns the metadata tables available for the given table. Currently only Iceberg
+   * metadata tables are supported. Only tables that match the pattern of 'matcher' are
+   * returned.
+   */
+  public List<String> getMetadataTableNames(String dbName, String tblName,
+      PatternMatcher matcher, User user) throws ImpalaException {
+    RetryTracker retries = new RetryTracker(
+        String.format("fetching %s table names to query metadata table list", dbName));
+    while (true) {
+      try {
+        return doGetMetadataTableNames(dbName, tblName, matcher, user);
       } catch(InconsistentMetadataFetchException e) {
         retries.handleRetryOrThrow(e);
       }
@@ -1193,12 +1216,8 @@ public class Frontend {
           "Check the server log for more details.");
   }
 
-  private List<String> doGetTableNames(String dbName, PatternMatcher matcher,
+  private void filterTablesIfAuthNeeded(String dbName, User user, List<String> tblNames)
-      User user, Set<TImpalaTableType> tableTypes)
       throws ImpalaException {
-    FeCatalog catalog = getCatalog();
-    List<String> tblNames = catalog.getTableNames(dbName, matcher, tableTypes);
-
     boolean needsAuthChecks = authzFactory_.getAuthorizationConfig().isEnabled()
                               && !userHasAccessForWholeDb(user, dbName);
 
@@ -1214,7 +1233,7 @@ public class Frontend {
         // 'owned' tables for a given user just because the metadata is not loaded.
         // TODO(IMPALA-8937): Figure out a way to load Table/Database ownership
         // information when fetching the table lists from HMS.
-        FeTable table = catalog.getTableIfCached(dbName, tblName);
+        FeTable table = getCatalog().getTableIfCached(dbName, tblName);
         String tableOwner = table.getOwnerUser();
         if (tableOwner == null) {
           LOG.info("Table {} not yet loaded, ignoring it in table listing.",
@@ -1226,10 +1245,39 @@ public class Frontend {
 
       filterUnaccessibleElements(pendingCheckTasks, tblNames);
     }
+  }
+
+  private List<String> doGetCatalogTableNames(String dbName, PatternMatcher matcher,
+      User user, Set<TImpalaTableType> tableTypes) throws ImpalaException {
+    FeCatalog catalog = getCatalog();
+    List<String> tblNames = catalog.getTableNames(dbName, matcher, tableTypes);
+
+    filterTablesIfAuthNeeded(dbName, user, tblNames);
 
     return tblNames;
   }
 
+  private List<String> doGetMetadataTableNames(String dbName, String catalogTblName,
+      PatternMatcher matcher, User user)
+      throws ImpalaException {
+    FeTable catalogTbl = getCatalog().getTable(dbName, catalogTblName);
+    // Analysis ensures that only Iceberg tables are passed to this function.
+    Preconditions.checkState(catalogTbl instanceof FeIcebergTable);
+
+    List<String> listToFilter = new ArrayList<>();
+    listToFilter.add(catalogTblName);
+
+    filterTablesIfAuthNeeded(dbName, user, listToFilter);
+    if (listToFilter.isEmpty()) return Collections.emptyList();
+
+    List<String> metadataTblNames = Arrays.stream(MetadataTableType.values())
+      .map(tblType -> tblType.toString().toLowerCase())
+      .collect(Collectors.toList());
+    List<String> filteredMetadataTblNames = Catalog.filterStringsByPattern(
+        metadataTblNames, matcher);
+    return filteredMetadataTblNames;
+  }
+
   /**
    * Returns a list of columns of a table using 'matcher' and are accessible
    * to the given user.

fe/src/main/java/org/apache/impala/service/JniFrontend.java

@@ -78,6 +78,7 @@ import org.apache.impala.thrift.TGetHadoopConfigResponse;
 import org.apache.impala.thrift.TGetHadoopGroupsRequest;
 import org.apache.impala.thrift.TGetHadoopGroupsResponse;
 import org.apache.impala.thrift.TGetTableHistoryResult;
+import org.apache.impala.thrift.TGetMetadataTablesParams;
 import org.apache.impala.thrift.TGetTablesParams;
 import org.apache.impala.thrift.TGetTablesResult;
 import org.apache.impala.thrift.TLoadDataReq;
@@ -94,6 +95,7 @@ import org.apache.impala.thrift.TShowRolesParams;
 import org.apache.impala.thrift.TShowStatsOp;
 import org.apache.impala.thrift.TShowStatsParams;
 import org.apache.impala.thrift.TDescribeHistoryParams;
+import org.apache.impala.thrift.TSessionState;
 import org.apache.impala.thrift.TTableName;
 import org.apache.impala.thrift.TUniqueId;
 import org.apache.impala.thrift.TUpdateCatalogCacheRequest;
@@ -256,10 +258,11 @@ public class JniFrontend {
   }
 
   /**
-   * Implement Hive's pattern-matching semantics for "SHOW TABLE [[LIKE] 'pattern']", and
+   * Returns a list of table names matching an optional pattern.
-   * return a list of table names matching an optional pattern.
+   *
-   * The only metacharacters are '*' which matches any string of characters, and '|'
+   * Implements Hive's pattern-matching semantics for "SHOW TABLE [[LIKE] 'pattern']". The
-   * which denotes choice.  Doing the work here saves loading tables or databases from the
+   * only metacharacters are '*' which matches any string of characters, and '|' which
+   * denotes choice. Doing the work here saves loading tables or databases from the
    * metastore (which Hive would do if we passed the call through to the metastore
    * client). If the pattern is null, all strings are considered to match. If it is an
    * empty string, no strings match.
@@ -272,10 +275,9 @@ public class JniFrontend {
     Preconditions.checkNotNull(frontend_);
     TGetTablesParams params = new TGetTablesParams();
     JniUtil.deserializeThrift(protocolFactory_, params, thriftGetTablesParams);
-    // If the session was not set it indicates this is an internal Impala call.
+
-    User user = params.isSetSession() ?
+    TSessionState session = params.isSetSession() ? params.getSession() : null;
-        new User(TSessionStateUtil.getEffectiveUser(params.getSession())) :
+    User user = getUser(session);
-        ImpalaInternalAdminUser.getInstance();
 
     Preconditions.checkState(!params.isSetSession() || user != null );
     List<String> tables = frontend_.getTableNames(params.db,
@@ -293,6 +295,40 @@ public class JniFrontend {
     }
   }
 
+  /**
+   * Returns the metadata tables available for the given table. Currently only Iceberg
+   * metadata tables are supported.
+   *
+   * Pattern matching is done as in getTableNames().
+   *
+   * The argument is a serialized TGetMetadataTablesParams object.
+   * The return type is a serialised TGetTablesResult object.
+   * @see Frontend#getTableNames
+   */
+  public byte[] getMetadataTableNames(byte[] thriftGetMetadataTablesParams)
+      throws ImpalaException {
+    Preconditions.checkNotNull(frontend_);
+    TGetMetadataTablesParams params = new TGetMetadataTablesParams();
+    JniUtil.deserializeThrift(protocolFactory_, params, thriftGetMetadataTablesParams);
+
+    TSessionState session = params.isSetSession() ? params.getSession() : null;
+    User user = getUser(session);
+
+    Preconditions.checkState(!params.isSetSession() || user != null );
+    List<String> tables = frontend_.getMetadataTableNames(params.db, params.tbl,
+        PatternMatcher.createHivePatternMatcher(params.pattern), user);
+
+    TGetTablesResult result = new TGetTablesResult();
+    result.setTables(tables);
+
+    try {
+      TSerializer serializer = new TSerializer(protocolFactory_);
+      return serializer.serialize(result);
+    } catch (TException e) {
+      throw new InternalException(e.getMessage());
+    }
+  }
+
   /**
    * Returns files info of a table or partition.
    * The argument is a serialized TShowFilesParams object.
@@ -326,10 +362,10 @@ public class JniFrontend {
     Preconditions.checkNotNull(frontend_);
     TGetDbsParams params = new TGetDbsParams();
     JniUtil.deserializeThrift(protocolFactory_, params, thriftGetTablesParams);
-    // If the session was not set it indicates this is an internal Impala call.
+
-    User user = params.isSetSession() ?
+    TSessionState session = params.isSetSession() ? params.getSession() : null;
-        new User(TSessionStateUtil.getEffectiveUser(params.getSession())) :
+    User user = getUser(session);
-        ImpalaInternalAdminUser.getInstance();
+
     List<? extends FeDb> dbs = frontend_.getDbs(
         PatternMatcher.createHivePatternMatcher(params.pattern), user);
     TGetDbsResult result = new TGetDbsResult();
@@ -923,6 +959,14 @@ public class JniFrontend {
     return output.toString();
   }
 
+  private User getUser(TSessionState session) {
+    // If the session was not set it indicates this is an internal Impala call.
+    User user = session == null ?
+        ImpalaInternalAdminUser.getInstance() :
+        new User(TSessionStateUtil.getEffectiveUser(session));
+    return user;
+  }
+
   /**
    * Return an empty string if the default FileSystem configured in CONF refers to a
    * DistributedFileSystem and Impala can list the root directory "/". Otherwise,

fe/src/test/java/org/apache/impala/analysis/AnalyzeDDLTest.java

@@ -4106,6 +4106,17 @@ public class AnalyzeDDLTest extends FrontendTestBase {
     AnalyzesOk("show tables");
     AnalyzesOk("show tables like '*pattern'");
 
+    AnalyzesOk("show tables in functional");
+    AnalyzesOk("show tables in functional like '*pattern'");
+
+    AnalyzesOk("show metadata tables in functional_parquet.iceberg_query_metadata");
+    AnalyzesOk(
+        "show metadata tables in functional_parquet.iceberg_query_metadata like 'e*|f*'");
+
+    AnalysisError("show metadata tables in functional_parquet.alltypes",
+        "The SHOW METADATA TABLES statement is only valid for Iceberg tables: " +
+        "'functional_parquet.alltypes' is not an Iceberg table.");
+
     for (String fnType: new String[]{"", "aggregate", "analytic"}) {
       AnalyzesOk(String.format("show %s functions", fnType));
       AnalyzesOk(String.format("show %s functions like '*pattern'", fnType));

fe/src/test/java/org/apache/impala/analysis/ParserTest.java

@@ -1994,6 +1994,13 @@ public class ParserTest extends FrontendTestBase {
     // Empty pattern ok
     ParsesOk("SHOW TABLES ''");
     ParsesOk("SHOW VIEWS ''");
+    // Querying metadata tables: SHOW METADATA TABLES IN db.tbl
+    ParsesOk("SHOW METADATA TABLES IN tbl");
+    ParsesOk("SHOW METADATA TABLES IN tbl 'e*'");
+    ParsesOk("SHOW METADATA TABLES IN tbl LIKE 'e*'");
+    ParsesOk("SHOW METADATA TABLES IN db.tbl");
+    ParsesOk("SHOW METADATA TABLES IN db.tbl 'e*'");
+    ParsesOk("SHOW METADATA TABLES IN db.tbl LIKE 'e*'");
     // Databases
     ParsesOk("SHOW DATABASES");
     ParsesOk("SHOW SCHEMAS");
@@ -2044,6 +2051,10 @@ public class ParserTest extends FrontendTestBase {
     // Malformed pattern (no quotes)
     ParserError("SHOW TABLES tablename");
     ParserError("SHOW VIEWS tablename");
+    // Missing keyword METADATA when listing metadata tables
+    ParserError("SHOW TABLES IN db.tbl");
+    // Trying to list the metadata tables of a metadata table
+    ParserError("SHOW METADATA TABLES IN db.tbl.files");
     // Invalid SHOW DATA SOURCE statements
     ParserError("SHOW DATA");
     ParserError("SHOW SOURCE");

fe/src/test/java/org/apache/impala/analysis/ToSqlTest.java

@@ -1897,7 +1897,7 @@ public class ToSqlTest extends FrontendTestBase {
   }
 
   /**
-   * Test SHOW TABLES statements are output correctly.
+   * Tests that SHOW TABLES statements are output correctly.
    */
   @Test
   public void testShowTables() {
@@ -1908,6 +1908,21 @@ public class ToSqlTest extends FrontendTestBase {
     testToSql("SHOW TABLES IN functional LIKE 'alltypes*'");
   }
 
+  /**
+   * Tests that SHOW METADATA TABLES statements are output correctly.
+   */
+  @Test
+  public void testShowMetadataTables() {
+    String q1 = "SHOW METADATA TABLES IN iceberg_query_metadata";
+    testToSql(q1, "functional_parquet", q1);
+    String q2 = "SHOW METADATA TABLES IN iceberg_query_metadata LIKE '*file*'";
+    testToSql(q2, "functional_parquet", q2);
+
+    testToSql("SHOW METADATA TABLES IN functional_parquet.iceberg_query_metadata");
+    testToSql("SHOW METADATA TABLES IN functional_parquet.iceberg_query_metadata " +
+        "LIKE '*file*'");
+  }
+
   /**
    * Test SHOW VIEWS statements are output correctly.
    */

fe/src/test/java/org/apache/impala/authorization/AuthorizationStmtTest.java

@@ -1253,6 +1253,18 @@ public class AuthorizationStmtTest extends AuthorizationTestBase {
     }
     test.error(accessError("functional.*.*"));
 
+    // Show metadata tables in table.
+    test = authorize("show metadata tables in functional_parquet.iceberg_query_metadata");
+    test.error(accessError("functional_parquet"));
+
+    for (TPrivilegeLevel privilege: allExcept(TPrivilegeLevel.RWSTORAGE)) {
+      // Test that even if we have privileges on a different table in the same db, we
+      // still can't access 'iceberg_query_metadata' if we don't have privileges on it.
+      test.error(accessError("functional_parquet.iceberg_query_metadata"),
+          onTable("functional_parquet", "alltypes", privilege));
+      test.ok(onTable("functional_parquet", "iceberg_query_metadata", privilege));
+    }
+
     // Show views.
     test = authorize("show views in functional");
     // We exclude TPrivilegeLevel.RWSTORAGE because of the same reason mentioned above.

testdata/datasets/functional/functional_schema_template.sql

@@ -3885,6 +3885,14 @@ DELETE FROM {db_name}{db_suffix}.{table_name} WHERE i = 2;
 ---- DATASET
 functional
 ---- BASE_TABLE_NAME
+iceberg_view
+---- CREATE
+CREATE VIEW {db_name}{db_suffix}.{table_name} AS
+SELECT * FROM  {db_name}{db_suffix}.iceberg_query_metadata;
+====
+---- DATASET
+functional
+---- BASE_TABLE_NAME
 iceberg_lineitem_multiblock
 ---- CREATE
 CREATE EXTERNAL TABLE IF NOT EXISTS {db_name}{db_suffix}.{table_name}

testdata/workloads/functional-query/queries/QueryTest/iceberg-metadata-tables.test

@@ -971,3 +971,35 @@ describe functional_parquet.iceberg_query_metadata.all_entries;
 ---- TYPES
 STRING,STRING,STRING,STRING
 ====
+---- QUERY
+show metadata tables in functional_parquet.iceberg_query_metadata;
+---- RESULTS
+'all_data_files'
+'all_delete_files'
+'all_entries'
+'all_files'
+'all_manifests'
+'data_files'
+'delete_files'
+'entries'
+'files'
+'history'
+'manifests'
+'metadata_log_entries'
+'partitions'
+'position_deletes'
+'refs'
+'snapshots'
+---- TYPES
+STRING
+====
+---- QUERY
+show metadata tables in functional_parquet.alltypestiny;
+---- CATCH
+AnalysisException: The SHOW METADATA TABLES statement is only valid for Iceberg tables: 'functional_parquet.alltypestiny' is not an Iceberg table.
+====
+---- QUERY
+show metadata tables in functional_parquet.iceberg_view;
+---- CATCH
+AnalysisException: The SHOW METADATA TABLES statement is only valid for Iceberg tables: 'functional_parquet.iceberg_view' is not an Iceberg table.
+====

tests/authorization/test_authorization.py

@@ -101,6 +101,47 @@ class TestAuthorization(CustomClusterTestSuite):
     assert_file_in_dir_contains(self.impala_log_dir, "Ignoring removed flag "
                                                      "authorization_policy_file")
 
+  @pytest.mark.execute_serially
+  @CustomClusterTestSuite.with_args(
+    impalad_args="--server-name=server1 --ranger_service_type=hive "
+                 "--ranger_app_id=impala --authorization_provider=ranger "
+                 "--min_privilege_set_for_show_stmts=select",
+    catalogd_args="--server-name=server1 --ranger_service_type=hive "
+                  "--ranger_app_id=impala --authorization_provider=ranger")
+  def test_ranger_show_iceberg_metadata_tables(self, unique_name):
+    unique_db = unique_name + "_db"
+    tbl_name = "ice"
+    full_tbl_name = "{}.{}".format(unique_db, tbl_name)
+    non_existent_suffix = "_a"
+    non_existent_tbl_name = full_tbl_name + non_existent_suffix
+    priv = "select"
+    user = getuser()
+    query = "show metadata tables in {}".format(full_tbl_name)
+    query_non_existent = "show metadata tables in {}".format(non_existent_tbl_name)
+    admin_client = self.create_impala_client()
+    try:
+      admin_client.execute("create database {}".format(unique_db), user=ADMIN)
+      admin_client.execute(
+          "create table {} (i int) stored as iceberg".format(full_tbl_name), user=ADMIN)
+
+      # Check that when the user has no privileges on the database, the error is the same
+      # if we try an existing or a non-existing table.
+      exc1_str = str(self.execute_query_expect_failure(self.client, query, user=user))
+      exc2_str = str(self.execute_query_expect_failure(self.client, query_non_existent,
+          user=user))
+      assert exc1_str == exc2_str
+      assert "AuthorizationException" in exc1_str
+      assert "does not have privileges to access"
+
+      # Check that there is no error when the user has access to the table.
+      admin_client.execute("grant {priv} on database {db} to user {user}".format(
+          priv=priv, db=unique_db, user=user))
+      self.execute_query_expect_success(self.client, query, user=user)
+    finally:
+      admin_client.execute("revoke {priv} on database {db} from user {user}".format(
+          priv=priv, db=unique_db, user=user), user=ADMIN)
+      admin_client.execute("drop database if exists {} cascade".format(unique_db))
+
   @staticmethod
   def _verify_show_dbs(result, unique_name, visibility_privileges=PRIVILEGES):
     """ Helper function for verifying the results of SHOW DATABASES below.