github-actions[bot] commented on code in PR #69076:
URL: https://github.com/apache/doris/pull/69076#discussion_r4272386916
##########
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:
[P2] Read counts from this rewrite's committed snapshot
After rewriting manifests A/B into X, an external rollback to the retained
A/B snapshot between the metadata commit and Iceberg's post-commit refresh
makes `currentSnapshot()` return A/B again. These differences then report `[0,
0]` although this command replaced two manifests and created one; the newly
computed `rewrittenCount` is incorrect. The SDK's initial refresh can also
include manifests absent from `beforePaths`.
[SnapshotProducer](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/SnapshotProducer.java)
explicitly loads its own committed snapshot by ID during cleanup for this
reason. Identify that snapshot and use its operation-specific
`manifests-replaced`/`manifests-created` summary counts rather than comparing
unrelated table snapshots.
##########
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:
[P2] Bound the number of simultaneously open manifest writers
With at least two input manifests containing thousands of distinct
first-field values and one small entry per value, this key creates thousands of
writers that `BaseRewriteManifests` keeps open until the whole scan finishes.
The target size rolls individual writers; it does not bound this fanout. Each
[S3OutputStream](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java)
immediately opens a local staging-file descriptor, so cardinality above the
FE's available descriptor budget causes `Too many open files` and a failed
rewrite even for small input manifests. Process bounded batches or complete
groups sequentially so the first-field grouping does not retain every output
stream simultaneously.
##########
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:
[P2] Exclude specs introduced after the rewrite update is constructed
On the default `spec_id=null` path, start with two spec-0 manifests,
construct `rm`, then let another engine evolve the partition spec and append a
spec-1 manifest before the first apply refresh.
[BaseRewriteManifests](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java)
captured only spec 0 in its constructor, but reads the refreshed spec-1 entry
and looks up its writer spec in that old map. This passes null to
`ManifestWriter`, whose `spec.specId()` dereference fails; commit retries only
handle `CommitFailedException`. Apply a `rewriteIf` predicate scoped to the
initial targeted manifest paths (plus the requested spec filter), or recreate
the update with the newer spec inventory.
--
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]
