From 781e52a4c49bc1580b607c2d9ee692b4bbdbc606 Mon Sep 17 00:00:00 2001 From: raashed Date: Fri, 16 Oct 2026 02:17:48 +0530 Subject: [PATCH 1/2] Core: Keep metadata_log_entries snapshot details after expiration The latest_* columns were derived from the current snapshot log, which expiring snapshots trims, so older entries started returning null. Read them from the metadata file each entry points to, only when one of those columns is projected. If that file is gone, fall back to the previous lookup instead of failing the scan. Co-authored-by: yangshangqing95 --- .../iceberg/MetadataLogEntriesTable.java | 83 +++++++++--- .../hadoop/TestMetadataLogEntriesTable.java | 120 ++++++++++++++++++ 2 files changed, 186 insertions(+), 17 deletions(-) create mode 100644 core/src/test/java/org/apache/iceberg/hadoop/TestMetadataLogEntriesTable.java diff --git a/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java b/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java index 3cffee37dc50..aba32fd3fb9c 100644 --- a/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java +++ b/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java @@ -19,20 +19,29 @@ 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; public class MetadataLogEntriesTable extends BaseMetadataTable { + 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 +73,62 @@ 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); + }); + } + + /** + * Returns the latest snapshot recorded in the metadata file of the given log entry. + * + *

The snapshot is read from the entry's own metadata file because expiring snapshots trims the + * current snapshot log. If that file no longer exists, this falls back to the current snapshot + * log. + */ + 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) { + return latestSnapshotAsOfTime(metadataLogEntry.timestampMillis()); + } + } + + // 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 Snapshot latestSnapshotAsOfTime(long timestampMillis) { + try { + return table().snapshot(SnapshotUtil.snapshotIdAsOfTime(table(), timestampMillis)); + } catch (IllegalArgumentException e) { + // no snapshot in the current snapshot log is older than the entry + return null; + } } private class MetadataLogScan extends StaticTableScan { @@ -103,21 +161,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..0f0473ddce17 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/hadoop/TestMetadataLogEntriesTable.java @@ -0,0 +1,120 @@ +/* + * 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 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.TableScan; +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.junit.jupiter.api.Test; + +class TestMetadataLogEntriesTable extends HadoopTableTestBase { + + @Test + void snapshotDetailsAfterExpiration() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + Snapshot first = table.currentSnapshot(); + 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(1)) + .containsExactly( + after.get(1).get(0), + after.get(1).get(1), + first.snapshotId(), + first.schemaId(), + first.sequenceNumber()); + } + + @Test + void snapshotColumnsNotProjected() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + table.newFastAppend().appendFile(FILE_B).commit(); + + List> before = rows(metadataLogEntries().newScan().select("timestamp", "file")); + deletePreviousMetadataFiles(); + + assertThat(rows(metadataLogEntries().newScan().select("timestamp", "file"))).isEqualTo(before); + } + + @Test + void missingMetadataFileFallsBackToSnapshotLog() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + table.newFastAppend().appendFile(FILE_B).commit(); + + List> before = rows(metadataLogEntries().newScan()); + deletePreviousMetadataFiles(); + + assertThat(rows(metadataLogEntries().newScan())).isEqualTo(before); + } + + private org.apache.iceberg.Table metadataLogEntries() { + return MetadataTableUtils.createMetadataTableInstance( + table, MetadataTableType.METADATA_LOG_ENTRIES); + } + + private void deletePreviousMetadataFiles() { + TableMetadata current = ((HasTableOperations) table).operations().current(); + for (TableMetadata.MetadataLogEntry entry : current.previousFiles()) { + table.io().deleteFile(entry.file()); + } + } + + 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; + } +} From 792b300251f63953b9fdc1fb065fc7c84a589ebb Mon Sep 17 00:00:00 2001 From: raashed Date: Tue, 20 Oct 2026 06:00:47 +0530 Subject: [PATCH 2/2] Core: Return null metadata_log_entries snapshot details for missing files Falling back to the time-based lookup when an entry's metadata file is gone mixed file-derived and time-derived rows in one result, and a missing file looked like a real entry without a snapshot. Return null and log a warning instead. Tighten the tests: verify historical metadata files are never opened when no latest_* column is projected, and assert the expired snapshot's details on the row written while it was current. --- .../iceberg/MetadataLogEntriesTable.java | 26 +++------ .../hadoop/TestMetadataLogEntriesTable.java | 54 ++++++++++++------- 2 files changed, 42 insertions(+), 38 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java b/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java index aba32fd3fb9c..6a3479ec62ca 100644 --- a/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java +++ b/core/src/main/java/org/apache/iceberg/MetadataLogEntriesTable.java @@ -24,9 +24,11 @@ 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()); @@ -93,13 +95,6 @@ private DataTask task(TableScan scan) { }); } - /** - * Returns the latest snapshot recorded in the metadata file of the given log entry. - * - *

The snapshot is read from the entry's own metadata file because expiring snapshots trims the - * current snapshot log. If that file no longer exists, this falls back to the current snapshot - * log. - */ private Snapshot latestSnapshot( TableMetadata.MetadataLogEntry metadataLogEntry, TableMetadata current) { TableMetadata metadata; @@ -109,7 +104,11 @@ private Snapshot latestSnapshot( try { metadata = TableMetadataParser.read(table().io(), metadataLogEntry.file()); } catch (NotFoundException e) { - return latestSnapshotAsOfTime(metadataLogEntry.timestampMillis()); + LOG.warn( + "Metadata file {} was not found, setting snapshot details to null", + metadataLogEntry.file(), + e); + return null; } } @@ -122,15 +121,6 @@ private Snapshot latestSnapshot( return metadata.snapshot(Iterables.getLast(snapshotLog).snapshotId()); } - private Snapshot latestSnapshotAsOfTime(long timestampMillis) { - try { - return table().snapshot(SnapshotUtil.snapshotIdAsOfTime(table(), timestampMillis)); - } catch (IllegalArgumentException e) { - // no snapshot in the current snapshot log is older than the entry - return null; - } - } - private class MetadataLogScan extends StaticTableScan { MetadataLogScan(Table table) { super( diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestMetadataLogEntriesTable.java b/core/src/test/java/org/apache/iceberg/hadoop/TestMetadataLogEntriesTable.java index 0f0473ddce17..f5dfa7fcd612 100644 --- a/core/src/test/java/org/apache/iceberg/hadoop/TestMetadataLogEntriesTable.java +++ b/core/src/test/java/org/apache/iceberg/hadoop/TestMetadataLogEntriesTable.java @@ -19,6 +19,11 @@ 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; @@ -30,8 +35,10 @@ 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; @@ -41,7 +48,6 @@ class TestMetadataLogEntriesTable extends HadoopTableTestBase { @Test void snapshotDetailsAfterExpiration() throws IOException { table.newFastAppend().appendFile(FILE_A).commit(); - Snapshot first = table.currentSnapshot(); table.newFastAppend().appendFile(FILE_B).commit(); Snapshot second = table.currentSnapshot(); table.newFastAppend().appendFile(FILE_C).commit(); @@ -57,35 +63,42 @@ void snapshotDetailsAfterExpiration() throws IOException { List> after = rows(metadataLogEntries().newScan()); assertThat(after).hasSize(before.size() + 1); assertThat(after.subList(0, before.size())).isEqualTo(before); - assertThat(after.get(1)) - .containsExactly( - after.get(1).get(0), - after.get(1).get(1), - first.snapshotId(), - first.schemaId(), - first.sequenceNumber()); + 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()); - List> before = rows(metadataLogEntries().newScan().select("timestamp", "file")); - deletePreviousMetadataFiles(); + rows(metadataLogEntries(current, io).newScan().select("timestamp", "file")); - assertThat(rows(metadataLogEntries().newScan().select("timestamp", "file"))).isEqualTo(before); + for (TableMetadata.MetadataLogEntry previous : current.previousFiles()) { + verify(io, never()).newInputFile(previous.file()); + } } @Test - void missingMetadataFileFallsBackToSnapshotLog() throws IOException { + 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()); - deletePreviousMetadataFiles(); + for (TableMetadata.MetadataLogEntry previous : current.previousFiles()) { + table.io().deleteFile(previous.file()); + } - assertThat(rows(metadataLogEntries().newScan())).isEqualTo(before); + 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() { @@ -93,11 +106,12 @@ private org.apache.iceberg.Table metadataLogEntries() { table, MetadataTableType.METADATA_LOG_ENTRIES); } - private void deletePreviousMetadataFiles() { - TableMetadata current = ((HasTableOperations) table).operations().current(); - for (TableMetadata.MetadataLogEntry entry : current.previousFiles()) { - table.io().deleteFile(entry.file()); - } + 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 {