Core: Add metadata table for discovering metadata tables

Author: yangshangqing95

core/src/main/java/org/apache/iceberg/MetadataTableType.java

@@ -18,7 +18,10 @@
  */
 package org.apache.iceberg;
 
+import java.util.Arrays;
+import java.util.List;
 import java.util.Locale;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 
 public enum MetadataTableType {
   ENTRIES,
@@ -36,7 +39,14 @@ public enum MetadataTableType {
   ALL_FILES,
   ALL_MANIFESTS,
   ALL_ENTRIES,
-  POSITION_DELETES;
+  POSITION_DELETES,
+  METADATA_TABLES;
+
+  private final String lower;
+
+  MetadataTableType() {
+    this.lower = name().toLowerCase(Locale.ROOT);
+  }
 
   public static MetadataTableType from(String name) {
     try {
@@ -45,4 +55,14 @@ public static MetadataTableType from(String name) {
       return null;
     }
   }
+
+  public String tableName() {
+    return this.lower;
+  }
+
+  public static List<String> tableNames() {
+    return Arrays.stream(values())
+        .map(MetadataTableType::tableName)
+        .collect(ImmutableList.toImmutableList());
+  }
 }

core/src/main/java/org/apache/iceberg/MetadataTableUtils.java

@@ -88,6 +88,8 @@ private static Table createMetadataTableInstance(
         return new AllEntriesTable(baseTable, metadataTableName);
       case POSITION_DELETES:
         return new PositionDeletesTable(baseTable, metadataTableName);
+      case METADATA_TABLES:
+        return new MetadataTablesTable(baseTable, metadataTableName);
       default:
         throw new NoSuchTableException(
             "Unknown metadata table type: %s for %s", type, metadataTableName);

core/src/main/java/org/apache/iceberg/MetadataTablesTable.java

@@ -0,0 +1,71 @@
+/*
+ * 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.iceberg;
+
+import java.util.Arrays;
+import org.apache.iceberg.types.Types;
+
+/** A {@link Table} implementation that exposes a table's metadata tables as rows. */
+public class MetadataTablesTable extends BaseMetadataTable {
+  MetadataTablesTable(Table table) {
+    this(table, table.name() + "." + MetadataTableType.METADATA_TABLES.tableName());
+  }
+
+  MetadataTablesTable(Table table, String name) {
+    super(table, name);
+  }
+
+  private static final Schema METADATA_TABLES_SCHEMA =
+      new Schema(Types.NestedField.required(1, "metadata_table_name", Types.StringType.get()));
+
+  @Override
+  public TableScan newScan() {
+    return new MetadataTablesScan(table());
+  }
+
+  @Override
+  public Schema schema() {
+    return METADATA_TABLES_SCHEMA;
+  }
+
+  @Override
+  MetadataTableType metadataTableType() {
+    return MetadataTableType.METADATA_TABLES;
+  }
+
+  private DataTask task(BaseTableScan scan) {
+    String location = table().operations().current().metadataFileLocation();
+    return StaticDataTask.of(
+        table().io().newInputFile(location != null ? location : scan.table().location()),
+        schema(),
+        scan.schema(),
+        Arrays.asList(MetadataTableType.values()),
+        metadataType -> StaticDataTask.Row.of(metadataType.tableName()));
+  }
+
+  private class MetadataTablesScan extends StaticTableScan {
+    MetadataTablesScan(Table table) {
+      super(
+          table,
+          METADATA_TABLES_SCHEMA,
+          MetadataTableType.METADATA_TABLES,
+          MetadataTablesTable.this::task);
+    }
+  }
+}

core/src/test/java/org/apache/iceberg/TestMetadataTableScans.java

@@ -1833,6 +1833,26 @@ public void testDeleteFilesTableScanOnBranch() throws IOException {
         .isEqualTo(2);
   }
 
+  @TestTemplate
+  public void testMetadataTablesTableScan() throws IOException {
+    table.newFastAppend().appendFile(FILE_A).commit();
+
+    Table metadataTablesTable = new MetadataTablesTable(table);
+    List<String> actualTables = Lists.newArrayList();
+    try (CloseableIterable<FileScanTask> tasks = metadataTablesTable.newScan().planFiles()) {
+      assertThat(tasks).hasSize(1);
+      for (FileScanTask task : tasks) {
+        try (CloseableIterable<StructLike> rows = task.asDataTask().rows()) {
+          for (StructLike row : rows) {
+            actualTables.add(row.get(0, String.class));
+          }
+        }
+      }
+    }
+    List<String> expectedTables = MetadataTableType.tableNames();
+    assertThat(actualTables).containsExactlyElementsOf(expectedTables);
+  }
+
   private int rowCount(TableScan scan) throws IOException {
     int count = 0;
     try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {

docs/docs/spark-queries.md

@@ -340,6 +340,32 @@ To inspect a table's history, snapshots, and other metadata, Iceberg supports me
 
 Metadata tables are identified by adding the metadata table name after the original table name. For example, history for `db.table` is read using `db.table.history`.
 
+### Metadata Tables
+To list the metadata tables supported by a table:
+
+```sql
+SELECT * FROM prod.db.table.metadata_tables;
+```
+
+| metadata_table_name  |
+|----------------------|
+| entries              |
+| files                |
+| data_files           |
+| history              |
+| metadata_log_entries |
+| snapshots            |
+| refs                 |
+| manifests            |
+| partitions           |
+| all_data_files       |
+| all_delete_files     |
+| all_files            |
+| all_manifests        |
+| all_entries          |
+| position_deletes     |
+| metadata_tables      |
+
 ### History
 
 To show table history: