morningman commented on code in PR #69076:
URL: https://github.com/apache/doris/pull/69076#discussion_r4277175779
##########
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:
Correcting my earlier reply, which was wrong about Trino. Trino's
`optimize_manifests` no longer keys `clusterBy` by every value of the first
partition field: trinodb/trino de101509c3 bounded the keys to avoid running the
coordinator out of memory, and 4a58ac208e reads the field's distinct values
from the manifests, sorts them and cuts them into at most `min(ceil(total
manifest size / commit.manifest.target-size-bytes), 200)` runs of adjacent
values. So the writers can be bounded without writing the manifests ourselves.
a1cea007a99 does the same here. For each targeted partition spec, the
executor reads the distinct values of the spec's first partition field from the
targeted manifests (only the `partition` column, in parallel on Iceberg's
worker pool), sorts them with NULL first, and gives the value at rank i cluster
`i * clusters / n`, where `clusters = min(ceil(size of the spec's targeted
manifests / target size), 200)`. Adjacent partitions share a manifest, so the
manifests keep narrow partition summaries, and the rewrite keeps at most that
many writers open per spec. A spec whose targeted manifests fit in one
target-size manifest is one cluster and is not read ahead, so such a table now
gets one manifest instead of one per value. Unlike Trino, each spec is
clustered on its own first field rather than the current spec's, because
`spec_id` can target an older spec, and the first fields of two specs can be of
different types.
Tests:
`IcebergRewriteManifestsActionTest.clustersAdjacentFirstPartitionValues` (files
appended with 100, 1, 50 and 7 become the runs {1, 7} and {50, 100}),
`clustersEachPartitionSpecByItsOwnFirstField`,
`clusterCountIsTheManifestSizeOverTheTargetUpTo200`,
`assignsSortedValuesToRunsOfNearlyEqualLength` and
`mergesPartitionedManifestsThatFitTheTargetSizeIntoOne`. Not sorting, dropping
the cap, rounding up in a way that overflows, clustering every spec on the
current spec's first field, and sorting without NULL first each turn one of
them red. On a local cluster, a table partitioned by `a` with four INSERTs of
100, 1, 50 and 7, and `commit.manifest.target-size-bytes` set to half the size
of its four manifests, returns `[4, 2]`, and the two manifests cover 1 to 7 and
50 to 100.
##########
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:
Not changing this. The mismatch is inside Iceberg's `BaseRewriteManifests`,
in 1.11.0 and on Iceberg main alike: the constructor keeps
`ops().current().specsById()` for its manifest writers, while `performRewrite`
reads the manifests with the specs of the metadata that
`SnapshotProducer.apply()` has just refreshed. Any caller that builds the
rewrite from a loaded table and then commits it has this window; Trino's
`optimize_manifests` builds it from a loaded `BaseTable` in the same way.
I reproduced it against Iceberg 1.11.0 with the executor's calls. After
another handle promotes the identity source column from `int` to `long`,
`commit()` fails with `IllegalArgumentException: Wrong class, expected
java.lang.Integer, but was java.lang.Long`. With a1cea007a99, a spec whose
targeted manifests are larger than the target size reads its first partition
values ahead, with the specs of the loaded table, so the rewrite's readers then
return `Long` values that are not among them, and the executor fails first,
with `the table's partitioning changed during the rewrite`.
`IcebergRewriteManifestsActionTest.failsBeforeCommitWhenTheFirstPartitionFieldIsPromotedDuringTheRewrite`
pins that case: the rewrite fails, and the snapshot and the manifests stay as
they were. Running the procedure again on a freshly loaded table rewrites them.
The window runs from loading the table to the first refresh inside
`commit()`: one manifest-list read, plus that read-ahead for such a spec. A
type promotion of a partition source column has to land in it, and the
procedure then fails before it commits, with an error that names the cause. The
alternatives on our side are a retry around the SDK commit, or leaving out the
manifests whose partition type changed. Here the second one would leave out
every targeted manifest, so the procedure would commit a snapshot that changes
nothing and report `[0, 0]`. I prefer the failure. The fix belongs in
`BaseRewriteManifests`: its writers should take their specs from the refreshed
metadata, as its readers do.
##########
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:
Not changing this, although the sequence is real. I reproduced it against
Iceberg 1.11.0 with table operations that let another writer fast append and
then expire our snapshot right after our commit. `commit()` returns, and
`rm.updateEvent()` then throws `NullPointerException: Cannot invoke
"org.apache.iceberg.Snapshot.sequenceNumber()" because "snapshot" is null`.
The two counts exist only in the summary of that snapshot, so once it is
expired there is nothing exact left to report. Iceberg's own
`CreateSnapshotEvent` and commit report for this commit fail in the same way
(`SnapshotProducer.notifyListeners` only logs it), and Trino's
`optimize_manifests`, which reads the counts from the commit report, returns
none in this case. Our result columns are non-nullable INTs, so returning
success would mean making the counts up, for example by comparing the manifest
lists before and after the commit, which the first round of this review showed
to be wrong under a concurrent commit.
The window is between our metadata commit and the refresh that
`SnapshotProducer.commit()` makes right after it. In it another writer has to
commit an append and then expire a snapshot that is a few milliseconds old; by
default `expire_snapshots` keeps the snapshots of the last five days. The
rewrite only replaces manifests, so the table's data is the same either way,
and running the procedure again rewrites the current manifests.
--
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]
