diff --git a/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java b/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java index 3cffee37dc50..6a3479ec62ca 100644 --- a/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java +++ b/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java @@ -19,20 +19,31 @@ package org.apache.iceberg; import java.util.List; +import org.apache.iceberg.exceptions.NotFoundException; import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.Iterables; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.types.Types; -import org.apache.iceberg.util.SnapshotUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class MetadataLogEntriesTable extends BaseMetadataTable { + private static final Logger LOG = LoggerFactory.getLogger(MetadataLogEntriesTable.class); + + private static final Types.NestedField LATEST_SNAPSHOT_ID = + Types.NestedField.optional(3, "latest_snapshot_id", Types.LongType.get()); + private static final Types.NestedField LATEST_SCHEMA_ID = + Types.NestedField.optional(4, "latest_schema_id", Types.IntegerType.get()); + private static final Types.NestedField LATEST_SEQUENCE_NUMBER = + Types.NestedField.optional(5, "latest_sequence_number", Types.LongType.get()); private static final Schema METADATA_LOG_ENTRIES_SCHEMA = new Schema( Types.NestedField.required(1, "timestamp", Types.TimestampType.withZone()), Types.NestedField.required(2, "file", Types.StringType.get()), - Types.NestedField.optional(3, "latest_snapshot_id", Types.LongType.get()), - Types.NestedField.optional(4, "latest_schema_id", Types.IntegerType.get()), - Types.NestedField.optional(5, "latest_sequence_number", Types.LongType.get())); + LATEST_SNAPSHOT_ID, + LATEST_SCHEMA_ID, + LATEST_SEQUENCE_NUMBER); MetadataLogEntriesTable(Table table) { this(table, table.name() + ".metadata_log_entries"); @@ -64,13 +75,50 @@ private DataTask task(TableScan scan) { metadataLogEntries.add( new TableMetadata.MetadataLogEntry( current.lastUpdatedMillis(), current.metadataFileLocation())); + + // reading each historical metadata file is only needed for the latest_* columns + Schema projection = scan.schema(); + boolean needsSnapshot = + projection.findField(LATEST_SNAPSHOT_ID.fieldId()) != null + || projection.findField(LATEST_SCHEMA_ID.fieldId()) != null + || projection.findField(LATEST_SEQUENCE_NUMBER.fieldId()) != null; + return StaticDataTask.of( table().io().newInputFile(current.metadataFileLocation()), schema(), - scan.schema(), + projection, metadataLogEntries, - metadataLogEntry -> - MetadataLogEntriesTable.metadataLogEntryToRow(metadataLogEntry, table())); + metadataLogEntry -> { + Snapshot latestSnapshot = + needsSnapshot ? latestSnapshot(metadataLogEntry, current) : null; + return metadataLogEntryToRow(metadataLogEntry, latestSnapshot); + }); + } + + private Snapshot latestSnapshot( + TableMetadata.MetadataLogEntry metadataLogEntry, TableMetadata current) { + TableMetadata metadata; + if (metadataLogEntry.file().equals(current.metadataFileLocation())) { + metadata = current; + } else { + try { + metadata = TableMetadataParser.read(table().io(), metadataLogEntry.file()); + } catch (NotFoundException e) { + LOG.warn( + "Metadata file {} was not found, setting snapshot details to null", + metadataLogEntry.file(), + e); + return null; + } + } + + // use the snapshot log rather than the current snapshot, which is null after a replace + List snapshotLog = metadata.snapshotLog(); + if (snapshotLog.isEmpty()) { + return null; + } + + return metadata.snapshot(Iterables.getLast(snapshotLog).snapshotId()); } private class MetadataLogScan extends StaticTableScan { @@ -103,21 +151,12 @@ public CloseableIterable planFiles() { } private static StaticDataTask.Row metadataLogEntryToRow( - TableMetadata.MetadataLogEntry metadataLogEntry, Table table) { - Long latestSnapshotId = null; - Snapshot latestSnapshot = null; - try { - latestSnapshotId = SnapshotUtil.snapshotIdAsOfTime(table, metadataLogEntry.timestampMillis()); - latestSnapshot = table.snapshot(latestSnapshotId); - } catch (IllegalArgumentException ignored) { - // implies this metadata file was created at table creation - } - + TableMetadata.MetadataLogEntry metadataLogEntry, Snapshot latestSnapshot) { return StaticDataTask.Row.of( metadataLogEntry.timestampMillis() * 1000, metadataLogEntry.file(), // latest snapshot in this file corresponding to the log entry - latestSnapshotId, + latestSnapshot != null ? latestSnapshot.snapshotId() : null, latestSnapshot != null ? latestSnapshot.schemaId() : null, latestSnapshot != null ? latestSnapshot.sequenceNumber() : null); } diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestMetadataLogEntriesTable.java b/core/src/test/java/org/apache/iceberg/hadoop/TestMetadataLogEntriesTable.java new file mode 100644 index 000000000000..f5dfa7fcd612 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/hadoop/TestMetadataLogEntriesTable.java @@ -0,0 +1,134 @@ +/* + * 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.hadoop; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.util.List; +import org.apache.iceberg.ExpireSnapshots; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.MetadataTableType; +import org.apache.iceberg.MetadataTableUtils; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.TableMetadata; +import org.apache.iceberg.TableOperations; +import org.apache.iceberg.TableScan; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.relocated.com.google.common.collect.Iterables; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.junit.jupiter.api.Test; + +class TestMetadataLogEntriesTable extends HadoopTableTestBase { + + @Test + void snapshotDetailsAfterExpiration() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + table.newFastAppend().appendFile(FILE_B).commit(); + Snapshot second = table.currentSnapshot(); + table.newFastAppend().appendFile(FILE_C).commit(); + + List> before = rows(metadataLogEntries().newScan()); + + table + .expireSnapshots() + .expireSnapshotId(second.snapshotId()) + .cleanupLevel(ExpireSnapshots.CleanupLevel.NONE) + .commit(); + + List> after = rows(metadataLogEntries().newScan()); + assertThat(after).hasSize(before.size() + 1); + assertThat(after.subList(0, before.size())).isEqualTo(before); + assertThat(after.get(2).subList(2, 5)) + .containsExactly(second.snapshotId(), second.schemaId(), second.sequenceNumber()); + } + + @Test + void snapshotColumnsNotProjected() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + table.newFastAppend().appendFile(FILE_B).commit(); + TableMetadata current = ((HasTableOperations) table).operations().current(); + FileIO io = spy(table.io()); + + rows(metadataLogEntries(current, io).newScan().select("timestamp", "file")); + + for (TableMetadata.MetadataLogEntry previous : current.previousFiles()) { + verify(io, never()).newInputFile(previous.file()); + } + } + + @Test + void missingMetadataFileReturnsNullSnapshotDetails() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + table.newFastAppend().appendFile(FILE_B).commit(); + TableMetadata current = ((HasTableOperations) table).operations().current(); + List> before = rows(metadataLogEntries().newScan()); + for (TableMetadata.MetadataLogEntry previous : current.previousFiles()) { + table.io().deleteFile(previous.file()); + } + + List> expected = Lists.newArrayList(); + for (List row : before.subList(0, before.size() - 1)) { + expected.add(Lists.newArrayList(row.get(0), row.get(1), null, null, null)); + } + + expected.add(Iterables.getLast(before)); + + assertThat(rows(metadataLogEntries().newScan())).isEqualTo(expected); + } + + private org.apache.iceberg.Table metadataLogEntries() { + return MetadataTableUtils.createMetadataTableInstance( + table, MetadataTableType.METADATA_LOG_ENTRIES); + } + + private static org.apache.iceberg.Table metadataLogEntries(TableMetadata current, FileIO io) { + TableOperations operations = mock(TableOperations.class); + when(operations.current()).thenReturn(current); + when(operations.io()).thenReturn(io); + return MetadataTableUtils.createMetadataTableInstance( + operations, "table", "table.metadata_log_entries", MetadataTableType.METADATA_LOG_ENTRIES); + } + + private static List> rows(TableScan scan) throws IOException { + List> rows = Lists.newArrayList(); + try (CloseableIterable tasks = scan.planFiles(); + CloseableIterable structs = + Iterables.getOnlyElement(tasks).asDataTask().rows()) { + for (StructLike struct : structs) { + List row = Lists.newArrayList(); + for (int pos = 0; pos < struct.size(); pos += 1) { + row.add(struct.get(pos, Object.class)); + } + + rows.add(row); + } + } + + return rows; + } +}