diff --git a/core/src/main/java/org/apache/iceberg/actions/BaseRepairTable.java b/core/src/main/java/org/apache/iceberg/actions/BaseRepairTable.java new file mode 100644 index 000000000000..929f3bd3194b --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/actions/BaseRepairTable.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.actions; + +import org.immutables.value.Value; + +@Value.Enclosing +@SuppressWarnings("ImmutablesStyle") +@Value.Style( + typeImmutableEnclosing = "ImmutableRepairTable", + visibilityString = "PUBLIC", + builderVisibilityString = "PUBLIC") +interface BaseRepairTable extends RepairTable { + + @Value.Immutable + interface Result extends RepairTable.Result {} +} diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RepairMetrics.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RepairMetrics.java new file mode 100644 index 000000000000..1d5f1c88bcc8 --- /dev/null +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RepairMetrics.java @@ -0,0 +1,195 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.spark.actions; + +import static org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING; + +import java.nio.ByteBuffer; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import org.apache.iceberg.ContentFile; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.FileMetadata; +import org.apache.iceberg.Metrics; +import org.apache.iceberg.MetricsConfig; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Table; +import org.apache.iceberg.avro.Avro; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.mapping.NameMapping; +import org.apache.iceberg.mapping.NameMappingParser; +import org.apache.iceberg.orc.OrcMetrics; +import org.apache.iceberg.parquet.ParquetUtil; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; + +/** + * Reads the statistics of data and delete files and compares them against the statistics recorded + * in manifest entries. + * + *

Recomputed statistics always respect the metrics config of the table so that they are + * comparable with the stored statistics. + */ +class RepairMetrics { + + private RepairMetrics() {} + + /** Returns the name mapping of the table, or null if the table does not define one. */ + static NameMapping nameMapping(Table table) { + String mapping = table.properties().get(DEFAULT_NAME_MAPPING); + return mapping != null ? NameMappingParser.fromJson(mapping) : null; + } + + /** + * Returns the metrics config to use when recomputing the statistics of the given file. + * + *

Position delete files record statistics for the path and position columns only, which is a + * fixed config rather than the config of the table. + */ + static MetricsConfig metricsConfig(Table table, FileContent content) { + return content == FileContent.POSITION_DELETES + ? MetricsConfig.forPositionDelete() + : MetricsConfig.forTable(table); + } + + /** + * Returns true if the statistics of the file can be recomputed by reading it. + * + *

Deletion vectors are stored as blobs inside a Puffin file, so their statistics cannot be + * derived by reading the file they are stored in. + */ + static boolean supportsMetrics(ContentFile file) { + FileFormat format = file.format(); + return format == FileFormat.PARQUET || format == FileFormat.ORC || format == FileFormat.AVRO; + } + + /** Recomputes the statistics of a file by reading it. */ + static Metrics readMetrics( + InputFile input, ContentFile file, MetricsConfig config, NameMapping mapping) { + switch (file.format()) { + case PARQUET: + return ParquetUtil.fileMetrics(input, config, mapping); + case ORC: + return OrcMetrics.fromInputFile(input, config, mapping); + case AVRO: + // Avro does not record column statistics, only the number of records is recoverable + return new Metrics(Avro.rowCount(input), null, null, null, null); + default: + throw new UnsupportedOperationException("Cannot read metrics of format: " + file.format()); + } + } + + /** + * Returns true if the statistics recorded for the file differ from the statistics of the file + * itself. + * + *

The record count and the file size are always compared. Column level statistics are only + * compared when requested, because a table whose metrics config changed after a file was written + * reports statistics that legitimately differ from the recomputed ones. + */ + static boolean statsAreIncorrect( + ContentFile file, Metrics metrics, long fileSizeInBytes, boolean compareColumnMetrics) { + if (file.fileSizeInBytes() != fileSizeInBytes) { + return true; + } + + if (metrics.recordCount() != null && file.recordCount() != metrics.recordCount()) { + return true; + } + + if (!compareColumnMetrics) { + return false; + } + + return !countsMatch(file.columnSizes(), metrics.columnSizes()) + || !countsMatch(file.valueCounts(), metrics.valueCounts()) + || !countsMatch(file.nullValueCounts(), metrics.nullValueCounts()) + || !countsMatch(file.nanValueCounts(), metrics.nanValueCounts()) + || !boundsMatch(file.lowerBounds(), metrics.lowerBounds()) + || !boundsMatch(file.upperBounds(), metrics.upperBounds()); + } + + /** + * Rebuilds the file with the given statistics, preserving every other field. + * + *

The file size is passed separately as it is not part of the metrics of a file. + */ + static ContentFile withStats( + ContentFile file, PartitionSpec spec, Metrics metrics, long fileSizeInBytes) { + if (file.content() == FileContent.DATA) { + return DataFiles.builder(spec) + .copy((DataFile) file) + .withMetrics(metrics) + .withFileSizeInBytes(fileSizeInBytes) + .build(); + } + + DeleteFile delete = (DeleteFile) file; + FileMetadata.Builder builder = + FileMetadata.deleteFileBuilder(spec) + .copy(delete) + .withMetrics(metrics) + .withFileSizeInBytes(fileSizeInBytes); + List equalityFieldIds = delete.equalityFieldIds(); + if (delete.content() == FileContent.EQUALITY_DELETES + && equalityFieldIds != null + && !equalityFieldIds.isEmpty()) { + // copy(DeleteFile) drops the equality field ids, so they must be set again. Otherwise the + // rewritten entry keeps content EQUALITY_DELETES with null equality ids, which makes reads + // fail once the delete is applied. An entry that already lacks them is carried through as is. + builder.ofEqualityDeletes(equalityFieldIds.stream().mapToInt(Integer::intValue).toArray()); + } + + return builder.build(); + } + + /** + * Returns metrics carrying the recomputed record count but the column-level statistics stored for + * the file, used when column metrics are not being repaired so that a flagged entry has only its + * record count and file size corrected. + */ + static Metrics recordCountOnly(ContentFile file, Metrics recomputed) { + return new Metrics( + recomputed.recordCount(), + file.columnSizes(), + file.valueCounts(), + file.nullValueCounts(), + file.nanValueCounts(), + file.lowerBounds(), + file.upperBounds()); + } + + private static boolean countsMatch(Map stored, Map actual) { + return Objects.equals(normalize(stored), normalize(actual)); + } + + private static boolean boundsMatch( + Map stored, Map actual) { + return Objects.equals(normalize(stored), normalize(actual)); + } + + /** Treats a missing map and an empty map as equivalent, as writers use both. */ + private static Map normalize(Map map) { + return map == null ? ImmutableMap.of() : map; + } +} diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RepairTableSparkAction.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RepairTableSparkAction.java new file mode 100644 index 000000000000..43c676b042fb --- /dev/null +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RepairTableSparkAction.java @@ -0,0 +1,820 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.spark.actions; + +import static org.apache.iceberg.MetadataTableType.ENTRIES; + +import java.io.Serializable; +import java.util.EnumMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.function.Function; +import java.util.stream.Collectors; +import org.apache.hadoop.fs.Path; +import org.apache.iceberg.ContentFile; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.HasTableOperations; +import org.apache.iceberg.ManifestContent; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestWriter; +import org.apache.iceberg.Metrics; +import org.apache.iceberg.MetricsConfig; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Partitioning; +import org.apache.iceberg.RollingManifestWriter; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableOperations; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.actions.ImmutableRepairTable; +import org.apache.iceberg.actions.RepairTable; +import org.apache.iceberg.exceptions.CleanableFailure; +import org.apache.iceberg.exceptions.CommitStateUnknownException; +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.apache.iceberg.mapping.NameMapping; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; +import org.apache.iceberg.relocated.com.google.common.collect.Iterables; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.spark.JobGroupInfo; +import org.apache.iceberg.spark.SparkContentFile; +import org.apache.iceberg.spark.SparkDataFile; +import org.apache.iceberg.spark.SparkDeleteFile; +import org.apache.iceberg.spark.source.SerializableTableWithSize; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.PropertyUtil; +import org.apache.iceberg.util.ThreadPools; +import org.apache.spark.api.java.function.MapPartitionsFunction; +import org.apache.spark.broadcast.Broadcast; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Encoder; +import org.apache.spark.sql.Encoders; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.functions; +import org.apache.spark.sql.types.StructType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import scala.Tuple2; + +/** + * An action that repairs incorrect statistics in the manifests of a table. + * + *

The statistics of every live manifest entry are compared against the file the entry refers to. + * Only manifests that contain at least one incorrect entry are rewritten, so the cost of the commit + * is proportional to the number of incorrect entries rather than to the size of the table. + */ +public class RepairTableSparkAction extends BaseSnapshotUpdateSparkAction + implements RepairTable { + + public static final String USE_CACHING = "use-caching"; + public static final boolean USE_CACHING_DEFAULT = false; + + /** + * Whether to compare and repair column level statistics. When disabled, only record counts and + * file sizes are compared and repaired. + * + *

This is disabled by default. Recomputed column statistics reflect the current metrics config + * of the table, but the config a file was written under is not recorded, so a table whose config + * changed reports column statistics that legitimately differ from the recomputed ones. Repairing + * them in that case would overwrite correct statistics. Reading the footer of every candidate + * file happens regardless of this option; it only controls whether column statistics are + * compared. + */ + public static final String REPAIR_COLUMN_METRICS = "repair-column-metrics"; + + public static final boolean REPAIR_COLUMN_METRICS_DEFAULT = false; + + private static final Logger LOG = LoggerFactory.getLogger(RepairTableSparkAction.class); + + private static final RepairTable.Result EMPTY_RESULT = + ImmutableRepairTable.Result.builder() + .repairedManifests(ImmutableList.of()) + .repairedEntryCount(0L) + .build(); + + private static final String NEW_MANIFEST_PREFIX = "repaired-m-"; + + private final Table table; + private final int formatVersion; + private final long targetManifestSizeBytes; + private final boolean shouldStageManifests; + private final String outputLocation; + + private boolean repairFileMetrics = false; + private boolean dryRun = false; + + RepairTableSparkAction(SparkSession spark, Table table) { + super(spark); + this.table = table; + this.targetManifestSizeBytes = + PropertyUtil.propertyAsLong( + table.properties(), + TableProperties.MANIFEST_TARGET_SIZE_BYTES, + TableProperties.MANIFEST_TARGET_SIZE_BYTES_DEFAULT); + + TableOperations ops = ((HasTableOperations) table).operations(); + Path metadataFilePath = new Path(ops.metadataFileLocation("file")); + this.outputLocation = metadataFilePath.getParent().toString(); + this.formatVersion = ops.current().formatVersion(); + + boolean snapshotIdInheritanceEnabled = + PropertyUtil.propertyAsBoolean( + table.properties(), + TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED, + TableProperties.SNAPSHOT_ID_INHERITANCE_ENABLED_DEFAULT); + this.shouldStageManifests = formatVersion == 1 && !snapshotIdInheritanceEnabled; + } + + @Override + protected RepairTableSparkAction self() { + return this; + } + + @Override + public RepairTableSparkAction repairFileMetrics() { + this.repairFileMetrics = true; + return this; + } + + @Override + public RepairTableSparkAction dryRun() { + this.dryRun = true; + return this; + } + + @Override + public RepairTable.Result execute() { + String desc = String.format("Repairing manifests in %s (dryRun=%s)", table.name(), dryRun); + JobGroupInfo info = newJobGroupInfo("REPAIR-TABLE", desc); + return withJobGroupInfo(info, this::doExecute); + } + + private RepairTable.Result doExecute() { + if (!repairFileMetrics) { + // no repair was selected through the configuration methods, so there is nothing to do + return EMPTY_RESULT; + } + + Snapshot currentSnapshot = table.currentSnapshot(); + if (currentSnapshot == null) { + return EMPTY_RESULT; + } + + List repairedManifests = Lists.newArrayList(); + List newManifests = Lists.newArrayList(); + long repairedCount = 0L; + + for (ManifestContent content : ManifestContent.values()) { + RepairedManifests repaired = repairTable(content, currentSnapshot); + repairedManifests.addAll(repaired.repairedManifests()); + newManifests.addAll(repaired.newManifests()); + repairedCount += repaired.repairedCount(); + } + + if (repairedManifests.isEmpty()) { + return EMPTY_RESULT; + } + + // a dry run writes no manifests, so there is nothing to commit or clean up + if (!dryRun) { + replaceManifests(repairedManifests, newManifests); + } + + LOG.info( + "Repaired the stats of {} manifest entries, rewriting {} manifests as {} (dryRun={})", + repairedCount, + repairedManifests.size(), + newManifests.size(), + dryRun); + + return ImmutableRepairTable.Result.builder() + .repairedManifests(repairedManifests) + .repairedEntryCount(repairedCount) + .build(); + } + + private RepairedManifests repairTable(ManifestContent content, Snapshot snapshot) { + List manifests = loadManifests(content, snapshot); + if (manifests.isEmpty()) { + return RepairedManifests.empty(); + } + + // A manifest is rewritten with the spec it was written under, so manifests are grouped by + // spec and each group is repaired separately. Rewriting a manifest of an older spec with the + // current spec of the table would change the partition data of its entries. + Map> manifestsBySpecId = + manifests.stream().collect(Collectors.groupingBy(ManifestFile::partitionSpecId)); + + List repairedManifests = Lists.newArrayList(); + List newManifests = Lists.newArrayList(); + long repairedCount = 0L; + + for (Map.Entry> group : manifestsBySpecId.entrySet()) { + RepairedManifests repaired = repairManifests(content, group.getKey(), group.getValue()); + repairedManifests.addAll(repaired.repairedManifests()); + newManifests.addAll(repaired.newManifests()); + repairedCount += repaired.repairedCount(); + } + + return RepairedManifests.of(repairedManifests, newManifests, repairedCount); + } + + private RepairedManifests repairManifests( + ManifestContent content, int specId, List manifests) { + Dataset entryDF = buildManifestEntryDF(manifests); + + return withReusableDS( + entryDF, + df -> { + // the entries whose stats disagree with the files they refer to, as (manifest, path). + // cached because it is small, one row per incorrect entry, and is read by several actions + // below, whereas recomputing it would re-read every file. + Dataset verdicts = + df.mapPartitions( + newCheckStatsFunc(content, specId), + Encoders.tuple(Encoders.STRING(), Encoders.STRING())) + .toDF("manifest", "path") + .cache(); + + try { + List manifestsToRewrite = + verdicts.select("manifest").distinct().as(Encoders.STRING()).collectAsList(); + + if (manifestsToRewrite.isEmpty()) { + return RepairedManifests.empty(); + } + + long repairedCount = verdicts.count(); + List rewritten = + manifests.stream() + .filter(manifest -> manifestsToRewrite.contains(manifest.path())) + .collect(Collectors.toList()); + + // a dry run reports what would be repaired without writing any manifests + if (dryRun) { + return RepairedManifests.of(rewritten, ImmutableList.of(), repairedCount); + } + + // mark every entry of the affected manifests with whether its stats need repair by + // joining on the file path, so the unbounded per file set stays distributed rather than + // being collected to the driver and broadcast back out + Dataset entriesToRewrite = + df.filter(df.col("manifest").isin(manifestsToRewrite.toArray())); + Dataset markedEntries = markEntriesToRepair(entriesToRewrite, verdicts); + List written = + writeManifests(content, specId, markedEntries, rewritten.size()); + + return RepairedManifests.of(rewritten, written, repairedCount); + } finally { + verdicts.unpersist(false); + } + }); + } + + /** + * Marks every entry with a boolean {@code repair} column that is true when the entry's file has + * incorrect statistics, by left joining the entries against the verdicts on the file path. + */ + private Dataset markEntriesToRepair(Dataset entries, Dataset verdicts) { + Dataset repairedPaths = + verdicts.select(verdicts.col("path").as("repaired_path")).distinct(); + return entries + .join( + repairedPaths, + entries.col("data_file.file_path").equalTo(repairedPaths.col("repaired_path")), + "left") + .withColumn("repair", functions.col("repaired_path").isNotNull()) + .drop("repaired_path") + .select( + "manifest", + "snapshot_id", + "sequence_number", + "file_sequence_number", + "data_file", + "repair"); + } + + /** + * Loads the live entries of the given manifests, keeping the manifest each entry was read from so + * that only the manifests containing an incorrect entry are rewritten. + */ + private Dataset buildManifestEntryDF(List manifests) { + Dataset manifestDF = + spark() + .createDataset(Lists.transform(manifests, ManifestFile::path), Encoders.STRING()) + .toDF("manifest"); + + Dataset entryDF = + loadMetadataTable(table, ENTRIES) + .filter("status < 2") // select only live entries + .selectExpr( + "input_file_name() as manifest", + "snapshot_id", + "sequence_number", + "file_sequence_number", + "data_file"); + + return entryDF.join( + manifestDF, manifestDF.col("manifest").equalTo(entryDF.col("manifest")), "left_semi"); + } + + private List writeManifests( + ManifestContent content, int specId, Dataset entryDF, int numManifests) { + StructType sparkType = (StructType) entryDF.schema().apply("data_file").dataType(); + Types.StructType combinedFileType = DataFile.getType(Partitioning.partitionType(table)); + Types.StructType fileType = DataFile.getType(table.specs().get(specId).partitionType()); + ManifestWriterFactory writers = manifestWriters(specId); + RepairContext context = newRepairContext(content, specId); + + WriteManifests writeFunc = + content == ManifestContent.DATA + ? new WriteDataManifests(writers, combinedFileType, fileType, sparkType, context) + : new WriteDeleteManifests(writers, combinedFileType, fileType, sparkType, context); + + // repartition by manifest so the entries of each manifest are written together and the layout + // of the table is preserved, rather than scattered round robin as a plain repartition(n) would. + // this produces about as many manifests as are being replaced. + return writeFunc + .apply(entryDF.repartition(numManifests, entryDF.col("manifest"))) + .collectAsList(); + } + + private CheckStats newCheckStatsFunc(ManifestContent content, int specId) { + return new CheckStats(newRepairContext(content, specId)); + } + + private RepairContext newRepairContext(ManifestContent content, int specId) { + boolean repairColumnMetrics = + PropertyUtil.propertyAsBoolean( + options(), REPAIR_COLUMN_METRICS, REPAIR_COLUMN_METRICS_DEFAULT); + return new RepairContext( + sparkContext().broadcast(SerializableTableWithSize.copyOf(table)), + content, + specId, + repairColumnMetrics); + } + + private List loadManifests(ManifestContent content, Snapshot snapshot) { + switch (content) { + case DATA: + return snapshot.dataManifests(table.io()); + case DELETES: + return snapshot.deleteManifests(table.io()); + default: + throw new IllegalArgumentException("Unknown manifest content: " + content); + } + } + + private void replaceManifests( + Iterable deletedManifests, Iterable addedManifests) { + try { + org.apache.iceberg.RewriteManifests rewriteManifests = table.rewriteManifests(); + deletedManifests.forEach(rewriteManifests::deleteManifest); + addedManifests.forEach(rewriteManifests::addManifest); + commit(rewriteManifests); + + if (shouldStageManifests) { + // delete new manifests as they were rewritten before the commit + deleteFiles(Iterables.transform(addedManifests, ManifestFile::path)); + } + } catch (CommitStateUnknownException e) { + // don't clean up added manifest files, because they may have been successfully committed + throw e; + } catch (Exception e) { + if (e instanceof CleanableFailure) { + deleteFiles(Iterables.transform(addedManifests, ManifestFile::path)); + } + + throw e; + } + } + + private void deleteFiles(Iterable locations) { + Iterable files = + Iterables.transform(locations, location -> new FileInfo(location, MANIFEST)); + if (table.io() instanceof SupportsBulkOperations) { + deleteFiles((SupportsBulkOperations) table.io(), files.iterator()); + } else { + deleteFiles( + ThreadPools.getWorkerPool(), file -> table.io().deleteFile(file), files.iterator()); + } + } + + private ManifestWriterFactory manifestWriters(int specId) { + return new ManifestWriterFactory( + sparkContext().broadcast(SerializableTableWithSize.copyOf(table)), + formatVersion, + specId, + outputLocation, + // allow the actual size of manifests to be 20% higher as the estimation is not precise + (long) (1.2 * targetManifestSizeBytes)); + } + + private U withReusableDS(Dataset ds, Function, U> func) { + boolean useCaching = + PropertyUtil.propertyAsBoolean(options(), USE_CACHING, USE_CACHING_DEFAULT); + Dataset reusableDS = useCaching ? ds.cache() : ds; + + try { + return func.apply(reusableDS); + } finally { + if (useCaching) { + reusableDS.unpersist(false); + } + } + } + + /** The outcome of repairing the manifests of one content type. */ + private static class RepairedManifests { + private final List repairedManifests; + private final List newManifests; + private final long repairedCount; + + private RepairedManifests( + List repairedManifests, List newManifests, long repairedCount) { + this.repairedManifests = repairedManifests; + this.newManifests = newManifests; + this.repairedCount = repairedCount; + } + + static RepairedManifests empty() { + return new RepairedManifests(ImmutableList.of(), ImmutableList.of(), 0L); + } + + static RepairedManifests of( + List repairedManifests, List newManifests, long repairedCount) { + return new RepairedManifests(repairedManifests, newManifests, repairedCount); + } + + List repairedManifests() { + return repairedManifests; + } + + List newManifests() { + return newManifests; + } + + long repairedCount() { + return repairedCount; + } + } + + /** + * The state needed to read the statistics of a file on an executor. + * + *

The table is broadcast so that the file IO, schema and metrics config are available without + * being resolved for every entry. + */ + private static class RepairContext implements Serializable { + private final Broadcast tableBroadcast; + private final ManifestContent content; + private final int specId; + private final boolean repairColumnMetrics; + + private transient Map lazyMetricsConfigs = null; + private transient NameMapping lazyNameMapping = null; + private transient boolean nameMappingResolved = false; + + RepairContext( + Broadcast
tableBroadcast, + ManifestContent content, + int specId, + boolean repairColumnMetrics) { + this.tableBroadcast = tableBroadcast; + this.content = content; + this.specId = specId; + this.repairColumnMetrics = repairColumnMetrics; + } + + Table table() { + return tableBroadcast.value(); + } + + FileIO io() { + return table().io(); + } + + ManifestContent content() { + return content; + } + + boolean repairColumnMetrics() { + return repairColumnMetrics; + } + + PartitionSpec spec(int id) { + return table().specs().get(id); + } + + /** + * Returns the metrics config for the content type of the given file. A delete manifest can hold + * both position and equality deletes, whose configs differ, so the config is cached per content + * type rather than once for the whole manifest. + */ + MetricsConfig metricsConfig(ContentFile file) { + if (lazyMetricsConfigs == null) { + this.lazyMetricsConfigs = new EnumMap<>(FileContent.class); + } + + return lazyMetricsConfigs.computeIfAbsent( + file.content(), fileContent -> RepairMetrics.metricsConfig(table(), fileContent)); + } + + NameMapping nameMapping() { + if (!nameMappingResolved) { + this.lazyNameMapping = RepairMetrics.nameMapping(table()); + this.nameMappingResolved = true; + } + + return lazyNameMapping; + } + + SparkContentFile newFileWrapper(Types.StructType combinedFileType, StructType sparkType) { + Types.StructType fileType = DataFile.getType(spec(specId).partitionType()); + return content == ManifestContent.DATA + ? new SparkDataFile(combinedFileType, fileType, sparkType) + : new SparkDeleteFile(combinedFileType, fileType, sparkType); + } + } + + /** + * Compares the statistics of every entry against the file the entry refers to, emitting the + * manifest path and file path of each entry whose statistics are incorrect. + */ + private static class CheckStats implements MapPartitionsFunction> { + private final RepairContext context; + + CheckStats(RepairContext context) { + this.context = context; + } + + @Override + public Iterator> call(Iterator rows) { + List> verdicts = Lists.newArrayList(); + // the combined file type and the wrapper are identical for every row of the partition + Types.StructType combinedFileType = + DataFile.getType(Partitioning.partitionType(context.table())); + SparkContentFile fileWrapper = null; + + while (rows.hasNext()) { + Row row = rows.next(); + String manifest = row.getString(0); + Row fileRow = row.getStruct(4); + if (fileWrapper == null) { + fileWrapper = context.newFileWrapper(combinedFileType, (StructType) fileRow.schema()); + } + + ContentFile file = (ContentFile) fileWrapper.wrap(fileRow); + + if (!RepairMetrics.supportsMetrics(file)) { + continue; + } + + String location = file.location().toString(); + + try { + InputFile input = context.io().newInputFile(location); + long fileSizeInBytes = input.getLength(); + Metrics metrics = + RepairMetrics.readMetrics( + input, file, context.metricsConfig(file), context.nameMapping()); + + if (RepairMetrics.statsAreIncorrect( + file, metrics, fileSizeInBytes, context.repairColumnMetrics())) { + verdicts.add(new Tuple2<>(manifest, location)); + } + } catch (Exception e) { + // the stats of a file that cannot be read are left alone, as whether they are + // correct cannot be told without reading it + LOG.warn("Skipping the entry of {} as its statistics could not be read", location, e); + } + } + + return verdicts.iterator(); + } + } + + private static class WriteDataManifests extends WriteManifests { + WriteDataManifests( + ManifestWriterFactory writers, + Types.StructType combinedFileType, + Types.StructType fileType, + StructType sparkFileType, + RepairContext context) { + super(writers, combinedFileType, fileType, sparkFileType, context); + } + + @Override + protected SparkContentFile newFileWrapper() { + return new SparkDataFile(combinedFileType(), fileType(), sparkFileType()); + } + + @Override + protected RollingManifestWriter newManifestWriter() { + return writers().newRollingManifestWriter(); + } + } + + private static class WriteDeleteManifests extends WriteManifests { + WriteDeleteManifests( + ManifestWriterFactory writers, + Types.StructType combinedFileType, + Types.StructType fileType, + StructType sparkFileType, + RepairContext context) { + super(writers, combinedFileType, fileType, sparkFileType, context); + } + + @Override + protected SparkContentFile newFileWrapper() { + return new SparkDeleteFile(combinedFileType(), fileType(), sparkFileType()); + } + + @Override + protected RollingManifestWriter newManifestWriter() { + return writers().newRollingDeleteManifestWriter(); + } + } + + /** + * Writes the entries of the manifests being repaired, replacing the statistics of the entries + * that were found to be incorrect. + * + *

Entries are always written with {@link RollingManifestWriter#existing}, carrying the + * original snapshot id and sequence numbers so that the lineage of the files, and therefore the + * delete files that apply to them, is preserved. + */ + private abstract static class WriteManifests> + implements MapPartitionsFunction { + + private static final Encoder MANIFEST_ENCODER = + Encoders.javaSerialization(ManifestFile.class); + + private final ManifestWriterFactory writers; + private final Types.StructType combinedFileType; + private final Types.StructType fileType; + private final StructType sparkFileType; + private final RepairContext context; + + WriteManifests( + ManifestWriterFactory writers, + Types.StructType combinedFileType, + Types.StructType fileType, + StructType sparkFileType, + RepairContext context) { + this.writers = writers; + this.combinedFileType = combinedFileType; + this.fileType = fileType; + this.sparkFileType = sparkFileType; + this.context = context; + } + + protected abstract SparkContentFile newFileWrapper(); + + protected abstract RollingManifestWriter newManifestWriter(); + + public Dataset apply(Dataset input) { + return input.mapPartitions(this, MANIFEST_ENCODER); + } + + @Override + @SuppressWarnings("unchecked") + public Iterator call(Iterator rows) throws Exception { + SparkContentFile fileWrapper = newFileWrapper(); + RollingManifestWriter writer = newManifestWriter(); + + try { + while (rows.hasNext()) { + Row row = rows.next(); + long snapshotId = row.getLong(1); + long sequenceNumber = row.getLong(2); + Long fileSequenceNumber = row.isNullAt(3) ? null : row.getLong(3); + Row fileRow = row.getStruct(4); + boolean repair = row.getBoolean(5); + + F file = fileWrapper.wrap(fileRow); + if (repair) { + file = (F) repairStats(file); + } + + writer.existing(file, snapshotId, sequenceNumber, fileSequenceNumber); + } + } finally { + writer.close(); + } + + return writer.toManifestFiles().iterator(); + } + + /** Rebuilds the file with the statistics read from the file itself. */ + private ContentFile repairStats(ContentFile file) { + InputFile input = context.io().newInputFile(file.location()); + long fileSizeInBytes = input.getLength(); + Metrics metrics = + RepairMetrics.readMetrics( + input, file, context.metricsConfig(file), context.nameMapping()); + if (!context.repairColumnMetrics()) { + // only the record count and file size are being repaired, so keep the stored column stats + metrics = RepairMetrics.recordCountOnly(file, metrics); + } + + return RepairMetrics.withStats(file, context.spec(file.specId()), metrics, fileSizeInBytes); + } + + protected ManifestWriterFactory writers() { + return writers; + } + + protected Types.StructType combinedFileType() { + return combinedFileType; + } + + protected Types.StructType fileType() { + return fileType; + } + + protected StructType sparkFileType() { + return sparkFileType; + } + } + + private static class ManifestWriterFactory implements Serializable { + private final Broadcast

tableBroadcast; + private final int formatVersion; + private final int specId; + private final String outputLocation; + private final long maxManifestSizeBytes; + + ManifestWriterFactory( + Broadcast
tableBroadcast, + int formatVersion, + int specId, + String outputLocation, + long maxManifestSizeBytes) { + this.tableBroadcast = tableBroadcast; + this.formatVersion = formatVersion; + this.specId = specId; + this.outputLocation = outputLocation; + this.maxManifestSizeBytes = maxManifestSizeBytes; + } + + RollingManifestWriter newRollingManifestWriter() { + return new RollingManifestWriter<>(this::newManifestWriter, maxManifestSizeBytes); + } + + private ManifestWriter newManifestWriter() { + return ManifestFiles.write(formatVersion, spec(), newOutputFile(), null); + } + + RollingManifestWriter newRollingDeleteManifestWriter() { + return new RollingManifestWriter<>(this::newDeleteManifestWriter, maxManifestSizeBytes); + } + + private ManifestWriter newDeleteManifestWriter() { + return ManifestFiles.writeDeleteManifest(formatVersion, spec(), newOutputFile(), null); + } + + private PartitionSpec spec() { + return table().specs().get(specId); + } + + private OutputFile newOutputFile() { + return table().io().newOutputFile(newManifestLocation()); + } + + private String newManifestLocation() { + String fileName = FileFormat.AVRO.addExtension(NEW_MANIFEST_PREFIX + UUID.randomUUID()); + Path filePath = new Path(outputLocation, fileName); + return filePath.toString(); + } + + private Table table() { + return tableBroadcast.value(); + } + } +} diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java index b7361c336a69..7e8cee437edc 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/SparkActions.java @@ -85,6 +85,11 @@ public RewriteManifestsSparkAction rewriteManifests(Table table) { return new RewriteManifestsSparkAction(spark, table); } + @Override + public RepairTableSparkAction repairTable(Table table) { + return new RepairTableSparkAction(spark, table); + } + @Override public ExpireSnapshotsSparkAction expireSnapshots(Table table) { return new ExpireSnapshotsSparkAction(spark, table); diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRepairTableAction.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRepairTableAction.java new file mode 100644 index 000000000000..a04af5c2ca22 --- /dev/null +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRepairTableAction.java @@ -0,0 +1,968 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.spark.actions; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assumptions.assumeThat; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.ContentFile; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.DeleteFile; +import org.apache.iceberg.FileContent; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.FileMetadata; +import org.apache.iceberg.Files; +import org.apache.iceberg.ManifestFile; +import org.apache.iceberg.ManifestFiles; +import org.apache.iceberg.ManifestWriter; +import org.apache.iceberg.Metrics; +import org.apache.iceberg.Parameter; +import org.apache.iceberg.ParameterizedTestExtension; +import org.apache.iceberg.Parameters; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.actions.RepairTable; +import org.apache.iceberg.data.FileHelpers; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.exceptions.CommitStateUnknownException; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.hadoop.HadoopTables; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.spark.TestBase; +import org.apache.iceberg.spark.source.ThreeColumnRecord; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.Pair; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.io.TempDir; + +@ExtendWith(ParameterizedTestExtension.class) +public class TestRepairTableAction extends TestBase { + + private static final HadoopTables TABLES = new HadoopTables(new Configuration()); + private static final Schema SCHEMA = + new Schema( + optional(1, "c1", Types.IntegerType.get()), + optional(2, "c2", Types.StringType.get()), + optional(3, "c3", Types.StringType.get())); + + @Parameters(name = "formatVersion = {0}") + public static Object[] parameters() { + return new Object[][] {new Object[] {1}, new Object[] {2}, new Object[] {3}}; + } + + @Parameter private int formatVersion; + + private String tableLocation = null; + + @TempDir private Path temp; + @TempDir private File tableDir; + + @BeforeEach + public void setupTableLocation() { + this.tableLocation = tableDir.toURI().toString(); + } + + @TestTemplate + public void testRepairEmptyTable() { + Table table = createTable(PartitionSpec.unpartitioned()); + + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); + + assertThat(result.repairedManifests()).isEmpty(); + assertThat(result.repairedEntryCount()).isEqualTo(0); + } + + @TestTemplate + public void testRepairTableWithCorrectStats() { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + Snapshot before = table.currentSnapshot(); + + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); + + assertThat(result.repairedManifests()).isEmpty(); + assertThat(result.repairedEntryCount()).isEqualTo(0); + + table.refresh(); + assertThat(table.currentSnapshot().snapshotId()) + .as("should not commit a snapshot when nothing is repaired") + .isEqualTo(before.snapshotId()); + } + + @TestTemplate + public void testNoRepairSelectedIsNoOp() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + DataFile original = onlyDataFile(table); + replaceManifestWithCorruptStats(table, original); + + table.refresh(); + Snapshot before = table.currentSnapshot(); + DataFile corrupt = onlyDataFile(table); + + // no repair was selected, so execute() must do nothing even though the stats are incorrect + RepairTable.Result result = SparkActions.get().repairTable(table).execute(); + + assertThat(result.repairedManifests()).isEmpty(); + assertThat(result.repairedEntryCount()).isEqualTo(0); + + table.refresh(); + assertThat(table.currentSnapshot().snapshotId()) + .as("a repair with nothing selected must not commit") + .isEqualTo(before.snapshotId()); + assertThat(onlyDataFile(table).recordCount()) + .as("a repair with nothing selected must leave the incorrect stats in place") + .isEqualTo(corrupt.recordCount()); + } + + @TestTemplate + public void testRepairIncorrectRecordCountAndFileSize() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + List expectedRows = currentRows(); + DataFile original = onlyDataFile(table); + + // replace the manifest with one whose entry records a wrong record count and file size + replaceManifestWithCorruptStats(table, original); + + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); + + assertThat(result.repairedEntryCount()).isEqualTo(1); + assertThat(result.repairedManifests()).hasSize(1); + + table.refresh(); + DataFile repaired = onlyDataFile(table); + assertThat(repaired.recordCount()).isEqualTo(original.recordCount()); + assertThat(repaired.fileSizeInBytes()).isEqualTo(original.fileSizeInBytes()); + assertThat(repaired.location()).isEqualTo(original.location()); + + assertThat(currentRows()) + .as("table contents must be unchanged by the repair") + .containsExactlyInAnyOrderElementsOf(expectedRows); + } + + @TestTemplate + public void testRepairPreservesEntryLineage() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + DataFile original = onlyDataFile(table); + List lineageBefore = entryLineage(); + + replaceManifestWithCorruptStats(table, original); + + SparkActions.get().repairTable(table).repairFileMetrics().execute(); + + table.refresh(); + assertThat(entryLineage()) + .as("snapshot id and sequence numbers must be carried through the repair") + .containsExactlyInAnyOrderElementsOf(lineageBefore); + } + + @TestTemplate + public void testDryRunDoesNotCommit() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + DataFile original = onlyDataFile(table); + replaceManifestWithCorruptStats(table, original); + + table.refresh(); + Snapshot before = table.currentSnapshot(); + DataFile corrupt = onlyDataFile(table); + + RepairTable.Result result = + SparkActions.get().repairTable(table).repairFileMetrics().dryRun().execute(); + + assertThat(result.repairedEntryCount()).isEqualTo(1); + assertThat(result.repairedManifests()).hasSize(1); + + table.refresh(); + assertThat(table.currentSnapshot().snapshotId()) + .as("dry run must not commit") + .isEqualTo(before.snapshotId()); + assertThat(onlyDataFile(table).recordCount()) + .as("dry run must leave the incorrect stats in place") + .isEqualTo(corrupt.recordCount()); + } + + @TestTemplate + public void testRepairOnlyRewritesAffectedManifests() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(2)); + appendRecords(table, records(2)); + + table.refresh(); + assertThat(table.currentSnapshot().dataManifests(table.io())).hasSize(2); + + List manifests = table.currentSnapshot().dataManifests(table.io()); + ManifestFile untouched = manifests.get(1); + + // corrupt the entry of one manifest only + DataFile fileToCorrupt = readDataFiles(table, manifests.get(0)).get(0); + corruptStats(table, manifests.get(0), fileToCorrupt.location()); + + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); + + assertThat(result.repairedManifests()).hasSize(1); + assertThat(result.repairedEntryCount()).isEqualTo(1); + + table.refresh(); + assertThat(table.currentSnapshot().dataManifests(table.io())) + .as("the manifest without incorrect entries must be left in place") + .anyMatch(manifest -> manifest.path().equals(untouched.path())); + } + + @TestTemplate + public void testRepairPartitionedTable() throws IOException { + Table table = createTable(PartitionSpec.builderFor(SCHEMA).identity("c1").build()); + + Dataset df = + spark + .createDataFrame( + Lists.newArrayList( + new ThreeColumnRecord(1, "AAAA", "A"), new ThreeColumnRecord(2, "BBBB", "B")), + ThreeColumnRecord.class) + .coalesce(1); + df.select("c1", "c2", "c3").write().format("iceberg").mode("append").save(tableLocation); + + table.refresh(); + List expectedRows = currentRows(); + ManifestFile manifest = table.currentSnapshot().dataManifests(table.io()).get(0); + List files = readDataFiles(table, manifest); + + corruptStats(table, manifest, files.get(0).location()); + + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); + + assertThat(result.repairedEntryCount()).isEqualTo(1); + assertThat(currentRows()).containsExactlyInAnyOrderElementsOf(expectedRows); + } + + @TestTemplate + public void testRepairSkipsColumnMetricsByDefault() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + DataFile original = onlyDataFile(table); + + // only the column level statistics are wrong, the record count and the file size are correct + ManifestFile manifest = table.currentSnapshot().dataManifests(table.io()).get(0); + corruptStats(table, manifest, original.location(), false); + + RepairTable.Result skipped = + SparkActions.get().repairTable(table).repairFileMetrics().execute(); + + assertThat(skipped.repairedEntryCount()) + .as("column metrics must not be compared by default") + .isEqualTo(0); + + RepairTable.Result repaired = + SparkActions.get() + .repairTable(table) + .repairFileMetrics() + .option(RepairTableSparkAction.REPAIR_COLUMN_METRICS, "true") + .execute(); + + assertThat(repaired.repairedEntryCount()) + .as("column metrics are compared when enabled") + .isEqualTo(1); + } + + @TestTemplate + public void testRepairPreservesColumnStatsWhenColumnMetricsDisabled() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + DataFile original = onlyDataFile(table); + + // the record count, file size and column stats of the entry are all wrong + ManifestFile manifest = table.currentSnapshot().dataManifests(table.io()).get(0); + corruptStats(table, manifest, original.location(), true); + DataFile corrupt = onlyDataFile(table); + + // repair with column metrics disabled: the record count and file size are corrected, but the + // wrong column stats must be left untouched rather than replaced with the recomputed ones + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); + + assertThat(result.repairedEntryCount()).isEqualTo(1); + + table.refresh(); + DataFile repaired = onlyDataFile(table); + assertThat(repaired.recordCount()) + .as("the record count must be repaired") + .isEqualTo(original.recordCount()); + assertThat(repaired.fileSizeInBytes()) + .as("the file size must be repaired") + .isEqualTo(original.fileSizeInBytes()); + assertThat(repaired.valueCounts()) + .as("value counts must be kept, not replaced with recomputed ones") + .isEqualTo(corrupt.valueCounts()); + assertThat(repaired.nullValueCounts()) + .as("null value counts must be kept, not replaced with recomputed ones") + .isEqualTo(corrupt.nullValueCounts()); + assertThat(repaired.columnSizes()) + .as("column sizes must be kept, not replaced with recomputed ones") + .isEqualTo(corrupt.columnSizes()); + } + + @TestTemplate + public void testWithStatsPreservesEqualityFieldIds() { + // a rebuilt equality delete must keep its equality field ids, otherwise reading the table fails + // when the delete is applied. FileMetadata.Builder.copy(DeleteFile) does not carry them. + PartitionSpec spec = PartitionSpec.unpartitioned(); + DeleteFile equalityDelete = + FileMetadata.deleteFileBuilder(spec) + .ofEqualityDeletes(2, 3) + .withPath(tableLocation + "/data/eq-delete.parquet") + .withFileSizeInBytes(1024) + .withFormat(FileFormat.PARQUET) + .withRecordCount(10) + .build(); + + Metrics recomputed = new Metrics(10L, null, null, null, null); + ContentFile rebuilt = RepairMetrics.withStats(equalityDelete, spec, recomputed, 1024L); + + assertThat(rebuilt.content()).isEqualTo(FileContent.EQUALITY_DELETES); + assertThat(((DeleteFile) rebuilt).equalityFieldIds()) + .as("equality field ids must survive a rebuild") + .containsExactly(2, 3); + } + + @TestTemplate + public void testRepairEqualityDeleteStats() throws IOException { + assumeThat(formatVersion) + .as("delete files require format version 2 or higher") + .isGreaterThanOrEqualTo(2); + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + // write a real equality delete file, whose statistics on disk are correct, then commit an entry + // for it that records the wrong statistics, mimicking a writer that recorded them incorrectly + DeleteFile delete = writeEqDeletes(table, "c1", 0); + DeleteFile corruptEntry = + FileMetadata.deleteFileBuilder(table.spec()) + .copy(delete) + .ofEqualityDeletes( + delete.equalityFieldIds().stream().mapToInt(Integer::intValue).toArray()) + .withRecordCount(delete.recordCount() + 100) + .withFileSizeInBytes(delete.fileSizeInBytes() + 4096) + .build(); + table.newRowDelta().addDeletes(corruptEntry).commit(); + table.refresh(); + + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); + + assertThat(result.repairedEntryCount()).isEqualTo(1); + + table.refresh(); + DeleteFile repaired = onlyDeleteFile(table); + assertThat(repaired.recordCount()) + .as("the record count must be repaired") + .isEqualTo(delete.recordCount()); + assertThat(repaired.fileSizeInBytes()) + .as("the file size must be repaired") + .isEqualTo(delete.fileSizeInBytes()); + assertThat(repaired.content()).isEqualTo(FileContent.EQUALITY_DELETES); + assertThat(repaired.equalityFieldIds()) + .as("equality field ids must survive the repair") + .isEqualTo(delete.equalityFieldIds()); + } + + @TestTemplate + public void testRepairPositionDeleteStats() throws IOException { + assumeThat(formatVersion) + .as("position deletes are written as parquet files in format version 2") + .isEqualTo(2); + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + DataFile dataFile = onlyDataFile(table); + + DeleteFile delete = + writePosDeletes(table, Lists.newArrayList(Pair.of(dataFile.location(), 0L))); + DeleteFile corruptEntry = + FileMetadata.deleteFileBuilder(table.spec()) + .copy(delete) + .withRecordCount(delete.recordCount() + 100) + .withFileSizeInBytes(delete.fileSizeInBytes() + 4096) + .build(); + table.newRowDelta().addDeletes(corruptEntry).commit(); + table.refresh(); + + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); + + assertThat(result.repairedEntryCount()).isEqualTo(1); + + table.refresh(); + DeleteFile repaired = onlyDeleteFile(table); + assertThat(repaired.recordCount()) + .as("the record count must be repaired") + .isEqualTo(delete.recordCount()); + assertThat(repaired.fileSizeInBytes()) + .as("the file size must be repaired") + .isEqualTo(delete.fileSizeInBytes()); + assertThat(repaired.content()).isEqualTo(FileContent.POSITION_DELETES); + } + + @TestTemplate + public void testRepairDeleteManifestHoldingBothDeleteTypes() throws IOException { + assumeThat(formatVersion) + .as("position deletes are written as parquet files in format version 2") + .isEqualTo(2); + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + DataFile dataFile = onlyDataFile(table); + + DeleteFile posDelete = + writePosDeletes(table, Lists.newArrayList(Pair.of(dataFile.location(), 0L))); + DeleteFile eqDelete = writeEqDeletes(table, "c1", 1); + + // commit both deletes together so they share a single delete manifest, whose entries then have + // two different content types and therefore two different metrics configs + DeleteFile posEntry = + FileMetadata.deleteFileBuilder(table.spec()) + .copy(posDelete) + .withRecordCount(posDelete.recordCount() + 100) + .withFileSizeInBytes(posDelete.fileSizeInBytes() + 4096) + .build(); + DeleteFile eqEntry = + FileMetadata.deleteFileBuilder(table.spec()) + .copy(eqDelete) + .ofEqualityDeletes( + eqDelete.equalityFieldIds().stream().mapToInt(Integer::intValue).toArray()) + .withRecordCount(eqDelete.recordCount() + 100) + .withFileSizeInBytes(eqDelete.fileSizeInBytes() + 4096) + .build(); + table.newRowDelta().addDeletes(posEntry).addDeletes(eqEntry).commit(); + table.refresh(); + assertThat(table.currentSnapshot().deleteManifests(table.io())) + .as("both deletes must land in a single manifest for this to exercise mixed content") + .hasSize(1); + + // enable column metrics so the metrics config actually matters: the equality delete must be + // repaired under the table's config, not the position delete's, which is what keying the config + // by content type ensures + RepairTable.Result result = + SparkActions.get() + .repairTable(table) + .repairFileMetrics() + .option(RepairTableSparkAction.REPAIR_COLUMN_METRICS, "true") + .execute(); + + assertThat(result.repairedEntryCount()).isEqualTo(2); + + table.refresh(); + Map repairedByPath = Maps.newHashMap(); + for (DeleteFile file : + readDeleteFiles(table, table.currentSnapshot().deleteManifests(table.io()).get(0))) { + repairedByPath.put(file.location(), file); + } + + DeleteFile repairedPos = repairedByPath.get(posDelete.location()); + assertThat(repairedPos.content()).isEqualTo(FileContent.POSITION_DELETES); + assertThat(repairedPos.recordCount()).isEqualTo(posDelete.recordCount()); + assertThat(repairedPos.fileSizeInBytes()).isEqualTo(posDelete.fileSizeInBytes()); + + DeleteFile repairedEq = repairedByPath.get(eqDelete.location()); + assertThat(repairedEq.content()).isEqualTo(FileContent.EQUALITY_DELETES); + assertThat(repairedEq.recordCount()).isEqualTo(eqDelete.recordCount()); + assertThat(repairedEq.fileSizeInBytes()).isEqualTo(eqDelete.fileSizeInBytes()); + assertThat(repairedEq.equalityFieldIds()) + .as("equality field ids must survive the repair of a mixed manifest") + .isEqualTo(eqDelete.equalityFieldIds()); + // the equality delete's column stats must be recomputed under the table's config; had the + // position delete's config been used for it, the value counts would differ from the file + assertThat(repairedEq.valueCounts()) + .as("equality delete column stats must be recomputed under its own metrics config") + .isEqualTo(eqDelete.valueCounts()); + } + + @TestTemplate + public void testRepairSucceedsWithConcurrentAppend() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + DataFile original = onlyDataFile(table); + replaceManifestWithCorruptStats(table, original); + + // append concurrently, after the repair has determined what to rewrite but before it commits + RepairTable.Result result = + repairWithConcurrentChange(table, () -> appendRecords(table, records(2))); + + assertThat(result.repairedEntryCount()).isEqualTo(1); + + table.refresh(); + assertThat(currentRows()) + .as("the concurrently appended records must survive the repair") + .hasSize(6); + assertThat(dataFiles(table)) + .as("the stats of the repaired entry must be corrected") + .anySatisfy( + file -> { + assertThat(file.location()).isEqualTo(original.location()); + assertThat(file.recordCount()).isEqualTo(original.recordCount()); + assertThat(file.fileSizeInBytes()).isEqualTo(original.fileSizeInBytes()); + }); + } + + @TestTemplate + public void testRepairFailsWhenRepairedManifestIsConcurrentlyReplaced() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + DataFile original = onlyDataFile(table); + replaceManifestWithCorruptRecordCount(table, original); + + table.refresh(); + List rowsBeforeRepair = currentRows(); + DataFile corrupt = onlyDataFile(table); + + // concurrently rewrite the very manifest the repair is about to replace + assertThatThrownBy( + () -> + repairWithConcurrentChange( + table, + () -> { + Table concurrent = TABLES.load(tableLocation); + concurrent.rewriteManifests().clusterBy(file -> "").commit(); + })) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("could not be found in the latest snapshot"); + + table.refresh(); + assertThat(currentRows()) + .as("a failed repair must leave the contents of the table unchanged") + .containsExactlyInAnyOrderElementsOf(rowsBeforeRepair); + assertThat(onlyDataFile(table).recordCount()) + .as("a failed repair must not correct any stats") + .isEqualTo(corrupt.recordCount()); + } + + @TestTemplate + public void testRepairCleansUpManifestsOnCommitFailure() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + DataFile original = onlyDataFile(table); + replaceManifestWithCorruptRecordCount(table, original); + table.refresh(); + + List rowsBeforeRepair = currentRows(); + DataFile corrupt = onlyDataFile(table); + + // fail the commit with a cleanable failure, as a table whose retries are exhausted would + org.apache.iceberg.RewriteManifests spyRewriteManifests = spy(table.rewriteManifests()); + doThrow(new CommitFailedException("Injected commit failure")) + .when(spyRewriteManifests) + .commit(); + + Table spyTable = spy(table); + when(spyTable.rewriteManifests()).thenReturn(spyRewriteManifests); + + assertThatThrownBy(() -> SparkActions.get().repairTable(spyTable).repairFileMetrics().execute()) + .isInstanceOf(CommitFailedException.class) + .hasMessage("Injected commit failure"); + + table.refresh(); + assertThat(currentRows()) + .as("a failed repair must leave the contents of the table unchanged") + .containsExactlyInAnyOrderElementsOf(rowsBeforeRepair); + assertThat(onlyDataFile(table).recordCount()) + .as("a failed repair must not correct any stats") + .isEqualTo(corrupt.recordCount()); + assertThat(repairedManifestPaths()) + .as("the manifests written by a failed repair must be deleted") + .isEmpty(); + } + + @TestTemplate + public void testRepairKeepsManifestsOnCommitStateUnknown() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + DataFile original = onlyDataFile(table); + replaceManifestWithCorruptStats(table, original); + table.refresh(); + + // commit successfully but report the outcome as unknown + org.apache.iceberg.RewriteManifests rewriteManifests = table.rewriteManifests(); + org.apache.iceberg.RewriteManifests spyRewriteManifests = spy(rewriteManifests); + doAnswer( + invocation -> { + rewriteManifests.commit(); + throw new CommitStateUnknownException(new RuntimeException("Datacenter on Fire")); + }) + .when(spyRewriteManifests) + .commit(); + + Table spyTable = spy(table); + when(spyTable.rewriteManifests()).thenReturn(spyRewriteManifests); + + assertThatThrownBy(() -> SparkActions.get().repairTable(spyTable).repairFileMetrics().execute()) + .cause() + .isInstanceOf(RuntimeException.class) + .hasMessage("Datacenter on Fire"); + + table.refresh(); + + // the commit did succeed, so the repaired manifests must not have been deleted + assertThat(onlyDataFile(table).recordCount()) + .as("the repair committed, so the corrected stats must be readable") + .isEqualTo(original.recordCount()); + for (ManifestFile manifest : table.currentSnapshot().dataManifests(table.io())) { + assertThat(table.io().newInputFile(manifest.path()).exists()) + .as("manifests of a possibly committed repair must not be deleted") + .isTrue(); + } + } + + @TestTemplate + public void testDryRunLeavesNoManifestsBehind() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + DataFile original = onlyDataFile(table); + replaceManifestWithCorruptStats(table, original); + table.refresh(); + + RepairTable.Result result = + SparkActions.get().repairTable(table).repairFileMetrics().dryRun().execute(); + + assertThat(result.repairedEntryCount()).isEqualTo(1); + assertThat(repairedManifestPaths()) + .as("a dry run must not leave the manifests it wrote behind") + .isEmpty(); + } + + /** + * Runs the repair, applying the given change to the table after the manifests to repair have been + * determined but before the repair commits. + */ + private RepairTable.Result repairWithConcurrentChange(Table table, Runnable change) { + Table spyTable = spy(table); + when(spyTable.rewriteManifests()) + .thenAnswer( + invocation -> { + change.run(); + return table.rewriteManifests(); + }); + + return SparkActions.get().repairTable(spyTable).repairFileMetrics().execute(); + } + + /** + * Returns the manifests written by the repair action that are still present in the metadata + * directory. + * + *

Only the manifests the action itself wrote are considered. A failed commit can also leave + * behind a copy of a manifest made by the format version 1 staging path, which is written and + * owned by the core rewrite manifests operation rather than by this action. + */ + private Set repairedManifestPaths() throws IOException { + Set paths = Sets.newHashSet(); + File metadataDir = new File(tableDir, "metadata"); + File[] files = metadataDir.listFiles(); + if (files != null) { + for (File file : files) { + if (file.getName().startsWith("repaired-m-")) { + paths.add(file.getCanonicalPath()); + } + } + } + + return paths; + } + + @TestTemplate + public void testRepairAfterPartitionSpecEvolution() throws IOException { + Table table = createTable(PartitionSpec.unpartitioned()); + appendRecords(table, records(4)); + + DataFile original = onlyDataFile(table); + assertThat(original.specId()).isEqualTo(0); + assertThat(original.partition().size()).isEqualTo(0); + + // evolve the table to a partitioned spec; the existing manifest keeps referring to spec 0 + table.updateSpec().addField("c1").commit(); + table.refresh(); + assertThat(table.spec().specId()).isEqualTo(1); + + ManifestFile oldManifest = table.currentSnapshot().dataManifests(table.io()).get(0); + assertThat(oldManifest.partitionSpecId()) + .as("the manifest written before the evolution must still be tagged with the old spec") + .isEqualTo(0); + + // corrupt the stats of the entry that still belongs to the original, unpartitioned spec + corruptStats(table, oldManifest, original.location()); + + SparkActions.get().repairTable(table).repairFileMetrics().execute(); + + table.refresh(); + DataFile repaired = onlyDataFile(table); + assertThat(repaired.recordCount()) + .as("the repair must still correct the stats") + .isEqualTo(original.recordCount()); + assertThat(repaired.specId()) + .as("the repaired entry must keep the spec it was originally written under") + .isEqualTo(0); + assertThat(repaired.partition().size()) + .as("an unpartitioned file's partition data must still have zero fields after repair") + .isEqualTo(0); + } + + private List dataFiles(Table table) throws IOException { + List files = Lists.newArrayList(); + for (ManifestFile manifest : table.currentSnapshot().dataManifests(table.io())) { + files.addAll(readDataFiles(table, manifest)); + } + + return files; + } + + private Table createTable(PartitionSpec spec) { + Map options = Maps.newHashMap(); + options.put(TableProperties.FORMAT_VERSION, String.valueOf(formatVersion)); + return TABLES.create(SCHEMA, spec, options, tableLocation); + } + + private List records(int count) { + List records = Lists.newArrayList(); + for (int i = 0; i < count; i++) { + records.add(new ThreeColumnRecord(i, "AAAA" + i, "A")); + } + + return records; + } + + private void appendRecords(Table table, List records) { + Dataset df = spark.createDataFrame(records, ThreeColumnRecord.class).coalesce(1); + df.select("c1", "c2", "c3").write().format("iceberg").mode("append").save(tableLocation); + table.refresh(); + } + + private List currentRows() { + return rowsToJava( + spark.read().format("iceberg").load(tableLocation).sort("c1", "c2", "c3").collectAsList()); + } + + /** Returns the snapshot id and sequence numbers of every live entry. */ + private List entryLineage() { + return spark + .read() + .format("iceberg") + .load(tableLocation + "#entries") + .filter("status < 2") + .selectExpr("snapshot_id", "sequence_number", "file_sequence_number", "data_file.file_path") + .collectAsList(); + } + + private DataFile onlyDataFile(Table table) throws IOException { + table.refresh(); + ManifestFile manifest = table.currentSnapshot().dataManifests(table.io()).get(0); + List files = readDataFiles(table, manifest); + assertThat(files).hasSize(1); + return files.get(0); + } + + private List readDataFiles(Table table, ManifestFile manifest) throws IOException { + List files = Lists.newArrayList(); + try (org.apache.iceberg.io.CloseableIterable reader = + ManifestFiles.read(manifest, table.io(), table.specs())) { + reader.forEach(file -> files.add(file.copy())); + } + + return files; + } + + private DeleteFile onlyDeleteFile(Table table) throws IOException { + table.refresh(); + List manifests = table.currentSnapshot().deleteManifests(table.io()); + assertThat(manifests).hasSize(1); + List files = readDeleteFiles(table, manifests.get(0)); + assertThat(files).hasSize(1); + return files.get(0); + } + + private List readDeleteFiles(Table table, ManifestFile manifest) throws IOException { + List files = Lists.newArrayList(); + try (org.apache.iceberg.io.CloseableIterable reader = + ManifestFiles.readDeleteManifest(manifest, table.io(), table.specs())) { + reader.forEach(file -> files.add(file.copy())); + } + + return files; + } + + private DeleteFile writeEqDeletes(Table table, String key, Object... values) throws IOException { + Schema deleteSchema = table.schema().select(key); + Record template = GenericRecord.create(deleteSchema); + List deletes = Lists.newArrayList(); + for (Object value : values) { + deletes.add(template.copy(key, value)); + } + + OutputFile output = + Files.localOutput(File.createTempFile("eq-deletes", ".parquet", temp.toFile())); + return FileHelpers.writeDeleteFile(table, output, null, deletes, deleteSchema); + } + + private DeleteFile writePosDeletes(Table table, List> deletes) + throws IOException { + OutputFile output = + Files.localOutput(File.createTempFile("pos-deletes", ".parquet", temp.toFile())); + return FileHelpers.writeDeleteFile(table, output, null, deletes, formatVersion).first(); + } + + private void replaceManifestWithCorruptStats(Table table, DataFile file) throws IOException { + ManifestFile manifest = table.currentSnapshot().dataManifests(table.io()).get(0); + corruptStats(table, manifest, file.location()); + } + + /** + * Corrupts the record count of the given file's entry while leaving its file size accurate. The + * table therefore stays readable by a scan even while the corruption is unrepaired, which lets a + * test that expects the repair to fail still read the table back afterwards. + */ + private void replaceManifestWithCorruptRecordCount(Table table, DataFile file) + throws IOException { + ManifestFile manifest = table.currentSnapshot().dataManifests(table.io()).get(0); + corruptStats(table, manifest, file.location(), true, false); + } + + /** + * Rewrites a manifest so that the entry of the given file records an incorrect record count, file + * size and column statistics, mimicking a writer that recorded them incorrectly. + * + *

Every other entry of the manifest is carried through unchanged, along with the lineage of + * all entries, so that the manifest differs from the original only in the statistics of one + * entry. + */ + private void corruptStats(Table table, ManifestFile manifest, String location) + throws IOException { + corruptStats(table, manifest, location, true); + } + + /** + * Rewrites a manifest, corrupting the statistics of the entry of the given file. When {@code + * corruptCounts} is false, only the column level statistics are dropped, leaving the record count + * and the file size correct. + */ + private void corruptStats( + Table table, ManifestFile manifest, String location, boolean corruptCounts) + throws IOException { + corruptStats(table, manifest, location, corruptCounts, corruptCounts); + } + + /** + * Rewrites a manifest, corrupting the statistics of the entry of the given file. The record count + * (along with the column statistics) and the file size are corrupted independently, so that a + * caller can leave the file size accurate and keep the file readable by a scan while its other + * statistics are wrong. + */ + private void corruptStats( + Table table, + ManifestFile manifest, + String location, + boolean corruptCounts, + boolean corruptSize) + throws IOException { + File manifestFile = File.createTempFile("corrupt-manifest", ".avro", temp.toFile()); + assertThat(manifestFile.delete()).isTrue(); + PartitionSpec spec = table.specs().get(manifest.partitionSpecId()); + + // the snapshot id is assigned during commit, so the manifest must be written without one + ManifestWriter writer = + ManifestFiles.write( + formatVersion, spec, table.io().newOutputFile(manifestFile.getCanonicalPath()), null); + + // read the lineage of each entry from the metadata table, it is not exposed by the reader + Map lineageByPath = Maps.newHashMap(); + for (Row row : entryLineage()) { + lineageByPath.put(row.getString(3), row); + } + + try { + for (DataFile file : readDataFiles(table, manifest)) { + DataFile toWrite = + file.location().equals(location) + ? corrupt(spec, file, corruptCounts, corruptSize) + : file.copy(); + Row lineage = lineageByPath.get(file.location()); + writer.existing( + toWrite, + lineage.getLong(0), + lineage.getLong(1), + lineage.isNullAt(2) ? null : lineage.getLong(2)); + } + } finally { + writer.close(); + } + + table.rewriteManifests().deleteManifest(manifest).addManifest(writer.toManifestFile()).commit(); + table.refresh(); + } + + private DataFile corrupt( + PartitionSpec spec, DataFile file, boolean corruptCounts, boolean corruptSize) { + DataFiles.Builder builder = + DataFiles.builder(spec) + .copy(file) + // drop the column level statistics, keeping the column sizes + .withMetrics( + new Metrics( + corruptCounts ? file.recordCount() + 100 : file.recordCount(), + file.columnSizes(), + Maps.newHashMap(), + Maps.newHashMap(), + Maps.newHashMap())); + + return builder + .withFileSizeInBytes(corruptSize ? file.fileSizeInBytes() + 4096 : file.fileSizeInBytes()) + .build(); + } +}