morningman commented on code in PR #69076:
URL: https://github.com/apache/doris/pull/69076#discussion_r4279454821
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/action/RewriteManifestExecutor.java:
##########
@@ -66,51 +97,135 @@ 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);
+ // 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);
}
- // Commit manifest rewrite
+ // 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)
+ .collect(Collectors.toSet());
+ Map<Integer, Map<Object, Integer>> clustersBySpec =
+ clustersBySpec(table, manifestsBeforeTargeted,
targetManifestSizeBytes);
+ RewriteManifests rm = table.rewriteManifests()
+ .clusterBy(file -> clusterOf(clustersBySpec, file))
+ .rewriteIf(manifest ->
targetedPaths.contains(manifest.path()));
rm.commit();
- // Refresh snapshot after rewrite
- Snapshot snapshotAfter = table.currentSnapshot();
- if (snapshotAfter == null) {
- return new Result(rewrittenCount, 0);
+ // 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();
+ return new
Result(Integer.parseInt(summary.get(SnapshotSummary.REPLACED_MANIFESTS_COUNT)),
+
Integer.parseInt(summary.get(SnapshotSummary.CREATED_MANIFESTS_COUNT)));
+ } catch (Exception e) {
+ LOG.warn("Failed to execute manifest rewrite for table: {}",
table.name(), e);
+ throw new DorisConnectorException("Failed to rewrite manifests: "
+ e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Groups the data files of each targeted partition spec into runs of
adjacent values of the spec's first
+ * partition field, as Trino's optimize_manifests does. A spec gets
min(ceil(size of its targeted manifests /
+ * target size), 200) clusters: the distinct values are read from its
targeted manifests, sorted with NULL first,
+ * and the value at rank i goes to cluster i * clusters / n. Adjacent
partitions then share a manifest, which
+ * keeps the manifests' partition summaries narrow for pruning, while the
rewrite keeps at most that many
+ * writers open for the spec; Iceberg still rolls each cluster over at the
target size. An unpartitioned spec, or
+ * one whose targeted manifests fit in one target-size manifest, is one
cluster and is not read ahead. Each spec
+ * is clustered on its own first field, which can be of another type than
another spec's: Iceberg keeps a writer
+ * per cluster and spec, so the specs' cluster numbers do not collide.
+ *
+ * @return the value-to-cluster map of each targeted spec, empty for a
spec that is one cluster
+ */
+ private static Map<Integer, Map<Object, Integer>> clustersBySpec(Table
table, List<ManifestFile> targeted,
+ long targetManifestSizeBytes) throws IOException {
+ Map<Integer, List<ManifestFile>> manifestsBySpec = targeted.stream()
+ .collect(Collectors.groupingBy(ManifestFile::partitionSpecId));
+ Map<Integer, Map<Object, Integer>> clustersBySpec = new HashMap<>();
+ for (Map.Entry<Integer, List<ManifestFile>> entry :
manifestsBySpec.entrySet()) {
+ PartitionSpec spec = table.specs().get(entry.getKey());
+ int clusters =
clusterCount(entry.getValue().stream().mapToLong(ManifestFile::length).sum(),
+ targetManifestSizeBytes);
+ if (spec.isUnpartitioned() || clusters == 1) {
+ clustersBySpec.put(entry.getKey(), Collections.emptyMap());
+ continue;
}
+ Comparator<Object> order =
Comparators.<Object>nullsFirst().thenComparing(
+
Comparators.forType(spec.partitionType().fields().get(0).type().asPrimitiveType()));
Review Comment:
Not changing this. It reproduces: after `id` is promoted from INT to LONG,
its partition field removed and the column dropped, the first field of spec 0
has type UNKNOWN, and the files of the two manifests carry an Integer and a
Long. But Iceberg itself fails on this table as soon as it writes those files
into one manifest: a rewrite with a single cluster, `clusterBy(file -> 0)`,
which is what this executor asks for when the targeted manifests fit in one
target-size manifest, fails with the same ClassCastException in Iceberg
1.11.0's manifest writer, and so does any rewrite that merges the two
manifests. As you note, widening the comparator here would not help for the
same reason. The rewrite fails before it commits and leaves the table as it is.
Avoiding it would mean keeping the value classes of every erased partition
field apart in the cluster key, a rule that only this sequence of a promotion,
a removed partition field and a dropped column needs.
--
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]
