Add metadata_log_entries table to Iceberg

docs/src/main/sphinx/connector/iceberg.md

@@ -825,6 +825,52 @@ The output of the query has the following columns:
   - Whether or not this snapshot is an ancestor of the current snapshot.
 :::
 
+##### `$metadata_log_entries` table
+
+The `$metadata_log_entries` table provides a view of metadata log entries
+of the Iceberg table.
+
+You can retrieve the information about the metadata log entries of the Iceberg
+table `test_table` by using the following query:
+
+```
+SELECT * FROM "test_table$metadata_log_entries"
+```
+
+```text
+             timestamp                 |                                                              file                                                          | latest_snapshot_id  | latest_schema_id | latest_sequence_number
+---------------------------------------+----------------------------------------------------------------------------------------------------------------------------+---------------------+------------------+------------------------
+ 2024-01-16 15:55:31.172 Europe/Vienna | hdfs://hadoop-master:9000/user/hive/warehouse/test_table/metadata/00000-39174715-be2a-48fa-9949-35413b8b736e.metadata.json | 1221802298419195590 |                0 |                      1
+ 2024-01-16 17:19:56.118 Europe/Vienna | hdfs://hadoop-master:9000/user/hive/warehouse/test_table/metadata/00001-e40178c9-271f-4a96-ad29-eed5e7aef9b0.metadata.json | 7124386610209126943 |                0 |                      2
+```
+
+The output of the query has the following columns:
+
+:::{list-table} Metadata log entries columns
+:widths: 30, 30, 40
+:header-rows: 1
+
+* - Name
+  - Type
+  - Description
+* - `timestamp`
+  - `TIMESTAMP(3) WITH TIME ZONE`
+  - The time when the metadata was created.
+* - `file`
+  - `VARCHAR`
+  - The location of the metadata file.
+* - `latest_snapshot_id`
+  - `BIGINT`
+  - The identifier of the latest snapshot when the metadata was updated.
+* - `latest_schema_id`
+  - `INTEGER`
+  - The identifier of the latest schema when the metadata was updated.
+* - `latest_sequence_number`
+  - `BIGINT`
+  - The data sequence number of the metadata file.
+:::
+
+
 ##### `$snapshots` table
 
 The `$snapshots` table provides a detailed view of snapshots of the Iceberg

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

@@ -0,0 +1,109 @@
+/*
+ * 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.trino.plugin.iceberg;
+
+import io.trino.plugin.iceberg.util.PageListBuilder;
+import io.trino.spi.Page;
+import io.trino.spi.connector.ConnectorPageSource;
+import io.trino.spi.connector.ConnectorSession;
+import io.trino.spi.connector.ConnectorTableMetadata;
+import io.trino.spi.connector.ConnectorTransactionHandle;
+import io.trino.spi.connector.FixedPageSource;
+import io.trino.spi.connector.SystemTable;
+import io.trino.spi.predicate.TupleDomain;
+import io.trino.spi.type.TimeZoneKey;
+import org.apache.iceberg.DataTask;
+import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableScan;
+import org.apache.iceberg.io.CloseableIterable;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.List;
+import java.util.Map;
+
+import static com.google.common.collect.ImmutableMap.toImmutableMap;
+import static com.google.common.collect.Maps.immutableEntry;
+import static com.google.common.collect.Streams.mapWithIndex;
+import static java.util.Objects.requireNonNull;
+import static org.apache.iceberg.MetadataTableUtils.createMetadataTableInstance;
+
+public abstract class BaseSystemTable
+        implements SystemTable
+{
+    private final Table icebergTable;
+    private final ConnectorTableMetadata tableMetadata;
+    private final MetadataTableType metadataTableType;
+
+    BaseSystemTable(Table icebergTable, ConnectorTableMetadata tableMetadata, MetadataTableType metadataTableType)
+    {
+        this.icebergTable = requireNonNull(icebergTable, "icebergTable is null");
+        this.tableMetadata = requireNonNull(tableMetadata, "tableMetadata is null");
+        this.metadataTableType = requireNonNull(metadataTableType, "metadataTableType is null");
+    }
+
+    @Override
+    public Distribution getDistribution()
+    {
+        return Distribution.SINGLE_COORDINATOR;
+    }
+
+    @Override
+    public ConnectorTableMetadata getTableMetadata()
+    {
+        return tableMetadata;
+    }
+
+    @Override
+    public ConnectorPageSource pageSource(ConnectorTransactionHandle transactionHandle, ConnectorSession session, TupleDomain<Integer> constraint)
+    {
+        return new FixedPageSource(buildPages(tableMetadata, session, icebergTable, metadataTableType));
+    }
+
+    private List<Page> buildPages(ConnectorTableMetadata tableMetadata, ConnectorSession session, Table icebergTable, MetadataTableType metadataTableType)
+    {
+        PageListBuilder pagesBuilder = PageListBuilder.forTable(tableMetadata);
+
+        TableScan tableScan = createMetadataTableInstance(icebergTable, metadataTableType).newScan();
+        TimeZoneKey timeZoneKey = session.getTimeZoneKey();
+
+        Map<String, Integer> columnNameToPosition = mapWithIndex(tableScan.schema().columns().stream(),
+                (column, position) -> immutableEntry(column.name(), Long.valueOf(position).intValue()))
+                .collect(toImmutableMap(Map.Entry::getKey, Map.Entry::getValue));
+
+        try (CloseableIterable<FileScanTask> fileScanTasks = tableScan.planFiles()) {
+            fileScanTasks.forEach(fileScanTask -> addRows((DataTask) fileScanTask, pagesBuilder, timeZoneKey, columnNameToPosition));
+        }
+        catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+
+        return pagesBuilder.build();
+    }
+
+    private void addRows(DataTask dataTask, PageListBuilder pagesBuilder, TimeZoneKey timeZoneKey, Map<String, Integer> columnNameToPositionInSchema)
+    {
+        try (CloseableIterable<StructLike> dataRows = dataTask.rows()) {
+            dataRows.forEach(dataTaskRow -> addRow(pagesBuilder, dataTaskRow, timeZoneKey, columnNameToPositionInSchema));
+        }
+        catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
+
+    protected abstract void addRow(PageListBuilder pagesBuilder, StructLike structLike, TimeZoneKey timeZoneKey, Map<String, Integer> columnNameToPositionInSchema);
+}

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

@@ -575,6 +575,7 @@ private Optional<SystemTable> getRawSystemTable(ConnectorSession session, Schema
         return switch (tableType) {
             case DATA, MATERIALIZED_VIEW_STORAGE -> throw new VerifyException("Unexpected table type: " + tableType); // Handled above.
             case HISTORY -> Optional.of(new HistoryTable(tableName, table));
+            case METADATA_LOG_ENTRIES -> Optional.of(new MetadataLogEntriesTable(tableName, table));
             case SNAPSHOTS -> Optional.of(new SnapshotsTable(tableName, typeManager, table));
             case PARTITIONS -> Optional.of(new PartitionTable(tableName, typeManager, table, getCurrentSnapshotId(table)));
             case MANIFESTS -> Optional.of(new ManifestsTable(tableName, table, getCurrentSnapshotId(table)));

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

@@ -55,7 +55,6 @@
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.HistoryEntry;
-import org.apache.iceberg.MetadataTableType;
 import org.apache.iceberg.PartitionField;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
@@ -66,7 +65,6 @@
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableOperations;
-import org.apache.iceberg.TableScan;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.io.LocationProvider;
 import org.apache.iceberg.types.Type.PrimitiveType;
@@ -100,8 +98,6 @@
 import static com.google.common.collect.ImmutableMap.toImmutableMap;
 import static com.google.common.collect.ImmutableSet.toImmutableSet;
 import static com.google.common.collect.Iterables.getOnlyElement;
-import static com.google.common.collect.Maps.immutableEntry;
-import static com.google.common.collect.Streams.mapWithIndex;
 import static io.airlift.slice.Slices.utf8Slice;
 import static io.trino.plugin.base.io.ByteBuffers.getWrappedBytes;
 import static io.trino.plugin.hive.HiveMetadata.TABLE_COMMENT;
@@ -157,7 +153,6 @@
 import static java.math.RoundingMode.UNNECESSARY;
 import static java.util.Comparator.comparing;
 import static java.util.Objects.requireNonNull;
-import static org.apache.iceberg.MetadataTableUtils.createMetadataTableInstance;
 import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
 import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
 import static org.apache.iceberg.TableProperties.FORMAT_VERSION;
@@ -844,18 +839,6 @@ public static void commit(SnapshotUpdate<?> update, ConnectorSession session)
         update.commit();
     }
 
-    public static TableScan buildTableScan(Table icebergTable, MetadataTableType metadataTableType)
-    {
-        return createMetadataTableInstance(icebergTable, metadataTableType).newScan();
-    }
-
-    public static Map<String, Integer> columnNameToPositionInSchema(Schema schema)
-    {
-        return mapWithIndex(schema.columns().stream(),
-                (column, position) -> immutableEntry(column.name(), Long.valueOf(position).intValue()))
-                .collect(toImmutableMap(Entry::getKey, Entry::getValue));
-    }
-
     public static String getLatestMetadataLocation(TrinoFileSystem fileSystem, String location)
     {
         List<Location> latestMetadataLocations = new ArrayList<>();

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

@@ -0,0 +1,79 @@
+/*
+ * 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.trino.plugin.iceberg;
+
+import com.google.common.collect.ImmutableList;
+import io.trino.plugin.iceberg.util.PageListBuilder;
+import io.trino.spi.connector.ColumnMetadata;
+import io.trino.spi.connector.ConnectorTableMetadata;
+import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.type.TimeZoneKey;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+
+import java.util.Map;
+
+import static io.trino.spi.type.BigintType.BIGINT;
+import static io.trino.spi.type.IntegerType.INTEGER;
+import static io.trino.spi.type.TimestampWithTimeZoneType.TIMESTAMP_TZ_MILLIS;
+import static io.trino.spi.type.Timestamps.MICROSECONDS_PER_MILLISECOND;
+import static io.trino.spi.type.VarcharType.VARCHAR;
+import static java.util.Objects.requireNonNull;
+import static org.apache.iceberg.MetadataTableType.METADATA_LOG_ENTRIES;
+
+public class MetadataLogEntriesTable
+        extends BaseSystemTable
+{
+    private static final String TIMESTAMP_COLUMN_NAME = "timestamp";
+    private static final String FILE_COLUMN_NAME = "file";
+    private static final String LATEST_SNAPSHOT_ID_COLUMN_NAME = "latest_snapshot_id";
+    private static final String LATEST_SCHEMA_ID_COLUMN_NAME = "latest_schema_id";
+    private static final String LATEST_SEQUENCE_NUMBER_COLUMN_NAME = "latest_sequence_number";
+
+    public MetadataLogEntriesTable(SchemaTableName tableName, Table icebergTable)
+    {
+        super(
+                requireNonNull(icebergTable, "icebergTable is null"),
+                createConnectorTableMetadata(requireNonNull(tableName, "tableName is null")),
+                METADATA_LOG_ENTRIES);
+    }
+
+    private static ConnectorTableMetadata createConnectorTableMetadata(SchemaTableName tableName)
+    {
+        return new ConnectorTableMetadata(
+                tableName,
+                ImmutableList.<ColumnMetadata>builder()
+                        .add(new ColumnMetadata(TIMESTAMP_COLUMN_NAME, TIMESTAMP_TZ_MILLIS))
+                        .add(new ColumnMetadata(FILE_COLUMN_NAME, VARCHAR))
+                        .add(new ColumnMetadata(LATEST_SNAPSHOT_ID_COLUMN_NAME, BIGINT))
+                        .add(new ColumnMetadata(LATEST_SCHEMA_ID_COLUMN_NAME, INTEGER))
+                        .add(new ColumnMetadata(LATEST_SEQUENCE_NUMBER_COLUMN_NAME, BIGINT))
+                        .build());
+    }
+
+    @Override
+    protected void addRow(PageListBuilder pagesBuilder, StructLike structLike, TimeZoneKey timeZoneKey, Map<String, Integer> columnNameToPositionInSchema)
+    {
+        pagesBuilder.beginRow();
+
+        pagesBuilder.appendTimestampTzMillis(
+                structLike.get(columnNameToPositionInSchema.get(TIMESTAMP_COLUMN_NAME), Long.class) / MICROSECONDS_PER_MILLISECOND,
+                timeZoneKey);
+        pagesBuilder.appendVarchar(structLike.get(columnNameToPositionInSchema.get(FILE_COLUMN_NAME), String.class));
+        pagesBuilder.appendBigint(structLike.get(columnNameToPositionInSchema.get(LATEST_SNAPSHOT_ID_COLUMN_NAME), Long.class));
+        pagesBuilder.appendInteger(structLike.get(columnNameToPositionInSchema.get(LATEST_SCHEMA_ID_COLUMN_NAME), Integer.class));
+        pagesBuilder.appendBigint(structLike.get(columnNameToPositionInSchema.get(LATEST_SEQUENCE_NUMBER_COLUMN_NAME), Long.class));
+        pagesBuilder.endRow();
+    }
+}