github-actions[bot] commented on code in PR #69076:
URL: https://github.com/apache/doris/pull/69076#discussion_r4274739816
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/RewriteManifestExecutor.java:
##########
@@ -66,60 +78,66 @@ 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);
}
-
- // Configure rewrite operation, optionally restricting manifests
by specId
- RewriteManifests rm = table.rewriteManifests();
-
- if (specId != null) {
- final int targetSpecId = specId;
- rm.rewriteIf(manifest -> manifest.partitionSpecId() ==
targetSpecId);
- }
-
- // Commit manifest rewrite
- rm.commit();
-
- // Refresh snapshot after rewrite
- Snapshot snapshotAfter = table.currentSnapshot();
- if (snapshotAfter == null) {
- return new Result(rewrittenCount, 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);
}
- // 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()
- .map(ManifestFile::path)
- .collect(java.util.stream.Collectors.toSet());
-
- int addedCount = (int) manifestsAfterTargeted.stream()
+ // Without a cluster function Iceberg rewrites nothing
(BaseRewriteManifests.requiresRewrite) and only
+ // commits a snapshot that keeps every manifest. Only the targeted
manifests are rewritten: a manifest
+ // that another writer commits before the rewrite applies is kept
as it is. Such a manifest can be of a
+ // partition spec added after the table was loaded, which the
rewrite cannot write, because it takes
+ // the specs of the table it is created from.
+ Set<String> targetedPaths = manifestsBeforeTargeted.stream()
.map(ManifestFile::path)
- .filter(path -> !beforePaths.contains(path))
- .count();
+ .collect(Collectors.toSet());
+ RewriteManifests rm = table.rewriteManifests()
+ .clusterBy(RewriteManifestExecutor::clusterKey)
+ .rewriteIf(manifest ->
targetedPaths.contains(manifest.path()));
+ rm.commit();
- return new Result(rewrittenCount, addedCount);
+ // Count the manifests replaced and written by the snapshot this
rewrite committed. Iceberg builds the
+ // event of that snapshot after the commit
(SnapshotProducer.notifyListeners), while the table's current
+ // snapshot can already be another one.
+ Map<String, String> summary = ((CreateSnapshotEvent)
rm.updateEvent()).summary();
Review Comment:
[P2] Keep event lookup from failing a successful rewrite
After this rewrite commits snapshot `S`, another writer can fast-append `T`
and explicitly expire `S` before Iceberg's cleanup refresh. [Expiration is
legal once S is no longer a snapshot-ref
target](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java),
and T retains the rewritten manifests.
[`BaseRewriteManifests.updateEvent`](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/BaseRewriteManifests.java)
then dereferences the missing snapshot.
[`SnapshotProducer.commit`](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/SnapshotProducer.java)
suppresses that notification failure and returns success, but this second
lookup raises `DorisConnectorException`, reports that the rewrite failed, and
skips `ConnectorExecuteAction`'s mutation refresh. Obtain the successful
attempt's counts independe
ntly of a lookup requiring the snapshot to remain in live metadata. Unlike
thread 4272386916's wrong counts with a retained rewrite snapshot, this is a
new exception after valid snapshot expiration.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/RewriteManifestExecutor.java:
##########
@@ -66,60 +78,66 @@ 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);
}
-
- // Configure rewrite operation, optionally restricting manifests
by specId
- RewriteManifests rm = table.rewriteManifests();
-
- if (specId != null) {
- final int targetSpecId = specId;
- rm.rewriteIf(manifest -> manifest.partitionSpecId() ==
targetSpecId);
- }
-
- // Commit manifest rewrite
- rm.commit();
-
- // Refresh snapshot after rewrite
- Snapshot snapshotAfter = table.currentSnapshot();
- if (snapshotAfter == null) {
- return new Result(rewrittenCount, 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);
}
- // 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()
- .map(ManifestFile::path)
- .collect(java.util.stream.Collectors.toSet());
-
- int addedCount = (int) manifestsAfterTargeted.stream()
+ // Without a cluster function Iceberg rewrites nothing
(BaseRewriteManifests.requiresRewrite) and only
+ // commits a snapshot that keeps every manifest. Only the targeted
manifests are rewritten: a manifest
+ // that another writer commits before the rewrite applies is kept
as it is. Such a manifest can be of a
+ // partition spec added after the table was loaded, which the
rewrite cannot write, because it takes
+ // the specs of the table it is created from.
+ Set<String> targetedPaths = manifestsBeforeTargeted.stream()
.map(ManifestFile::path)
- .filter(path -> !beforePaths.contains(path))
- .count();
+ .collect(Collectors.toSet());
+ RewriteManifests rm = table.rewriteManifests()
+ .clusterBy(RewriteManifestExecutor::clusterKey)
Review Comment:
[P2] Keep reader and writer specs consistent across schema promotion
With two manifests identity-partitioned by `id INT`, load the table and let
another engine promote `id` to LONG before this rewrite applies. [Iceberg
rebuilds the existing specs against the new
schema](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/TableMetadata.java)
without changing their IDs or manifest paths, so the path predicate still
targets both. `BaseRewriteManifests` captured the old INT writer spec, but its
readers use refreshed specs and promote partition values to `Long`. The
writer's
[`PartitionSummary`](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/PartitionSummary.java)
still requests `Integer.class`, so `PartitionData.get` throws and the rewrite
fails before commit. Unlike thread 4272386934's new-spec manifests, these are
original target paths with the same spec ID. Keep reader/writer spec schemas
aligned across refresh/retry, or recreate th
e update when its captured schema becomes stale, and cover this promotion case.
--
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]
