mbutrovich commented on code in PR #15470:
URL: https://github.com/apache/iceberg/pull/15470#discussion_r3453872520
##########
core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java:
##########
@@ -445,14 +498,21 @@ private static RewriteResult<DeleteFile>
writeDeleteFileEntry(
String sourcePrefix,
String targetPrefix,
String stagingLocation,
- ManifestWriter<DeleteFile> writer) {
+ ManifestWriter<DeleteFile> writer,
+ Map<String, Long> rewrittenDeleteFileSizes) {
DeleteFile file = entry.file();
RewriteResult<DeleteFile> result = new RewriteResult<>();
switch (file.content()) {
case POSITION_DELETES:
- DeleteFile posDeleteFile = newPositionDeleteEntry(file, spec,
sourcePrefix, targetPrefix);
+ // Rewriting the embedded data file paths changes the file size, so
record the actual size
+ // measured when the file was rewritten. Falls back to the original
size for entries whose
+ // file was not rewritten (e.g. deleted entries that are not copied to
the target).
+ long fileSizeInBytes =
+ rewrittenDeleteFileSizes.getOrDefault(file.location(),
file.fileSizeInBytes());
Review Comment:
Added `testRewriteDeleteManifestFallsBackToOriginalSizeForDeletedEntries` in
`TestRewriteTablePathUtil` (a manifest with one live and one DELETED entry, a
size map with only the live file; asserts the DELETED entry keeps its original
size). Did it at the core level since a DELETED entry's file is usually still
live in an earlier manifest, so a full copy rewrites it and the fallback branch
never runs.
##########
spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -682,27 +704,120 @@ private static RewriteResult<DeleteFile>
writeDeleteManifest(
specsById,
sourcePrefix,
targetPrefix,
- stagingLocation);
+ stagingLocation,
+ rewrittenDeleteFileSizes.value());
} catch (IOException e) {
throw new RuntimeIOException(e);
}
}
- private void rewritePositionDeletes(Set<DeleteFile> toRewrite) {
+ /**
+ * Enumerate the distinct position delete files referenced by the delete
manifests being
+ * rewritten. Equality delete files are excluded because they hold no
absolute paths and are not
+ * rewritten.
+ */
+ private Set<DeleteFile> positionDeletesToRewrite(Set<ManifestFile>
metaFiles) {
+ Set<ManifestFile> deleteManifests =
+ metaFiles.stream()
+ .filter(manifest -> manifest.content() == ManifestContent.DELETES)
+ .collect(Collectors.toSet());
+ if (deleteManifests.isEmpty()) {
+ return Collections.emptySet();
+ }
+
+ Encoder<ManifestFile> manifestFileEncoder =
Encoders.javaSerialization(ManifestFile.class);
+ Dataset<ManifestFile> manifestDS =
+ spark().createDataset(Lists.newArrayList(deleteManifests),
manifestFileEncoder);
+ Encoder<DeleteFile> deleteFileEncoder =
Encoders.javaSerialization(DeleteFile.class);
+
+ List<DeleteFile> referencedDeleteFiles =
+ manifestDS
+ .repartition(deleteManifests.size())
+ .flatMap(positionDeletesInManifest(tableBroadcast()),
deleteFileEncoder)
+ .collectAsList();
+
+ // DeleteFile does not override equals(); DeleteFileSet dedupes by path so
a delete file shared
+ // across manifests is rewritten only once.
Review Comment:
Done. Used `DeleteFileSet.of(referencedDeleteFiles)` for enumeration and
dropped the comment. The physical-rewrite dedup by `location()` is kept
separately so multi-DV Puffin files don't collide (kevin's @819).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
