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..a7f90f4b26bf --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/DeleteFilesDoFn.java @@ -0,0 +1,189 @@ +/* + * 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.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.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; +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. + * + *

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; + + 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(); + Preconditions.checkState( + tableIdString != null && !tableIdString.isEmpty(), + "Received a file batch without a table identifier"); + + 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()) { + FileIO io = + TableCache.get(catalogConfig, IcebergUtils.parseTableIdentifier(tableIdString)).io(); + failedPaths = deletePaths(io, paths); + DELETED_FILES.inc(paths.size() - failedPaths.size()); + if (!failedPaths.isEmpty()) { + FAILED_FILE_DELETIONS.inc(failedPaths.size()); + } + } else { + LOG.info( + ExpireSnapshots.PREFIX + + "cleanFiles=false; leaving {} unreferenced file(s) of table '{}' in storage.", + paths.size(), + tableIdString); + } + + 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..3b190f3708b5 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshots.java @@ -0,0 +1,331 @@ +/* + * 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.Distinct; +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.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; +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): 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. + *
  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}. + *
+ * + *

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> { + + 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.addIfNotNull( + DisplayData.item("retainLast", config.getRetainLast()).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. Duplicate identifiers are removed so that each table + // is committed against exactly once. + PCollectionTuple planned = + 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); + + 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(), TypeDescriptor.of(FileInfo.class))) + .via(file -> KV.of(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() + .setCleanFiles(true) + .setDeleteBatchSize(10_000); + } + + @SchemaFieldDescription( + "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. If unset, the" + + " table's history.expire.min-snapshots-to-keep property is used.") + @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, 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(); + + @SchemaFieldDescription("Batch size for bulk file deletion calls. Default is 10,000.") + @Pure + public abstract @Nullable Integer getDeleteBatchSize(); + + 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..535b98c62e85 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsResult.java @@ -0,0 +1,153 @@ +/* + * 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}, 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) +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..2dadb42ad879 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ManifestFileBean.java @@ -0,0 +1,341 @@ +/* + * 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.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; + +/** + * 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; + private final byte @Nullable [] keyMetadata; + private final @Nullable String partitionSpecJson; + private final @Nullable String partitionSchemaJson; + + private transient @Nullable PartitionSpec cachedPartitionSpec; + + 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, + 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; + 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; + 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(), + 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, + 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() { + 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 keyMetadata != null ? ByteBuffer.wrap(keyMetadata).asReadOnlyBuffer() : 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..3d1120aa0f71 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/PlanExpireSnapshotsDoFn.java @@ -0,0 +1,256 @@ +/* + * 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.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; +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.PartitionSpec; +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. + * + *

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 { + + 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); + // 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( + 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(); + Long expireOlderThan = config.getExpireOlderThan(); + if (expireOlderThan != null) { + expire = expire.expireOlderThan(expireOlderThan); + } + // 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); + } + List snapshotIds = config.getSnapshotIds(); + if (snapshotIds != null) { + for (Long id : snapshotIds) { + expire = expire.expireSnapshotId(id); + } + } + if (config.getCleanExpiredMetadata() != null) { + 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()); + 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<>(); + Map currentSpecs = table.specs(); + 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())) { + 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 new file mode 100644 index 000000000000..9f381e0becb6 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/maintenance/ReadManifestDoFn.java @@ -0,0 +1,123 @@ +/* + * 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.HashMap; +import java.util.List; +import java.util.Map; +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.catalog.TableIdentifier; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.FileIO; + +/** + * 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> { + + private static final List PROJECTION = + ImmutableList.of(DataFile.FILE_PATH.name(), DataFile.CONTENT.name()); + + private final IcebergCatalogConfig catalogConfig; + + 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(); + + 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) { + 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); + } + } + + 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/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..56488077cee4 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/maintenance/ExpireSnapshotsTest.java @@ -0,0 +1,638 @@ +/* + * 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.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; + +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.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; +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.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; +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 testCleanFilesDisabledExpiresMetadataButKeepsFiles() throws IOException { + TableIdentifier tableId = TableIdentifier.of("default", "no_clean_" + 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) + .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()); + // Reports the unreferenced file even though it is not deleted + assertEquals(1L, r.getDeletedDataFilesCount()); + return null; + }); + + pipeline.run(); + + // 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 + 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. + * + *

+ */ + 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. + * + * + * + * 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. + * + * + */ + 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/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)); + } +} 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;