This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new d974dd8920f [fix](iceberg) Count expired manifests as manifest files
in expire_snapshots (#69043)
d974dd8920f is described below
commit d974dd8920fe910cf15dea723ccc57e01031dc43
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Thu Oct 15 14:48:11 2026 +0800
[fix](iceberg) Count expired manifests as manifest files in
expire_snapshots (#69043)
### What problem does this PR solve?
Issue Number: None
Related PR: #68902
Problem Summary:
**In short.** The Iceberg `expire_snapshots` procedure counted every
expired manifest as a deleted data file, so `deleted_data_files_count`
was too high and `deleted_manifest_files_count` was almost always 0.
This came up in a unit test written for #68902.
**Background.** `expire_snapshots` returns the six counters of Spark's
`ExpireSnapshotsSparkAction`: deleted data files, position delete files,
equality delete files, manifests, manifest lists and statistics files.
`IcebergExpireSnapshotsAction` fills them in the `deleteWith` callback
that Iceberg calls for each file it removes. Delete files are recognized
by path, from a map the action builds from every snapshot's delete
manifests. Manifests and manifest lists were recognized by name: a path
that contains `-m-` and ends in `.avro` was a manifest, one that
contains `snap-` was a manifest list, and anything else that is not a
statistics file was a data file.
**The problem, and what it cost.** Iceberg names a manifest `<commit
UUID>-m<N>.avro` (`SnapshotProducer.newManifestOutputFile`), so a
manifest written by a normal commit never contains `-m-`; only the
`optimized-m-` names of `rewrite_manifests` did. Every manifest that
`expire_snapshots` deleted was therefore counted as a data file. For
example, after appending `f1`, deleting it and appending `f2`,
`expire_snapshots("retain_last" = "1")` reclaims `f1`, two manifests and
two manifest lists:
| Counter | master | with this PR |
|---|---|---|
| `deleted_data_files_count` | 3 | 1 |
| `deleted_manifest_files_count` | 0 | 2 |
| `deleted_manifest_lists_count` | 2 | 2 |
Only the reported counts were wrong; the files that Iceberg deletes are
the same.
**How this PR fixes it.** The callback classifies manifests and manifest
lists by path, as it already does for delete files. While the action
reads every snapshot's manifest list to find the delete files, it now
also records the paths of all manifests (`Snapshot.allManifests`, which
the same read already loaded) and every snapshot's
`manifestListLocation()`. A single map from path to file kind replaces
the delete-file content map. The expiration refreshes the table when it
commits (`RemoveSnapshots.internalApply`), so it can also expire
snapshots that another writer committed after the map was built. For a
path the map does not know, the callback falls back to the file name,
and the whole name has to match the name Iceberg gives the file, UUID
included: `snap-<snapshot id>-<attempt>-<commit UUID>.avro` is a
manifest list, and `<commit UUID>-m<N>.avro` or Spark's
`optimized-m-<UUID>.avro` is a manifest. Data files take this fallback
too, as the map holds none, so a looser rule would count an Avro data
file such as `sales-m1.avro` as a manifest. A delete file of such a
snapshot still counts as a data file, because its name does not tell the
two apart. Statistics files are still told apart by name.
**Results.**
`IcebergExpireSnapshotsActionTest.countsExpiredManifestsAsManifestFiles`
builds the table above and asserts `[1, 0, 0, 2, 2, 0]`; on master it
gets `[3, 0, 0, 0, 2, 0]`.
`countsMetadataOfSnapshotsCommittedAfterTheTableWasLoaded` loads the
table, lets another writer append, delete and append, and then expires
snapshots through the loaded table: it asserts `[1, 0, 0, 2, 3, 0]`, and
without the file-name fallback it gets `[5, 0, 0, 0, 1, 0]`.
`countsAnAvroDataFileNamedLikeAManifestAsADataFile` appends an Avro data
file named `sales-m1.avro`, deletes it and appends `f2`: it asserts `[1,
0, 0, 2, 2, 0]`, and a rule that takes any `-m<N>.avro` name for a
manifest gets `[0, 0, 0, 3, 2, 0]`.
`classifiesUnmappedMetadataFilesByIcebergFileNames` pins the name rules,
including Avro data files whose names only resemble them. The existing
tests keep their results, including the append-only case (`[0, 0, 0, 0,
2, 0]`) and the delete-manifest dedup. The regression
`test_iceberg_expire_snapshots` records `0 0 0 0 2 0` for an append-only
table, where no manifest expires, so its output does not change.
---
.../action/IcebergExpireSnapshotsAction.java | 80 ++++++++--
.../action/IcebergExpireSnapshotsActionTest.java | 171 ++++++++++++++++++++-
2 files changed, 232 insertions(+), 19 deletions(-)
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergExpireSnapshotsAction.java
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergExpireSnapshotsAction.java
index 3fbcd454c4e..d7ab40214d9 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergExpireSnapshotsAction.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/IcebergExpireSnapshotsAction.java
@@ -50,13 +50,14 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.regex.Pattern;
/**
* Removes old snapshots from an iceberg table to free storage and improve
metadata performance. Connector
* port of legacy {@code IcebergExpireSnapshotsAction} — the most involved of
the snapshot procedures (five
* optional arguments, a custom validation pass, an FE-local fixed thread pool
for concurrent deletes, a
* {@code deleteWith} callback that classifies deleted files into the six
Spark-compatible counters, and the
- * delete-file content map). The validation messages and the SDK call chain
are verbatim; the validation
+ * file kind map it classifies them by). The validation messages and the SDK
call chain are verbatim; the validation
* failures throw {@link DorisConnectorException} in place of the legacy
{@code AnalysisException}/
* {@code UserException} (message-identical, T08 byte-parity). {@code
parseTimestamp} keeps the legacy
* {@code ZoneId.systemDefault()} (this procedure, unlike {@code
rollback_to_timestamp}, never used the
@@ -70,11 +71,26 @@ public class IcebergExpireSnapshotsAction extends
BaseIcebergAction {
public static final String SNAPSHOT_IDS = "snapshot_ids";
public static final String CLEAN_EXPIRED_METADATA =
"clean_expired_metadata";
+ // The names Iceberg gives a manifest list and a manifest; see
metadataFileKindByName.
+ private static final String UUID_REGEX =
"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}";
+ private static final Pattern MANIFEST_LIST_FILE_NAME =
Pattern.compile("snap-\\d+-\\d+-" + UUID_REGEX + "\\.avro");
+ private static final Pattern MANIFEST_FILE_NAME =
+ Pattern.compile("(" + UUID_REGEX + "-m\\d+|optimized-m-" +
UUID_REGEX + ")\\.avro");
+
// Test-only gate for the delete-manifest dedup: the number of DISTINCT
delete manifests read by the most
- // recent buildDeleteFileContentMap call. Asserts each manifest is read
once, not once per referencing snapshot.
+ // recent buildFileKindMap call. Asserts each manifest is read once, not
once per referencing snapshot.
@VisibleForTesting
int lastDeleteManifestReadCount;
+ /** The kinds of file that expiring snapshots deletes, other than data and
statistics files. */
+ @VisibleForTesting
+ enum FileKind {
+ POSITION_DELETES,
+ EQUALITY_DELETES,
+ MANIFEST,
+ MANIFEST_LIST
+ }
+
public IcebergExpireSnapshotsAction(Map<String, String> properties,
List<String> partitionNames,
ConnectorPredicate whereCondition) {
super("expire_snapshots", properties, partitionNames, whereCondition);
@@ -173,8 +189,7 @@ public class IcebergExpireSnapshotsAction extends
BaseIcebergAction {
ExecutorService deleteExecutor = null;
try {
- Map<String, FileContent> deleteFileContentByPath =
- buildDeleteFileContentMap(icebergTable);
+ Map<String, FileKind> fileKindByPath =
buildFileKindMap(icebergTable);
ExpireSnapshots expireSnapshots = icebergTable.expireSnapshots();
// Configure older_than timestamp
@@ -221,14 +236,17 @@ public class IcebergExpireSnapshotsAction extends
BaseIcebergAction {
// Set up delete callback to count files by type
expireSnapshots.deleteWith(path -> {
- FileContent deleteContent = deleteFileContentByPath.get(path);
- if (deleteContent == FileContent.POSITION_DELETES) {
+ FileKind fileKind = fileKindByPath.get(path);
+ if (fileKind == null) {
+ fileKind = metadataFileKindByName(path);
+ }
+ if (fileKind == FileKind.POSITION_DELETES) {
deletedPositionDeleteFilesCount.incrementAndGet();
- } else if (deleteContent == FileContent.EQUALITY_DELETES) {
+ } else if (fileKind == FileKind.EQUALITY_DELETES) {
deletedEqualityDeleteFilesCount.incrementAndGet();
- } else if (path.contains("-m-") && path.endsWith(".avro")) {
+ } else if (fileKind == FileKind.MANIFEST) {
deletedManifestFilesCount.incrementAndGet();
- } else if (path.contains("snap-") && path.endsWith(".avro")) {
+ } else if (fileKind == FileKind.MANIFEST_LIST) {
deletedManifestListsCount.incrementAndGet();
} else if (path.endsWith(".stats") ||
path.contains("statistics")) {
deletedStatisticsFilesCount.incrementAndGet();
@@ -276,9 +294,14 @@ public class IcebergExpireSnapshotsAction extends
BaseIcebergAction {
}
}
+ /**
+ * Maps the delete files, manifests and manifest lists of every snapshot
to their kind, so that the delete
+ * callback can count each file it is handed. Manifests are found by path,
not by name: SnapshotProducer
+ * names a manifest {@code <commit UUID>-m<N>.avro}.
+ */
@VisibleForTesting
- Map<String, FileContent> buildDeleteFileContentMap(Table icebergTable) {
- Map<String, FileContent> deleteFileContentByPath = new HashMap<>();
+ Map<String, FileKind> buildFileKindMap(Table icebergTable) {
+ Map<String, FileKind> fileKindByPath = new HashMap<>();
// Dedup delete-manifest reads across snapshots. Iceberg manifests are
immutable and adjacent snapshots
// carry the same delete manifests forward unchanged, so re-reading
one yields the identical DeleteFile
// set (putIfAbsent already made the re-read a no-op). Reading each
DISTINCT manifest exactly once
@@ -289,6 +312,12 @@ public class IcebergExpireSnapshotsAction extends
BaseIcebergAction {
int reads = 0;
try {
for (Snapshot snapshot : icebergTable.snapshots()) {
+ if (snapshot.manifestListLocation() != null) {
+ fileKindByPath.put(snapshot.manifestListLocation(),
FileKind.MANIFEST_LIST);
+ }
+ for (ManifestFile manifest :
snapshot.allManifests(icebergTable.io())) {
+ fileKindByPath.put(manifest.path(), FileKind.MANIFEST);
+ }
List<ManifestFile> deleteManifests =
snapshot.deleteManifests(icebergTable.io());
if (deleteManifests == null || deleteManifests.isEmpty()) {
continue;
@@ -301,8 +330,9 @@ public class IcebergExpireSnapshotsAction extends
BaseIcebergAction {
try (CloseableIterable<DeleteFile> deleteFiles =
ManifestFiles.readDeleteManifest(
manifest, icebergTable.io(),
icebergTable.specs())) {
for (DeleteFile deleteFile : deleteFiles) {
- deleteFileContentByPath.putIfAbsent(
- deleteFile.location(),
deleteFile.content());
+ fileKindByPath.putIfAbsent(deleteFile.location(),
+ deleteFile.content() ==
FileContent.POSITION_DELETES
+ ? FileKind.POSITION_DELETES :
FileKind.EQUALITY_DELETES);
}
}
}
@@ -311,7 +341,29 @@ public class IcebergExpireSnapshotsAction extends
BaseIcebergAction {
throw new DorisConnectorException("Failed to build delete file
content map: " + e.getMessage(), e);
}
lastDeleteManifestReadCount = reads;
- return deleteFileContentByPath;
+ return fileKindByPath;
+ }
+
+ /**
+ * Tells a manifest or manifest list that {@link #buildFileKindMap} did
not see apart by its file name. The
+ * expiration refreshes the table when it commits, so it also expires
snapshots that another writer committed
+ * after the map was built. SnapshotProducer names a manifest list
+ * {@code snap-<snapshot id>-<attempt>-<commit UUID>.avro} and a manifest
{@code <commit UUID>-m<N>.avro};
+ * Spark's {@code rewrite_manifests} names a manifest {@code
optimized-m-<UUID>.avro}. Returns null for any
+ * other file. The whole name must match, UUID included: every data file
takes this path, as the map holds
+ * no data files, so a looser rule would count an Avro data file such as
{@code sales-m1.avro} as a manifest.
+ * A delete file of such a snapshot still counts as a data file, as its
name does not tell the two apart.
+ */
+ @VisibleForTesting
+ static FileKind metadataFileKindByName(String path) {
+ String fileName = path.substring(path.lastIndexOf('/') + 1);
+ if (MANIFEST_LIST_FILE_NAME.matcher(fileName).matches()) {
+ return FileKind.MANIFEST_LIST;
+ }
+ if (MANIFEST_FILE_NAME.matcher(fileName).matches()) {
+ return FileKind.MANIFEST;
+ }
+ return null;
}
@Override
diff --git
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergExpireSnapshotsActionTest.java
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergExpireSnapshotsActionTest.java
index 9b5ffc981e7..373d1ced7ae 100644
---
a/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergExpireSnapshotsActionTest.java
+++
b/fe/fe-connector/fe-connector-iceberg/src/test/java/org/apache/doris/connector/iceberg/action/IcebergExpireSnapshotsActionTest.java
@@ -27,8 +27,11 @@ import
org.apache.doris.connector.spi.pushdown.ConnectorPredicate;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
-import org.apache.iceberg.FileContent;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.ManifestFile;
+import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.TableIdentifier;
@@ -110,6 +113,162 @@ public class IcebergExpireSnapshotsActionTest {
"retain_last=1 must leave exactly one snapshot");
}
+ @Test
+ public void countsExpiredManifestsAsManifestFiles() {
+ // Iceberg names a manifest <commit UUID>-m<N>.avro
(SnapshotProducer.newManifestOutputFile), so a "-m-"
+ // file name check never matched one and counted every expired
manifest as a deleted data file. Removing
+ // f1 and then appending f2 leaves the first snapshot's manifest
unreferenced by the last snapshot, so
+ // retain_last=1 deletes that manifest together with f1 and the two
older manifest lists.
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ TableIdentifier id = ActionTestTables.id("t");
+ ActionTestTables.createTable(catalog, "t");
+ ActionTestTables.appendSnapshot(catalog, "t", "f1.parquet", 1L);
+
catalog.loadTable(id).newDelete().deleteFile("s3://b/db1/f1.parquet").commit();
+ ActionTestTables.appendSnapshot(catalog, "t", "f2.parquet", 2L);
+
+ Table table = catalog.loadTable(id);
+ Set<String> retained = new HashSet<>();
+ for (ManifestFile manifest :
table.currentSnapshot().allManifests(table.io())) {
+ retained.add(manifest.path());
+ }
+ Set<String> expired = new HashSet<>();
+ for (Snapshot snapshot : table.snapshots()) {
+ for (ManifestFile manifest : snapshot.allManifests(table.io())) {
+ if (!retained.contains(manifest.path())) {
+ expired.add(manifest.path());
+ }
+ }
+ }
+ // Sanity: the fixture must expire a manifest, and no manifest name
carries "-m-".
+ Assertions.assertFalse(expired.isEmpty(), "the last snapshot must not
reference every manifest");
+ Assertions.assertTrue(expired.stream().noneMatch(path ->
path.contains("-m-")), expired.toString());
+
+ IcebergExpireSnapshotsAction action =
action(ImmutableMap.of("retain_last", "1"));
+ action.validate();
+ List<String> row = action.execute(table,
ActionTestTables.session("UTC")).getRows().get(0);
+ Assertions.assertEquals(ImmutableList.of("1", "0", "0",
String.valueOf(expired.size()), "2", "0"), row,
+ "f1 is the only deleted data file; expired manifests " +
expired + " count as manifest files");
+ }
+
+ @Test
+ public void countsMetadataOfSnapshotsCommittedAfterTheTableWasLoaded() {
+ // The expiration refreshes the table when it commits
(RemoveSnapshots.internalApply), so it also expires
+ // snapshots that another writer committed after the action loaded the
table and built its file kind map.
+ // Their manifests and manifest lists are not in the map and must
still count as such, not as deleted
+ // data files.
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ TableIdentifier id = ActionTestTables.id("t");
+ ActionTestTables.createTable(catalog, "t");
+ ActionTestTables.appendSnapshot(catalog, "t", "f0.parquet", 1L);
+ Table loaded = catalog.loadTable(id);
+ // Another writer appends f1, deletes it and appends f2, which leaves
f1 and the manifests that listed it
+ // unreferenced by the last snapshot.
+ ActionTestTables.appendSnapshot(catalog, "t", "f1.parquet", 1L);
+
catalog.loadTable(id).newDelete().deleteFile("s3://b/db1/f1.parquet").commit();
+ ActionTestTables.appendSnapshot(catalog, "t", "f2.parquet", 2L);
+
+ Table current = catalog.loadTable(id);
+ Set<String> retained = new HashSet<>();
+ for (ManifestFile manifest :
current.currentSnapshot().allManifests(current.io())) {
+ retained.add(manifest.path());
+ }
+ Set<String> expired = new HashSet<>();
+ for (Snapshot snapshot : current.snapshots()) {
+ for (ManifestFile manifest : snapshot.allManifests(current.io())) {
+ if (!retained.contains(manifest.path())) {
+ expired.add(manifest.path());
+ }
+ }
+ }
+ // Sanity: manifests expire, and the map built from the loaded table
knows none of them.
+ Map<String, IcebergExpireSnapshotsAction.FileKind> kindByPath =
+ action(ImmutableMap.of("retain_last",
"1")).buildFileKindMap(loaded);
+ Assertions.assertFalse(expired.isEmpty(), "the last snapshot must not
reference every manifest");
+
Assertions.assertTrue(expired.stream().noneMatch(kindByPath::containsKey),
expired.toString());
+
+ // An older_than in 2100 expires every snapshot but the retained one,
however fast the test runs.
+ IcebergExpireSnapshotsAction action =
+ action(ImmutableMap.of("older_than", "4102444800000",
"retain_last", "1"));
+ action.validate();
+ List<String> row = action.execute(loaded,
ActionTestTables.session("UTC")).getRows().get(0);
+ Assertions.assertEquals(ImmutableList.of("1", "0", "0",
String.valueOf(expired.size()), "3", "0"), row,
+ "f1 is the only deleted data file; expired manifests " +
expired
+ + " and the manifest lists of the three expired
snapshots count as such");
+ Assertions.assertEquals(1,
Iterables.size(catalog.loadTable(id).snapshots()),
+ "retain_last=1 must leave exactly one snapshot");
+ }
+
+ @Test
+ public void classifiesUnmappedMetadataFilesByIcebergFileNames() {
+ String dir = "s3://b/db1/t/metadata/";
+
Assertions.assertEquals(IcebergExpireSnapshotsAction.FileKind.MANIFEST_LIST,
+ IcebergExpireSnapshotsAction.metadataFileKindByName(
+ dir +
"snap-1234-1-0a1b2c3d-4e5f-6071-8293-a4b5c6d7e8f9.avro"));
+ Assertions.assertEquals(IcebergExpireSnapshotsAction.FileKind.MANIFEST,
+ IcebergExpireSnapshotsAction.metadataFileKindByName(
+ dir +
"0a1b2c3d-4e5f-6071-8293-a4b5c6d7e8f9-m12.avro"));
+ Assertions.assertEquals(IcebergExpireSnapshotsAction.FileKind.MANIFEST,
+ IcebergExpireSnapshotsAction.metadataFileKindByName(
+ dir +
"optimized-m-0a1b2c3d-4e5f-6071-8293-a4b5c6d7e8f9.avro"));
+ // Data files, including Avro ones, and statistics files are left to
the callback's other branches.
+
Assertions.assertNull(IcebergExpireSnapshotsAction.metadataFileKindByName(
+
"s3://b/db1/t/data/00000-0-0a1b2c3d-4e5f-6071-8293-a4b5c6d7e8f9-00001.avro"));
+
Assertions.assertNull(IcebergExpireSnapshotsAction.metadataFileKindByName(
+
"s3://b/db1/t/data/00000-0-0a1b2c3d-4e5f-6071-8293-a4b5c6d7e8f9-00001.parquet"));
+
Assertions.assertNull(IcebergExpireSnapshotsAction.metadataFileKindByName(
+ dir + "1234-0a1b2c3d-4e5f-6071-8293-a4b5c6d7e8f9.stats"));
+ // Only the file name counts, not a directory that looks like a
manifest list.
+
Assertions.assertNull(IcebergExpireSnapshotsAction.metadataFileKindByName(
+
"s3://b/snap-dir/00000-0-0a1b2c3d-4e5f-6071-8293-a4b5c6d7e8f9-00001.avro"));
+ // Every data file is classified by name, so Avro data files whose
names only resemble metadata names are
+ // not metadata: the whole name has to match, UUID included.
+ for (String dataFile : ImmutableList.of("sales-m1.avro",
"sales-m-1.avro", "optimized-m-1.avro",
+ "snap-1.avro", "snap-1234-1-sales.avro")) {
+ Assertions.assertNull(
+
IcebergExpireSnapshotsAction.metadataFileKindByName("s3://b/db1/t/data/" +
dataFile), dataFile);
+ }
+ }
+
+ @Test
+ public void countsAnAvroDataFileNamedLikeAManifestAsADataFile() {
+ // The file kind map holds no data files, so every expired data file
is classified by its name. An Avro
+ // data file whose name ends in -m<N>.avro is still a data file.
+ InMemoryCatalog catalog = ActionTestTables.freshCatalog();
+ TableIdentifier id = ActionTestTables.id("t");
+ ActionTestTables.createTable(catalog, "t");
+ DataFile avroDataFile =
DataFiles.builder(PartitionSpec.unpartitioned())
+ .withPath("s3://b/db1/t/data/sales-m1.avro")
+ .withFileSizeInBytes(1024)
+ .withRecordCount(1L)
+ .withFormat(FileFormat.AVRO)
+ .build();
+ catalog.loadTable(id).newAppend().appendFile(avroDataFile).commit();
+
catalog.loadTable(id).newDelete().deleteFile(avroDataFile.location()).commit();
+ ActionTestTables.appendSnapshot(catalog, "t", "f2.parquet", 2L);
+
+ Table table = catalog.loadTable(id);
+ Set<String> retained = new HashSet<>();
+ for (ManifestFile manifest :
table.currentSnapshot().allManifests(table.io())) {
+ retained.add(manifest.path());
+ }
+ Set<String> expired = new HashSet<>();
+ for (Snapshot snapshot : table.snapshots()) {
+ for (ManifestFile manifest : snapshot.allManifests(table.io())) {
+ if (!retained.contains(manifest.path())) {
+ expired.add(manifest.path());
+ }
+ }
+ }
+ Assertions.assertFalse(expired.isEmpty(), "the last snapshot must not
reference every manifest");
+
+ IcebergExpireSnapshotsAction action =
+ action(ImmutableMap.of("older_than", "4102444800000",
"retain_last", "1"));
+ action.validate();
+ List<String> row = action.execute(table,
ActionTestTables.session("UTC")).getRows().get(0);
+ Assertions.assertEquals(ImmutableList.of("1", "0", "0",
String.valueOf(expired.size()), "2", "0"), row,
+ "sales-m1.avro is the only deleted data file; expired
manifests " + expired);
+ }
+
@Test
public void resultSchemaIsSixBigintCounters() {
List<ConnectorColumn> schema = action(ImmutableMap.of("retain_last",
"1")).getResultSchema();
@@ -124,7 +283,7 @@ public class IcebergExpireSnapshotsActionTest {
}
@Test
- public void
buildDeleteFileContentMapDedupsManifestsSharedAcrossSnapshots() {
+ public void buildFileKindMapDedupsManifestsSharedAcrossSnapshots() {
// WHY: expire_snapshots builds the delete-file -> content
classification map by scanning every snapshot's
// delete manifests. Iceberg carries immutable delete manifests
forward across snapshots, so the naive
// per-snapshot scan re-reads the same manifest once per referencing
snapshot. Deduping by manifest path
@@ -159,13 +318,15 @@ public class IcebergExpireSnapshotsActionTest {
+ " distinct=" + distinctDeleteManifests.size());
IcebergExpireSnapshotsAction action =
action(ImmutableMap.of("retain_last", "1"));
- Map<String, FileContent> contentByPath =
action.buildDeleteFileContentMap(table);
+ Map<String, IcebergExpireSnapshotsAction.FileKind> kindByPath =
action.buildFileKindMap(table);
Assertions.assertEquals(distinctDeleteManifests.size(),
action.lastDeleteManifestReadCount,
"each distinct delete manifest must be read exactly once
(dedup), not once per referencing snapshot");
- Assertions.assertEquals(FileContent.POSITION_DELETES,
contentByPath.get("s3://b/db1/pos-del.parquet"),
+
Assertions.assertEquals(IcebergExpireSnapshotsAction.FileKind.POSITION_DELETES,
+ kindByPath.get("s3://b/db1/pos-del.parquet"),
"the position-delete file must stay classified as
POSITION_DELETES");
- Assertions.assertEquals(FileContent.EQUALITY_DELETES,
contentByPath.get("s3://b/db1/eq-del.parquet"),
+
Assertions.assertEquals(IcebergExpireSnapshotsAction.FileKind.EQUALITY_DELETES,
+ kindByPath.get("s3://b/db1/eq-del.parquet"),
"the equality-delete file must stay classified as
EQUALITY_DELETES");
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
