amogh-jahagirdar commented on code in PR #15470:
URL: https://github.com/apache/iceberg/pull/15470#discussion_r3444893363
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -310,21 +317,29 @@ private Result rebuildMetadata() {
// rebuild manifest files
Set<ManifestFile> metaFiles = rewriteManifestListResult.toRewrite();
- RewriteContentFileResult rewriteManifestResult =
- rewriteManifests(deltaSnapshots, endMetadata, metaFiles);
- // rebuild position delete files
- Set<DeleteFile> deleteFiles =
- rewriteManifestResult.toRewrite().stream()
- .filter(e -> e instanceof DeleteFile)
- .map(e -> (DeleteFile) e)
- .collect(Collectors.toCollection(DeleteFileSet::create));
- rewritePositionDeletes(deleteFiles);
+ // Enumerate the distinct position delete files referenced by the delete
manifests being
+ // rewritten (metadata-only pass).
Review Comment:
I don't think the comment is particularly helpful, it's pretty apparent from
the code what's happening here, I'd remove it.
##########
spark/v4.1/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) {
Review Comment:
I would probably just have the caller pass in the Set<ManifestFile>
deleteManifests. I'm not sure why the original code kept calling this
"metaFiles" but that's not a good name for it.
##########
spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -310,21 +317,29 @@ private Result rebuildMetadata() {
// rebuild manifest files
Set<ManifestFile> metaFiles = rewriteManifestListResult.toRewrite();
- RewriteContentFileResult rewriteManifestResult =
- rewriteManifests(deltaSnapshots, endMetadata, metaFiles);
- // rebuild position delete files
- Set<DeleteFile> deleteFiles =
- rewriteManifestResult.toRewrite().stream()
- .filter(e -> e instanceof DeleteFile)
- .map(e -> (DeleteFile) e)
- .collect(Collectors.toCollection(DeleteFileSet::create));
- rewritePositionDeletes(deleteFiles);
+ // Enumerate the distinct position delete files referenced by the delete
manifests being
+ // rewritten (metadata-only pass).
+ Set<DeleteFile> deleteFilesToRewrite = positionDeletesToRewrite(metaFiles);
+
+ // Rewrite those delete files in parallel (deduped by path) and collect
the actual size of each
+ // rewritten file. The size is measured from the writer on the executor
that produced the file,
+ // avoiding both the end-of-job burst of getLength()/HEAD calls and the
file system races where
+ // an in-progress write underreports its length.
+ Map<String, Long> rewrittenDeleteFileSizes =
rewritePositionDeletes(deleteFilesToRewrite);
+
+ // Rewrite manifests (metadata-only), stamping file_size_in_bytes from the
measured sizes.
Review Comment:
Same as above, I'd just remove this comment.
##########
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.
Review Comment:
Not sure this comment is really necessary (especially the second statement
since that's apparent from the name)
##########
core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java:
##########
Review Comment:
Why don't we deprecate this one as well?
##########
core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java:
##########
@@ -617,6 +682,37 @@ public static void rewritePositionDeleteFile(
String targetPrefix,
PositionDeleteReaderWriter posDeleteReaderWriter)
throws IOException {
+ rewritePositionDeleteFileReturningLength(
+ deleteFile, outputFile, io, spec, sourcePrefix, targetPrefix,
posDeleteReaderWriter);
+ }
+
+ /**
+ * Rewrite a position delete file, replacing path references, and return the
size of the rewritten
+ * file.
+ *
+ * <p>The size is measured from the writer after it is closed (rather than
via a separate {@code
+ * getLength()}/HEAD call), so it is accurate even on file systems where the
length of an
+ * in-progress write underreports. Callers record this size as {@code
file_size_in_bytes} in the
+ * rewritten manifest.
+ *
+ * @param deleteFile source position delete file to rewrite
+ * @param outputFile output file to write the rewritten delete file to
+ * @param io file io
+ * @param spec spec of delete file
+ * @param sourcePrefix source prefix that will be replaced
+ * @param targetPrefix target prefix to replace it
+ * @param posDeleteReaderWriter class to read and write position delete files
+ * @return the size in bytes of the rewritten file
+ */
+ public static long rewritePositionDeleteFileReturningLength(
Review Comment:
If we deprecate the old one, I think we can just give this a shorter name
like rewritePositionDelete.
##########
spark/v3.5/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.
+ DeleteFileSet distinct = DeleteFileSet.create();
+ distinct.addAll(referencedDeleteFiles);
+ return distinct;
+ }
+
+ private static FlatMapFunction<ManifestFile, DeleteFile>
positionDeletesInManifest(
+ Broadcast<Table> tableArg) {
+ return manifestFile -> {
+ Table table = tableArg.getValue();
+ List<DeleteFile> deleteFiles = Lists.newArrayList();
+ try (ManifestReader<DeleteFile> reader =
+ ManifestFiles.readDeleteManifest(manifestFile, table.io(),
table.specs())) {
+ for (DeleteFile deleteFile : reader) {
+ if (deleteFile.content() == FileContent.POSITION_DELETES) {
+ deleteFiles.add(deleteFile.copy());
+ }
+ }
+ }
+ return deleteFiles.iterator();
+ };
+ }
+
+ /**
+ * Rewrite the given position delete files in parallel, returning a map from
each source delete
+ * file path to the size of its rewritten file.
+ */
+ private Map<String, Long> rewritePositionDeletes(Set<DeleteFile> toRewrite) {
if (toRewrite.isEmpty()) {
- return;
+ return Collections.emptyMap();
}
Encoder<DeleteFile> deleteFileEncoder =
Encoders.javaSerialization(DeleteFile.class);
- Dataset<DeleteFile> deleteFileDs =
+ Dataset<DeleteFile> deleteFileDS =
spark().createDataset(Lists.newArrayList(toRewrite),
deleteFileEncoder);
PositionDeleteReaderWriter posDeleteReaderWriter = new
SparkPositionDeleteReaderWriter();
- deleteFileDs
- .repartition(toRewrite.size())
- .foreach(
- rewritePositionDelete(
- tableBroadcast(), sourcePrefix, targetPrefix, stagingDir,
posDeleteReaderWriter));
+ List<Tuple2<String, Long>> rewrittenSizes =
+ deleteFileDS
+ .repartition(toRewrite.size())
+ .map(
+ rewritePositionDelete(
+ tableBroadcast(),
+ sourcePrefix,
+ targetPrefix,
+ stagingDir,
+ posDeleteReaderWriter),
+ Encoders.tuple(Encoders.STRING(), Encoders.LONG()))
+ .collectAsList();
+
+ Map<String, Long> sizesBySourcePath = Maps.newHashMap();
Review Comment:
Minor: I'd presize the map to the length of the rewrittenSizes.
##########
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:
Do we have a test which exercises the fallback to using
file.fileSizeInBytes() for the DELETED entries? I couldn't find one..
##########
core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java:
##########
@@ -377,7 +378,12 @@ public static RewriteResult<DataFile> rewriteDataManifest(
* @param stagingLocation staging location for rewritten files (referred
delete file will be
* rewritten here)
* @return a copy plan of content files in the manifest that was rewritten
+ * @deprecated since 1.11.0, will be removed in 1.12.0; use the overload
that accepts the map of
Review Comment:
I think you're going to have to change to 1.12.0 and 1.13.0 respectively now
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/actions/TestRewriteTablePathsAction.java:
##########
@@ -629,26 +632,222 @@ public void testPositionDeletesDeduplication() throws
Exception {
// in a new manifest, which will cause duplicate DeleteFile objects when
processing
tableWithPosDeletes.newRowDelta().addDeletes(positionDeletes).commit();
- // This should NOT throw AlreadyExistsException - the fix uses
DeleteFileSet to deduplicate
- // Without the fix (using Collectors.toSet()), this would fail because:
- // 1. Both manifests contain entries for the same delete file
- // 2. Processing returns two different DeleteFile objects for the same file
- // 3. HashSet doesn't deduplicate them (DeleteFile doesn't override
equals())
- // 4. rewritePositionDeletes tries to write the same file twice ->
AlreadyExistsException
+ // This should NOT throw AlreadyExistsException
RewriteTablePath.Result result =
actions()
.rewriteTablePath(tableWithPosDeletes)
.stagingLocation(stagingLocation())
.rewriteLocationPrefix(tableWithPosDeletes.location(),
targetTableLocation())
.execute();
- // Verify the rewrite completed successfully - should have rewritten
exactly 1 delete file
- // (the duplicate should be deduplicated by DeleteFileSet)
assertThat(result.rewrittenDeleteFilePathsCount())
.as("Should have rewritten exactly 1 delete file after deduplication")
.isEqualTo(1);
}
+ // Regression test: when the same position delete file is referenced from
manifests in different
Review Comment:
Since we have to change the other spark modules in this PR why not include
the tests for those as well?
##########
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:
Don't think the comment is neccessary here, also I think you can use
`DeleteFileSet.of(referencedDeleteFiles)` below.
--
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]
