kevinjqliu commented on code in PR #15470:
URL: https://github.com/apache/iceberg/pull/15470#discussion_r3446986819
##########
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) {
+ 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();
+ for (Tuple2<String, Long> entry : rewrittenSizes) {
+ sizesBySourcePath.put(entry._1(), entry._2());
+ }
+ return sizesBySourcePath;
+ }
+
+ private static MapFunction<DeleteFile, Tuple2<String, Long>>
rewritePositionDelete(
+ Broadcast<Table> tableArg,
+ String sourcePrefixArg,
+ String targetPrefixArg,
+ String stagingLocationArg,
+ PositionDeleteReaderWriter posDeleteReaderWriter) {
+ return deleteFile -> {
+ FileIO io = tableArg.getValue().io();
+ String newPath =
+ RewriteTablePathUtil.stagingPath(
+ deleteFile.location(), sourcePrefixArg, stagingLocationArg);
+ OutputFile outputFile = io.newOutputFile(newPath);
+ PartitionSpec spec =
tableArg.getValue().specs().get(deleteFile.specId());
+ long rewrittenLength =
+ RewriteTablePathUtil.rewritePositionDeleteFileReturningLength(
+ deleteFile,
+ outputFile,
+ io,
+ spec,
+ sourcePrefixArg,
+ targetPrefixArg,
+ posDeleteReaderWriter);
+ return new Tuple2<>(deleteFile.location(), rewrittenLength);
Review Comment:
For v3 tables, multiple DV `DeleteFile`s can point to the same Puffin file
while using different content offsets/sizes. In this action, the staged rewrite
path is derived only from `deleteFile.location()`, so those logical DVs can try
to rewrite the same physical Puffin file to the same staged path.
I verified this locally with a v3 table containing two DVs in one Puffin;
`rewrite_table_path` failed with `AlreadyExistsException`.
Can we dedupe the physical DV rewrite work by location, while still applying
the rewritten Puffin file size to each manifest entry?
##########
core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java:
##########
@@ -669,7 +769,7 @@ record = recordIt.next();
* @param sourcePrefix source prefix that will be replaced
* @param targetPrefix target prefix to replace it
*/
Review Comment:
nit: add return to java docs
```
* @return the size in bytes of the rewritten DV file
```
##########
spark/v4.1/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:
nit: on comments, i feel shorter comments are better.
a lot of of the additional context can be move to the function javadoc if
necessary.
i feel just this is enough
// rebuild position delete files
// rebuild manifest files
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
Review Comment:
```suggestion
* <ul>
* <li>Rebuild version files to staging
* <li>Rebuild manifest list files to staging
* <li>Rewrite referenced position delete files to staging
* <li>Rebuild manifests to staging
* <li>Get all files needed to move
* </ul>
```
update the docs
--
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]
