morningman commented on code in PR #69076:
URL: https://github.com/apache/doris/pull/69076#discussion_r4274462609
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/RewriteManifestExecutor.java:
##########
@@ -86,25 +101,16 @@ public Result execute(Table table, Integer specId) {
// Commit manifest rewrite
rm.commit();
- // Refresh snapshot after rewrite
- Snapshot snapshotAfter = table.currentSnapshot();
- if (snapshotAfter == null) {
- return new Result(rewrittenCount, 0);
- }
-
- // Collect manifests after rewrite and filter by specId
- List<ManifestFile> manifestsAfter =
snapshotAfter.dataManifests(table.io());
- List<ManifestFile> manifestsAfterTargeted =
filterBySpecId(manifestsAfter, specId);
-
- // Compute addedCount as newly produced manifests (path not in
before set)
- java.util.Set<String> beforePaths =
manifestsBeforeTargeted.stream()
+ // Count the targeted manifests the rewrite replaced and the
manifests it wrote in their place
+ Set<String> beforePaths = manifestsBeforeTargeted.stream()
.map(ManifestFile::path)
- .collect(java.util.stream.Collectors.toSet());
-
- int addedCount = (int) manifestsAfterTargeted.stream()
+ .collect(Collectors.toSet());
+ Set<String> afterPaths =
filterBySpecId(table.currentSnapshot().dataManifests(table.io()), specId)
+ .stream()
.map(ManifestFile::path)
- .filter(path -> !beforePaths.contains(path))
- .count();
+ .collect(Collectors.toSet());
+ int rewrittenCount = (int) beforePaths.stream().filter(path ->
!afterPaths.contains(path)).count();
Review Comment:
Fixed in 8afa3e3ce7a. The counts are now `manifests-replaced` and
`manifests-created` of the snapshot this rewrite committed. After `rm.commit()`
the executor reads them from `rm.updateEvent()`, the event Iceberg itself
builds for that snapshot after the commit (`SnapshotProducer.notifyListeners`).
`BaseRewriteManifests.updateEvent()` loads the snapshot by its id, so a commit
or a rollback by another writer after ours does not change the counts.
Test: `keepsAManifestCommittedAfterTheTableWasLoaded` loads the table, lets
another writer append a third manifest, and rewrites through the loaded table.
It expects `[2, 1]` with the appended manifest kept. Comparing the table's
manifests before and after the commit gives `[2, 2]` there.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/RewriteManifestExecutor.java:
##########
@@ -113,13 +119,27 @@ public Result execute(Table table, Integer specId) {
}
}
+ /**
+ * Groups data files by the value of their first partition field (one
group for an unpartitioned table), as
+ * Trino's optimize_manifests does, so a rewrite of a multi-field
partition does not write one manifest per
+ * partition. The key holds the value PartitionData returns (immutable, or
a copy for binary values) rather
+ * than the partition, so it does not depend on whether ManifestReader
reuses the partition between rows (its
+ * iterator() copies files for that reason); the list lets a NULL value be
a key, which Iceberg rejects.
+ */
+ private static Object clusterKey(DataFile file) {
+ StructLike partition = file.partition();
+ return partition.size() == 0
+ ? Collections.emptyList()
+ : Collections.singletonList(partition.get(0, Object.class));
Review Comment:
Not changing this. One open writer per cluster key until the scan ends is
how Iceberg's `RewriteManifests.clusterBy` works: `BaseRewriteManifests` keeps
a `WriterWrapper` per key and closes them all at the end. Trino's
`optimize_manifests` uses the same API with the same first-partition-field key,
and this PR follows Trino on purpose. To bound the open writers we would have
to write the manifests ourselves, as Spark's `RewriteManifestsSparkAction`
does: read every entry, sort or group it, write the manifests and commit them
with `deleteManifest` / `addManifest`. Spark distributes that sort, but on the
FE it means holding every entry in memory first. The case needs thousands of
distinct first-field values in one table, and such a table gets at least one
manifest per value from this rewrite in any case, as the PR description says,
so the procedure is a poor fit for it.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/RewriteManifestExecutor.java:
##########
@@ -66,17 +75,23 @@ public Result execute(Table table, Integer specId) {
}
// Collect manifests before rewrite and filter by specId if
provided
- List<ManifestFile> manifestsBefore =
currentSnapshot.dataManifests(table.io());
- List<ManifestFile> manifestsBeforeTargeted =
filterBySpecId(manifestsBefore, specId);
-
- int rewrittenCount = manifestsBeforeTargeted.size();
-
- if (rewrittenCount == 0) {
+ List<ManifestFile> manifestsBeforeTargeted =
+ filterBySpecId(currentSnapshot.dataManifests(table.io()),
specId);
+ if (manifestsBeforeTargeted.isEmpty()) {
+ return new Result(0, 0);
+ }
+ // One manifest below the target size has nothing to merge with,
as in Spark's and Trino's manifest
+ // rewrites: leave the table alone instead of committing a
snapshot that changes nothing.
+ long targetManifestSizeBytes =
PropertyUtil.propertyAsLong(table.properties(),
+ TableProperties.MANIFEST_TARGET_SIZE_BYTES,
TableProperties.MANIFEST_TARGET_SIZE_BYTES_DEFAULT);
+ if (manifestsBeforeTargeted.size() == 1
+ && manifestsBeforeTargeted.get(0).length() <
targetManifestSizeBytes) {
return new Result(0, 0);
}
- // Configure rewrite operation, optionally restricting manifests
by specId
- RewriteManifests rm = table.rewriteManifests();
+ // Without a cluster function Iceberg rewrites nothing
(BaseRewriteManifests.requiresRewrite) and only
+ // commits a snapshot that keeps every manifest. Optionally
restrict the rewrite to one spec.
+ RewriteManifests rm =
table.rewriteManifests().clusterBy(RewriteManifestExecutor::clusterKey);
Review Comment:
Fixed in 8afa3e3ce7a, as you suggest. `rewriteIf` now accepts only the
manifests that were targeted when the rewrite started, by path, which also
applies `spec_id`. A manifest that another writer commits in between is kept as
it is, whatever its spec.
Test: `keepsAManifestOfAPartitionSpecAddedAfterTheTableWasLoaded` loads the
table. Another writer then adds an identity partition field and fast appends a
file of the new spec. The rewrite through the loaded table returns `[2, 1]` and
keeps the new spec's manifest; before the fix it failed with `Cannot invoke
"org.apache.iceberg.PartitionSpec.specId()" because "spec" is null`. The other
writer fast appends because a merge append also merges the old spec's
manifests, which leaves the rewrite nothing to target.
--
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]
