|
| 1 | +/* |
| 2 | + * Licensed to the Apache Software Foundation (ASF) under one |
| 3 | + * or more contributor license agreements. See the NOTICE file |
| 4 | + * distributed with this work for additional information |
| 5 | + * regarding copyright ownership. The ASF licenses this file |
| 6 | + * to you under the Apache License, Version 2.0 (the |
| 7 | + * "License"); you may not use this file except in compliance |
| 8 | + * with the License. You may obtain a copy of the License at |
| 9 | + * |
| 10 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | + * |
| 12 | + * Unless required by applicable law or agreed to in writing, |
| 13 | + * software distributed under the License is distributed on an |
| 14 | + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 15 | + * KIND, either express or implied. See the License for the |
| 16 | + * specific language governing permissions and limitations |
| 17 | + * under the License. |
| 18 | + */ |
| 19 | +package org.apache.iceberg.spark.actions; |
| 20 | + |
| 21 | +import static org.apache.spark.sql.functions.countDistinct; |
| 22 | +import static org.apache.spark.sql.functions.min; |
| 23 | + |
| 24 | +import java.util.List; |
| 25 | +import java.util.stream.Collectors; |
| 26 | +import org.apache.iceberg.DuplicateRegistrationRepair; |
| 27 | +import org.apache.iceberg.FileContent; |
| 28 | +import org.apache.iceberg.HasTableOperations; |
| 29 | +import org.apache.iceberg.MetadataTableType; |
| 30 | +import org.apache.iceberg.Table; |
| 31 | +import org.apache.iceberg.TableUtil; |
| 32 | +import org.apache.iceberg.exceptions.ValidationException; |
| 33 | +import org.apache.iceberg.spark.SparkTableUtil; |
| 34 | +import org.apache.spark.sql.Dataset; |
| 35 | +import org.apache.spark.sql.Row; |
| 36 | +import org.apache.spark.sql.SparkSession; |
| 37 | +import org.slf4j.Logger; |
| 38 | +import org.slf4j.LoggerFactory; |
| 39 | + |
| 40 | +/** |
| 41 | + * AFFIRM: shared duplicate-file-registration detection and repair, used by both {@link |
| 42 | + * RewriteDataFilesSparkAction} and {@link RewritePositionDeleteFilesSparkAction}. |
| 43 | + * |
| 44 | + * <p>Pulled out of {@code RewriteDataFilesSparkAction} per review feedback on Affirm/iceberg#6 |
| 45 | + * (pullrequestreview-5105776910): the underlying defect -- a file registered twice, at two data |
| 46 | + * sequence numbers, silently corrupting the next compaction because {@code DataFileSet}/{@code |
| 47 | + * DeleteFileSet} collapse it by location alone -- is reachable through any action that commits |
| 48 | + * through {@code ManifestFilterManager}, not just {@code rewrite_data_files}. {@code |
| 49 | + * rewrite_position_delete_files} is run standalone in production oncall/backfill notebooks with |
| 50 | + * no dependency on {@code rewrite_data_files} ever having run against the same table, so it needs |
| 51 | + * the identical guard rather than a second, drifting copy of this logic. |
| 52 | + */ |
| 53 | +final class DuplicateFileRegistrationGuard { |
| 54 | + |
| 55 | + private static final Logger LOG = LoggerFactory.getLogger(DuplicateFileRegistrationGuard.class); |
| 56 | + private static final int DUPLICATE_REGISTRATION_REPORT_LIMIT = 10; |
| 57 | + |
| 58 | + private DuplicateFileRegistrationGuard() {} |
| 59 | + |
| 60 | + /** |
| 61 | + * Detects live file paths (data or delete) registered at more than one data sequence number |
| 62 | + * and, depending on {@code resolve}, either repairs them or throws. |
| 63 | + * |
| 64 | + * @param spark the caller's Spark session |
| 65 | + * @param table the table being scanned |
| 66 | + * @param validate if false, skip the check entirely |
| 67 | + * @param resolve if true, repair a detected duplicate instead of failing; forced to a no-op |
| 68 | + * for any duplicate this guard cannot safely auto-repair (see {@link |
| 69 | + * #hasUnsafeDeletionVectorDuplicate}) |
| 70 | + * @param validatePropertyName the caller's validate-option name, used only in error messages |
| 71 | + * @param resolvePropertyName the caller's resolve-option name, used only in error messages |
| 72 | + */ |
| 73 | + static void validateOrRepair( |
| 74 | + SparkSession spark, |
| 75 | + Table table, |
| 76 | + boolean validate, |
| 77 | + boolean resolve, |
| 78 | + String validatePropertyName, |
| 79 | + String resolvePropertyName) { |
| 80 | + if (!validate) { |
| 81 | + return; |
| 82 | + } |
| 83 | + |
| 84 | + List<Row> duplicates = findDuplicateFileRegistrations(spark, table); |
| 85 | + if (duplicates.isEmpty()) { |
| 86 | + return; |
| 87 | + } |
| 88 | + |
| 89 | + boolean unsafeForAutoRepair = hasUnsafeDeletionVectorDuplicate(table, duplicates); |
| 90 | + if (resolve && !unsafeForAutoRepair) { |
| 91 | + repairDuplicateFileRegistrations(table, duplicates, resolvePropertyName); |
| 92 | + return; |
| 93 | + } |
| 94 | + |
| 95 | + throwDuplicateRegistrationError( |
| 96 | + table, duplicates, unsafeForAutoRepair, validatePropertyName, resolvePropertyName); |
| 97 | + } |
| 98 | + |
| 99 | + /** |
| 100 | + * AFFIRM: returns one row per live file path (data or delete) that is registered at more than |
| 101 | + * one distinct live data sequence number, with columns {@code file_path}, {@code content} (the |
| 102 | + * lowest live registration's content type -- content does not vary across duplicate |
| 103 | + * registrations of the same physical file), {@code keep_sequence_number} (the lowest live data |
| 104 | + * sequence number registered for that path), and {@code distinct_sequence_numbers} (how many |
| 105 | + * distinct sequence numbers that path is live at). Empty if the table has no such path. |
| 106 | + * |
| 107 | + * <p>Uses the {@code Dataset} groupBy/agg API (same shape as {@link |
| 108 | + * RemoveDanglingDeletesSparkAction#findDanglingDeletes}), not a temp view plus raw SQL -- |
| 109 | + * cheaper (no extra catalog round-trip) and compile-time column-safe. |
| 110 | + */ |
| 111 | + private static List<Row> findDuplicateFileRegistrations(SparkSession spark, Table table) { |
| 112 | + Dataset<Row> liveEntries = |
| 113 | + SparkTableUtil.loadMetadataTable(spark, table, MetadataTableType.ENTRIES) |
| 114 | + .filter("status < 2") // live entries only; status 2 is DELETED |
| 115 | + .selectExpr( |
| 116 | + "data_file.file_path as file_path", |
| 117 | + "data_file.content as content", |
| 118 | + "sequence_number") |
| 119 | + .distinct(); |
| 120 | + |
| 121 | + return liveEntries |
| 122 | + .groupBy("file_path") |
| 123 | + .agg( |
| 124 | + min("content").as("content"), |
| 125 | + min("sequence_number").as("keep_sequence_number"), |
| 126 | + countDistinct("sequence_number").as("distinct_sequence_numbers")) |
| 127 | + .filter("distinct_sequence_numbers > 1") |
| 128 | + .collectAsList(); |
| 129 | + } |
| 130 | + |
| 131 | + /** |
| 132 | + * AFFIRM: true if any duplicated path in {@code duplicates} is a delete file on a table that |
| 133 | + * may carry V3+ deletion vectors. |
| 134 | + * |
| 135 | + * <p>A V3 deletion vector packs multiple DISTINCT delete files (one per referenced data file) |
| 136 | + * into a single Puffin file that they all share as {@code location()} -- {@code |
| 137 | + * DeleteFileSet}'s equality key is {@code (location, contentOffset, contentSizeInBytes)} |
| 138 | + * specifically because of this; {@code DataFileSet} has no such concern and keys on {@code |
| 139 | + * location()} alone. This guard's matching ({@code ManifestFilterManager |
| 140 | + * #isDuplicateRegistrationToDrop}) is location-only, so on a V3+ table it cannot distinguish |
| 141 | + * those legitimately-distinct DVs sharing one Puffin path from an actual duplicate registration |
| 142 | + * of that whole path. Auto-repairing in that case risks dropping a DV that still legitimately |
| 143 | + * covers a different data file than the one actually duplicated -- reproducing the exact class |
| 144 | + * of defect this guard exists to prevent, through a path this guard cannot see. Refuse to |
| 145 | + * auto-repair the whole batch when this is possible; always require manual remediation for it, |
| 146 | + * regardless of the caller's resolve option. |
| 147 | + */ |
| 148 | + private static boolean hasUnsafeDeletionVectorDuplicate(Table table, List<Row> duplicates) { |
| 149 | + if (TableUtil.formatVersion(table) < 3) { |
| 150 | + return false; |
| 151 | + } |
| 152 | + |
| 153 | + return duplicates.stream() |
| 154 | + .anyMatch(row -> ((int) row.getAs("content")) != FileContent.DATA.id()); |
| 155 | + } |
| 156 | + |
| 157 | + /** |
| 158 | + * AFFIRM: repairs every duplicated file path found by {@link |
| 159 | + * #findDuplicateFileRegistrations(SparkSession, Table)} in a single commit, then refreshes |
| 160 | + * {@code table} so the caller plans its rewrite against the corrected metadata. |
| 161 | + * |
| 162 | + * <p>For each path, keeps the lowest live data sequence number and drops every other live |
| 163 | + * registration of that path. This is metadata-only: no physical file is touched, and the |
| 164 | + * underlying file is never at risk of being treated as orphaned, because the kept registration |
| 165 | + * still references its exact location. See {@link DuplicateRegistrationRepair} for why the |
| 166 | + * lowest sequence number, specifically, must be the one kept. |
| 167 | + */ |
| 168 | + private static void repairDuplicateFileRegistrations( |
| 169 | + Table table, List<Row> duplicates, String resolvePropertyName) { |
| 170 | + DuplicateRegistrationRepair repair = |
| 171 | + new DuplicateRegistrationRepair(table.name(), ((HasTableOperations) table).operations()); |
| 172 | + |
| 173 | + for (Row row : duplicates) { |
| 174 | + String path = row.getAs("file_path"); |
| 175 | + int content = row.getAs("content"); |
| 176 | + long keepSequenceNumber = row.getAs("keep_sequence_number"); |
| 177 | + if (content == FileContent.DATA.id()) { |
| 178 | + repair.keepDataFile(path, keepSequenceNumber); |
| 179 | + } else { |
| 180 | + // FileContent.POSITION_DELETES or FileContent.EQUALITY_DELETES |
| 181 | + repair.keepDeleteFile(path, keepSequenceNumber); |
| 182 | + } |
| 183 | + } |
| 184 | + |
| 185 | + repair.commit(); |
| 186 | + table.refresh(); |
| 187 | + |
| 188 | + LOG.warn( |
| 189 | + "Repaired {} duplicate file registration(s) in {} before rewriting: kept the lowest " |
| 190 | + + "live data sequence number for each path and dropped every other live " |
| 191 | + + "registration. This usually follows a commit retried after its outcome became " |
| 192 | + + "unknown. Set {}=false to require manual remediation instead of automatic repair.", |
| 193 | + duplicates.size(), |
| 194 | + table.name(), |
| 195 | + resolvePropertyName); |
| 196 | + } |
| 197 | + |
| 198 | + private static void throwDuplicateRegistrationError( |
| 199 | + Table table, |
| 200 | + List<Row> duplicates, |
| 201 | + boolean unsafeForAutoRepair, |
| 202 | + String validatePropertyName, |
| 203 | + String resolvePropertyName) { |
| 204 | + boolean truncated = duplicates.size() > DUPLICATE_REGISTRATION_REPORT_LIMIT; |
| 205 | + String sample = |
| 206 | + duplicates.stream() |
| 207 | + .limit(DUPLICATE_REGISTRATION_REPORT_LIMIT) |
| 208 | + .map( |
| 209 | + row -> { |
| 210 | + String path = row.getAs("file_path"); |
| 211 | + long distinctSequenceNumbers = row.getAs("distinct_sequence_numbers"); |
| 212 | + return String.format("%s (%d sequence numbers)", path, distinctSequenceNumbers); |
| 213 | + }) |
| 214 | + .collect(Collectors.joining(", ")); |
| 215 | + |
| 216 | + String remediation = |
| 217 | + unsafeForAutoRepair |
| 218 | + ? String.format( |
| 219 | + "At least one duplicated path is a delete file on a table that may carry V3+ " |
| 220 | + + "deletion vectors, where a single Puffin file legitimately holds multiple " |
| 221 | + + "distinct delete files at the same location and cannot safely auto-repair " |
| 222 | + + "using location-only matching -- refusing regardless of %s. Resolve the " |
| 223 | + + "duplicate registrations manually.", |
| 224 | + resolvePropertyName) |
| 225 | + : String.format( |
| 226 | + "Resolve the duplicate registrations first, set %s=true to repair " |
| 227 | + + "automatically, or set %s=false to proceed anyway and accept the risk.", |
| 228 | + resolvePropertyName, validatePropertyName); |
| 229 | + |
| 230 | + throw new ValidationException( |
| 231 | + "Cannot rewrite %s: %d file path(s) (data or delete) are registered at more than one " |
| 232 | + + "data sequence number. Compaction cannot handle that safely and would turn it into " |
| 233 | + + "duplicate rows. This usually follows a commit retried after its outcome became " |
| 234 | + + "unknown. Affected (showing up to %d): %s%s. %s", |
| 235 | + table.name(), |
| 236 | + duplicates.size(), |
| 237 | + DUPLICATE_REGISTRATION_REPORT_LIMIT, |
| 238 | + sample, |
| 239 | + truncated ? ", ..." : "", |
| 240 | + remediation); |
| 241 | + } |
| 242 | +} |
0 commit comments