From d3dbc7f891048906c0e426e7f83f44a5f847eb16 Mon Sep 17 00:00:00 2001 From: jrmccluskey Date: Thu, 10 Sep 2026 13:31:11 +0000 Subject: [PATCH 1/2] [Gemini] Add ExpireSnapshots maintenance operation to Iceberg IO --- .../iceberg/maintenance/DeleteFilesDoFn.java | 182 +++++++ .../iceberg/maintenance/ExpireSnapshots.java | 323 ++++++++++++ .../maintenance/ExpireSnapshotsResult.java | 152 ++++++ .../io/iceberg/maintenance/FileCategory.java | 39 ++ .../sdk/io/iceberg/maintenance/FileInfo.java | 79 +++ .../iceberg/maintenance/ManifestFileBean.java | 242 +++++++++ .../maintenance/PlanExpireSnapshotsDoFn.java | 232 +++++++++ .../iceberg/maintenance/ReadManifestDoFn.java | 115 +++++ .../io/iceberg/maintenance/package-info.java | 20 + .../maintenance/DeleteFilesDoFnTest.java | 337 ++++++++++++ .../ExpireSnapshotsResultTest.java | 141 ++++++ .../maintenance/ExpireSnapshotsTest.java | 478 ++++++++++++++++++ .../ExpireSnapshotsTestFixtures.java | 197 ++++++++ .../PlanExpireSnapshotsDoFnTest.java | 177 +++++++ .../maintenance/ReadManifestDoFnTest.java | 154 ++++++ .../io/iceberg/maintenance/package-info.java | 20 + 16 files changed, 2888 insertions(+) create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFn.java create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshots.java create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResult.java create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/FileCategory.java create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/FileInfo.java create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBean.java create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFn.java create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFn.java create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/package-info.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFnTest.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResultTest.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTest.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTestFixtures.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFnTest.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFnTest.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/package-info.java diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFn.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFn.java new file mode 100644 index 000000000000..1bb5447b1b2c --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFn.java @@ -0,0 +1,182 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; +import org.apache.beam.sdk.io.iceberg.IcebergUtils; +import org.apache.beam.sdk.io.iceberg.TableCache; +import org.apache.beam.sdk.metrics.Metrics; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.util.ShardedKey; +import org.apache.beam.sdk.values.KV; +import org.apache.iceberg.Table; +import org.apache.iceberg.exceptions.NotFoundException; +import org.apache.iceberg.io.BulkDeletionFailureException; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.SupportsBulkOperations; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Worker DoFn that deletes batched unreferenced files from storage and emits partial deletion + * metrics. + */ +public class DeleteFilesDoFn + extends DoFn, Iterable>, ExpireSnapshotsResult> { + + private static final Logger LOG = LoggerFactory.getLogger(DeleteFilesDoFn.class); + + private final IcebergCatalogConfig catalogConfig; + private final ExpireSnapshots.Configuration config; + + public DeleteFilesDoFn(IcebergCatalogConfig catalogConfig, ExpireSnapshots.Configuration config) { + this.catalogConfig = catalogConfig; + this.config = config; + } + + @ProcessElement + public void processElement( + @Element KV, Iterable> element, + OutputReceiver out) { + String tableIdString = element.getKey().getKey(); + if (tableIdString == null || tableIdString.isEmpty()) { + return; + } + + Table table = + TableCache.getAndRefreshIfStale( + catalogConfig, IcebergUtils.parseTableIdentifier(tableIdString)); + FileIO io = table.io(); + + List fileList = new ArrayList<>(); + List paths = new ArrayList<>(); + for (FileInfo file : element.getValue()) { + fileList.add(file); + paths.add(file.getPath()); + } + + if (paths.isEmpty()) { + return; + } + + Set failedPaths = Collections.emptySet(); + if (config.cleanFiles()) { + failedPaths = deletePaths(io, paths); + if (!failedPaths.isEmpty()) { + Metrics.counter(DeleteFilesDoFn.class, "failed_file_deletions").inc(failedPaths.size()); + } + } else { + LOG.info( + ExpireSnapshots.PREFIX + + "Dry run enabled (cleanFiles=false); skipping physical deletion of {} file(s).", + paths.size()); + } + + long dataCount = 0; + long posDeleteCount = 0; + long eqDeleteCount = 0; + long manifestCount = 0; + long manifestListCount = 0; + long statsCount = 0; + + for (FileInfo file : fileList) { + if (failedPaths.contains(file.getPath())) { + continue; + } + switch (file.fileCategory()) { + case DATA: + dataCount++; + break; + case POSITION_DELETES: + posDeleteCount++; + break; + case EQUALITY_DELETES: + eqDeleteCount++; + break; + case MANIFEST: + manifestCount++; + break; + case MANIFEST_LIST: + manifestListCount++; + break; + case STATISTICS: + statsCount++; + break; + } + } + + out.output( + ExpireSnapshotsResult.builder() + .setDeletedDataFilesCount(dataCount) + .setDeletedPositionDeleteFilesCount(posDeleteCount) + .setDeletedEqualityDeleteFilesCount(eqDeleteCount) + .setDeletedManifestsCount(manifestCount) + .setDeletedManifestListsCount(manifestListCount) + .setDeletedStatisticsFilesCount(statsCount) + .build()); + } + + /** + * Deletes {@code paths} using {@link SupportsBulkOperations} when supported by {@link FileIO}, + * falling back to per-file deletion. Silently ignores {@link NotFoundException}. + * + * @return set of paths that failed to be deleted + */ + static Set deletePaths(FileIO io, List paths) { + Set failedPaths = new HashSet<>(); + if (paths.isEmpty()) { + return failedPaths; + } + if (io instanceof SupportsBulkOperations) { + try { + ((SupportsBulkOperations) io).deleteFiles(paths); + return failedPaths; + } catch (BulkDeletionFailureException e) { + LOG.warn( + ExpireSnapshots.PREFIX + + "Bulk delete failed for {} of {} files. Retrying individually.", + e.numberFailedObjects(), + paths.size(), + e); + } catch (RuntimeException e) { + LOG.warn( + ExpireSnapshots.PREFIX + + "Bulk delete raised non-bulk exception; falling back to per-file deletion.", + e); + } + } + + for (String path : paths) { + try { + io.deleteFile(path); + } catch (NotFoundException e) { + LOG.debug( + ExpireSnapshots.PREFIX + "File {} not found during deletion (already removed).", path); + } catch (Exception e) { + LOG.warn(ExpireSnapshots.PREFIX + "Failed to delete file {}.", path, e); + failedPaths.add(path); + } + } + return failedPaths; + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshots.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshots.java new file mode 100644 index 000000000000..8a06c809f722 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshots.java @@ -0,0 +1,323 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import com.google.auto.value.AutoValue; +import java.io.Serializable; +import java.util.List; +import org.apache.beam.sdk.coders.IterableCoder; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.SerializableCoder; +import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; +import org.apache.beam.sdk.schemas.AutoValueSchema; +import org.apache.beam.sdk.schemas.NoSuchSchemaException; +import org.apache.beam.sdk.schemas.SchemaCoder; +import org.apache.beam.sdk.schemas.annotations.DefaultSchema; +import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription; +import org.apache.beam.sdk.transforms.Combine; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.transforms.Flatten; +import org.apache.beam.sdk.transforms.GroupByKey; +import org.apache.beam.sdk.transforms.GroupIntoBatches; +import org.apache.beam.sdk.transforms.MapElements; +import org.apache.beam.sdk.transforms.PTransform; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.transforms.Redistribute; +import org.apache.beam.sdk.transforms.display.DisplayData; +import org.apache.beam.sdk.util.ShardedKey; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.sdk.values.PCollectionList; +import org.apache.beam.sdk.values.PCollectionTuple; +import org.apache.beam.sdk.values.TupleTagList; +import org.apache.beam.sdk.values.TypeDescriptors; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.checkerframework.dataflow.qual.Pure; + +/** + * Distributed snapshot expiration maintenance operation for Apache Iceberg tables. + * + *

Prunes snapshots older than a configured retention threshold, retains the last $N$ ancestors, + * removes unreferenced manifests and manifest lists, and physically deletes obsolete data and + * delete files from storage. + * + *

Execution Model

+ * + *
    + *
  1. Phase 1 (Planning): Runs inside an initial worker {@link PlanExpireSnapshotsDoFn}. + * Validates table GC settings and commits the metadata change via {@code + * table.expireSnapshots().cleanExpiredFiles(false).commit()}. Zero file deletions happen on + * the driver. + *
  2. Phase 2 (Manifest Scanning): Manifest files from candidate and retained snapshots + * are redistributed and read in parallel across workers via {@link ReadManifestDoFn}. + *
  3. Phase 3 (Anti-Join): Candidate and retained files are keyed by file path and + * evaluated in a distributed anti-join. Any file referenced by any active snapshot is + * preserved. + *
  4. Phase 4 (Physical Deletion): Unreferenced files are batched and deleted in parallel + * across workers via key-sharded {@link GroupIntoBatches} and {@link DeleteFilesDoFn}, + * utilizing bulk object deletion where supported. + *
  5. Phase 5 (Aggregation): Deletion counts and expired snapshot counts are merged into a + * single {@link ExpireSnapshotsResult}. + *
+ */ +public class ExpireSnapshots + extends PTransform, PCollection> { + + public static final String PREFIX = "[ExpireSnapshots] "; + + private final IcebergCatalogConfig catalogConfig; + private final Configuration config; + + ExpireSnapshots(IcebergCatalogConfig catalogConfig, Configuration config) { + this.catalogConfig = catalogConfig; + this.config = config; + } + + public static ExpireSnapshots create(IcebergCatalogConfig catalogConfig) { + return new ExpireSnapshots(catalogConfig, Configuration.builder().build()); + } + + public static ExpireSnapshots create(IcebergCatalogConfig catalogConfig, Configuration config) { + return new ExpireSnapshots(catalogConfig, config); + } + + @Override + public void populateDisplayData(DisplayData.Builder builder) { + super.populateDisplayData(builder); + builder.addIfNotNull( + DisplayData.item("expireOlderThan", config.getExpireOlderThan()) + .withLabel("Expire Older Than (Millis)")); + builder.add( + DisplayData.item("retainLast", config.retainLast()).withLabel("Retain Last Snapshots")); + builder.add( + DisplayData.item("cleanFiles", config.cleanFiles()) + .withLabel("Clean Files (Physical Deletion)")); + builder.add( + DisplayData.item("cleanExpiredMetadata", config.cleanExpiredMetadata()) + .withLabel("Clean Expired Metadata")); + builder.add( + DisplayData.item("deleteBatchSize", config.deleteBatchSize()) + .withLabel("Delete Batch Size")); + } + + @Override + public PCollection expand(PCollection tableIdentifiers) { + Preconditions.checkArgument( + tableIdentifiers.isBounded() == PCollection.IsBounded.BOUNDED, + "ExpireSnapshots only supports bounded (batch) input."); + config.validate(); + + SchemaCoder fileInfoCoder; + SchemaCoder resultCoder; + try { + fileInfoCoder = + tableIdentifiers.getPipeline().getSchemaRegistry().getSchemaCoder(FileInfo.class); + resultCoder = + tableIdentifiers + .getPipeline() + .getSchemaRegistry() + .getSchemaCoder(ExpireSnapshotsResult.class); + } catch (NoSuchSchemaException e) { + throw new RuntimeException("Failed to load schema coders for ExpireSnapshots", e); + } + + KvCoder kvFileInfoCoder = KvCoder.of(StringUtf8Coder.of(), fileInfoCoder); + + // Phase 1: Planning and metadata commit + PCollectionTuple planned = + tableIdentifiers.apply( + "Plan Expire Snapshots", + ParDo.of(new PlanExpireSnapshotsDoFn(catalogConfig, config)) + .withOutputTags( + PlanExpireSnapshotsDoFn.PLAN_SUMMARY, + TupleTagList.of(PlanExpireSnapshotsDoFn.MANIFESTS) + .and(PlanExpireSnapshotsDoFn.DIRECT_FILES))); + + PCollection planSummary = + planned.get(PlanExpireSnapshotsDoFn.PLAN_SUMMARY).setCoder(resultCoder); + + PCollection> directFiles = + planned.get(PlanExpireSnapshotsDoFn.DIRECT_FILES).setCoder(kvFileInfoCoder); + + // Phase 2: Distributed manifest scanning + PCollection> manifestEntries = + planned + .get(PlanExpireSnapshotsDoFn.MANIFESTS) + .setCoder( + KvCoder.of(StringUtf8Coder.of(), SerializableCoder.of(ManifestFileBean.class))) + .apply("Redistribute Manifests", Redistribute.arbitrarily()) + .apply("Read Manifest Entries", ParDo.of(new ReadManifestDoFn(catalogConfig))) + .setCoder(kvFileInfoCoder); + + // Phase 3: Distributed anti-join + PCollection filesToDelete = + PCollectionList.of(directFiles) + .and(manifestEntries) + .apply("Flatten All Files", Flatten.pCollections()) + .setCoder(kvFileInfoCoder) + .apply("Group by Path", GroupByKey.create()) + .apply("Anti-Join Filter", ParDo.of(new AntiJoinFilterFn())) + .setCoder(fileInfoCoder); + + // Phase 4: Batched file deletion + PCollection deletionSummary = + filesToDelete + .apply( + "Key for Batching", + MapElements.into( + TypeDescriptors.kvs( + TypeDescriptors.strings(), + org.apache.beam.sdk.values.TypeDescriptor.of(FileInfo.class))) + .via( + file -> + KV.of(MoreObjects.firstNonNull(file.getTableIdentifier(), ""), file))) + .setCoder(KvCoder.of(StringUtf8Coder.of(), fileInfoCoder)) + .apply( + "Batch Files", + GroupIntoBatches.ofSize(config.deleteBatchSize()) + .withShardedKey()) + .setCoder( + KvCoder.of( + ShardedKey.Coder.of(StringUtf8Coder.of()), IterableCoder.of(fileInfoCoder))) + .apply("Delete Files", ParDo.of(new DeleteFilesDoFn(catalogConfig, config))) + .setCoder(resultCoder); + + // Phase 5: Global metric aggregation + return PCollectionList.of(planSummary) + .and(deletionSummary) + .apply("Flatten Result Fragments", Flatten.pCollections()) + .setCoder(resultCoder) + .apply("Merge into Final Result", Combine.globally(new ExpireSnapshotsResult.Merge())); + } + + /** Filters grouped file entries: emits candidate if NO valid reference exists. */ + static class AntiJoinFilterFn extends DoFn>, FileInfo> { + @ProcessElement + public void process( + @Element KV> element, OutputReceiver out) { + boolean isValid = false; + FileInfo candidate = null; + + for (FileInfo info : element.getValue()) { + if (info.getValid()) { + isValid = true; + break; + } + if (candidate == null) { + candidate = info; + } + } + + if (!isValid && candidate != null) { + out.output(candidate); + } + } + } + + /** Configuration options for {@link ExpireSnapshots}. */ + @AutoValue + @DefaultSchema(AutoValueSchema.class) + public abstract static class Configuration implements Serializable { + + public static Builder builder() { + return new AutoValue_ExpireSnapshots_Configuration.Builder() + .setRetainLast(1) + .setCleanFiles(true) + .setDeleteBatchSize(10_000); + } + + @SchemaFieldDescription( + "Cutoff timestamp in milliseconds. Snapshots older than this are expired.") + @Pure + public abstract @Nullable Long getExpireOlderThan(); + + @SchemaFieldDescription( + "Safety floor: minimum number of ancestor snapshots to retain. Must be >= 1.") + @Pure + public abstract @Nullable Integer getRetainLast(); + + @SchemaFieldDescription("Explicit snapshot IDs to expire.") + @Pure + public abstract @Nullable List getSnapshotIds(); + + @SchemaFieldDescription( + "Whether to clean up unused partition specs and schemas no longer referenced by any snapshot.") + @Pure + public abstract @Nullable Boolean getCleanExpiredMetadata(); + + @SchemaFieldDescription( + "Whether to physically delete unreferenced files from storage. If false, acts as a dry run.") + @Pure + public abstract @Nullable Boolean getCleanFiles(); + + @SchemaFieldDescription("Batch size for bulk file deletion calls. Default is 10,000.") + @Pure + public abstract @Nullable Integer getDeleteBatchSize(); + + public int retainLast() { + return MoreObjects.firstNonNull(getRetainLast(), 1); + } + + public boolean cleanExpiredMetadata() { + return MoreObjects.firstNonNull(getCleanExpiredMetadata(), false); + } + + public boolean cleanFiles() { + return MoreObjects.firstNonNull(getCleanFiles(), true); + } + + public int deleteBatchSize() { + return MoreObjects.firstNonNull(getDeleteBatchSize(), 10_000); + } + + public void validate() { + if (getRetainLast() != null) { + Preconditions.checkArgument( + getRetainLast() >= 1, + "retainLast must be at least 1 to prevent deleting current table state, got %s", + getRetainLast()); + } + if (getDeleteBatchSize() != null) { + Preconditions.checkArgument( + getDeleteBatchSize() > 0, + "deleteBatchSize must be positive, got %s", + getDeleteBatchSize()); + } + } + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setExpireOlderThan(@Nullable Long millis); + + public abstract Builder setRetainLast(@Nullable Integer retainLast); + + public abstract Builder setSnapshotIds(@Nullable List snapshotIds); + + public abstract Builder setCleanExpiredMetadata(@Nullable Boolean clean); + + public abstract Builder setCleanFiles(@Nullable Boolean cleanFiles); + + public abstract Builder setDeleteBatchSize(@Nullable Integer size); + + public abstract Configuration build(); + } + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResult.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResult.java new file mode 100644 index 000000000000..e8166f530e99 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResult.java @@ -0,0 +1,152 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import com.google.auto.value.AutoValue; +import java.io.Serializable; +import org.apache.beam.sdk.schemas.AutoValueSchema; +import org.apache.beam.sdk.schemas.annotations.DefaultSchema; +import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; +import org.apache.beam.sdk.transforms.Combine; + +/** + * Structured summary metrics of an {@link ExpireSnapshots} run. + * + *

All counts default to 0. An empty or no-op run produces an all-zeros result. When {@code + * cleanFiles} is set to {@code false} (dry run mode), file counts reflect unreferenced candidate + * files identified for deletion rather than files actually deleted from storage. + */ +@AutoValue +@DefaultSchema(AutoValueSchema.class) +public abstract class ExpireSnapshotsResult implements Serializable { + + /** Number of physical data files removed from storage. */ + @SchemaFieldNumber("0") + public abstract long getDeletedDataFilesCount(); + + /** Number of position delete files removed from storage. */ + @SchemaFieldNumber("1") + public abstract long getDeletedPositionDeleteFilesCount(); + + /** Number of equality delete files removed from storage. */ + @SchemaFieldNumber("2") + public abstract long getDeletedEqualityDeleteFilesCount(); + + /** Number of manifest files removed from storage. */ + @SchemaFieldNumber("3") + public abstract long getDeletedManifestsCount(); + + /** Number of manifest lists removed from storage. */ + @SchemaFieldNumber("4") + public abstract long getDeletedManifestListsCount(); + + /** Number of Puffin statistics files removed from storage. */ + @SchemaFieldNumber("5") + public abstract long getDeletedStatisticsFilesCount(); + + /** Number of snapshots expired from table metadata. */ + @SchemaFieldNumber("6") + public abstract long getExpiredSnapshotsCount(); + + /** A builder with every count pre-set to 0. */ + public static Builder builder() { + return new AutoValue_ExpireSnapshotsResult.Builder() + .setDeletedDataFilesCount(0L) + .setDeletedPositionDeleteFilesCount(0L) + .setDeletedEqualityDeleteFilesCount(0L) + .setDeletedManifestsCount(0L) + .setDeletedManifestListsCount(0L) + .setDeletedStatisticsFilesCount(0L) + .setExpiredSnapshotsCount(0L); + } + + /** The all-zeros identity result. */ + public static ExpireSnapshotsResult zeros() { + return builder().build(); + } + + /** Field-wise sum of two result fragments. */ + public static ExpireSnapshotsResult merge(ExpireSnapshotsResult a, ExpireSnapshotsResult b) { + return builder() + .setDeletedDataFilesCount(a.getDeletedDataFilesCount() + b.getDeletedDataFilesCount()) + .setDeletedPositionDeleteFilesCount( + a.getDeletedPositionDeleteFilesCount() + b.getDeletedPositionDeleteFilesCount()) + .setDeletedEqualityDeleteFilesCount( + a.getDeletedEqualityDeleteFilesCount() + b.getDeletedEqualityDeleteFilesCount()) + .setDeletedManifestsCount(a.getDeletedManifestsCount() + b.getDeletedManifestsCount()) + .setDeletedManifestListsCount( + a.getDeletedManifestListsCount() + b.getDeletedManifestListsCount()) + .setDeletedStatisticsFilesCount( + a.getDeletedStatisticsFilesCount() + b.getDeletedStatisticsFilesCount()) + .setExpiredSnapshotsCount(a.getExpiredSnapshotsCount() + b.getExpiredSnapshotsCount()) + .build(); + } + + /** + * Sums per-stage or per-worker result fragments into a single final result. The identity is + * {@link #zeros()}, so {@code Combine.globally} in a bounded global window emits one all-zeros + * row even on empty input. + */ + public static class Merge + extends Combine.CombineFn< + ExpireSnapshotsResult, ExpireSnapshotsResult, ExpireSnapshotsResult> { + @Override + public ExpireSnapshotsResult createAccumulator() { + return zeros(); + } + + @Override + public ExpireSnapshotsResult addInput( + ExpireSnapshotsResult accumulator, ExpireSnapshotsResult input) { + return merge(accumulator, input); + } + + @Override + public ExpireSnapshotsResult mergeAccumulators(Iterable accumulators) { + ExpireSnapshotsResult merged = zeros(); + for (ExpireSnapshotsResult acc : accumulators) { + merged = merge(merged, acc); + } + return merged; + } + + @Override + public ExpireSnapshotsResult extractOutput(ExpireSnapshotsResult accumulator) { + return accumulator; + } + } + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setDeletedDataFilesCount(long count); + + public abstract Builder setDeletedPositionDeleteFilesCount(long count); + + public abstract Builder setDeletedEqualityDeleteFilesCount(long count); + + public abstract Builder setDeletedManifestsCount(long count); + + public abstract Builder setDeletedManifestListsCount(long count); + + public abstract Builder setDeletedStatisticsFilesCount(long count); + + public abstract Builder setExpiredSnapshotsCount(long count); + + public abstract ExpireSnapshotsResult build(); + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/FileCategory.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/FileCategory.java new file mode 100644 index 000000000000..622e46a167bd --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/FileCategory.java @@ -0,0 +1,39 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +/** Categorization of files managed during Iceberg table maintenance operations. */ +public enum FileCategory { + /** Physical data file containing user rows. */ + DATA, + + /** Row-level position delete file. */ + POSITION_DELETES, + + /** Row-level equality delete file. */ + EQUALITY_DELETES, + + /** Iceberg manifest file referencing data or delete files. */ + MANIFEST, + + /** Iceberg manifest list file referencing manifest files. */ + MANIFEST_LIST, + + /** Iceberg Puffin statistics file. */ + STATISTICS +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/FileInfo.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/FileInfo.java new file mode 100644 index 000000000000..b4e24b73925f --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/FileInfo.java @@ -0,0 +1,79 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import com.google.auto.value.AutoValue; +import java.io.Serializable; +import org.apache.beam.sdk.schemas.AutoValueSchema; +import org.apache.beam.sdk.schemas.annotations.DefaultSchema; +import org.apache.beam.sdk.schemas.annotations.SchemaFieldNumber; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions; + +/** Represents a file entry evaluated during snapshot expiration. */ +@AutoValue +@DefaultSchema(AutoValueSchema.class) +public abstract class FileInfo implements Serializable { + + /** Fully qualified URI or storage path of the file. */ + @SchemaFieldNumber("0") + public abstract String getPath(); + + /** Category of the file (data, delete, manifest, manifest list, statistics). */ + @SchemaFieldNumber("1") + public abstract String getCategory(); + + /** Whether this file is reachable and valid in a retained snapshot. */ + @SchemaFieldNumber("2") + public abstract boolean getValid(); + + /** Associated Iceberg table identifier string. */ + @SchemaFieldNumber("3") + public abstract String getTableIdentifier(); + + public static Builder builder() { + return new AutoValue_FileInfo.Builder(); + } + + public static FileInfo of( + String path, FileCategory category, boolean valid, String tableIdentifier) { + Preconditions.checkNotNull(tableIdentifier, "tableIdentifier must not be null"); + return builder() + .setPath(path) + .setCategory(category.name()) + .setValid(valid) + .setTableIdentifier(tableIdentifier) + .build(); + } + + public FileCategory fileCategory() { + return FileCategory.valueOf(getCategory()); + } + + @AutoValue.Builder + public abstract static class Builder { + public abstract Builder setPath(String path); + + public abstract Builder setCategory(String category); + + public abstract Builder setValid(boolean valid); + + public abstract Builder setTableIdentifier(String tableIdentifier); + + public abstract FileInfo build(); + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBean.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBean.java new file mode 100644 index 000000000000..bd3897a1dd65 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBean.java @@ -0,0 +1,242 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import java.io.Serializable; +import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; +import org.apache.iceberg.ManifestContent; +import org.apache.iceberg.ManifestFile; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Lightweight serializable implementation of {@link ManifestFile} used to fan out manifest reading + * across Beam workers. + */ +@SuppressWarnings("nullness") +public class ManifestFileBean implements ManifestFile, Serializable { + + private final String path; + private final long length; + private final int partitionSpecId; + private final ManifestContent content; + private final @Nullable Long snapshotId; + private final @Nullable Integer addedFilesCount; + private final @Nullable Long addedRowsCount; + private final @Nullable Integer existingFilesCount; + private final @Nullable Long existingRowsCount; + private final @Nullable Integer deletedFilesCount; + private final @Nullable Long deletedRowsCount; + private final long sequenceNumber; + private final long minSequenceNumber; + private final boolean valid; + + public ManifestFileBean( + String path, + long length, + int partitionSpecId, + ManifestContent content, + @Nullable Long snapshotId, + boolean valid) { + this( + path, + length, + partitionSpecId, + content, + snapshotId, + null, + null, + null, + null, + null, + null, + 0L, + 0L, + valid); + } + + public ManifestFileBean( + String path, + long length, + int partitionSpecId, + ManifestContent content, + @Nullable Long snapshotId, + @Nullable Integer addedFilesCount, + @Nullable Long addedRowsCount, + @Nullable Integer existingFilesCount, + @Nullable Long existingRowsCount, + @Nullable Integer deletedFilesCount, + @Nullable Long deletedRowsCount, + long sequenceNumber, + long minSequenceNumber, + boolean valid) { + this.path = path; + this.length = length; + this.partitionSpecId = partitionSpecId; + this.content = content; + this.snapshotId = snapshotId; + this.addedFilesCount = addedFilesCount; + this.addedRowsCount = addedRowsCount; + this.existingFilesCount = existingFilesCount; + this.existingRowsCount = existingRowsCount; + this.deletedFilesCount = deletedFilesCount; + this.deletedRowsCount = deletedRowsCount; + this.sequenceNumber = sequenceNumber; + this.minSequenceNumber = minSequenceNumber; + this.valid = valid; + } + + public static ManifestFileBean fromManifestFile(ManifestFile manifest, boolean valid) { + return new ManifestFileBean( + manifest.path(), + manifest.length(), + manifest.partitionSpecId(), + manifest.content() != null ? manifest.content() : ManifestContent.DATA, + manifest.snapshotId(), + manifest.addedFilesCount(), + manifest.addedRowsCount(), + manifest.existingFilesCount(), + manifest.existingRowsCount(), + manifest.deletedFilesCount(), + manifest.deletedRowsCount(), + manifest.sequenceNumber(), + manifest.minSequenceNumber(), + valid); + } + + public boolean isValid() { + return valid; + } + + @Override + public String path() { + return path; + } + + @Override + public long length() { + return length; + } + + @Override + public int partitionSpecId() { + return partitionSpecId; + } + + @Override + public ManifestContent content() { + return content; + } + + @Override + public @Nullable Long snapshotId() { + return snapshotId; + } + + @Override + public @Nullable Integer addedFilesCount() { + return addedFilesCount; + } + + @Override + public @Nullable Long addedRowsCount() { + return addedRowsCount; + } + + @Override + public @Nullable Integer existingFilesCount() { + return existingFilesCount; + } + + @Override + public @Nullable Long existingRowsCount() { + return existingRowsCount; + } + + @Override + public @Nullable Integer deletedFilesCount() { + return deletedFilesCount; + } + + @Override + public @Nullable Long deletedRowsCount() { + return deletedRowsCount; + } + + @Override + public List partitions() { + return Collections.emptyList(); + } + + @Override + public @Nullable ByteBuffer keyMetadata() { + return null; + } + + @Override + public ManifestFile copy() { + return this; + } + + @Override + public long sequenceNumber() { + return sequenceNumber; + } + + @Override + public long minSequenceNumber() { + return minSequenceNumber; + } + + @Override + public boolean equals(@Nullable Object o) { + if (this == o) { + return true; + } + if (!(o instanceof ManifestFileBean)) { + return false; + } + ManifestFileBean that = (ManifestFileBean) o; + return length == that.length + && partitionSpecId == that.partitionSpecId + && valid == that.valid + && Objects.equals(path, that.path) + && content == that.content + && Objects.equals(snapshotId, that.snapshotId); + } + + @Override + public int hashCode() { + return Objects.hash(path, length, partitionSpecId, content, snapshotId, valid); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("path", path) + .add("length", length) + .add("partitionSpecId", partitionSpecId) + .add("content", content) + .add("snapshotId", snapshotId) + .add("valid", valid) + .toString(); + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFn.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFn.java new file mode 100644 index 000000000000..351826887d77 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFn.java @@ -0,0 +1,232 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; +import org.apache.beam.sdk.io.iceberg.IcebergUtils; +import org.apache.beam.sdk.io.iceberg.TableCache; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.TupleTag; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.PartitionStatisticsFile; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.StatisticsFile; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.util.PropertyUtil; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Execution-time planning DoFn for {@link ExpireSnapshots}. + * + *

Validates {@code GC_ENABLED}, invokes {@code + * table.expireSnapshots().cleanExpiredFiles(false).commit()} to atomically prune snapshots from + * metadata, and emits candidate and valid file descriptors for distributed content resolution. + */ +public class PlanExpireSnapshotsDoFn extends DoFn { + + private static final Logger LOG = LoggerFactory.getLogger(PlanExpireSnapshotsDoFn.class); + + public static final TupleTag PLAN_SUMMARY = new TupleTag<>() {}; + public static final TupleTag> MANIFESTS = new TupleTag<>() {}; + public static final TupleTag> DIRECT_FILES = new TupleTag<>() {}; + + private final IcebergCatalogConfig catalogConfig; + private final ExpireSnapshots.Configuration config; + + public PlanExpireSnapshotsDoFn( + IcebergCatalogConfig catalogConfig, ExpireSnapshots.Configuration config) { + this.catalogConfig = catalogConfig; + this.config = config; + } + + @ProcessElement + public void processElement(@Element String tableIdString, MultiOutputReceiver out) { + TableIdentifier tableId = IcebergUtils.parseTableIdentifier(tableIdString); + Table table = TableCache.getAndRefreshIfStale(catalogConfig, tableId); + + boolean gcEnabled = + PropertyUtil.propertyAsBoolean( + table.properties(), TableProperties.GC_ENABLED, TableProperties.GC_ENABLED_DEFAULT); + if (!gcEnabled) { + throw new ValidationException( + "Cannot expire snapshots: GC is disabled (deleting files may corrupt other tables)"); + } + + List originalSnapshots = Lists.newArrayList(table.snapshots()); + if (originalSnapshots.isEmpty()) { + LOG.info( + ExpireSnapshots.PREFIX + "Table '{}' has no snapshots; expiration is a no-op.", tableId); + out.get(PLAN_SUMMARY).output(ExpireSnapshotsResult.zeros()); + return; + } + + org.apache.iceberg.ExpireSnapshots expire = table.expireSnapshots(); + if (config.getExpireOlderThan() != null) { + expire = expire.expireOlderThan(config.getExpireOlderThan()); + } + if (config.getRetainLast() != null) { + expire = expire.retainLast(config.retainLast()); + } + if (config.getSnapshotIds() != null) { + for (Long id : config.getSnapshotIds()) { + expire = expire.expireSnapshotId(id); + } + } + if (config.getCleanExpiredMetadata() != null) { + expire = expire.cleanExpiredMetadata(config.cleanExpiredMetadata()); + } + + List originalStats = Lists.newArrayList(table.statisticsFiles()); + List originalPartitionStats = + Lists.newArrayList(table.partitionStatisticsFiles()); + LOG.info( + ExpireSnapshots.PREFIX + + "Committing snapshot expiration with cleanExpiredFiles(false) on table '{}'.", + tableId); + expire.cleanExpiredFiles(false).commit(); + + table.refresh(); + Set retainedSnapshotIds = new HashSet<>(); + for (Snapshot s : table.snapshots()) { + retainedSnapshotIds.add(s.snapshotId()); + } + + Set deletedSnapshotIds = new HashSet<>(); + for (Snapshot s : originalSnapshots) { + if (!retainedSnapshotIds.contains(s.snapshotId())) { + deletedSnapshotIds.add(s.snapshotId()); + } + } + + if (deletedSnapshotIds.isEmpty()) { + LOG.info( + ExpireSnapshots.PREFIX + "No snapshots expired for table '{}'; expiration is a no-op.", + tableId); + out.get(PLAN_SUMMARY).output(ExpireSnapshotsResult.zeros()); + return; + } + + LOG.info( + ExpireSnapshots.PREFIX + "Expired {} snapshot(s) {} from table '{}'.", + deletedSnapshotIds.size(), + deletedSnapshotIds, + tableId); + + out.get(PLAN_SUMMARY) + .output( + ExpireSnapshotsResult.builder() + .setExpiredSnapshotsCount((long) deletedSnapshotIds.size()) + .build()); + + // 1. Emit Manifest Lists + for (Snapshot s : table.snapshots()) { + if (s.manifestListLocation() != null) { + String path = s.manifestListLocation(); + out.get(DIRECT_FILES) + .output( + KV.of(path, FileInfo.of(path, FileCategory.MANIFEST_LIST, true, tableIdString))); + } + } + + for (Snapshot s : originalSnapshots) { + if (deletedSnapshotIds.contains(s.snapshotId()) && s.manifestListLocation() != null) { + String path = s.manifestListLocation(); + out.get(DIRECT_FILES) + .output( + KV.of(path, FileInfo.of(path, FileCategory.MANIFEST_LIST, false, tableIdString))); + } + } + + // 2. Emit Statistics Files + for (StatisticsFile sf : originalStats) { + if (retainedSnapshotIds.contains(sf.snapshotId())) { + out.get(DIRECT_FILES) + .output( + KV.of( + sf.path(), + FileInfo.of(sf.path(), FileCategory.STATISTICS, true, tableIdString))); + } else if (deletedSnapshotIds.contains(sf.snapshotId())) { + out.get(DIRECT_FILES) + .output( + KV.of( + sf.path(), + FileInfo.of(sf.path(), FileCategory.STATISTICS, false, tableIdString))); + } + } + + for (PartitionStatisticsFile psf : originalPartitionStats) { + if (retainedSnapshotIds.contains(psf.snapshotId())) { + out.get(DIRECT_FILES) + .output( + KV.of( + psf.path(), + FileInfo.of(psf.path(), FileCategory.STATISTICS, true, tableIdString))); + } else if (deletedSnapshotIds.contains(psf.snapshotId())) { + out.get(DIRECT_FILES) + .output( + KV.of( + psf.path(), + FileInfo.of(psf.path(), FileCategory.STATISTICS, false, tableIdString))); + } + } + + // 3. Emit Manifest Files + Set validManifestPaths = new HashSet<>(); + for (Snapshot s : table.snapshots()) { + for (ManifestFile m : s.allManifests(table.io())) { + if (validManifestPaths.add(m.path())) { + out.get(DIRECT_FILES) + .output( + KV.of( + m.path(), FileInfo.of(m.path(), FileCategory.MANIFEST, true, tableIdString))); + out.get(MANIFESTS) + .output(KV.of(tableIdString, ManifestFileBean.fromManifestFile(m, true))); + } + } + } + + Set candidateManifestPaths = new HashSet<>(); + for (Snapshot s : originalSnapshots) { + if (deletedSnapshotIds.contains(s.snapshotId())) { + for (ManifestFile m : s.allManifests(table.io())) { + if (candidateManifestPaths.add(m.path())) { + out.get(DIRECT_FILES) + .output( + KV.of( + m.path(), + FileInfo.of(m.path(), FileCategory.MANIFEST, false, tableIdString))); + // If the manifest is already part of a retained snapshot, its files are all valid + if (!validManifestPaths.contains(m.path())) { + out.get(MANIFESTS) + .output(KV.of(tableIdString, ManifestFileBean.fromManifestFile(m, false))); + } + } + } + } + } + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFn.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFn.java new file mode 100644 index 000000000000..b5fa5ae458b0 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFn.java @@ -0,0 +1,115 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; +import org.apache.beam.sdk.io.iceberg.IcebergUtils; +import org.apache.beam.sdk.io.iceberg.TableCache; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.ManifestContent; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Table; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.FileIO; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Worker DoFn that reads entries from a manifest file and emits tagged {@link FileInfo} records. + */ +public class ReadManifestDoFn extends DoFn, KV> { + + private static final List PROJECTION = + ImmutableList.of(DataFile.FILE_PATH.name(), DataFile.CONTENT.name()); + + private final IcebergCatalogConfig catalogConfig; + + private transient @Nullable String cachedTableIdString; + private transient @Nullable Table cachedTable; + private transient @Nullable FileIO cachedIo; + private transient @Nullable Map cachedSpecs; + + public ReadManifestDoFn(IcebergCatalogConfig catalogConfig) { + this.catalogConfig = catalogConfig; + } + + @ProcessElement + public void processElement( + @Element KV element, OutputReceiver> out) + throws IOException { + String tableIdString = element.getKey(); + ManifestFileBean manifest = element.getValue(); + + Table table; + FileIO io; + Map specs; + if (Objects.equals(cachedTableIdString, tableIdString) + && cachedTable != null + && cachedIo != null + && cachedSpecs != null) { + table = cachedTable; + io = cachedIo; + specs = cachedSpecs; + } else { + table = + TableCache.getAndRefreshIfStale( + catalogConfig, IcebergUtils.parseTableIdentifier(tableIdString)); + io = table.io(); + specs = table.specs(); + cachedTableIdString = tableIdString; + cachedTable = table; + cachedIo = io; + cachedSpecs = specs; + } + + ManifestContent content = manifest.content(); + if (content == ManifestContent.DATA) { + try (CloseableIterable reader = + ManifestFiles.read(manifest, io, specs).select(PROJECTION)) { + for (DataFile file : reader) { + String path = file.path().toString(); + out.output( + KV.of(path, FileInfo.of(path, FileCategory.DATA, manifest.isValid(), tableIdString))); + } + } + } else if (content == ManifestContent.DELETES) { + try (CloseableIterable reader = + ManifestFiles.readDeleteManifest(manifest, io, specs).select(PROJECTION)) { + for (DeleteFile file : reader) { + String path = file.path().toString(); + FileCategory category = + file.content() == FileContent.POSITION_DELETES + ? FileCategory.POSITION_DELETES + : FileCategory.EQUALITY_DELETES; + out.output(KV.of(path, FileInfo.of(path, category, manifest.isValid(), tableIdString))); + } + } + } else { + throw new IllegalArgumentException("Unsupported manifest content: " + content); + } + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/package-info.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/package-info.java new file mode 100644 index 000000000000..95350c07ee30 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/package-info.java @@ -0,0 +1,20 @@ +/* + * 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. + */ + +/** Iceberg maintenance transforms. */ +package org.apache.beam.sdk.io.iceberg.maintenance; diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFnTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFnTest.java new file mode 100644 index 000000000000..763226182543 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFnTest.java @@ -0,0 +1,337 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import org.apache.beam.sdk.coders.IterableCoder; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; +import org.apache.beam.sdk.io.iceberg.TestDataWarehouse; +import org.apache.beam.sdk.io.iceberg.TestFixtures; +import org.apache.beam.sdk.schemas.NoSuchSchemaException; +import org.apache.beam.sdk.schemas.SchemaCoder; +import org.apache.beam.sdk.schemas.SchemaRegistry; +import org.apache.beam.sdk.testing.PAssert; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.util.ShardedKey; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.hadoop.fs.Path; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.NotFoundException; +import org.apache.iceberg.io.BulkDeletionFailureException; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.SupportsBulkOperations; +import org.junit.ClassRule; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class DeleteFilesDoFnTest { + + @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder(); + @Rule public TestDataWarehouse warehouse = new TestDataWarehouse(TEMPORARY_FOLDER, "default"); + @Rule public TestPipeline pipeline = TestPipeline.create(); + + private IcebergCatalogConfig getCatalogConfig() { + return IcebergCatalogConfig.builder() + .setCatalogProperties(ImmutableMap.of("type", "hadoop", "warehouse", warehouse.location)) + .build(); + } + + @Test + public void testDeletesFilesPhysicallyWhenCleanFilesTrue() + throws IOException, NoSuchSchemaException { + TableIdentifier tableId = TableIdentifier.of("default", "delete_phys_" + System.nanoTime()); + Table table = warehouse.createTable(tableId, TestFixtures.SCHEMA); + + DataFile file = + warehouse.writeRecords( + "delete_target_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(ExpireSnapshotsTestFixtures.createRecord(1L, "val-1"))); + + File diskFile = new File(new Path(file.path().toString()).toUri()); + assertTrue("Target file must exist prior to deletion", diskFile.exists()); + + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder().setCleanFiles(true).build(); + + SchemaCoder fileInfoCoder = + SchemaRegistry.createDefault().getSchemaCoder(FileInfo.class); + + PCollection output = + pipeline + .apply( + Create.of( + KV.of( + ShardedKey.of(tableId.toString(), new byte[0]), + (Iterable) + Collections.singletonList( + FileInfo.of( + file.path().toString(), + FileCategory.DATA, + false, + tableId.toString())))) + .withCoder( + KvCoder.of( + ShardedKey.Coder.of(StringUtf8Coder.of()), + IterableCoder.of(fileInfoCoder)))) + .apply(ParDo.of(new DeleteFilesDoFn(getCatalogConfig(), config))); + + PAssert.that(output) + .containsInAnyOrder(ExpireSnapshotsResult.builder().setDeletedDataFilesCount(1L).build()); + + pipeline.run(); + + assertFalse("File must be physically deleted from storage", diskFile.exists()); + } + + @Test + public void testDryRunDoesNotDeleteFromStorage() throws IOException, NoSuchSchemaException { + TableIdentifier tableId = TableIdentifier.of("default", "dry_run_" + System.nanoTime()); + Table table = warehouse.createTable(tableId, TestFixtures.SCHEMA); + + DataFile file = + warehouse.writeRecords( + "dry_target_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(ExpireSnapshotsTestFixtures.createRecord(1L, "val-1"))); + + File diskFile = new File(new Path(file.path().toString()).toUri()); + assertTrue("Target file must exist prior to deletion", diskFile.exists()); + + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder().setCleanFiles(false).build(); + + SchemaCoder fileInfoCoder = + SchemaRegistry.createDefault().getSchemaCoder(FileInfo.class); + + PCollection output = + pipeline + .apply( + Create.of( + KV.of( + ShardedKey.of(tableId.toString(), new byte[0]), + (Iterable) + Collections.singletonList( + FileInfo.of( + file.path().toString(), + FileCategory.DATA, + false, + tableId.toString())))) + .withCoder( + KvCoder.of( + ShardedKey.Coder.of(StringUtf8Coder.of()), + IterableCoder.of(fileInfoCoder)))) + .apply(ParDo.of(new DeleteFilesDoFn(getCatalogConfig(), config))); + + PAssert.that(output) + .containsInAnyOrder(ExpireSnapshotsResult.builder().setDeletedDataFilesCount(1L).build()); + + pipeline.run(); + + assertTrue("File must still exist on storage during dry run", diskFile.exists()); + } + + @Test + public void testIgnoresNotFoundException() throws NoSuchSchemaException { + TableIdentifier tableId = TableIdentifier.of("default", "missing_file_" + System.nanoTime()); + warehouse.createTable(tableId, TestFixtures.SCHEMA); + + String missingPath = warehouse.location + "/non_existent.parquet"; + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder().setCleanFiles(true).build(); + + SchemaCoder fileInfoCoder = + SchemaRegistry.createDefault().getSchemaCoder(FileInfo.class); + + PCollection output = + pipeline + .apply( + Create.of( + KV.of( + ShardedKey.of(tableId.toString(), new byte[0]), + (Iterable) + Collections.singletonList( + FileInfo.of( + missingPath, + FileCategory.DATA, + false, + tableId.toString())))) + .withCoder( + KvCoder.of( + ShardedKey.Coder.of(StringUtf8Coder.of()), + IterableCoder.of(fileInfoCoder)))) + .apply(ParDo.of(new DeleteFilesDoFn(getCatalogConfig(), config))); + + PAssert.that(output) + .containsInAnyOrder(ExpireSnapshotsResult.builder().setDeletedDataFilesCount(1L).build()); + + // Should complete cleanly without exception + pipeline.run(); + } + + @Test + public void testDeletesPositionAndEqualityDeleteFiles() + throws IOException, NoSuchSchemaException { + TableIdentifier tableId = + TableIdentifier.of("default", "delete_row_level_" + System.nanoTime()); + warehouse.createTable(tableId, TestFixtures.SCHEMA); + + File posDiskFile = TEMPORARY_FOLDER.newFile("pos_del_" + System.nanoTime() + ".parquet"); + File eqDiskFile = TEMPORARY_FOLDER.newFile("eq_del_" + System.nanoTime() + ".parquet"); + assertTrue(posDiskFile.exists()); + assertTrue(eqDiskFile.exists()); + + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder().setCleanFiles(true).build(); + + SchemaCoder fileInfoCoder = + SchemaRegistry.createDefault().getSchemaCoder(FileInfo.class); + + List files = + Arrays.asList( + FileInfo.of( + posDiskFile.getAbsolutePath(), + FileCategory.POSITION_DELETES, + false, + tableId.toString()), + FileInfo.of( + eqDiskFile.getAbsolutePath(), + FileCategory.EQUALITY_DELETES, + false, + tableId.toString())); + + PCollection output = + pipeline + .apply( + Create.of( + KV.of( + ShardedKey.of(tableId.toString(), new byte[0]), + (Iterable) files)) + .withCoder( + KvCoder.of( + ShardedKey.Coder.of(StringUtf8Coder.of()), + IterableCoder.of(fileInfoCoder)))) + .apply(ParDo.of(new DeleteFilesDoFn(getCatalogConfig(), config))); + + PAssert.that(output) + .containsInAnyOrder( + ExpireSnapshotsResult.builder() + .setDeletedPositionDeleteFilesCount(1L) + .setDeletedEqualityDeleteFilesCount(1L) + .build()); + + pipeline.run(); + + assertFalse("Position delete file must be deleted", posDiskFile.exists()); + assertFalse("Equality delete file must be deleted", eqDiskFile.exists()); + } + + @Test + public void testDeletePathsBulkFallbackAndPartialFailure() { + Set deleted = new HashSet<>(); + FileIO io = + new FileIO() { + @Override + public InputFile newInputFile(String path) { + throw new UnsupportedOperationException(); + } + + @Override + public OutputFile newOutputFile(String path) { + throw new UnsupportedOperationException(); + } + + @Override + public void deleteFile(String path) { + if ("fail.parquet".equals(path)) { + throw new RuntimeException("permission denied"); + } else if ("missing.parquet".equals(path)) { + throw new NotFoundException("not found"); + } else { + deleted.add(path); + } + } + }; + + // Standard FileIO without SupportsBulkOperations + Set failed = + DeleteFilesDoFn.deletePaths( + io, Arrays.asList("ok.parquet", "missing.parquet", "fail.parquet")); + assertEquals(Collections.singleton("fail.parquet"), failed); + assertTrue(deleted.contains("ok.parquet")); + } + + @Test + public void testDeletePathsBulkOperationsWithFailure() { + Set deleted = new HashSet<>(); + class BulkFileIO implements FileIO, SupportsBulkOperations { + @Override + public InputFile newInputFile(String path) { + throw new UnsupportedOperationException(); + } + + @Override + public OutputFile newOutputFile(String path) { + throw new UnsupportedOperationException(); + } + + @Override + public void deleteFile(String path) { + if ("fail.parquet".equals(path)) { + throw new RuntimeException("permission denied"); + } + deleted.add(path); + } + + @Override + public void deleteFiles(Iterable pathsToDelete) throws BulkDeletionFailureException { + throw new BulkDeletionFailureException(1); + } + } + + BulkFileIO io = new BulkFileIO(); + Set failed = + DeleteFilesDoFn.deletePaths(io, Arrays.asList("ok.parquet", "fail.parquet")); + assertEquals(Collections.singleton("fail.parquet"), failed); + assertTrue(deleted.contains("ok.parquet")); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResultTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResultTest.java new file mode 100644 index 000000000000..97c79f6206d8 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResultTest.java @@ -0,0 +1,141 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import static org.junit.Assert.assertEquals; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.Arrays; +import java.util.Collections; +import org.apache.beam.sdk.coders.Coder; +import org.apache.beam.sdk.schemas.NoSuchSchemaException; +import org.apache.beam.sdk.schemas.SchemaRegistry; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class ExpireSnapshotsResultTest { + + @Test + public void testZerosIdentity() { + ExpireSnapshotsResult zeros = ExpireSnapshotsResult.zeros(); + assertEquals(0L, zeros.getDeletedDataFilesCount()); + assertEquals(0L, zeros.getDeletedPositionDeleteFilesCount()); + assertEquals(0L, zeros.getDeletedEqualityDeleteFilesCount()); + assertEquals(0L, zeros.getDeletedManifestsCount()); + assertEquals(0L, zeros.getDeletedManifestListsCount()); + assertEquals(0L, zeros.getDeletedStatisticsFilesCount()); + assertEquals(0L, zeros.getExpiredSnapshotsCount()); + } + + @Test + public void testMergeFragments() { + ExpireSnapshotsResult a = + ExpireSnapshotsResult.builder() + .setDeletedDataFilesCount(5L) + .setDeletedPositionDeleteFilesCount(2L) + .setDeletedEqualityDeleteFilesCount(1L) + .setDeletedManifestsCount(3L) + .setDeletedManifestListsCount(1L) + .setDeletedStatisticsFilesCount(0L) + .setExpiredSnapshotsCount(1L) + .build(); + + ExpireSnapshotsResult b = + ExpireSnapshotsResult.builder() + .setDeletedDataFilesCount(10L) + .setDeletedPositionDeleteFilesCount(0L) + .setDeletedEqualityDeleteFilesCount(3L) + .setDeletedManifestsCount(2L) + .setDeletedManifestListsCount(1L) + .setDeletedStatisticsFilesCount(1L) + .setExpiredSnapshotsCount(1L) + .build(); + + ExpireSnapshotsResult merged = ExpireSnapshotsResult.merge(a, b); + assertEquals(15L, merged.getDeletedDataFilesCount()); + assertEquals(2L, merged.getDeletedPositionDeleteFilesCount()); + assertEquals(4L, merged.getDeletedEqualityDeleteFilesCount()); + assertEquals(5L, merged.getDeletedManifestsCount()); + assertEquals(2L, merged.getDeletedManifestListsCount()); + assertEquals(1L, merged.getDeletedStatisticsFilesCount()); + assertEquals(2L, merged.getExpiredSnapshotsCount()); + } + + @Test + public void testCombineFn() { + ExpireSnapshotsResult.Merge mergeFn = new ExpireSnapshotsResult.Merge(); + ExpireSnapshotsResult acc = mergeFn.createAccumulator(); + assertEquals(ExpireSnapshotsResult.zeros(), acc); + + ExpireSnapshotsResult item1 = + ExpireSnapshotsResult.builder() + .setDeletedDataFilesCount(3L) + .setExpiredSnapshotsCount(1L) + .build(); + ExpireSnapshotsResult item2 = + ExpireSnapshotsResult.builder() + .setDeletedDataFilesCount(7L) + .setDeletedManifestsCount(2L) + .build(); + + acc = mergeFn.addInput(acc, item1); + acc = mergeFn.addInput(acc, item2); + + ExpireSnapshotsResult output = mergeFn.extractOutput(acc); + assertEquals(10L, output.getDeletedDataFilesCount()); + assertEquals(2L, output.getDeletedManifestsCount()); + assertEquals(1L, output.getExpiredSnapshotsCount()); + + ExpireSnapshotsResult mergedAcc = + mergeFn.mergeAccumulators(Arrays.asList(item1, item2, ExpireSnapshotsResult.zeros())); + assertEquals(10L, mergedAcc.getDeletedDataFilesCount()); + assertEquals(2L, mergedAcc.getDeletedManifestsCount()); + assertEquals(1L, mergedAcc.getExpiredSnapshotsCount()); + + ExpireSnapshotsResult emptyMerge = mergeFn.mergeAccumulators(Collections.emptyList()); + assertEquals(ExpireSnapshotsResult.zeros(), emptyMerge); + } + + @Test + public void testSchemaCoderSerialization() throws NoSuchSchemaException, IOException { + Coder coder = + SchemaRegistry.createDefault().getSchemaCoder(ExpireSnapshotsResult.class); + + ExpireSnapshotsResult original = + ExpireSnapshotsResult.builder() + .setDeletedDataFilesCount(123L) + .setDeletedPositionDeleteFilesCount(45L) + .setDeletedEqualityDeleteFilesCount(6L) + .setDeletedManifestsCount(78L) + .setDeletedManifestListsCount(9L) + .setDeletedStatisticsFilesCount(10L) + .setExpiredSnapshotsCount(11L) + .build(); + + ByteArrayOutputStream out = new ByteArrayOutputStream(); + coder.encode(original, out); + ByteArrayInputStream in = new ByteArrayInputStream(out.toByteArray()); + ExpireSnapshotsResult decoded = coder.decode(in); + + assertEquals(original, decoded); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTest.java new file mode 100644 index 000000000000..7bb92e1cc660 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTest.java @@ -0,0 +1,478 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.io.File; +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import org.apache.beam.sdk.Pipeline; +import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; +import org.apache.beam.sdk.io.iceberg.TestDataWarehouse; +import org.apache.beam.sdk.io.iceberg.TestFixtures; +import org.apache.beam.sdk.testing.PAssert; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; +import org.apache.hadoop.fs.Path; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.FileMetadata; +import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.io.CloseableIterable; +import org.junit.ClassRule; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class ExpireSnapshotsTest { + + @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder(); + @Rule public TestDataWarehouse warehouse = new TestDataWarehouse(TEMPORARY_FOLDER, "default"); + @Rule public TestPipeline pipeline = TestPipeline.create(); + + private IcebergCatalogConfig getCatalogConfig() { + return IcebergCatalogConfig.builder() + .setCatalogProperties(ImmutableMap.of("type", "hadoop", "warehouse", warehouse.location)) + .build(); + } + + @Test + public void testStandardSnapshotExpiration() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "standard_expire_" + System.nanoTime()); + Table table = ExpireSnapshotsTestFixtures.createTableWithMultiSnapshots(warehouse, tableId); + + List originalSnapshots = Lists.newArrayList(table.snapshots()); + assertEquals(3, originalSnapshots.size()); + + // Record data file paths from snapshot 1, 2, and 3 + String s1Path = getDataFilePath(table, originalSnapshots.get(0)); + String s2Path = getDataFilePath(table, originalSnapshots.get(1)); + String s3Path = getDataFilePath(table, originalSnapshots.get(2)); + + assertTrue("S1 file must exist", fileExists(s1Path)); + assertTrue("S2 file must exist", fileExists(s2Path)); + assertTrue("S3 file must exist", fileExists(s3Path)); + + long cutoff = originalSnapshots.get(2).timestampMillis(); + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder() + .setExpireOlderThan(cutoff) + .setRetainLast(1) + .setCleanFiles(true) + .build(); + + PCollection result = + pipeline + .apply(Create.of(tableId.toString())) + .apply(ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result) + .satisfies( + results -> { + ExpireSnapshotsResult r = results.iterator().next(); + assertEquals(2L, r.getExpiredSnapshotsCount()); + assertEquals(0L, r.getDeletedDataFilesCount()); + assertTrue(r.getDeletedManifestListsCount() >= 2L); + return null; + }); + + pipeline.run(); + + table.refresh(); + List remainingSnapshots = Lists.newArrayList(table.snapshots()); + assertEquals(1, remainingSnapshots.size()); + assertEquals(originalSnapshots.get(2).snapshotId(), remainingSnapshots.get(0).snapshotId()); + + // In an append-only sequence, S3 still references files from S1 and S2; all data files must be + // preserved! + assertTrue("S1 file must still exist as it is referenced in S3", fileExists(s1Path)); + assertTrue("S2 file must still exist as it is referenced in S3", fileExists(s2Path)); + assertTrue("S3 file must still exist as it is referenced in S3", fileExists(s3Path)); + } + + @Test + public void testSharedFilesPreservedAcrossSnapshots() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "shared_files_" + System.nanoTime()); + Table table = ExpireSnapshotsTestFixtures.createTableWithOverwrites(warehouse, tableId); + + List snapshots = Lists.newArrayList(table.snapshots()); + assertEquals(3, snapshots.size()); + + // In createTableWithOverwrites: + // S1: File A + // S2: File B + // S3: File A overwritten with File C (active: B, C; obsolete: A) + long cutoff = snapshots.get(2).timestampMillis(); + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder() + .setExpireOlderThan(cutoff) + .setRetainLast(1) + .setCleanFiles(true) + .build(); + + PCollection result = + pipeline + .apply(Create.of(tableId.toString())) + .apply(ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result) + .satisfies( + results -> { + ExpireSnapshotsResult r = results.iterator().next(); + assertEquals(2L, r.getExpiredSnapshotsCount()); + // Only File A is deleted; File B was retained into S3 and File C was added in S3 + assertEquals(1L, r.getDeletedDataFilesCount()); + return null; + }); + + pipeline.run(); + + table.refresh(); + assertEquals(1, Lists.newArrayList(table.snapshots()).size()); + + // Verify active files in current snapshot + try (CloseableIterable tasks = table.newScan().planFiles()) { + int activeFileCount = 0; + for (FileScanTask task : tasks) { + activeFileCount++; + assertTrue( + "Active file must exist on disk: " + task.file().path(), + fileExists(task.file().path().toString())); + } + assertEquals(2, activeFileCount); + } + } + + @Test + public void testBranchAndTagProtection() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "branch_protect_" + System.nanoTime()); + String branchName = "test_branch"; + Table table = ExpireSnapshotsTestFixtures.createTableWithBranch(warehouse, tableId, branchName); + + table.refresh(); + assertEquals(3, Lists.newArrayList(table.snapshots()).size()); + + Snapshot branchHead = table.snapshot(table.refs().get(branchName).snapshotId()); + String branchFilePath = getDataFilePath(table, branchHead); + assertTrue("Branch data file must exist", fileExists(branchFilePath)); + + // Expire on main branch older than current time, retaining 1 + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder() + .setExpireOlderThan(System.currentTimeMillis() + 100_000L) + .setRetainLast(1) + .setCleanFiles(true) + .build(); + + PCollection result = + pipeline + .apply(Create.of(tableId.toString())) + .apply(ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result) + .satisfies( + results -> { + ExpireSnapshotsResult r = results.iterator().next(); + assertTrue(r.getExpiredSnapshotsCount() >= 1); + return null; + }); + + pipeline.run(); + + table.refresh(); + // Branch file must be completely untouched and preserved! + assertTrue("Branch data file must NOT be deleted", fileExists(branchFilePath)); + assertTrue("Branch ref must still exist", table.refs().containsKey(branchName)); + } + + @Test + public void testRetainLastSafetyFloor() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "retain_floor_" + System.nanoTime()); + Table table = ExpireSnapshotsTestFixtures.createTableWithMultiSnapshots(warehouse, tableId); + + List snapshots = Lists.newArrayList(table.snapshots()); + assertEquals(3, snapshots.size()); + long cutoff = snapshots.get(2).timestampMillis(); + + // Snapshots 0 and 1 are older than cutoff, but retainLast = 2 preserves Snapshot 1 + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder() + .setExpireOlderThan(cutoff) + .setRetainLast(2) + .setCleanFiles(true) + .build(); + + PCollection result = + pipeline + .apply(Create.of(tableId.toString())) + .apply(ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result) + .satisfies( + results -> { + ExpireSnapshotsResult r = results.iterator().next(); + assertEquals(1L, r.getExpiredSnapshotsCount()); + assertEquals(0L, r.getDeletedDataFilesCount()); + return null; + }); + + pipeline.run(); + + table.refresh(); + assertEquals(2, Lists.newArrayList(table.snapshots()).size()); + } + + @Test + public void testIdempotencyRepeatedExecutions() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "idempotent_" + System.nanoTime()); + Table table = ExpireSnapshotsTestFixtures.createTableWithMultiSnapshots(warehouse, tableId); + + List snapshots = Lists.newArrayList(table.snapshots()); + long cutoff = snapshots.get(2).timestampMillis(); + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder() + .setExpireOlderThan(cutoff) + .setRetainLast(1) + .setCleanFiles(true) + .build(); + + // Run 1: expires 2 snapshots + pipeline + .apply("Input 1", Create.of(tableId.toString())) + .apply("Expire 1", ExpireSnapshots.create(getCatalogConfig(), config)); + pipeline.run(); + + table.refresh(); + assertEquals(1, Lists.newArrayList(table.snapshots()).size()); + + // Run 2: pipeline runs again on the same table + Pipeline pipeline2 = Pipeline.create(); + PCollection result2 = + pipeline2 + .apply("Input 2", Create.of(tableId.toString())) + .apply("Expire 2", ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result2).containsInAnyOrder(ExpireSnapshotsResult.zeros()); + pipeline2.run().waitUntilFinish(); + + table.refresh(); + assertEquals(1, Lists.newArrayList(table.snapshots()).size()); + } + + @Test + public void testDryRunDoesNotDeleteFromStorage() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "dry_run_" + System.nanoTime()); + Table table = ExpireSnapshotsTestFixtures.createTableWithOverwrites(warehouse, tableId); + + List snapshots = Lists.newArrayList(table.snapshots()); + String fileAPath = getDataFilePath(table, snapshots.get(0)); + + long cutoff = snapshots.get(2).timestampMillis(); + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder() + .setExpireOlderThan(cutoff) + .setRetainLast(1) + .setCleanFiles(false) // Dry run + .build(); + + PCollection result = + pipeline + .apply(Create.of(tableId.toString())) + .apply(ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result) + .satisfies( + results -> { + ExpireSnapshotsResult r = results.iterator().next(); + assertEquals(2L, r.getExpiredSnapshotsCount()); + assertEquals(1L, r.getDeletedDataFilesCount()); + return null; + }); + + pipeline.run(); + + // Files must still physically exist on storage after dry run! + assertTrue("File A must still exist after dry run", fileExists(fileAPath)); + } + + @Test + public void testEmptyTableAndSingleSnapshot() { + TableIdentifier tableId = TableIdentifier.of("default", "empty_table_" + System.nanoTime()); + warehouse.createTable(tableId, TestFixtures.SCHEMA); + + ExpireSnapshots.Configuration config = ExpireSnapshots.Configuration.builder().build(); + + PCollection result = + pipeline + .apply(Create.of(tableId.toString())) + .apply(ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result).containsInAnyOrder(ExpireSnapshotsResult.zeros()); + pipeline.run(); + } + + @Test + public void testMultiTablePipeline() throws IOException { + TableIdentifier table1 = TableIdentifier.of("default", "multi_table_1_" + System.nanoTime()); + TableIdentifier table2 = TableIdentifier.of("default", "multi_table_2_" + System.nanoTime()); + + Table t1 = ExpireSnapshotsTestFixtures.createTableWithMultiSnapshots(warehouse, table1); + Table t2 = ExpireSnapshotsTestFixtures.createTableWithMultiSnapshots(warehouse, table2); + + List s1 = Lists.newArrayList(t1.snapshots()); + List s2 = Lists.newArrayList(t2.snapshots()); + long cutoff = Math.max(s1.get(2).timestampMillis(), s2.get(2).timestampMillis()); + + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder() + .setExpireOlderThan(cutoff) + .setRetainLast(1) + .setCleanFiles(true) + .build(); + + PCollection result = + pipeline + .apply(Create.of(table1.toString(), table2.toString())) + .apply(ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result) + .satisfies( + results -> { + ExpireSnapshotsResult r = results.iterator().next(); + // 2 snapshots expired per table * 2 tables = 4 + assertEquals(4L, r.getExpiredSnapshotsCount()); + assertTrue(r.getDeletedManifestListsCount() >= 4L); + return null; + }); + + pipeline.run(); + + t1.refresh(); + t2.refresh(); + assertEquals(1, Lists.newArrayList(t1.snapshots()).size()); + assertEquals(1, Lists.newArrayList(t2.snapshots()).size()); + } + + @Test + public void testExpireSnapshotsWithDeleteFiles() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "expire_deletes_" + System.nanoTime()); + Table table = warehouse.createTable(tableId, TestFixtures.SCHEMA); + table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit(); + table.refresh(); + + // S1: Data file A + DataFile fileA = + warehouse.writeRecords( + "file_a_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(ExpireSnapshotsTestFixtures.createRecord(0L, "val-0"))); + table.newAppend().appendFile(fileA).commit(); + table.refresh(); + + // S2: Add position delete file + File posDeleteDiskFile = TEMPORARY_FOLDER.newFile("pos_del_" + System.nanoTime() + ".parquet"); + assertTrue(posDeleteDiskFile.exists()); + + DeleteFile posDelete = + FileMetadata.deleteFileBuilder(table.spec()) + .ofPositionDeletes() + .withPath(posDeleteDiskFile.getAbsolutePath()) + .withFormat(FileFormat.PARQUET) + .withFileSizeInBytes(100L) + .withRecordCount(1L) + .build(); + + ExpireSnapshotsTestFixtures.waitUntilAfter(table.currentSnapshot().timestampMillis()); + table.newRowDelta().addDeletes(posDelete).commit(); + table.refresh(); + + // S3: Overwrite / rewrite data file (clean up / replace) + DataFile fileB = + warehouse.writeRecords( + "file_b_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(ExpireSnapshotsTestFixtures.createRecord(1L, "val-1"))); + ExpireSnapshotsTestFixtures.waitUntilAfter(table.currentSnapshot().timestampMillis()); + table + .newRewrite() + .validateFromSnapshot(table.currentSnapshot().snapshotId()) + .deleteFile(fileA) + .deleteFile(posDelete) + .addFile(fileB) + .commit(); + table.refresh(); + + List snapshots = Lists.newArrayList(table.snapshots()); + assertEquals(3, snapshots.size()); + long cutoff = snapshots.get(2).timestampMillis(); + + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder() + .setExpireOlderThan(cutoff) + .setRetainLast(1) + .setCleanFiles(true) + .build(); + + PCollection result = + pipeline + .apply(Create.of(tableId.toString())) + .apply(ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result) + .satisfies( + results -> { + ExpireSnapshotsResult r = results.iterator().next(); + assertEquals(2L, r.getExpiredSnapshotsCount()); + assertEquals(1L, r.getDeletedDataFilesCount()); + assertEquals(1L, r.getDeletedPositionDeleteFilesCount()); + return null; + }); + + pipeline.run(); + + assertFalse( + "Position delete file must be physically deleted from storage", posDeleteDiskFile.exists()); + } + + private static boolean fileExists(String path) { + return new File(new Path(path).toUri()).exists(); + } + + private static String getDataFilePath(Table table, Snapshot snapshot) throws IOException { + List files = Lists.newArrayList(snapshot.addedDataFiles(table.io())); + if (!files.isEmpty()) { + return files.get(0).path().toString(); + } + return null; + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTestFixtures.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTestFixtures.java new file mode 100644 index 000000000000..f91f29dad22a --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTestFixtures.java @@ -0,0 +1,197 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import org.apache.beam.sdk.io.iceberg.TestDataWarehouse; +import org.apache.beam.sdk.io.iceberg.TestFixtures; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.OverwriteFiles; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; + +/** Test fixtures and data generators for testing {@link ExpireSnapshots}. */ +public class ExpireSnapshotsTestFixtures { + + public static Record createRecord(long id, String data) { + Record r = GenericRecord.create(TestFixtures.SCHEMA); + r.setField("id", id); + r.setField("data", data); + return r; + } + + /** + * Builds a table with 3 successive append snapshots. + * + *

    + *
  • Snapshot 1: File 1 (rows 0, 1) + *
  • Snapshot 2: File 2 (rows 2, 3) + *
  • Snapshot 3: File 3 (rows 4, 5) + *
+ */ + public static Table createTableWithMultiSnapshots( + TestDataWarehouse warehouse, TableIdentifier tableId) throws IOException { + Table table = warehouse.createTable(tableId, TestFixtures.SCHEMA); + + for (int i = 1; i <= 3; i++) { + List records = new ArrayList<>(); + records.add(createRecord((i - 1) * 2L, "val-" + ((i - 1) * 2L))); + records.add(createRecord((i - 1) * 2L + 1L, "val-" + ((i - 1) * 2L + 1L))); + + DataFile file = + warehouse.writeRecords( + "file_" + i + "_" + System.nanoTime() + ".parquet", table.schema(), records); + + if (table.currentSnapshot() != null) { + waitUntilAfter(table.currentSnapshot().timestampMillis()); + } + AppendFiles append = table.newAppend(); + append.appendFile(file); + append.commit(); + table.refresh(); + } + + return table; + } + + /** + * Builds a table with overwriting snapshots. + * + *
    + *
  • Snapshot 1 adds File A (id 0) + *
  • Snapshot 2 adds File B (id 1) + *
  • Snapshot 3 overwrites File A with File C (id 2) + *
+ * + * Active files in Snapshot 3: File B and File C. File A is obsolete. + */ + public static Table createTableWithOverwrites( + TestDataWarehouse warehouse, TableIdentifier tableId) throws IOException { + Table table = warehouse.createTable(tableId, TestFixtures.SCHEMA); + + // Snapshot 1: File A + DataFile fileA = + warehouse.writeRecords( + "file_a_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(createRecord(0L, "val-0"))); + table.newAppend().appendFile(fileA).commit(); + table.refresh(); + + // Snapshot 2: File B + DataFile fileB = + warehouse.writeRecords( + "file_b_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(createRecord(1L, "val-1"))); + if (table.currentSnapshot() != null) { + waitUntilAfter(table.currentSnapshot().timestampMillis()); + } + table.newAppend().appendFile(fileB).commit(); + table.refresh(); + + // Snapshot 3: Overwrite File A with File C + DataFile fileC = + warehouse.writeRecords( + "file_c_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(createRecord(2L, "val-2"))); + if (table.currentSnapshot() != null) { + waitUntilAfter(table.currentSnapshot().timestampMillis()); + } + OverwriteFiles overwrite = table.newOverwrite(); + overwrite.deleteFile(fileA); + overwrite.addFile(fileC); + overwrite.commit(); + table.refresh(); + + return table; + } + + /** + * Builds a table with a branch. + * + *
    + *
  • Snapshot 1 on main (File A) + *
  • Branch "branch_a" created from Snapshot 1 + *
  • Snapshot 2 appended to main (File B) + *
  • Snapshot 3 appended to "branch_a" (File C) + *
+ */ + public static Table createTableWithBranch( + TestDataWarehouse warehouse, TableIdentifier tableId, String branchName) throws IOException { + Table table = warehouse.createTable(tableId, TestFixtures.SCHEMA); + + // Snapshot 1: File A + DataFile fileA = + warehouse.writeRecords( + "file_main_1_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(createRecord(0L, "main-0"))); + table.newAppend().appendFile(fileA).commit(); + table.refresh(); + long snapshot1Id = table.currentSnapshot().snapshotId(); + + // Create branch from snapshot 1 + table.manageSnapshots().createBranch(branchName, snapshot1Id).commit(); + table.refresh(); + + // Snapshot 2: File B on main + DataFile fileB = + warehouse.writeRecords( + "file_main_2_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(createRecord(1L, "main-1"))); + if (table.currentSnapshot() != null) { + waitUntilAfter(table.currentSnapshot().timestampMillis()); + } + table.newAppend().appendFile(fileB).commit(); + table.refresh(); + + // Snapshot 3: File C on branch + DataFile fileC = + warehouse.writeRecords( + "file_branch_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(createRecord(2L, "branch-0"))); + waitUntilAfter(table.currentSnapshot().timestampMillis()); + table.newAppend().toBranch(branchName).appendFile(fileC).commit(); + table.refresh(); + + return table; + } + + public static void waitUntilAfter(long timestampMillis) { + long current = System.currentTimeMillis(); + while (current <= timestampMillis) { + try { + Thread.sleep(2); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + current = System.currentTimeMillis(); + } + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFnTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFnTest.java new file mode 100644 index 000000000000..7b2ed9180a9f --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFnTest.java @@ -0,0 +1,177 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; + +import java.io.IOException; +import java.util.List; +import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; +import org.apache.beam.sdk.io.iceberg.TestDataWarehouse; +import org.apache.beam.sdk.io.iceberg.TestFixtures; +import org.apache.beam.sdk.testing.PAssert; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.values.PCollectionTuple; +import org.apache.beam.sdk.values.TupleTagList; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.catalog.TableIdentifier; +import org.junit.ClassRule; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class PlanExpireSnapshotsDoFnTest { + + @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder(); + @Rule public TestDataWarehouse warehouse = new TestDataWarehouse(TEMPORARY_FOLDER, "default"); + @Rule public TestPipeline pipeline = TestPipeline.create(); + + private IcebergCatalogConfig getCatalogConfig() { + return IcebergCatalogConfig.builder() + .setCatalogProperties(ImmutableMap.of("type", "hadoop", "warehouse", warehouse.location)) + .build(); + } + + @Test + public void testPlanExpiresSnapshotsWithCleanupLevelNone() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "plan_test_" + System.nanoTime()); + Table table = ExpireSnapshotsTestFixtures.createTableWithMultiSnapshots(warehouse, tableId); + + List snapshots = Lists.newArrayList(table.snapshots()); + assertEquals(3, snapshots.size()); + long cutoff = snapshots.get(2).timestampMillis(); + + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder().setExpireOlderThan(cutoff).setRetainLast(1).build(); + + PCollectionTuple planned = + pipeline + .apply(Create.of(tableId.toString())) + .apply( + ParDo.of(new PlanExpireSnapshotsDoFn(getCatalogConfig(), config)) + .withOutputTags( + PlanExpireSnapshotsDoFn.PLAN_SUMMARY, + TupleTagList.of(PlanExpireSnapshotsDoFn.MANIFESTS) + .and(PlanExpireSnapshotsDoFn.DIRECT_FILES))); + + PAssert.that(planned.get(PlanExpireSnapshotsDoFn.PLAN_SUMMARY)) + .containsInAnyOrder(ExpireSnapshotsResult.builder().setExpiredSnapshotsCount(2L).build()); + + pipeline.run(); + + table.refresh(); + List remainingSnapshots = Lists.newArrayList(table.snapshots()); + assertEquals(1, remainingSnapshots.size()); + assertEquals(snapshots.get(2).snapshotId(), remainingSnapshots.get(0).snapshotId()); + } + + @Test + public void testPlanThrowsWhenGcDisabled() { + TableIdentifier tableId = TableIdentifier.of("default", "gc_disabled_" + System.nanoTime()); + warehouse.createTable( + tableId, TestFixtures.SCHEMA, null, ImmutableMap.of(TableProperties.GC_ENABLED, "false")); + + ExpireSnapshots.Configuration config = ExpireSnapshots.Configuration.builder().build(); + + pipeline + .apply(Create.of(tableId.toString())) + .apply( + ParDo.of(new PlanExpireSnapshotsDoFn(getCatalogConfig(), config)) + .withOutputTags( + PlanExpireSnapshotsDoFn.PLAN_SUMMARY, + TupleTagList.of(PlanExpireSnapshotsDoFn.MANIFESTS) + .and(PlanExpireSnapshotsDoFn.DIRECT_FILES))); + + assertThrows( + Exception.class, + () -> { + pipeline.run(); + }); + } + + @Test + public void testPlanZeroSnapshotsExpiredIsNoOp() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "noop_test_" + System.nanoTime()); + Table table = ExpireSnapshotsTestFixtures.createTableWithMultiSnapshots(warehouse, tableId); + table.refresh(); + + // Cutoff far in the past + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder().setExpireOlderThan(1L).build(); + + PCollectionTuple planned = + pipeline + .apply(Create.of(tableId.toString())) + .apply( + ParDo.of(new PlanExpireSnapshotsDoFn(getCatalogConfig(), config)) + .withOutputTags( + PlanExpireSnapshotsDoFn.PLAN_SUMMARY, + TupleTagList.of(PlanExpireSnapshotsDoFn.MANIFESTS) + .and(PlanExpireSnapshotsDoFn.DIRECT_FILES))); + + PAssert.that(planned.get(PlanExpireSnapshotsDoFn.PLAN_SUMMARY)) + .containsInAnyOrder(ExpireSnapshotsResult.zeros()); + + pipeline.run(); + + table.refresh(); + assertEquals(3, Lists.newArrayList(table.snapshots()).size()); + } + + @Test + public void testPlanHonorsRetainLast() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "retain_last_" + System.nanoTime()); + Table table = ExpireSnapshotsTestFixtures.createTableWithMultiSnapshots(warehouse, tableId); + table.refresh(); + + // Cutoff in the future so all 3 are older than cutoff, but retainLast = 2 + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder() + .setExpireOlderThan(System.currentTimeMillis() + 100_000L) + .setRetainLast(2) + .build(); + + PCollectionTuple planned = + pipeline + .apply(Create.of(tableId.toString())) + .apply( + ParDo.of(new PlanExpireSnapshotsDoFn(getCatalogConfig(), config)) + .withOutputTags( + PlanExpireSnapshotsDoFn.PLAN_SUMMARY, + TupleTagList.of(PlanExpireSnapshotsDoFn.MANIFESTS) + .and(PlanExpireSnapshotsDoFn.DIRECT_FILES))); + + PAssert.that(planned.get(PlanExpireSnapshotsDoFn.PLAN_SUMMARY)) + .containsInAnyOrder(ExpireSnapshotsResult.builder().setExpiredSnapshotsCount(1L).build()); + + pipeline.run(); + + table.refresh(); + assertEquals(2, Lists.newArrayList(table.snapshots()).size()); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFnTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFnTest.java new file mode 100644 index 000000000000..05c248fd9043 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFnTest.java @@ -0,0 +1,154 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import static org.junit.Assert.assertEquals; + +import java.io.IOException; +import java.util.Collections; +import java.util.List; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.SerializableCoder; +import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; +import org.apache.beam.sdk.io.iceberg.TestDataWarehouse; +import org.apache.beam.sdk.io.iceberg.TestFixtures; +import org.apache.beam.sdk.testing.PAssert; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.FileMetadata; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.catalog.TableIdentifier; +import org.junit.ClassRule; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class ReadManifestDoFnTest { + + @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder(); + @Rule public TestDataWarehouse warehouse = new TestDataWarehouse(TEMPORARY_FOLDER, "default"); + @Rule public TestPipeline pipeline = TestPipeline.create(); + + private IcebergCatalogConfig getCatalogConfig() { + return IcebergCatalogConfig.builder() + .setCatalogProperties(ImmutableMap.of("type", "hadoop", "warehouse", warehouse.location)) + .build(); + } + + @Test + public void testReadsDataFilesFromManifest() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "read_manifest_" + System.nanoTime()); + Table table = warehouse.createTable(tableId, TestFixtures.SCHEMA); + + DataFile file = + warehouse.writeRecords( + "data_1_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(ExpireSnapshotsTestFixtures.createRecord(1L, "val-1"))); + AppendFiles append = table.newAppend(); + append.appendFile(file); + append.commit(); + table.refresh(); + + List manifests = table.currentSnapshot().allManifests(table.io()); + assertEquals(1, manifests.size()); + ManifestFile manifest = manifests.get(0); + + PCollection> output = + pipeline + .apply( + Create.of( + KV.of( + tableId.toString(), ManifestFileBean.fromManifestFile(manifest, true))) + .withCoder( + KvCoder.of( + StringUtf8Coder.of(), SerializableCoder.of(ManifestFileBean.class)))) + .apply(ParDo.of(new ReadManifestDoFn(getCatalogConfig()))); + + PAssert.that(output) + .containsInAnyOrder( + KV.of( + file.path().toString(), + FileInfo.of(file.path().toString(), FileCategory.DATA, true, tableId.toString()))); + + pipeline.run(); + } + + @Test + public void testReadsDeleteFilesFromManifest() { + TableIdentifier tableId = + TableIdentifier.of("default", "read_delete_manifest_" + System.nanoTime()); + Table table = warehouse.createTable(tableId, TestFixtures.SCHEMA); + table.updateProperties().set(TableProperties.FORMAT_VERSION, "2").commit(); + table.refresh(); + + DeleteFile posDelete = + FileMetadata.deleteFileBuilder(table.spec()) + .ofPositionDeletes() + .withPath(warehouse.location + "/pos_delete_" + System.nanoTime() + ".parquet") + .withFormat(FileFormat.PARQUET) + .withFileSizeInBytes(100L) + .withRecordCount(1L) + .build(); + + table.newRowDelta().addDeletes(posDelete).commit(); + table.refresh(); + + List deleteManifests = table.currentSnapshot().deleteManifests(table.io()); + assertEquals(1, deleteManifests.size()); + ManifestFile deleteManifest = deleteManifests.get(0); + + PCollection> output = + pipeline + .apply( + Create.of( + KV.of( + tableId.toString(), + ManifestFileBean.fromManifestFile(deleteManifest, true))) + .withCoder( + KvCoder.of( + StringUtf8Coder.of(), SerializableCoder.of(ManifestFileBean.class)))) + .apply(ParDo.of(new ReadManifestDoFn(getCatalogConfig()))); + + PAssert.that(output) + .containsInAnyOrder( + KV.of( + posDelete.path().toString(), + FileInfo.of( + posDelete.path().toString(), + FileCategory.POSITION_DELETES, + true, + tableId.toString()))); + + pipeline.run(); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/package-info.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/package-info.java new file mode 100644 index 000000000000..8b9f0f6647a9 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/package-info.java @@ -0,0 +1,20 @@ +/* + * 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. + */ + +/** Tests for Iceberg maintenance transforms. */ +package org.apache.beam.sdk.io.iceberg.maintenance; From 77c2d5e23b44a0f26c799675dd613001b909f4bd Mon Sep 17 00:00:00 2001 From: jrmccluskey Date: Thu, 15 Oct 2026 17:58:18 +0000 Subject: [PATCH 2/2] agentic review --- .../iceberg/maintenance/DeleteFilesDoFn.java | 31 ++-- .../iceberg/maintenance/ExpireSnapshots.java | 64 ++++--- .../maintenance/ExpireSnapshotsResult.java | 5 +- .../iceberg/maintenance/ManifestFileBean.java | 103 ++++++++++- .../maintenance/PlanExpireSnapshotsDoFn.java | 42 ++++- .../iceberg/maintenance/ReadManifestDoFn.java | 64 ++++--- .../maintenance/ExpireSnapshotsTest.java | 170 +++++++++++++++++- .../maintenance/ManifestFileBeanTest.java | 94 ++++++++++ 8 files changed, 487 insertions(+), 86 deletions(-) create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBeanTest.java diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFn.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFn.java index 1bb5447b1b2c..a7f90f4b26bf 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFn.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFn.java @@ -25,11 +25,12 @@ import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; import org.apache.beam.sdk.io.iceberg.IcebergUtils; import org.apache.beam.sdk.io.iceberg.TableCache; +import org.apache.beam.sdk.metrics.Counter; import org.apache.beam.sdk.metrics.Metrics; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.util.ShardedKey; import org.apache.beam.sdk.values.KV; -import org.apache.iceberg.Table; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions; import org.apache.iceberg.exceptions.NotFoundException; import org.apache.iceberg.io.BulkDeletionFailureException; import org.apache.iceberg.io.FileIO; @@ -40,11 +41,18 @@ /** * Worker DoFn that deletes batched unreferenced files from storage and emits partial deletion * metrics. + * + *

When {@code cleanFiles} is disabled, no files are deleted and the emitted counts reflect the + * unreferenced files that were identified for deletion. */ public class DeleteFilesDoFn extends DoFn, Iterable>, ExpireSnapshotsResult> { private static final Logger LOG = LoggerFactory.getLogger(DeleteFilesDoFn.class); + private static final Counter DELETED_FILES = + Metrics.counter(DeleteFilesDoFn.class, "deleted_files"); + private static final Counter FAILED_FILE_DELETIONS = + Metrics.counter(DeleteFilesDoFn.class, "failed_file_deletions"); private final IcebergCatalogConfig catalogConfig; private final ExpireSnapshots.Configuration config; @@ -59,14 +67,9 @@ public void processElement( @Element KV, Iterable> element, OutputReceiver out) { String tableIdString = element.getKey().getKey(); - if (tableIdString == null || tableIdString.isEmpty()) { - return; - } - - Table table = - TableCache.getAndRefreshIfStale( - catalogConfig, IcebergUtils.parseTableIdentifier(tableIdString)); - FileIO io = table.io(); + Preconditions.checkState( + tableIdString != null && !tableIdString.isEmpty(), + "Received a file batch without a table identifier"); List fileList = new ArrayList<>(); List paths = new ArrayList<>(); @@ -81,15 +84,19 @@ public void processElement( Set failedPaths = Collections.emptySet(); if (config.cleanFiles()) { + FileIO io = + TableCache.get(catalogConfig, IcebergUtils.parseTableIdentifier(tableIdString)).io(); failedPaths = deletePaths(io, paths); + DELETED_FILES.inc(paths.size() - failedPaths.size()); if (!failedPaths.isEmpty()) { - Metrics.counter(DeleteFilesDoFn.class, "failed_file_deletions").inc(failedPaths.size()); + FAILED_FILE_DELETIONS.inc(failedPaths.size()); } } else { LOG.info( ExpireSnapshots.PREFIX - + "Dry run enabled (cleanFiles=false); skipping physical deletion of {} file(s).", - paths.size()); + + "cleanFiles=false; leaving {} unreferenced file(s) of table '{}' in storage.", + paths.size(), + tableIdString); } long dataCount = 0; diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshots.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshots.java index 8a06c809f722..3b190f3708b5 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshots.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshots.java @@ -31,6 +31,7 @@ import org.apache.beam.sdk.schemas.annotations.DefaultSchema; import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription; import org.apache.beam.sdk.transforms.Combine; +import org.apache.beam.sdk.transforms.Distinct; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.Flatten; import org.apache.beam.sdk.transforms.GroupByKey; @@ -46,6 +47,7 @@ import org.apache.beam.sdk.values.PCollectionList; import org.apache.beam.sdk.values.PCollectionTuple; import org.apache.beam.sdk.values.TupleTagList; +import org.apache.beam.sdk.values.TypeDescriptor; import org.apache.beam.sdk.values.TypeDescriptors; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions; @@ -55,17 +57,17 @@ /** * Distributed snapshot expiration maintenance operation for Apache Iceberg tables. * - *

Prunes snapshots older than a configured retention threshold, retains the last $N$ ancestors, + *

Prunes snapshots older than a configured retention threshold, retains the last N ancestors, * removes unreferenced manifests and manifest lists, and physically deletes obsolete data and * delete files from storage. * *

Execution Model

* *
    - *
  1. Phase 1 (Planning): Runs inside an initial worker {@link PlanExpireSnapshotsDoFn}. - * Validates table GC settings and commits the metadata change via {@code - * table.expireSnapshots().cleanExpiredFiles(false).commit()}. Zero file deletions happen on - * the driver. + *
  2. Phase 1 (Planning): Table identifiers are deduplicated, then each table is planned + * on a worker by {@link PlanExpireSnapshotsDoFn}. Validates table GC settings and commits the + * metadata change via {@code table.expireSnapshots().cleanExpiredFiles(false).commit()}. No + * file deletions happen during planning. *
  3. Phase 2 (Manifest Scanning): Manifest files from candidate and retained snapshots * are redistributed and read in parallel across workers via {@link ReadManifestDoFn}. *
  4. Phase 3 (Anti-Join): Candidate and retained files are keyed by file path and @@ -77,6 +79,13 @@ *
  5. Phase 5 (Aggregation): Deletion counts and expired snapshot counts are merged into a * single {@link ExpireSnapshotsResult}. *
+ * + *

Failure Semantics

+ * + *

The metadata commit in Phase 1 is a non-idempotent side effect. If a planning bundle fails + * after the commit succeeds and is retried, the retry observes the already-expired table state and + * plans no deletions, so files unreferenced by that commit are left in storage. Such files are + * orphans and can be reclaimed by a subsequent orphan file cleanup. */ public class ExpireSnapshots extends PTransform, PCollection> { @@ -105,8 +114,8 @@ public void populateDisplayData(DisplayData.Builder builder) { builder.addIfNotNull( DisplayData.item("expireOlderThan", config.getExpireOlderThan()) .withLabel("Expire Older Than (Millis)")); - builder.add( - DisplayData.item("retainLast", config.retainLast()).withLabel("Retain Last Snapshots")); + builder.addIfNotNull( + DisplayData.item("retainLast", config.getRetainLast()).withLabel("Retain Last Snapshots")); builder.add( DisplayData.item("cleanFiles", config.cleanFiles()) .withLabel("Clean Files (Physical Deletion)")); @@ -141,15 +150,18 @@ public PCollection expand(PCollection tableIdenti KvCoder kvFileInfoCoder = KvCoder.of(StringUtf8Coder.of(), fileInfoCoder); - // Phase 1: Planning and metadata commit + // Phase 1: Planning and metadata commit. Duplicate identifiers are removed so that each table + // is committed against exactly once. PCollectionTuple planned = - tableIdentifiers.apply( - "Plan Expire Snapshots", - ParDo.of(new PlanExpireSnapshotsDoFn(catalogConfig, config)) - .withOutputTags( - PlanExpireSnapshotsDoFn.PLAN_SUMMARY, - TupleTagList.of(PlanExpireSnapshotsDoFn.MANIFESTS) - .and(PlanExpireSnapshotsDoFn.DIRECT_FILES))); + tableIdentifiers + .apply("Deduplicate Table Identifiers", Distinct.create()) + .apply( + "Plan Expire Snapshots", + ParDo.of(new PlanExpireSnapshotsDoFn(catalogConfig, config)) + .withOutputTags( + PlanExpireSnapshotsDoFn.PLAN_SUMMARY, + TupleTagList.of(PlanExpireSnapshotsDoFn.MANIFESTS) + .and(PlanExpireSnapshotsDoFn.DIRECT_FILES))); PCollection planSummary = planned.get(PlanExpireSnapshotsDoFn.PLAN_SUMMARY).setCoder(resultCoder); @@ -184,11 +196,8 @@ public PCollection expand(PCollection tableIdenti "Key for Batching", MapElements.into( TypeDescriptors.kvs( - TypeDescriptors.strings(), - org.apache.beam.sdk.values.TypeDescriptor.of(FileInfo.class))) - .via( - file -> - KV.of(MoreObjects.firstNonNull(file.getTableIdentifier(), ""), file))) + TypeDescriptors.strings(), TypeDescriptor.of(FileInfo.class))) + .via(file -> KV.of(file.getTableIdentifier(), file))) .setCoder(KvCoder.of(StringUtf8Coder.of(), fileInfoCoder)) .apply( "Batch Files", @@ -239,18 +248,19 @@ public abstract static class Configuration implements Serializable { public static Builder builder() { return new AutoValue_ExpireSnapshots_Configuration.Builder() - .setRetainLast(1) .setCleanFiles(true) .setDeleteBatchSize(10_000); } @SchemaFieldDescription( - "Cutoff timestamp in milliseconds. Snapshots older than this are expired.") + "Cutoff timestamp in milliseconds. Snapshots older than this are expired. If unset, the" + + " table's history.expire.max-snapshot-age-ms property is used.") @Pure public abstract @Nullable Long getExpireOlderThan(); @SchemaFieldDescription( - "Safety floor: minimum number of ancestor snapshots to retain. Must be >= 1.") + "Safety floor: minimum number of ancestor snapshots to retain. Must be >= 1. If unset, the" + + " table's history.expire.min-snapshots-to-keep property is used.") @Pure public abstract @Nullable Integer getRetainLast(); @@ -264,7 +274,9 @@ public static Builder builder() { public abstract @Nullable Boolean getCleanExpiredMetadata(); @SchemaFieldDescription( - "Whether to physically delete unreferenced files from storage. If false, acts as a dry run.") + "Whether to physically delete unreferenced files from storage. If false, snapshots are" + + " still expired from table metadata and unreferenced files are reported in the result," + + " but are left in storage.") @Pure public abstract @Nullable Boolean getCleanFiles(); @@ -272,10 +284,6 @@ public static Builder builder() { @Pure public abstract @Nullable Integer getDeleteBatchSize(); - public int retainLast() { - return MoreObjects.firstNonNull(getRetainLast(), 1); - } - public boolean cleanExpiredMetadata() { return MoreObjects.firstNonNull(getCleanExpiredMetadata(), false); } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResult.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResult.java index e8166f530e99..535b98c62e85 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResult.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResult.java @@ -28,8 +28,9 @@ * Structured summary metrics of an {@link ExpireSnapshots} run. * *

All counts default to 0. An empty or no-op run produces an all-zeros result. When {@code - * cleanFiles} is set to {@code false} (dry run mode), file counts reflect unreferenced candidate - * files identified for deletion rather than files actually deleted from storage. + * cleanFiles} is set to {@code false}, snapshots are still expired from table metadata, but file + * counts reflect the unreferenced files identified for deletion rather than files actually deleted + * from storage. */ @AutoValue @DefaultSchema(AutoValueSchema.class) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBean.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBean.java index bd3897a1dd65..2dadb42ad879 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBean.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBean.java @@ -23,8 +23,13 @@ import java.util.List; import java.util.Objects; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.MoreObjects; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions; import org.apache.iceberg.ManifestContent; import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.PartitionSpecParser; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.util.ByteBuffers; import org.checkerframework.checker.nullness.qual.Nullable; /** @@ -48,6 +53,11 @@ public class ManifestFileBean implements ManifestFile, Serializable { private final long sequenceNumber; private final long minSequenceNumber; private final boolean valid; + private final byte @Nullable [] keyMetadata; + private final @Nullable String partitionSpecJson; + private final @Nullable String partitionSchemaJson; + + private transient @Nullable PartitionSpec cachedPartitionSpec; public ManifestFileBean( String path, @@ -88,6 +98,44 @@ public ManifestFileBean( long sequenceNumber, long minSequenceNumber, boolean valid) { + this( + path, + length, + partitionSpecId, + content, + snapshotId, + addedFilesCount, + addedRowsCount, + existingFilesCount, + existingRowsCount, + deletedFilesCount, + deletedRowsCount, + sequenceNumber, + minSequenceNumber, + valid, + null, + null, + null); + } + + private ManifestFileBean( + String path, + long length, + int partitionSpecId, + ManifestContent content, + @Nullable Long snapshotId, + @Nullable Integer addedFilesCount, + @Nullable Long addedRowsCount, + @Nullable Integer existingFilesCount, + @Nullable Long existingRowsCount, + @Nullable Integer deletedFilesCount, + @Nullable Long deletedRowsCount, + long sequenceNumber, + long minSequenceNumber, + boolean valid, + byte @Nullable [] keyMetadata, + @Nullable String partitionSpecJson, + @Nullable String partitionSchemaJson) { this.path = path; this.length = length; this.partitionSpecId = partitionSpecId; @@ -102,9 +150,13 @@ public ManifestFileBean( this.sequenceNumber = sequenceNumber; this.minSequenceNumber = minSequenceNumber; this.valid = valid; + this.keyMetadata = keyMetadata; + this.partitionSpecJson = partitionSpecJson; + this.partitionSchemaJson = partitionSchemaJson; } public static ManifestFileBean fromManifestFile(ManifestFile manifest, boolean valid) { + ByteBuffer manifestKeyMetadata = manifest.keyMetadata(); return new ManifestFileBean( manifest.path(), manifest.length(), @@ -119,7 +171,54 @@ public static ManifestFileBean fromManifestFile(ManifestFile manifest, boolean v manifest.deletedRowsCount(), manifest.sequenceNumber(), manifest.minSequenceNumber(), - valid); + valid, + manifestKeyMetadata != null ? ByteBuffers.toByteArray(manifestKeyMetadata) : null, + null, + null); + } + + /** + * Returns a copy of this bean that carries the given partition spec. Used for manifests whose + * spec may no longer be present in the table metadata (e.g. after expired metadata cleanup), so + * that the manifest can still be read on workers. + */ + public ManifestFileBean withPartitionSpec(PartitionSpec spec) { + Preconditions.checkArgument( + spec.specId() == partitionSpecId, + "Partition spec id %s does not match manifest spec id %s", + spec.specId(), + partitionSpecId); + return new ManifestFileBean( + path, + length, + partitionSpecId, + content, + snapshotId, + addedFilesCount, + addedRowsCount, + existingFilesCount, + existingRowsCount, + deletedFilesCount, + deletedRowsCount, + sequenceNumber, + minSequenceNumber, + valid, + keyMetadata, + PartitionSpecParser.toJson(spec), + SchemaParser.toJson(spec.schema())); + } + + /** + * Returns the partition spec carried by this bean, or {@code null} if none was attached via + * {@link #withPartitionSpec(PartitionSpec)}. + */ + public @Nullable PartitionSpec partitionSpec() { + if (cachedPartitionSpec == null && partitionSpecJson != null && partitionSchemaJson != null) { + cachedPartitionSpec = + PartitionSpecParser.fromJson( + SchemaParser.fromJson(partitionSchemaJson), partitionSpecJson); + } + return cachedPartitionSpec; } public boolean isValid() { @@ -188,7 +287,7 @@ public List partitions() { @Override public @Nullable ByteBuffer keyMetadata() { - return null; + return keyMetadata != null ? ByteBuffer.wrap(keyMetadata).asReadOnlyBuffer() : null; } @Override diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFn.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFn.java index 351826887d77..3d1120aa0f71 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFn.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFn.java @@ -17,8 +17,10 @@ */ package org.apache.beam.sdk.io.iceberg.maintenance; +import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; import org.apache.beam.sdk.io.iceberg.IcebergUtils; @@ -28,6 +30,7 @@ import org.apache.beam.sdk.values.TupleTag; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.PartitionStatisticsFile; import org.apache.iceberg.Snapshot; import org.apache.iceberg.StatisticsFile; @@ -45,6 +48,10 @@ *

Validates {@code GC_ENABLED}, invokes {@code * table.expireSnapshots().cleanExpiredFiles(false).commit()} to atomically prune snapshots from * metadata, and emits candidate and valid file descriptors for distributed content resolution. + * + *

The metadata commit is a non-idempotent side effect: if the bundle is retried after a + * successful commit, the retry plans no deletions and the files unreferenced by the original commit + * remain in storage as orphans. */ public class PlanExpireSnapshotsDoFn extends DoFn { @@ -66,7 +73,9 @@ public PlanExpireSnapshotsDoFn( @ProcessElement public void processElement(@Element String tableIdString, MultiOutputReceiver out) { TableIdentifier tableId = IcebergUtils.parseTableIdentifier(tableIdString); - Table table = TableCache.getAndRefreshIfStale(catalogConfig, tableId); + // Planning must start from the latest table metadata so that the expired snapshot set (and + // therefore the reported counts and deletion candidates) is computed against current state. + Table table = TableCache.getRefreshed(catalogConfig, tableId); boolean gcEnabled = PropertyUtil.propertyAsBoolean( @@ -85,14 +94,18 @@ public void processElement(@Element String tableIdString, MultiOutputReceiver ou } org.apache.iceberg.ExpireSnapshots expire = table.expireSnapshots(); - if (config.getExpireOlderThan() != null) { - expire = expire.expireOlderThan(config.getExpireOlderThan()); + Long expireOlderThan = config.getExpireOlderThan(); + if (expireOlderThan != null) { + expire = expire.expireOlderThan(expireOlderThan); } - if (config.getRetainLast() != null) { - expire = expire.retainLast(config.retainLast()); + // Only override the table's history.expire.min-snapshots-to-keep property when explicitly set. + Integer retainLast = config.getRetainLast(); + if (retainLast != null) { + expire = expire.retainLast(retainLast); } - if (config.getSnapshotIds() != null) { - for (Long id : config.getSnapshotIds()) { + List snapshotIds = config.getSnapshotIds(); + if (snapshotIds != null) { + for (Long id : snapshotIds) { expire = expire.expireSnapshotId(id); } } @@ -100,6 +113,9 @@ public void processElement(@Element String tableIdString, MultiOutputReceiver ou expire = expire.cleanExpiredMetadata(config.cleanExpiredMetadata()); } + // Capture pre-commit metadata. Expired manifests must be read with their original partition + // specs, which may be removed from table metadata when cleanExpiredMetadata is enabled. + Map originalSpecs = new HashMap<>(table.specs()); List originalStats = Lists.newArrayList(table.statisticsFiles()); List originalPartitionStats = Lists.newArrayList(table.partitionStatisticsFiles()); @@ -210,6 +226,7 @@ public void processElement(@Element String tableIdString, MultiOutputReceiver ou } Set candidateManifestPaths = new HashSet<>(); + Map currentSpecs = table.specs(); for (Snapshot s : originalSnapshots) { if (deletedSnapshotIds.contains(s.snapshotId())) { for (ManifestFile m : s.allManifests(table.io())) { @@ -221,8 +238,15 @@ public void processElement(@Element String tableIdString, MultiOutputReceiver ou FileInfo.of(m.path(), FileCategory.MANIFEST, false, tableIdString))); // If the manifest is already part of a retained snapshot, its files are all valid if (!validManifestPaths.contains(m.path())) { - out.get(MANIFESTS) - .output(KV.of(tableIdString, ManifestFileBean.fromManifestFile(m, false))); + ManifestFileBean bean = ManifestFileBean.fromManifestFile(m, false); + // Carry the original spec for manifests whose spec was removed by the commit + if (!currentSpecs.containsKey(m.partitionSpecId())) { + PartitionSpec originalSpec = originalSpecs.get(m.partitionSpecId()); + if (originalSpec != null) { + bean = bean.withPartitionSpec(originalSpec); + } + } + out.get(MANIFESTS).output(KV.of(tableIdString, bean)); } } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFn.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFn.java index b5fa5ae458b0..9f381e0becb6 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFn.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFn.java @@ -18,9 +18,9 @@ package org.apache.beam.sdk.io.iceberg.maintenance; import java.io.IOException; +import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Objects; import org.apache.beam.sdk.io.iceberg.IcebergCatalogConfig; import org.apache.beam.sdk.io.iceberg.IcebergUtils; import org.apache.beam.sdk.io.iceberg.TableCache; @@ -34,12 +34,16 @@ import org.apache.iceberg.ManifestFiles; import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.FileIO; -import org.checkerframework.checker.nullness.qual.Nullable; /** * Worker DoFn that reads entries from a manifest file and emits tagged {@link FileInfo} records. + * + *

Partition specs are resolved from the spec carried on the {@link ManifestFileBean} when + * present (for manifests whose spec was removed from table metadata), otherwise from the cached + * table, refreshing the table if the manifest references a spec newer than the cached metadata. */ public class ReadManifestDoFn extends DoFn, KV> { @@ -48,11 +52,6 @@ public class ReadManifestDoFn extends DoFn, KV cachedSpecs; - public ReadManifestDoFn(IcebergCatalogConfig catalogConfig) { this.catalogConfig = catalogConfig; } @@ -64,27 +63,10 @@ public void processElement( String tableIdString = element.getKey(); ManifestFileBean manifest = element.getValue(); - Table table; - FileIO io; - Map specs; - if (Objects.equals(cachedTableIdString, tableIdString) - && cachedTable != null - && cachedIo != null - && cachedSpecs != null) { - table = cachedTable; - io = cachedIo; - specs = cachedSpecs; - } else { - table = - TableCache.getAndRefreshIfStale( - catalogConfig, IcebergUtils.parseTableIdentifier(tableIdString)); - io = table.io(); - specs = table.specs(); - cachedTableIdString = tableIdString; - cachedTable = table; - cachedIo = io; - cachedSpecs = specs; - } + TableIdentifier tableId = IcebergUtils.parseTableIdentifier(tableIdString); + Table table = TableCache.get(catalogConfig, tableId); + Map specs = resolveSpecs(table, tableId, manifest); + FileIO io = table.io(); ManifestContent content = manifest.content(); if (content == ManifestContent.DATA) { @@ -112,4 +94,30 @@ public void processElement( throw new IllegalArgumentException("Unsupported manifest content: " + content); } } + + private Map resolveSpecs( + Table table, TableIdentifier tableId, ManifestFileBean manifest) { + int specId = manifest.partitionSpecId(); + Map specs = table.specs(); + if (specs.containsKey(specId)) { + return specs; + } + + PartitionSpec carriedSpec = manifest.partitionSpec(); + if (carriedSpec != null) { + Map merged = new HashMap<>(specs); + merged.put(specId, carriedSpec); + return merged; + } + + // The cached table may predate the spec referenced by this manifest. + specs = TableCache.getRefreshed(catalogConfig, tableId).specs(); + if (!specs.containsKey(specId)) { + throw new IllegalStateException( + String.format( + "Cannot read manifest %s: partition spec %s not found in table %s", + manifest.path(), specId, tableId)); + } + return specs; + } } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTest.java index 7bb92e1cc660..56488077cee4 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTest.java @@ -17,6 +17,9 @@ */ package org.apache.beam.sdk.io.iceberg.maintenance; +import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.not; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -32,6 +35,7 @@ import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.display.DisplayData; import org.apache.beam.sdk.values.PCollection; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; @@ -41,10 +45,13 @@ import org.apache.iceberg.FileFormat; import org.apache.iceberg.FileMetadata; import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.PartitionKey; +import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.Record; import org.apache.iceberg.io.CloseableIterable; import org.junit.ClassRule; import org.junit.Rule; @@ -290,8 +297,8 @@ public void testIdempotencyRepeatedExecutions() throws IOException { } @Test - public void testDryRunDoesNotDeleteFromStorage() throws IOException { - TableIdentifier tableId = TableIdentifier.of("default", "dry_run_" + System.nanoTime()); + public void testCleanFilesDisabledExpiresMetadataButKeepsFiles() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "no_clean_" + System.nanoTime()); Table table = ExpireSnapshotsTestFixtures.createTableWithOverwrites(warehouse, tableId); List snapshots = Lists.newArrayList(table.snapshots()); @@ -302,7 +309,7 @@ public void testDryRunDoesNotDeleteFromStorage() throws IOException { ExpireSnapshots.Configuration.builder() .setExpireOlderThan(cutoff) .setRetainLast(1) - .setCleanFiles(false) // Dry run + .setCleanFiles(false) .build(); PCollection result = @@ -315,14 +322,167 @@ public void testDryRunDoesNotDeleteFromStorage() throws IOException { results -> { ExpireSnapshotsResult r = results.iterator().next(); assertEquals(2L, r.getExpiredSnapshotsCount()); + // Reports the unreferenced file even though it is not deleted assertEquals(1L, r.getDeletedDataFilesCount()); return null; }); pipeline.run(); - // Files must still physically exist on storage after dry run! - assertTrue("File A must still exist after dry run", fileExists(fileAPath)); + // Snapshots are still expired from metadata + table.refresh(); + assertEquals(1, Lists.newArrayList(table.snapshots()).size()); + // Files must still physically exist on storage + assertTrue("File A must still exist when cleanFiles=false", fileExists(fileAPath)); + } + + @Test + public void testUnsetRetainLastHonorsTableProperty() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "min_keep_prop_" + System.nanoTime()); + Table table = ExpireSnapshotsTestFixtures.createTableWithMultiSnapshots(warehouse, tableId); + table.updateProperties().set(TableProperties.MIN_SNAPSHOTS_TO_KEEP, "3").commit(); + table.refresh(); + + // All 3 snapshots are older than the cutoff, but the table requires keeping 3. + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder() + .setExpireOlderThan(System.currentTimeMillis() + 100_000L) + .build(); + + PCollection result = + pipeline + .apply(Create.of(tableId.toString())) + .apply(ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result).containsInAnyOrder(ExpireSnapshotsResult.zeros()); + pipeline.run(); + + table.refresh(); + assertEquals(3, Lists.newArrayList(table.snapshots()).size()); + } + + @Test + public void testDuplicateTableIdentifiersAreExpiredOnce() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "dup_ids_" + System.nanoTime()); + Table table = ExpireSnapshotsTestFixtures.createTableWithOverwrites(warehouse, tableId); + + List snapshots = Lists.newArrayList(table.snapshots()); + long cutoff = snapshots.get(2).timestampMillis(); + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder().setExpireOlderThan(cutoff).setRetainLast(1).build(); + + PCollection result = + pipeline + .apply(Create.of(tableId.toString(), tableId.toString(), tableId.toString())) + .apply(ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result) + .satisfies( + results -> { + ExpireSnapshotsResult r = results.iterator().next(); + assertEquals(2L, r.getExpiredSnapshotsCount()); + assertEquals(1L, r.getDeletedDataFilesCount()); + return null; + }); + + pipeline.run(); + + table.refresh(); + assertEquals(1, Lists.newArrayList(table.snapshots()).size()); + } + + @Test + public void testCleanExpiredMetadataReadsManifestsOfRemovedSpecs() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "removed_spec_" + System.nanoTime()); + Table table = + warehouse.createTable( + tableId, + TestFixtures.SCHEMA, + PartitionSpec.unpartitioned(), + ImmutableMap.of(TableProperties.FORMAT_VERSION, "2")); + + // S1: File A under the unpartitioned spec (spec 0) + DataFile fileA = + warehouse.writeRecords( + "file_a_" + System.nanoTime() + ".parquet", + table.schema(), + Collections.singletonList(ExpireSnapshotsTestFixtures.createRecord(0L, "a"))); + table.newAppend().appendFile(fileA).commit(); + table.refresh(); + int originalSpecId = table.spec().specId(); + + // Evolve to a partitioned spec (spec 1) + table.updateSpec().addField("data").commit(); + table.refresh(); + PartitionSpec newSpec = table.spec(); + + // S2: Delete File A + ExpireSnapshotsTestFixtures.waitUntilAfter(table.currentSnapshot().timestampMillis()); + table.newDelete().deleteFile(fileA).commit(); + table.refresh(); + + // S3: File B under the partitioned spec + Record recordB = ExpireSnapshotsTestFixtures.createRecord(1L, "b"); + PartitionKey partitionKey = new PartitionKey(newSpec, table.schema()); + partitionKey.partition(recordB); + DataFile fileB = + warehouse.writeRecords( + "file_b_" + System.nanoTime() + ".parquet", + table.schema(), + newSpec, + partitionKey, + Collections.singletonList(recordB)); + ExpireSnapshotsTestFixtures.waitUntilAfter(table.currentSnapshot().timestampMillis()); + table.newAppend().appendFile(fileB).commit(); + table.refresh(); + + String fileAPath = fileA.path().toString(); + String fileBPath = fileB.path().toString(); + List snapshots = Lists.newArrayList(table.snapshots()); + assertEquals(3, snapshots.size()); + long cutoff = snapshots.get(2).timestampMillis(); + + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder() + .setExpireOlderThan(cutoff) + .setRetainLast(1) + .setCleanExpiredMetadata(true) + .build(); + + PCollection result = + pipeline + .apply(Create.of(tableId.toString())) + .apply(ExpireSnapshots.create(getCatalogConfig(), config)); + + PAssert.that(result) + .satisfies( + results -> { + ExpireSnapshotsResult r = results.iterator().next(); + assertEquals(2L, r.getExpiredSnapshotsCount()); + assertEquals(1L, r.getDeletedDataFilesCount()); + return null; + }); + + pipeline.run(); + + table.refresh(); + assertFalse( + "Original spec must have been removed by expired metadata cleanup", + table.specs().containsKey(originalSpecId)); + assertFalse("File A must be deleted", fileExists(fileAPath)); + assertTrue("File B must be preserved", fileExists(fileBPath)); + } + + @Test + public void testDisplayData() { + ExpireSnapshots.Configuration config = + ExpireSnapshots.Configuration.builder().setExpireOlderThan(1234L).build(); + DisplayData displayData = DisplayData.from(ExpireSnapshots.create(getCatalogConfig(), config)); + + assertThat(displayData, hasDisplayItem("expireOlderThan", 1234L)); + assertThat(displayData, hasDisplayItem("cleanFiles", true)); + assertThat(displayData, hasDisplayItem("deleteBatchSize", 10_000)); + assertThat(displayData, not(hasDisplayItem("retainLast"))); } @Test diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBeanTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBeanTest.java new file mode 100644 index 000000000000..93302bb7d453 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBeanTest.java @@ -0,0 +1,94 @@ +/* + * 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.beam.sdk.io.iceberg.maintenance; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import org.apache.beam.sdk.coders.SerializableCoder; +import org.apache.beam.sdk.io.iceberg.TestFixtures; +import org.apache.beam.sdk.util.CoderUtils; +import org.apache.iceberg.ManifestContent; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.util.ByteBuffers; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class ManifestFileBeanTest { + + private static final PartitionSpec SPEC = + PartitionSpec.builderFor(TestFixtures.SCHEMA).withSpecId(3).identity("data").build(); + + @Test + public void testCarriedPartitionSpecSurvivesSerialization() throws Exception { + ManifestFileBean bean = + new ManifestFileBean( + "/tmp/manifest.avro", 100L, SPEC.specId(), ManifestContent.DATA, 1L, false) + .withPartitionSpec(SPEC); + + ManifestFileBean decoded = CoderUtils.clone(SerializableCoder.of(ManifestFileBean.class), bean); + + PartitionSpec carried = decoded.partitionSpec(); + assertNotNull(carried); + assertEquals(SPEC.specId(), carried.specId()); + assertEquals(SPEC.fields(), carried.fields()); + assertEquals(SPEC.schema().asStruct(), carried.schema().asStruct()); + } + + @Test + public void testNoCarriedPartitionSpecByDefault() { + ManifestFileBean bean = + new ManifestFileBean("/tmp/manifest.avro", 100L, 0, ManifestContent.DATA, 1L, true); + assertNull(bean.partitionSpec()); + assertNull(bean.keyMetadata()); + } + + @Test + public void testWithPartitionSpecRejectsMismatchedSpecId() { + ManifestFileBean bean = + new ManifestFileBean("/tmp/manifest.avro", 100L, 0, ManifestContent.DATA, 1L, true); + assertThrows(IllegalArgumentException.class, () -> bean.withPartitionSpec(SPEC)); + } + + @Test + public void testKeyMetadataPreservedFromManifestFile() throws Exception { + byte[] key = "encryption-key-metadata".getBytes(StandardCharsets.UTF_8); + ManifestFile source = + new ManifestFileBean("/tmp/manifest.avro", 100L, 0, ManifestContent.DATA, 1L, true) { + @Override + public ByteBuffer keyMetadata() { + return ByteBuffer.wrap(key); + } + }; + + ManifestFileBean bean = ManifestFileBean.fromManifestFile(source, true); + ManifestFileBean decoded = CoderUtils.clone(SerializableCoder.of(ManifestFileBean.class), bean); + + ByteBuffer keyMetadata = decoded.keyMetadata(); + assertNotNull(keyMetadata); + assertArrayEquals(key, ByteBuffers.toByteArray(keyMetadata)); + } +}