varun-lakhyani commented on code in PR #16910:
URL: https://github.com/apache/iceberg/pull/16910#discussion_r3789976357
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -494,12 +523,18 @@ private Set<Pair<String, String>>
partitionStatsFileCopyPlan(
*
* @param snapshot snapshot represented by the manifest list
* @param tableMetadata metadata of table
+ * @param manifestFiles the manifests the snapshot's manifest list references
* @param manifestsToRewrite filter of manifests to rewrite.
+ * @param rewrittenManifestLengths map from source manifest path to its
rewritten byte length
Review Comment:
I would make these three same as rewriteManifestList in Util.
https://github.com/apache/iceberg/pull/16910/changes#diff-285b4520a0f058b7fe2a0103efe3f15c0021a6acd9031cdc22c3ddae6151b467R299-R307
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -302,21 +302,31 @@ private Result rebuildMetadata() {
Set<Snapshot> validSnapshots =
Sets.difference(snapshotSet(endMetadata), snapshotSet(startMetadata));
- // rebuild manifest-list files
- Set<RewriteResult<ManifestFile>> manifestListResults =
Sets.newConcurrentHashSet();
+ // Read every valid snapshot's manifest list once, before writing
anything. The rewritten
+ // manifest list must record each referenced manifest's rewritten length
(manifest_length),
+ // which is only known after the manifests are rewritten below.
+ Map<Snapshot, List<ManifestFile>> manifestsBySnapshot =
Maps.newConcurrentMap();
Tasks.foreach(validSnapshots)
.noRetry()
.throwFailureWhenFinished()
.executeWith(executorService)
.run(
snapshot ->
- manifestListResults.add(
- rewriteManifestList(snapshot, endMetadata,
manifestsToRewrite)));
-
- RewriteResult<ManifestFile> rewriteManifestListResult = new
RewriteResult<>();
- manifestListResults.forEach(rewriteManifestListResult::append);
-
- Set<ManifestFile> manifestFiles = rewriteManifestListResult.toRewrite();
+ manifestsBySnapshot.put(
+ snapshot,
+ RewriteTablePathUtil.manifestsInSnapshot(snapshot,
table.io(), sourcePrefix)));
+
+ // Manifests selected for rewrite. In an incremental run this is only the
manifests added by the
+ // delta snapshots, so a manifest carried over from an earlier run is not
rewritten here and
+ // keeps its source length in the manifest list. See the note on
rewriteManifestList.
+ Set<ManifestFile> manifestFiles = Sets.newHashSet();
Review Comment:
nit maybe code is self explaining here, if required this can be made one
liner
##########
core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java:
##########
@@ -351,17 +431,64 @@ public static RewriteResult<DataFile> rewriteDataManifest(
String sourcePrefix,
String targetPrefix)
throws IOException {
+ return rewriteDataManifestAndMeasureLength(
+ manifestFile,
+ snapshotIds,
+ outputFile,
+ io,
+ format,
+ specsById,
+ sourcePrefix,
+ targetPrefix)
+ .first();
+ }
+
+ /**
+ * Rewrite a data manifest, replacing path references, and return the
rewritten manifest's byte
+ * length.
+ *
+ * <p>The length is read from the closed manifest writer rather than via a
separate {@code
+ * getLength()} call, which would cost a stat request per manifest against
object storage. Callers
+ * record this length as the {@code manifest_length} of the rewritten
manifest in the manifest
+ * list.
+ *
+ * @param manifestFile source manifest file to rewrite
+ * @param snapshotIds snapshot ids for filtering returned data manifest
entries
+ * @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 the copy plan of content files in the rewritten manifest, paired
with the rewritten
+ * manifest's byte length
+ */
+ public static Pair<RewriteResult<DataFile>, Long>
rewriteDataManifestAndMeasureLength(
+ ManifestFile manifestFile,
+ Set<Long> snapshotIds,
+ OutputFile outputFile,
+ FileIO io,
+ int format,
+ Map<Integer, PartitionSpec> specsById,
+ String sourcePrefix,
+ String targetPrefix)
+ throws IOException {
PartitionSpec spec = specsById.get(manifestFile.partitionSpecId());
- try (ManifestWriter<DataFile> writer =
- ManifestFiles.write(format, spec, outputFile,
manifestFile.snapshotId());
+ ManifestWriter<DataFile> writer =
+ ManifestFiles.write(format, spec, outputFile,
manifestFile.snapshotId());
+ RewriteResult<DataFile> result;
+ try (writer;
ManifestReader<DataFile> reader =
ManifestFiles.read(manifestFile, io,
specsById).select(Arrays.asList("*"))) {
- return StreamSupport.stream(reader.entries().spliterator(), false)
- .map(
- entry ->
- writeDataFileEntry(entry, snapshotIds, spec, sourcePrefix,
targetPrefix, writer))
- .reduce(new RewriteResult<>(), RewriteResult::append);
+ result =
+ StreamSupport.stream(reader.entries().spliterator(), false)
+ .map(
+ entry ->
+ writeDataFileEntry(
+ entry, snapshotIds, spec, sourcePrefix,
targetPrefix, writer))
+ .reduce(new RewriteResult<>(), RewriteResult::append);
}
+ return Pair.of(result, writer.length());
Review Comment:
Just wondering - did you consider carrying the length on RewriteResult
instead of returning a Pair? it would merge in append the same way
RewriteContentFileResult does, and rewriteDataManifest / rewriteDeleteManifest
could keep their names and signatures — no new methods or deprecations.
Might not be worth at this point, just wondering if it was considered.
##########
core/src/test/java/org/apache/iceberg/TestRewriteTablePathUtil.java:
##########
@@ -376,4 +385,70 @@ private ManifestFile
deleteManifestWithLiveAndDeletedEntry(DeleteFile live, Dele
return writer.toManifestFile();
}
+
+ /**
+ * One manifest is measured and one is not, in a single manifest list. The
measured one must take
+ * the mapped length and the unmeasured one must keep its source length,
which pins both the
+ * stamping and the fact that the map is keyed by the source path, not the
rewritten one.
+ */
+ @TestTemplate
+ public void testRewriteManifestListStampsMeasuredManifestLengths() throws
IOException {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ table.newFastAppend().appendFile(FILE_B).commit();
+ Snapshot snapshot = table.currentSnapshot();
+ List<ManifestFile> manifests = snapshot.allManifests(table.io());
+ assertThat(manifests).hasSize(2);
+ ManifestFile measured = manifests.get(0);
+ ManifestFile unmeasured = manifests.get(1);
+
+ String manifestPath = measured.path();
+ String sourcePrefix = manifestPath.substring(0,
manifestPath.indexOf("/metadata/"));
+ String targetPrefix = sourcePrefix + "/relocated";
+ String stagingDir = temp.resolve("staging").toString();
+ String outputPath = temp.resolve("rewritten-list-" + System.nanoTime() +
".avro").toString();
+
+ // a value distinct from the source length so we can tell it was applied
+ long rewrittenLength = measured.length() + 4242L;
+
+ RewriteTablePathUtil.rewriteManifestList(
+ snapshot,
+ table.io(),
+ table.ops().current(),
+ manifests,
+ Set.of(measured.path(), unmeasured.path()),
+ sourcePrefix,
+ targetPrefix,
+ stagingDir,
+ outputPath,
+ Map.of(measured.path(), rewrittenLength));
Review Comment:
nit
ImmutableMap just to be consistent?
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -334,6 +344,25 @@ private Result rebuildMetadata() {
manifestFiles,
sparkContext().broadcast(rewrittenDeleteFileSizes));
+ // rebuild manifest-list files last, stamping manifest_length with the
rewritten manifest sizes
Review Comment:
nit
similar maybe just keep rebuild manifest-list files,
updated manifest file size is the base functionality we might not need to
specify separately
##########
spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/actions/RewriteTablePathSparkAction.java:
##########
@@ -302,21 +302,31 @@ private Result rebuildMetadata() {
Set<Snapshot> validSnapshots =
Sets.difference(snapshotSet(endMetadata), snapshotSet(startMetadata));
- // rebuild manifest-list files
- Set<RewriteResult<ManifestFile>> manifestListResults =
Sets.newConcurrentHashSet();
+ // Read every valid snapshot's manifest list once, before writing
anything. The rewritten
+ // manifest list must record each referenced manifest's rewritten length
(manifest_length),
+ // which is only known after the manifests are rewritten below.
+ Map<Snapshot, List<ManifestFile>> manifestsBySnapshot =
Maps.newConcurrentMap();
Review Comment:
nit. maybe just Read every valid snapshot's manifest list works.
i think the reason for reordering doesn't belong here
##########
core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java:
##########
@@ -251,38 +251,77 @@ private static List<Snapshot> updatePathInSnapshots(
/**
* Rewrite a manifest list representing a snapshot, replacing path
references.
*
+ * <p>Every entry keeps its source {@code manifest_length}, which does not
match the rewritten
+ * manifest when the target prefix differs in length from the source.
Callers that rewrite to a
+ * different-length prefix should use {@link #rewriteManifestList(Snapshot,
FileIO, TableMetadata,
+ * List, Set, String, String, String, String, Map)} and pass the measured
lengths.
+ */
+ public static RewriteResult<ManifestFile> rewriteManifestList(
Review Comment:
Can we make it like `@Deprecated` and `@deprecated` since 1.12.0, will be
removed in 1.13.0; use the overload....
The stale-size rewriteDeleteManifest in the same file is `@Deprecated` for
the same reason (#15470).
##########
core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java:
##########
@@ -301,6 +340,15 @@ public static RewriteResult<ManifestFile>
rewriteManifestList(
for (ManifestFile file : manifestFiles) {
ManifestFile newFile = file.copy();
((StructLike) newFile).set(0, newPath(newFile.path(), sourcePrefix,
targetPrefix));
+ Long rewrittenLength = rewrittenManifestLengths.get(file.path());
+ if (rewrittenLength == null && !rewrittenManifestLengths.isEmpty()) {
Review Comment:
I assume `!rewrittenManifestLengths.isEmpty()` is put just to ignore the opt
out case of 8 params args rewriteManifestList but it silently skips the case
where map is empty, maybe case where no manifest is selected for rewrite.
Maybe a boolean or something for the case when it's called via the older
rewriteManifestList - or if that one gets `@Deprecated`, just let the empty
map warn too and drop the special case.
##########
core/src/test/java/org/apache/iceberg/TestRewriteTablePathUtil.java:
##########
@@ -376,4 +385,70 @@ private ManifestFile
deleteManifestWithLiveAndDeletedEntry(DeleteFile live, Dele
return writer.toManifestFile();
}
+
+ /**
+ * One manifest is measured and one is not, in a single manifest list. The
measured one must take
+ * the mapped length and the unmeasured one must keep its source length,
which pins both the
+ * stamping and the fact that the map is keyed by the source path, not the
rewritten one.
+ */
+ @TestTemplate
+ public void testRewriteManifestListStampsMeasuredManifestLengths() throws
IOException {
+ table.newFastAppend().appendFile(FILE_A).commit();
+ table.newFastAppend().appendFile(FILE_B).commit();
+ Snapshot snapshot = table.currentSnapshot();
+ List<ManifestFile> manifests = snapshot.allManifests(table.io());
+ assertThat(manifests).hasSize(2);
+ ManifestFile measured = manifests.get(0);
+ ManifestFile unmeasured = manifests.get(1);
+
+ String manifestPath = measured.path();
+ String sourcePrefix = manifestPath.substring(0,
manifestPath.indexOf("/metadata/"));
Review Comment:
nit lastIndexOf instead of indexOf?
##########
core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java:
##########
@@ -328,6 +376,36 @@ private static List<ManifestFile>
manifestFilesInSnapshot(FileIO io, Snapshot sn
return manifestFiles;
}
+ /**
+ * Read the manifests referenced by a snapshot's manifest list, without
writing anything.
+ *
+ * <p>Callers need this before producing the rewritten manifest list, which
records each
+ * referenced manifest's rewritten length. Reading once here and passing the
result to {@link
+ * #rewriteManifestList(Snapshot, FileIO, TableMetadata, List, Set, String,
String, String,
+ * String, Map)} also avoids reading the manifest list twice.
+ *
+ * <p>An unreadable manifest list is logged and treated as empty, matching
the behaviour this
+ * method was extracted from.
+ *
+ * @param snapshot snapshot whose manifest list is read
+ * @param io file io
+ * @param sourcePrefix source prefix every referenced manifest must live
under
+ * @return the manifests referenced by the snapshot's manifest list, or an
empty list if the
+ * manifest list could not be read
+ */
+ public static List<ManifestFile> manifestsInSnapshot(
Review Comment:
Confusing with `manifestFilesInSnapshot`, but this one also validates via
Preconditions and throws, and both stay.
Maybe something like `manifestFilesInSnapshotValidatingPrefix` so the
difference is visible at the callsite?
##########
core/src/main/java/org/apache/iceberg/RewriteTablePathUtil.java:
##########
@@ -340,6 +418,8 @@ private static List<ManifestFile>
manifestFilesInSnapshot(FileIO io, Snapshot sn
* @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
+ * @see #rewriteDataManifestAndMeasureLength which also returns the
rewritten manifest length, for
+ * callers that need to record manifest_length in the manifest list
*/
public static RewriteResult<DataFile> rewriteDataManifest(
Review Comment:
Should this get `@Deprecated` like the old 9-arg rewriteDeleteManifest?
--
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]
