Skip to content

Core, Spark: Fix invalid snapshot ID filtering in RewriteTablePath copy plan #16220

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
krisnaru wants to merge 1 commit into
base: main
Choose a base branch
from
+216 −48
Open

Core, Spark: Fix invalid snapshot ID filtering in RewriteTablePath copy plan #16220

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
100 changes: 84 additions & 16 deletions core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -332,7 +332,9 @@ private static List<ManifestFile> manifestFilesInSnapshot(FileIO io, Snapshot sn
* Rewrite a data manifest, replacing path references.
*
* @param manifestFile source manifest file to rewrite
* @param snapshotIds snapshot ids for filtering returned data manifest entries
* @param snapshotIds snapshot ids for filtering copy-plan entries; live entries whose adding
* snapshot is not in this set are written to the rewritten manifest but excluded from the
* copy plan
* @param outputFile output file to rewrite manifest file to
* @param io file io
* @param format format of the manifest file
Expand Down Expand Up @@ -364,11 +366,37 @@ public static RewriteResult<DataFile> rewriteDataManifest(
}
}

/**
* Rewrite a data manifest, replacing path references. All live entries are included in the copy
* plan regardless of the snapshot that added them.
*
* @param manifestFile source manifest file to rewrite
* @param outputFile output file to rewrite manifest file to
* @param io file io
* @param format format of the manifest file
* @param specsById map of partition specs by id
* @param sourcePrefix source prefix that will be replaced
* @param targetPrefix target prefix that will replace it
* @return a copy plan of content files in the manifest that was rewritten
*/
public static RewriteResult<DataFile> rewriteDataManifest(
ManifestFile manifestFile,
OutputFile outputFile,
FileIO io,
int format,
Map<Integer, PartitionSpec> specsById,
String sourcePrefix,
String targetPrefix)
throws IOException {
return rewriteDataManifest(
manifestFile, null, outputFile, io, format, specsById, sourcePrefix, targetPrefix);
}

/**
* Rewrite a delete manifest, replacing path references.
*
* @param manifestFile source delete manifest to rewrite
* @param snapshotIds snapshot ids for filtering returned delete manifest entries
* @param snapshotIds snapshot ids for filtering copy-plan entries
* @param outputFile output file to rewrite manifest file to
* @param io file io
* @param format format of the manifest file
Expand Down Expand Up @@ -417,7 +445,7 @@ public static RewriteResult<DeleteFile> rewriteDeleteManifest(
* file_size_in_bytes} stays consistent with the rewritten file on disk.
*
* @param manifestFile source delete manifest to rewrite
* @param snapshotIds snapshot ids for filtering returned delete manifest entries
* @param snapshotIds snapshot ids for filtering copy-plan entries
* @param outputFile output file to rewrite manifest file to
* @param io file io
* @param format format of the manifest file
Expand Down Expand Up @@ -463,6 +491,46 @@ public static RewriteResult<DeleteFile> rewriteDeleteManifest(
}
}

/**
* Rewrite a delete manifest, replacing path references. All live entries are included in the copy
* plan regardless of the snapshot that added them.
*
* @param manifestFile source delete manifest to rewrite
* @param outputFile output file to rewrite manifest file to
* @param io file io
* @param format format of the manifest file
* @param specsById map of partition specs by id
* @param sourcePrefix source prefix that will be replaced
* @param targetPrefix target prefix that will replace it
* @param stagingLocation staging location for rewritten position delete files
* @param rewrittenDeleteFileSizes map from source position delete file path to the actual size of
* the rewritten file; entries absent from the map keep their original size
* @return a copy plan of content files in the manifest that was rewritten
*/
public static RewriteResult<DeleteFile> rewriteDeleteManifest(
ManifestFile manifestFile,
OutputFile outputFile,
FileIO io,
int format,
Map<Integer, PartitionSpec> specsById,
String sourcePrefix,
String targetPrefix,
String stagingLocation,
Map<String, Long> rewrittenDeleteFileSizes)
throws IOException {
return rewriteDeleteManifest(
manifestFile,
null,
outputFile,
io,
format,
specsById,
sourcePrefix,
targetPrefix,
stagingLocation,
rewrittenDeleteFileSizes);
}

private static RewriteResult<DataFile> writeDataFileEntry(
ManifestEntry<DataFile> entry,
Set<Long> snapshotIds,
Expand All @@ -482,15 +550,22 @@ private static RewriteResult<DataFile> writeDataFileEntry(
DataFile newDataFile =
DataFiles.builder(spec).copy(entry.file()).withPath(targetDataFilePath).build();
appendEntryWithFile(entry, writer, newDataFile);
// keep the following entries in metadata but exclude them from copyPlan
// 1) deleted data files
// 2) entries not changed by snapshotIds
if (entry.isLive() && snapshotIds.contains(entry.snapshotId())) {
if (includeInCopyPlan(entry, snapshotIds)) {
result.copyPlan().add(Pair.of(sourceDataFilePath, newDataFile.location()));
}
return result;
}

/**
* Whether a manifest entry should be added to the copy plan. Every live entry is included; when
* {@code snapshotIds} is non-null, inclusion is further restricted to entries added by one of
* those snapshots (legacy filtering). A {@code null} {@code snapshotIds} means all live entries
* are included.
*/
private static boolean includeInCopyPlan(ManifestEntry<?> entry, Set<Long> snapshotIds) {
return entry.isLive() && (snapshotIds == null || snapshotIds.contains(entry.snapshotId()));
}

private static RewriteResult<DeleteFile> writeDeleteFileEntry(
ManifestEntry<DeleteFile> entry,
Set<Long> snapshotIds,
Expand All @@ -513,10 +588,7 @@ private static RewriteResult<DeleteFile> writeDeleteFileEntry(
DeleteFile posDeleteFile =
newPositionDeleteEntry(file, spec, sourcePrefix, targetPrefix, fileSizeInBytes);
appendEntryWithFile(entry, writer, posDeleteFile);
// keep the following entries in metadata but exclude them from copyPlan
// 1) deleted position delete files
// 2) entries not changed by snapshotIds
if (entry.isLive() && snapshotIds.contains(entry.snapshotId())) {
if (includeInCopyPlan(entry, snapshotIds)) {
result
.copyPlan()
.add(
Expand All @@ -529,11 +601,7 @@ private static RewriteResult<DeleteFile> writeDeleteFileEntry(
case EQUALITY_DELETES:
DeleteFile eqDeleteFile = newEqualityDeleteEntry(file, spec, sourcePrefix, targetPrefix);
appendEntryWithFile(entry, writer, eqDeleteFile);
// keep the following entries in metadata but exclude them from copyPlan
// 1) deleted equality delete files
// 2) entries not changed by snapshotIds
if (entry.isLive() && snapshotIds.contains(entry.snapshotId())) {
// No need to rewrite equality delete files as they do not contain absolute file paths.
if (includeInCopyPlan(entry, snapshotIds)) {
result.copyPlan().add(Pair.of(file.location(), eqDeleteFile.location()));
}
return result;
Expand Down
51 changes: 21 additions & 30 deletions ...4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -329,10 +329,25 @@ private Result rebuildMetadata() {
// rebuild manifest files
RewriteContentFileResult rewriteManifestResult =
rewriteManifests(
deltaSnapshots,
endMetadata,
manifestFiles,
sparkContext().broadcast(rewrittenDeleteFileSizes));
endMetadata, manifestFiles, sparkContext().broadcast(rewrittenDeleteFileSizes));

// For incremental copies, filter out files that were already copied in previous replication
if (startMetadata != null && !rewriteManifestResult.copyPlan().isEmpty()) {
try {
Table startTable = newStaticTable(startVersionName, table.io());
Set<String> previouslyCopiedPaths =
Sets.newHashSet(
contentFileDS(startTable).select("path").as(Encoders.STRING()).collectAsList());
rewriteManifestResult
.copyPlan()
.removeIf(pair -> previouslyCopiedPaths.contains(pair.first()));
} catch (Exception e) {
LOG.warn(
"Unable to read content files from start version for incremental filtering, "
+ "falling back to full copy plan.",
e);
}
}

ImmutableRewriteTablePath.Result.Builder builder =
ImmutableRewriteTablePath.Result.builder()
Expand Down Expand Up @@ -574,7 +589,6 @@ public RewriteContentFileResult appendDeleteFile(RewriteResult<DeleteFile> r1) {

/** Rewrite manifest files in a distributed manner and return rewritten data files path pairs. */
private RewriteContentFileResult rewriteManifests(
Set<Snapshot> deltaSnapshots,
TableMetadata tableMetadata,
Set<ManifestFile> toRewrite,
Broadcast<Map<String, Long>> rewrittenDeleteFileSizes) {
Expand All @@ -585,15 +599,12 @@ private RewriteContentFileResult rewriteManifests(
Encoder<ManifestFile> manifestFileEncoder = Encoders.javaSerialization(ManifestFile.class);
Dataset<ManifestFile> manifestDS =
spark().createDataset(Lists.newArrayList(toRewrite), manifestFileEncoder);
Set<Long> deltaSnapshotIds =
deltaSnapshots.stream().map(Snapshot::snapshotId).collect(Collectors.toSet());

return manifestDS
.repartition(toRewrite.size())
.map(
toManifests(
tableBroadcast(),
sparkContext().broadcast(deltaSnapshotIds),
stagingDir,
tableMetadata.formatVersion(),
sourcePrefix,
Expand All @@ -607,7 +618,6 @@ private RewriteContentFileResult rewriteManifests(

private static MapFunction<ManifestFile, RewriteContentFileResult> toManifests(
Broadcast<Table> table,
Broadcast<Set<Long>> deltaSnapshotIds,
String stagingLocation,
int format,
String sourcePrefix,
Expand All @@ -620,20 +630,13 @@ private static MapFunction<ManifestFile, RewriteContentFileResult> toManifests(
case DATA:
result.appendDataFile(
writeDataManifest(
manifestFile,
table,
deltaSnapshotIds,
stagingLocation,
format,
sourcePrefix,
targetPrefix));
manifestFile, table, stagingLocation, format, sourcePrefix, targetPrefix));
break;
case DELETES:
result.appendDeleteFile(
writeDeleteManifest(
manifestFile,
table,
deltaSnapshotIds,
stagingLocation,
format,
sourcePrefix,
Expand All @@ -651,7 +654,6 @@ private static MapFunction<ManifestFile, RewriteContentFileResult> toManifests(
private static RewriteResult<DataFile> writeDataManifest(
ManifestFile manifestFile,
Broadcast<Table> table,
Broadcast<Set<Long>> snapshotIds,
String stagingLocation,
int format,
String sourcePrefix,
Expand All @@ -662,16 +664,8 @@ private static RewriteResult<DataFile> writeDataManifest(
FileIO io = table.getValue().io();
OutputFile outputFile = io.newOutputFile(stagingPath);
Map<Integer, PartitionSpec> specsById = table.getValue().specs();
Set<Long> deltaSnapshotIds = snapshotIds.value();
return RewriteTablePathUtil.rewriteDataManifest(
manifestFile,
deltaSnapshotIds,
outputFile,
io,
format,
specsById,
sourcePrefix,
targetPrefix);
manifestFile, outputFile, io, format, specsById, sourcePrefix, targetPrefix);
} catch (IOException e) {
throw new RuntimeIOException(e);
}
Expand All @@ -680,7 +674,6 @@ private static RewriteResult<DataFile> writeDataManifest(
private static RewriteResult<DeleteFile> writeDeleteManifest(
ManifestFile manifestFile,
Broadcast<Table> table,
Broadcast<Set<Long>> snapshotIds,
String stagingLocation,
int format,
String sourcePrefix,
Expand All @@ -692,10 +685,8 @@ private static RewriteResult<DeleteFile> writeDeleteManifest(
FileIO io = table.getValue().io();
OutputFile outputFile = io.newOutputFile(stagingPath);
Map<Integer, PartitionSpec> specsById = table.getValue().specs();
Set<Long> deltaSnapshotIds = snapshotIds.value();
return RewriteTablePathUtil.rewriteDeleteManifest(
manifestFile,
deltaSnapshotIds,
outputFile,
io,
format,
Expand Down
Loading
Loading