From 50cc3c1158d4cbc083e7dc0bc3a3e99952208bd9 Mon Sep 17 00:00:00 2001 From: rahulsmahadev Date: Tue, 28 Jul 2026 20:56:03 +0000 Subject: [PATCH 1/6] Spark 4.1: Implement RepairTable action for entry stats Adds a Spark implementation of the RepairTable action, which repairs manifest entries whose statistics disagree with the files they refer to. The statistics of every live entry are compared against the file by reading its footer, and only the 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. Entries are rewritten with ManifestWriter#existing, carrying through the original snapshot id and data and file sequence numbers. This preserves the lineage of the files, and therefore which delete files apply to them, so the repair leaves the contents of the table unchanged. Repairing statistics does not change the number of live files, so the commit goes through the existing rewrite manifests validation unchanged. - RepairMetrics reads and compares statistics per format (Parquet, ORC and Avro) and rebuilds a file with corrected statistics. - Files whose statistics cannot be read are carried through unchanged and counted as incorrect but not repaired, which is what distinguishes entryStatsIncorrectCount from entryStatsRepairedCount. - The repair-column-metrics option skips the footer reads and repairs only record counts and file sizes, for tables where only those are suspect. - dryRun reports what would be repaired without committing. --- .../iceberg/actions/BaseRepairTable.java | 33 + .../iceberg/spark/actions/RepairMetrics.java | 165 ++++ .../spark/actions/RepairTableSparkAction.java | 747 ++++++++++++++++++ .../iceberg/spark/actions/SparkActions.java | 5 + .../spark/actions/TestRepairTableAction.java | 397 ++++++++++ 5 files changed, 1347 insertions(+) create mode 100644 core/src/main/java/org/apache/iceberg/actions/BaseRepairTable.java create mode 100644 spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RepairMetrics.java create mode 100644 spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RepairTableSparkAction.java create mode 100644 spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRepairTableAction.java 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..9f926e21c69c --- /dev/null +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RepairMetrics.java @@ -0,0 +1,165 @@ +/* + * 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 java.nio.ByteBuffer; +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(org.apache.iceberg.TableProperties.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(); + } else { + return FileMetadata.deleteFileBuilder(spec) + .copy((DeleteFile) file) + .withMetrics(metrics) + .withFileSizeInBytes(fileSizeInBytes) + .build(); + } + } + + 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..1884f83a9fe7 --- /dev/null +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RepairTableSparkAction.java @@ -0,0 +1,747 @@ +/* + * 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.Iterator; +import java.util.List; +import java.util.Set; +import java.util.UUID; +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.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.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.types.StructType; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * 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, which requires reading the footer of + * every file. When disabled, only record counts and file sizes are repaired. + */ + public static final String REPAIR_COLUMN_METRICS = "repair-column-metrics"; + + public static final boolean REPAIR_COLUMN_METRICS_DEFAULT = true; + + 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 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() { + // repairing entry stats is currently the only repair this action performs + 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() { + 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; + } + + if (dryRun) { + // the new manifests were written to determine what the repair would produce + deleteFiles(Iterables.transform(newManifests, ManifestFile::path)); + } else { + 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(); + } + + Dataset entryDF = buildManifestEntryDF(manifests); + + return withReusableDS( + entryDF, + df -> { + // find the entries whose stats disagree with the files they refer to + List verdicts = + df.mapPartitions(newCheckStatsFunc(content), Encoders.bean(EntryVerdict.class)) + .collectAsList(); + + if (verdicts.isEmpty()) { + return RepairedManifests.empty(); + } + + long repairedCount = verdicts.size(); + + Set manifestsToRewrite = + verdicts.stream().map(EntryVerdict::getManifest).collect(Collectors.toSet()); + List rewritten = + manifests.stream() + .filter(manifest -> manifestsToRewrite.contains(manifest.path())) + .collect(Collectors.toList()); + + Set repairedPaths = + verdicts.stream().map(EntryVerdict::getPath).collect(Collectors.toSet()); + + // rewrite every entry of the affected manifests, repairing the incorrect ones + Dataset entriesToRewrite = + df.filter(df.col("manifest").isin(manifestsToRewrite.toArray())); + List written = + writeManifests(content, entriesToRewrite, rewritten.size(), repairedPaths); + + return RepairedManifests.of(rewritten, written, repairedCount); + }); + } + + /** + * 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, Dataset entryDF, int numManifests, Set repairedPaths) { + StructType sparkType = (StructType) entryDF.schema().apply("data_file").dataType(); + Types.StructType combinedFileType = DataFile.getType(Partitioning.partitionType(table)); + ManifestWriterFactory writers = manifestWriters(); + Broadcast> repaired = sparkContext().broadcast(repairedPaths); + RepairContext context = newRepairContext(content); + + WriteManifests writeFunc = + content == ManifestContent.DATA + ? new WriteDataManifests(writers, combinedFileType, sparkType, repaired, context) + : new WriteDeleteManifests(writers, combinedFileType, sparkType, repaired, context); + + // preserve the entry order of the manifests being rewritten + return writeFunc.apply(entryDF.repartition(numManifests)).collectAsList(); + } + + private CheckStats newCheckStatsFunc(ManifestContent content) { + return new CheckStats(newRepairContext(content)); + } + + private RepairContext newRepairContext(ManifestContent content) { + boolean repairColumnMetrics = + PropertyUtil.propertyAsBoolean( + options(), REPAIR_COLUMN_METRICS, REPAIR_COLUMN_METRICS_DEFAULT); + return new RepairContext( + sparkContext().broadcast(SerializableTableWithSize.copyOf(table)), + content, + 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() { + return new ManifestWriterFactory( + sparkContext().broadcast(SerializableTableWithSize.copyOf(table)), + formatVersion, + table.spec().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, java.util.function.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 boolean repairColumnMetrics; + + private transient MetricsConfig lazyMetricsConfig = null; + private transient NameMapping lazyNameMapping = null; + private transient boolean nameMappingResolved = false; + + RepairContext( + Broadcast
tableBroadcast, ManifestContent content, boolean repairColumnMetrics) { + this.tableBroadcast = tableBroadcast; + this.content = content; + this.repairColumnMetrics = repairColumnMetrics; + } + + Table table() { + return tableBroadcast.value(); + } + + FileIO io() { + return table().io(); + } + + ManifestContent content() { + return content; + } + + boolean repairColumnMetrics() { + return repairColumnMetrics; + } + + PartitionSpec spec(int specId) { + return table().specs().get(specId); + } + + MetricsConfig metricsConfig(ContentFile file) { + if (lazyMetricsConfig == null) { + this.lazyMetricsConfig = RepairMetrics.metricsConfig(table(), file.content()); + } + + return lazyMetricsConfig; + } + + 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(table().spec().partitionType()); + return content == ManifestContent.DATA + ? new SparkDataFile(combinedFileType, fileType, sparkType) + : new SparkDeleteFile(combinedFileType, fileType, sparkType); + } + } + + /** A manifest entry whose statistics disagree with the file it refers to. */ + public static class EntryVerdict implements Serializable { + private String manifest; + private String path; + + public EntryVerdict() {} + + EntryVerdict(String manifest, String path) { + this.manifest = manifest; + this.path = path; + } + + public String getManifest() { + return manifest; + } + + public void setManifest(String manifest) { + this.manifest = manifest; + } + + public String getPath() { + return path; + } + + public void setPath(String path) { + this.path = path; + } + } + + /** Compares the statistics of every entry against the file the entry refers to. */ + 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(); + + while (rows.hasNext()) { + Row row = rows.next(); + String manifest = row.getString(0); + Row fileRow = row.getStruct(4); + StructType sparkType = (StructType) fileRow.schema(); + Types.StructType combinedFileType = + DataFile.getType(Partitioning.partitionType(context.table())); + ContentFile file = + (ContentFile) context.newFileWrapper(combinedFileType, sparkType).wrap(fileRow); + + if (!RepairMetrics.supportsMetrics(file)) { + continue; + } + + String location = file.location().toString(); + + try { + org.apache.iceberg.io.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 EntryVerdict(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, + StructType sparkFileType, + Broadcast> repairedPaths, + RepairContext context) { + super(writers, combinedFileType, sparkFileType, repairedPaths, 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, + StructType sparkFileType, + Broadcast> repairedPaths, + RepairContext context) { + super(writers, combinedFileType, sparkFileType, repairedPaths, 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 StructType sparkFileType; + private final Broadcast> repairedPaths; + private final RepairContext context; + + WriteManifests( + ManifestWriterFactory writers, + Types.StructType combinedFileType, + StructType sparkFileType, + Broadcast> repairedPaths, + RepairContext context) { + this.writers = writers; + this.combinedFileType = combinedFileType; + this.sparkFileType = sparkFileType; + this.repairedPaths = repairedPaths; + 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(); + Set repaired = repairedPaths.value(); + + 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); + + F file = fileWrapper.wrap(fileRow); + String location = file.location().toString(); + + if (repaired.contains(location)) { + 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) { + org.apache.iceberg.io.InputFile input = context.io().newInputFile(file.location()); + long fileSizeInBytes = input.getLength(); + Metrics metrics = + RepairMetrics.readMetrics( + input, file, context.metricsConfig(file), context.nameMapping()); + 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 DataFile.getType(context.table().spec().partitionType()); + } + + 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..026e697c5fd3 --- /dev/null +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRepairTableAction.java @@ -0,0 +1,397 @@ +/* + * 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 java.io.File; +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +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.hadoop.HadoopTables; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.spark.TestBase; +import org.apache.iceberg.spark.source.ThreeColumnRecord; +import org.apache.iceberg.types.Types; +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).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).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 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).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).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).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).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).execute(); + + assertThat(result.repairedEntryCount()).isEqualTo(1); + assertThat(currentRows()).containsExactlyInAnyOrderElementsOf(expectedRows); + } + + @TestTemplate + public void testRepairSkipsColumnMetricsWhenDisabled() 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) + .option(RepairTableSparkAction.REPAIR_COLUMN_METRICS, "false") + .execute(); + + assertThat(skipped.repairedEntryCount()) + .as("column metrics must not be compared when disabled") + .isEqualTo(0); + + RepairTable.Result repaired = SparkActions.get().repairTable(table).execute(); + + assertThat(repaired.repairedEntryCount()) + .as("column metrics must be compared by default") + .isEqualTo(1); + } + + 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 void replaceManifestWithCorruptStats(Table table, DataFile file) throws IOException { + ManifestFile manifest = table.currentSnapshot().dataManifests(table.io()).get(0); + corruptStats(table, manifest, file.location()); + } + + /** + * 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 { + 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) : 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) { + 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(corruptCounts ? file.fileSizeInBytes() + 4096 : file.fileSizeInBytes()) + .build(); + } +} From 2e8f0210b07fe56377770ce0ee9aa713b0ef1c17 Mon Sep 17 00:00:00 2001 From: rahulsmahadev Date: Tue, 28 Jul 2026 21:08:47 +0000 Subject: [PATCH 2/6] Spark 4.1: Add concurrency tests for RepairTable Covers what happens when the table changes underneath a repair, and what is left behind when the commit does not succeed: - a concurrent append commits between planning and the repair commit: the repair succeeds and the appended records survive, since the appended data lands in a new manifest and the manifests being repaired are still present - a concurrent operation replaces the very manifest being repaired: the commit fails validation in BaseRewriteManifests#validateDeletedManifests and the table is left untouched, rather than dropping the concurrent change - a failed commit deletes the manifests the repair wrote - a commit reported as CommitStateUnknownException keeps them, as the commit may have succeeded - a dry run leaves none of them behind Note the cleanup tests assert on the manifests written by the action itself. A failed commit on a format version 1 table can also leave behind a copy of a manifest made by the core staging path in BaseRewriteManifests, which is only cleaned up after a successful commit and is not owned by this action. --- .../spark/actions/TestRepairTableAction.java | 214 ++++++++++++++++++ 1 file changed, 214 insertions(+) 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 index 026e697c5fd3..d7bb4c260410 100644 --- 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 @@ -20,12 +20,18 @@ 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.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.DataFile; import org.apache.iceberg.DataFiles; @@ -42,9 +48,13 @@ import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; import org.apache.iceberg.actions.RepairTable; +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.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; @@ -262,6 +272,210 @@ public void testRepairSkipsColumnMetricsWhenDisabled() throws IOException { .isEqualTo(1); } + @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); + replaceManifestWithCorruptStats(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); + replaceManifestWithCorruptStats(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).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).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).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).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; + } + + 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)); From 49b5d65c20c0ecb0124ccf6713f7f42b8912fe01 Mon Sep 17 00:00:00 2001 From: rahulsmahadev Date: Tue, 18 Aug 2026 22:13:46 +0000 Subject: [PATCH 3/6] Spark 4.1: Repair manifests with the spec they were written under Manifests of the current snapshot were loaded regardless of their partition spec, but the writer was bound to the current default spec of the table. After partition evolution, rewriting a manifest of an older spec therefore wrote the current spec id and partition type to its entries. Group the manifests by partition spec id and repair each group with a writer bound to that spec, so an entry keeps the spec it was written under. Also pass the spec specific file type to the writer instead of deriving it from the current spec of the table. --- .../spark/actions/RepairTableSparkAction.java | 58 +++++++++++++++---- .../spark/actions/TestRepairTableAction.java | 37 ++++++++++++ 2 files changed, 84 insertions(+), 11 deletions(-) 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 index 1884f83a9fe7..a9c67b57e3ea 100644 --- 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 @@ -23,6 +23,7 @@ import java.io.Serializable; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.UUID; import java.util.stream.Collectors; @@ -207,6 +208,28 @@ private RepairedManifests repairTable(ManifestContent content, Snapshot snapshot 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( @@ -237,7 +260,7 @@ private RepairedManifests repairTable(ManifestContent content, Snapshot snapshot Dataset entriesToRewrite = df.filter(df.col("manifest").isin(manifestsToRewrite.toArray())); List written = - writeManifests(content, entriesToRewrite, rewritten.size(), repairedPaths); + writeManifests(content, specId, entriesToRewrite, rewritten.size(), repairedPaths); return RepairedManifests.of(rewritten, written, repairedCount); }); @@ -268,19 +291,27 @@ private Dataset buildManifestEntryDF(List manifests) { } private List writeManifests( - ManifestContent content, Dataset entryDF, int numManifests, Set repairedPaths) { + ManifestContent content, + int specId, + Dataset entryDF, + int numManifests, + Set repairedPaths) { StructType sparkType = (StructType) entryDF.schema().apply("data_file").dataType(); Types.StructType combinedFileType = DataFile.getType(Partitioning.partitionType(table)); - ManifestWriterFactory writers = manifestWriters(); + Types.StructType fileType = DataFile.getType(table.specs().get(specId).partitionType()); + ManifestWriterFactory writers = manifestWriters(specId); Broadcast> repaired = sparkContext().broadcast(repairedPaths); RepairContext context = newRepairContext(content); WriteManifests writeFunc = content == ManifestContent.DATA - ? new WriteDataManifests(writers, combinedFileType, sparkType, repaired, context) - : new WriteDeleteManifests(writers, combinedFileType, sparkType, repaired, context); + ? new WriteDataManifests( + writers, combinedFileType, fileType, sparkType, repaired, context) + : new WriteDeleteManifests( + writers, combinedFileType, fileType, sparkType, repaired, context); - // preserve the entry order of the manifests being rewritten + // write about as many manifests as are being replaced, so repairing does not change the + // manifest layout of the table return writeFunc.apply(entryDF.repartition(numManifests)).collectAsList(); } @@ -344,11 +375,11 @@ private void deleteFiles(Iterable locations) { } } - private ManifestWriterFactory manifestWriters() { + private ManifestWriterFactory manifestWriters(int specId) { return new ManifestWriterFactory( sparkContext().broadcast(SerializableTableWithSize.copyOf(table)), formatVersion, - table.spec().specId(), + specId, outputLocation, // allow the actual size of manifests to be 20% higher as the estimation is not precise (long) (1.2 * targetManifestSizeBytes)); @@ -553,10 +584,11 @@ private static class WriteDataManifests extends WriteManifests { WriteDataManifests( ManifestWriterFactory writers, Types.StructType combinedFileType, + Types.StructType fileType, StructType sparkFileType, Broadcast> repairedPaths, RepairContext context) { - super(writers, combinedFileType, sparkFileType, repairedPaths, context); + super(writers, combinedFileType, fileType, sparkFileType, repairedPaths, context); } @Override @@ -574,10 +606,11 @@ private static class WriteDeleteManifests extends WriteManifests { WriteDeleteManifests( ManifestWriterFactory writers, Types.StructType combinedFileType, + Types.StructType fileType, StructType sparkFileType, Broadcast> repairedPaths, RepairContext context) { - super(writers, combinedFileType, sparkFileType, repairedPaths, context); + super(writers, combinedFileType, fileType, sparkFileType, repairedPaths, context); } @Override @@ -607,6 +640,7 @@ private abstract static class WriteManifests> private final ManifestWriterFactory writers; private final Types.StructType combinedFileType; + private final Types.StructType fileType; private final StructType sparkFileType; private final Broadcast> repairedPaths; private final RepairContext context; @@ -614,11 +648,13 @@ private abstract static class WriteManifests> WriteManifests( ManifestWriterFactory writers, Types.StructType combinedFileType, + Types.StructType fileType, StructType sparkFileType, Broadcast> repairedPaths, RepairContext context) { this.writers = writers; this.combinedFileType = combinedFileType; + this.fileType = fileType; this.sparkFileType = sparkFileType; this.repairedPaths = repairedPaths; this.context = context; @@ -682,7 +718,7 @@ protected Types.StructType combinedFileType() { } protected Types.StructType fileType() { - return DataFile.getType(context.table().spec().partitionType()); + return fileType; } protected StructType sparkFileType() { 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 index d7bb4c260410..7d3fad5ab98e 100644 --- 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 @@ -467,6 +467,43 @@ private Set repairedManifestPaths() throws IOException { 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).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())) { From 96fcfa548965770a5072c9b9eaa8109fa861d641 Mon Sep 17 00:00:00 2001 From: rahulsmahadev Date: Wed, 2 Sep 2026 01:29:38 +0000 Subject: [PATCH 4/6] Spark 4.1: Correct delete-file repair and gate the RepairTable action Correctness: - Equality deletes now keep their equality field ids when rebuilt. FileMetadata.Builder.copy(DeleteFile) drops them, so a repaired entry otherwise kept content EQUALITY_DELETES with null equality ids, and reading the table failed once the delete was applied. - The metrics config is resolved per file content, so a delete manifest holding both position and equality deletes no longer compares equality deletes against the position-delete config and rewrites them needlessly. - The check-side file wrapper binds to the manifest's own spec rather than the table's current spec. - When column-metrics repair is disabled, only the record count and file size of a flagged entry are corrected; the stored column stats are kept instead of being replaced with the recomputed ones. Behavior: - repairFileMetrics() now selects the repair; execute() is a no-op when no repair has been selected, as the interface describes. - repair-column-metrics defaults to false. Recomputed column stats reflect the current metrics config, which the file may not have been written under, so repairing them can overwrite correct statistics. - A dry run no longer writes manifests only to delete them; it reports what would be repaired without writing anything. Also hoist the combined file type and file wrapper out of the per-row loop in CheckStats, and replace inline-qualified references with imports. --- .../iceberg/spark/actions/RepairMetrics.java | 43 ++++-- .../spark/actions/RepairTableSparkAction.java | 100 ++++++++++---- .../spark/actions/TestRepairTableAction.java | 130 +++++++++++++++--- 3 files changed, 216 insertions(+), 57 deletions(-) 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 index 9f926e21c69c..c1e803ef5040 100644 --- 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 @@ -18,6 +18,8 @@ */ package org.apache.iceberg.spark.actions; +import static org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING; + import java.nio.ByteBuffer; import java.util.Map; import java.util.Objects; @@ -53,8 +55,7 @@ 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(org.apache.iceberg.TableProperties.DEFAULT_NAME_MAPPING); + String mapping = table.properties().get(DEFAULT_NAME_MAPPING); return mapping != null ? NameMappingParser.fromJson(mapping) : null; } @@ -140,13 +141,39 @@ static ContentFile withStats( .withMetrics(metrics) .withFileSizeInBytes(fileSizeInBytes) .build(); - } else { - return FileMetadata.deleteFileBuilder(spec) - .copy((DeleteFile) file) - .withMetrics(metrics) - .withFileSizeInBytes(fileSizeInBytes) - .build(); } + + DeleteFile delete = (DeleteFile) file; + FileMetadata.Builder builder = + FileMetadata.deleteFileBuilder(spec) + .copy(delete) + .withMetrics(metrics) + .withFileSizeInBytes(fileSizeInBytes); + if (delete.content() == FileContent.EQUALITY_DELETES) { + // 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. + builder.ofEqualityDeletes( + delete.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) { 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 index a9c67b57e3ea..6ef9e0b6ee6e 100644 --- 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 @@ -21,16 +21,19 @@ 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.Set; 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; @@ -51,6 +54,7 @@ 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; @@ -90,12 +94,19 @@ public class RepairTableSparkAction extends BaseSnapshotUpdateSparkActionThis 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 = true; + public static final boolean REPAIR_COLUMN_METRICS_DEFAULT = false; private static final Logger LOG = LoggerFactory.getLogger(RepairTableSparkAction.class); @@ -113,6 +124,7 @@ public class RepairTableSparkAction extends BaseSnapshotUpdateSparkAction { // find the entries whose stats disagree with the files they refer to List verdicts = - df.mapPartitions(newCheckStatsFunc(content), Encoders.bean(EntryVerdict.class)) + df.mapPartitions( + newCheckStatsFunc(content, specId), Encoders.bean(EntryVerdict.class)) .collectAsList(); if (verdicts.isEmpty()) { @@ -253,6 +269,11 @@ private RepairedManifests repairManifests( .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); + } + Set repairedPaths = verdicts.stream().map(EntryVerdict::getPath).collect(Collectors.toSet()); @@ -301,7 +322,7 @@ private List writeManifests( Types.StructType fileType = DataFile.getType(table.specs().get(specId).partitionType()); ManifestWriterFactory writers = manifestWriters(specId); Broadcast> repaired = sparkContext().broadcast(repairedPaths); - RepairContext context = newRepairContext(content); + RepairContext context = newRepairContext(content, specId); WriteManifests writeFunc = content == ManifestContent.DATA @@ -315,17 +336,18 @@ private List writeManifests( return writeFunc.apply(entryDF.repartition(numManifests)).collectAsList(); } - private CheckStats newCheckStatsFunc(ManifestContent content) { - return new CheckStats(newRepairContext(content)); + private CheckStats newCheckStatsFunc(ManifestContent content, int specId) { + return new CheckStats(newRepairContext(content, specId)); } - private RepairContext newRepairContext(ManifestContent content) { + 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); } @@ -385,7 +407,7 @@ private ManifestWriterFactory manifestWriters(int specId) { (long) (1.2 * targetManifestSizeBytes)); } - private U withReusableDS(Dataset ds, java.util.function.Function, U> func) { + private U withReusableDS(Dataset ds, Function, U> func) { boolean useCaching = PropertyUtil.propertyAsBoolean(options(), USE_CACHING, USE_CACHING_DEFAULT); Dataset reusableDS = useCaching ? ds.cache() : ds; @@ -443,16 +465,21 @@ long repairedCount() { private static class RepairContext implements Serializable { private final Broadcast

tableBroadcast; private final ManifestContent content; + private final int specId; private final boolean repairColumnMetrics; - private transient MetricsConfig lazyMetricsConfig = null; + private transient Map lazyMetricsConfigs = null; private transient NameMapping lazyNameMapping = null; private transient boolean nameMappingResolved = false; RepairContext( - Broadcast
tableBroadcast, ManifestContent content, boolean repairColumnMetrics) { + Broadcast
tableBroadcast, + ManifestContent content, + int specId, + boolean repairColumnMetrics) { this.tableBroadcast = tableBroadcast; this.content = content; + this.specId = specId; this.repairColumnMetrics = repairColumnMetrics; } @@ -472,16 +499,22 @@ boolean repairColumnMetrics() { return repairColumnMetrics; } - PartitionSpec spec(int specId) { - return table().specs().get(specId); + 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 (lazyMetricsConfig == null) { - this.lazyMetricsConfig = RepairMetrics.metricsConfig(table(), file.content()); + if (lazyMetricsConfigs == null) { + this.lazyMetricsConfigs = new EnumMap<>(FileContent.class); } - return lazyMetricsConfig; + return lazyMetricsConfigs.computeIfAbsent( + file.content(), fileContent -> RepairMetrics.metricsConfig(table(), fileContent)); } NameMapping nameMapping() { @@ -494,7 +527,7 @@ NameMapping nameMapping() { } SparkContentFile newFileWrapper(Types.StructType combinedFileType, StructType sparkType) { - Types.StructType fileType = DataFile.getType(table().spec().partitionType()); + Types.StructType fileType = DataFile.getType(spec(specId).partitionType()); return content == ManifestContent.DATA ? new SparkDataFile(combinedFileType, fileType, sparkType) : new SparkDeleteFile(combinedFileType, fileType, sparkType); @@ -541,16 +574,20 @@ private static class CheckStats implements MapPartitionsFunction 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); - StructType sparkType = (StructType) fileRow.schema(); - Types.StructType combinedFileType = - DataFile.getType(Partitioning.partitionType(context.table())); - ContentFile file = - (ContentFile) context.newFileWrapper(combinedFileType, sparkType).wrap(fileRow); + if (fileWrapper == null) { + fileWrapper = context.newFileWrapper(combinedFileType, (StructType) fileRow.schema()); + } + + ContentFile file = (ContentFile) fileWrapper.wrap(fileRow); if (!RepairMetrics.supportsMetrics(file)) { continue; @@ -559,7 +596,7 @@ public Iterator call(Iterator rows) { String location = file.location().toString(); try { - org.apache.iceberg.io.InputFile input = context.io().newInputFile(location); + InputFile input = context.io().newInputFile(location); long fileSizeInBytes = input.getLength(); Metrics metrics = RepairMetrics.readMetrics( @@ -701,11 +738,16 @@ public Iterator call(Iterator rows) throws Exception { /** Rebuilds the file with the statistics read from the file itself. */ private ContentFile repairStats(ContentFile file) { - org.apache.iceberg.io.InputFile input = context.io().newInputFile(file.location()); + 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); } 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 index 7d3fad5ab98e..9e4a9cf9bfc3 100644 --- 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 @@ -33,8 +33,13 @@ 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.ManifestFile; import org.apache.iceberg.ManifestFiles; import org.apache.iceberg.ManifestWriter; @@ -96,7 +101,7 @@ public void setupTableLocation() { public void testRepairEmptyTable() { Table table = createTable(PartitionSpec.unpartitioned()); - RepairTable.Result result = SparkActions.get().repairTable(table).execute(); + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); assertThat(result.repairedManifests()).isEmpty(); assertThat(result.repairedEntryCount()).isEqualTo(0); @@ -109,7 +114,7 @@ public void testRepairTableWithCorrectStats() { Snapshot before = table.currentSnapshot(); - RepairTable.Result result = SparkActions.get().repairTable(table).execute(); + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); assertThat(result.repairedManifests()).isEmpty(); assertThat(result.repairedEntryCount()).isEqualTo(0); @@ -120,6 +125,33 @@ public void testRepairTableWithCorrectStats() { .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()); @@ -131,7 +163,7 @@ public void testRepairIncorrectRecordCountAndFileSize() throws IOException { // 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).execute(); + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); assertThat(result.repairedEntryCount()).isEqualTo(1); assertThat(result.repairedManifests()).hasSize(1); @@ -157,7 +189,7 @@ public void testRepairPreservesEntryLineage() throws IOException { replaceManifestWithCorruptStats(table, original); - SparkActions.get().repairTable(table).execute(); + SparkActions.get().repairTable(table).repairFileMetrics().execute(); table.refresh(); assertThat(entryLineage()) @@ -177,7 +209,8 @@ public void testDryRunDoesNotCommit() throws IOException { Snapshot before = table.currentSnapshot(); DataFile corrupt = onlyDataFile(table); - RepairTable.Result result = SparkActions.get().repairTable(table).dryRun().execute(); + RepairTable.Result result = + SparkActions.get().repairTable(table).repairFileMetrics().dryRun().execute(); assertThat(result.repairedEntryCount()).isEqualTo(1); assertThat(result.repairedManifests()).hasSize(1); @@ -207,7 +240,7 @@ public void testRepairOnlyRewritesAffectedManifests() throws IOException { DataFile fileToCorrupt = readDataFiles(table, manifests.get(0)).get(0); corruptStats(table, manifests.get(0), fileToCorrupt.location()); - RepairTable.Result result = SparkActions.get().repairTable(table).execute(); + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); assertThat(result.repairedManifests()).hasSize(1); assertThat(result.repairedEntryCount()).isEqualTo(1); @@ -238,14 +271,14 @@ public void testRepairPartitionedTable() throws IOException { corruptStats(table, manifest, files.get(0).location()); - RepairTable.Result result = SparkActions.get().repairTable(table).execute(); + RepairTable.Result result = SparkActions.get().repairTable(table).repairFileMetrics().execute(); assertThat(result.repairedEntryCount()).isEqualTo(1); assertThat(currentRows()).containsExactlyInAnyOrderElementsOf(expectedRows); } @TestTemplate - public void testRepairSkipsColumnMetricsWhenDisabled() throws IOException { + public void testRepairSkipsColumnMetricsByDefault() throws IOException { Table table = createTable(PartitionSpec.unpartitioned()); appendRecords(table, records(4)); @@ -256,22 +289,78 @@ public void testRepairSkipsColumnMetricsWhenDisabled() throws IOException { corruptStats(table, manifest, original.location(), false); RepairTable.Result skipped = - SparkActions.get() - .repairTable(table) - .option(RepairTableSparkAction.REPAIR_COLUMN_METRICS, "false") - .execute(); + SparkActions.get().repairTable(table).repairFileMetrics().execute(); assertThat(skipped.repairedEntryCount()) - .as("column metrics must not be compared when disabled") + .as("column metrics must not be compared by default") .isEqualTo(0); - RepairTable.Result repaired = SparkActions.get().repairTable(table).execute(); + RepairTable.Result repaired = + SparkActions.get() + .repairTable(table) + .repairFileMetrics() + .option(RepairTableSparkAction.REPAIR_COLUMN_METRICS, "true") + .execute(); assertThat(repaired.repairedEntryCount()) - .as("column metrics must be compared by default") + .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("the stored column stats must be kept, not replaced with recomputed ones") + .isEqualTo(corrupt.valueCounts()); + } + + @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 testRepairSucceedsWithConcurrentAppend() throws IOException { Table table = createTable(PartitionSpec.unpartitioned()); @@ -354,7 +443,7 @@ public void testRepairCleansUpManifestsOnCommitFailure() throws IOException { Table spyTable = spy(table); when(spyTable.rewriteManifests()).thenReturn(spyRewriteManifests); - assertThatThrownBy(() -> SparkActions.get().repairTable(spyTable).execute()) + assertThatThrownBy(() -> SparkActions.get().repairTable(spyTable).repairFileMetrics().execute()) .isInstanceOf(CommitFailedException.class) .hasMessage("Injected commit failure"); @@ -393,7 +482,7 @@ public void testRepairKeepsManifestsOnCommitStateUnknown() throws IOException { Table spyTable = spy(table); when(spyTable.rewriteManifests()).thenReturn(spyRewriteManifests); - assertThatThrownBy(() -> SparkActions.get().repairTable(spyTable).execute()) + assertThatThrownBy(() -> SparkActions.get().repairTable(spyTable).repairFileMetrics().execute()) .cause() .isInstanceOf(RuntimeException.class) .hasMessage("Datacenter on Fire"); @@ -420,7 +509,8 @@ public void testDryRunLeavesNoManifestsBehind() throws IOException { replaceManifestWithCorruptStats(table, original); table.refresh(); - RepairTable.Result result = SparkActions.get().repairTable(table).dryRun().execute(); + RepairTable.Result result = + SparkActions.get().repairTable(table).repairFileMetrics().dryRun().execute(); assertThat(result.repairedEntryCount()).isEqualTo(1); assertThat(repairedManifestPaths()) @@ -441,7 +531,7 @@ private RepairTable.Result repairWithConcurrentChange(Table table, Runnable chan return table.rewriteManifests(); }); - return SparkActions.get().repairTable(spyTable).execute(); + return SparkActions.get().repairTable(spyTable).repairFileMetrics().execute(); } /** @@ -489,7 +579,7 @@ public void testRepairAfterPartitionSpecEvolution() throws IOException { // corrupt the stats of the entry that still belongs to the original, unpartitioned spec corruptStats(table, oldManifest, original.location()); - SparkActions.get().repairTable(table).execute(); + SparkActions.get().repairTable(table).repairFileMetrics().execute(); table.refresh(); DataFile repaired = onlyDataFile(table); From 82d8ede38be48c5a90ae033f0f73614ad20800b1 Mon Sep 17 00:00:00 2001 From: rahulsmahadev Date: Wed, 2 Sep 2026 02:13:06 +0000 Subject: [PATCH 5/6] Spark 4.1: Join instead of broadcasting the incorrect file set, cover delete manifests The set of incorrect file paths was collected to the driver and broadcast back out. That set is unbounded, since a writer that recorded stats incorrectly usually did so for every file it wrote. Mark the entries to repair by joining the entries against the verdicts on the file path instead, so the set stays distributed. Only the set of manifests to rewrite, which is naturally small, is still collected. Verdicts are emitted as tuples rather than a bean and cached, as they are read more than once. Repartition the entries by manifest when writing so the manifest layout of the table is preserved, rather than scattered round robin by a plain repartition(n). Guard the equality field ids in RepairMetrics.withStats: an entry that is an equality delete but records no equality ids is carried through as is rather than throwing, so a malformed entry does not abort the whole repair. Add end-to-end coverage of the delete manifest path, which was previously unexercised: repair a position delete, an equality delete, and a manifest holding both, asserting the statistics are corrected and the equality field ids survive. --- .../iceberg/spark/actions/RepairMetrics.java | 11 +- .../spark/actions/RepairTableSparkAction.java | 179 ++++++++-------- .../spark/actions/TestRepairTableAction.java | 202 +++++++++++++++++- 3 files changed, 295 insertions(+), 97 deletions(-) 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 index c1e803ef5040..1d5f1c88bcc8 100644 --- 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 @@ -21,6 +21,7 @@ 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; @@ -149,12 +150,14 @@ static ContentFile withStats( .copy(delete) .withMetrics(metrics) .withFileSizeInBytes(fileSizeInBytes); - if (delete.content() == FileContent.EQUALITY_DELETES) { + 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. - builder.ofEqualityDeletes( - delete.equalityFieldIds().stream().mapToInt(Integer::intValue).toArray()); + // 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(); 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 index 6ef9e0b6ee6e..43c676b042fb 100644 --- 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 @@ -25,7 +25,6 @@ import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.Set; import java.util.UUID; import java.util.function.Function; import java.util.stream.Collectors; @@ -76,9 +75,11 @@ 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. @@ -250,43 +251,74 @@ private RepairedManifests repairManifests( return withReusableDS( entryDF, df -> { - // find the entries whose stats disagree with the files they refer to - List verdicts = + // 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.bean(EntryVerdict.class)) - .collectAsList(); - - if (verdicts.isEmpty()) { - return RepairedManifests.empty(); + 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); } - - long repairedCount = verdicts.size(); - - Set manifestsToRewrite = - verdicts.stream().map(EntryVerdict::getManifest).collect(Collectors.toSet()); - 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); - } - - Set repairedPaths = - verdicts.stream().map(EntryVerdict::getPath).collect(Collectors.toSet()); - - // rewrite every entry of the affected manifests, repairing the incorrect ones - Dataset entriesToRewrite = - df.filter(df.col("manifest").isin(manifestsToRewrite.toArray())); - List written = - writeManifests(content, specId, entriesToRewrite, rewritten.size(), repairedPaths); - - return RepairedManifests.of(rewritten, written, repairedCount); }); } + /** + * 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. @@ -312,28 +344,24 @@ private Dataset buildManifestEntryDF(List manifests) { } private List writeManifests( - ManifestContent content, - int specId, - Dataset entryDF, - int numManifests, - Set repairedPaths) { + 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); - Broadcast> repaired = sparkContext().broadcast(repairedPaths); RepairContext context = newRepairContext(content, specId); WriteManifests writeFunc = content == ManifestContent.DATA - ? new WriteDataManifests( - writers, combinedFileType, fileType, sparkType, repaired, context) - : new WriteDeleteManifests( - writers, combinedFileType, fileType, sparkType, repaired, context); - - // write about as many manifests as are being replaced, so repairing does not change the - // manifest layout of the table - return writeFunc.apply(entryDF.repartition(numManifests)).collectAsList(); + ? 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) { @@ -534,37 +562,11 @@ SparkContentFile newFileWrapper(Types.StructType combinedFileType, StructType } } - /** A manifest entry whose statistics disagree with the file it refers to. */ - public static class EntryVerdict implements Serializable { - private String manifest; - private String path; - - public EntryVerdict() {} - - EntryVerdict(String manifest, String path) { - this.manifest = manifest; - this.path = path; - } - - public String getManifest() { - return manifest; - } - - public void setManifest(String manifest) { - this.manifest = manifest; - } - - public String getPath() { - return path; - } - - public void setPath(String path) { - this.path = path; - } - } - - /** Compares the statistics of every entry against the file the entry refers to. */ - private static class CheckStats implements MapPartitionsFunction { + /** + * 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) { @@ -572,8 +574,8 @@ private static class CheckStats implements MapPartitionsFunction call(Iterator rows) { - List verdicts = Lists.newArrayList(); + 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())); @@ -604,7 +606,7 @@ public Iterator call(Iterator rows) { if (RepairMetrics.statsAreIncorrect( file, metrics, fileSizeInBytes, context.repairColumnMetrics())) { - verdicts.add(new EntryVerdict(manifest, location)); + 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 @@ -623,9 +625,8 @@ private static class WriteDataManifests extends WriteManifests { Types.StructType combinedFileType, Types.StructType fileType, StructType sparkFileType, - Broadcast> repairedPaths, RepairContext context) { - super(writers, combinedFileType, fileType, sparkFileType, repairedPaths, context); + super(writers, combinedFileType, fileType, sparkFileType, context); } @Override @@ -645,9 +646,8 @@ private static class WriteDeleteManifests extends WriteManifests { Types.StructType combinedFileType, Types.StructType fileType, StructType sparkFileType, - Broadcast> repairedPaths, RepairContext context) { - super(writers, combinedFileType, fileType, sparkFileType, repairedPaths, context); + super(writers, combinedFileType, fileType, sparkFileType, context); } @Override @@ -679,7 +679,6 @@ private abstract static class WriteManifests> private final Types.StructType combinedFileType; private final Types.StructType fileType; private final StructType sparkFileType; - private final Broadcast> repairedPaths; private final RepairContext context; WriteManifests( @@ -687,13 +686,11 @@ private abstract static class WriteManifests> Types.StructType combinedFileType, Types.StructType fileType, StructType sparkFileType, - Broadcast> repairedPaths, RepairContext context) { this.writers = writers; this.combinedFileType = combinedFileType; this.fileType = fileType; this.sparkFileType = sparkFileType; - this.repairedPaths = repairedPaths; this.context = context; } @@ -710,7 +707,6 @@ public Dataset apply(Dataset input) { public Iterator call(Iterator rows) throws Exception { SparkContentFile fileWrapper = newFileWrapper(); RollingManifestWriter writer = newManifestWriter(); - Set repaired = repairedPaths.value(); try { while (rows.hasNext()) { @@ -719,11 +715,10 @@ public Iterator call(Iterator rows) throws Exception { 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); - String location = file.location().toString(); - - if (repaired.contains(location)) { + if (repair) { file = (F) repairStats(file); } 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 index 9e4a9cf9bfc3..ff7cfde96bd2 100644 --- 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 @@ -21,6 +21,7 @@ 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; @@ -40,6 +41,7 @@ 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; @@ -53,16 +55,21 @@ 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; @@ -334,8 +341,14 @@ public void testRepairPreservesColumnStatsWhenColumnMetricsDisabled() throws IOE .as("the file size must be repaired") .isEqualTo(original.fileSizeInBytes()); assertThat(repaired.valueCounts()) - .as("the stored column stats must be kept, not replaced with recomputed ones") + .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 @@ -361,6 +374,154 @@ public void testWithStatsPreservesEqualityFieldIds() { .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()); @@ -658,6 +819,45 @@ private List readDataFiles(Table table, ManifestFile manifest) throws 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()); From aaf6b9cbc6d8d57ae87c39b80f49b7f7bd3b079f Mon Sep 17 00:00:00 2001 From: rahulsmahadev Date: Wed, 2 Sep 2026 21:58:08 +0000 Subject: [PATCH 6/6] Spark 4.1: Keep file size accurate in repair failure-path tests The two tests asserting that a failed repair leaves the table unchanged scan the table while the corruption is still present. Parquet now fetches small files eagerly using the recorded file size, so an inflated file_size_in_bytes makes the scan read past the end of the file. Corrupt only the record count in these tests, keeping the file size accurate, so the table stays readable. --- .../spark/actions/TestRepairTableAction.java | 40 ++++++++++++++++--- 1 file changed, 35 insertions(+), 5 deletions(-) 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 index ff7cfde96bd2..a04af5c2ca22 100644 --- 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 @@ -556,7 +556,7 @@ public void testRepairFailsWhenRepairedManifestIsConcurrentlyReplaced() throws I appendRecords(table, records(4)); DataFile original = onlyDataFile(table); - replaceManifestWithCorruptStats(table, original); + replaceManifestWithCorruptRecordCount(table, original); table.refresh(); List rowsBeforeRepair = currentRows(); @@ -589,7 +589,7 @@ public void testRepairCleansUpManifestsOnCommitFailure() throws IOException { appendRecords(table, records(4)); DataFile original = onlyDataFile(table); - replaceManifestWithCorruptStats(table, original); + replaceManifestWithCorruptRecordCount(table, original); table.refresh(); List rowsBeforeRepair = currentRows(); @@ -863,6 +863,17 @@ private void replaceManifestWithCorruptStats(Table table, DataFile file) throws 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. @@ -884,6 +895,22 @@ private void corruptStats(Table table, ManifestFile manifest, String location) 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()); @@ -902,7 +929,9 @@ private void corruptStats( try { for (DataFile file : readDataFiles(table, manifest)) { DataFile toWrite = - file.location().equals(location) ? corrupt(spec, file, corruptCounts) : file.copy(); + file.location().equals(location) + ? corrupt(spec, file, corruptCounts, corruptSize) + : file.copy(); Row lineage = lineageByPath.get(file.location()); writer.existing( toWrite, @@ -918,7 +947,8 @@ private void corruptStats( table.refresh(); } - private DataFile corrupt(PartitionSpec spec, DataFile file, boolean corruptCounts) { + private DataFile corrupt( + PartitionSpec spec, DataFile file, boolean corruptCounts, boolean corruptSize) { DataFiles.Builder builder = DataFiles.builder(spec) .copy(file) @@ -932,7 +962,7 @@ private DataFile corrupt(PartitionSpec spec, DataFile file, boolean corruptCount Maps.newHashMap())); return builder - .withFileSizeInBytes(corruptCounts ? file.fileSizeInBytes() + 4096 : file.fileSizeInBytes()) + .withFileSizeInBytes(corruptSize ? file.fileSizeInBytes() + 4096 : file.fileSizeInBytes()) .build(); } }