github-actions[bot] commented on code in PR #69076:
URL: https://github.com/apache/doris/pull/69076#discussion_r4277250590
##########
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:
[P2] Handle dropped historical partition sources before sorting
Consider a v2 table with an `identity(id)` spec: fast-append while `id` is
INT, promote it to LONG and fast-append again under the same spec, then remove
the partition field, drop `id`, and reload. Keep both manifests and choose a
target below their total size.
[PartitionSpec](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/api/src/main/java/org/apache/iceberg/PartitionSpec.java)
gives the historical field type UNKNOWN. The [manifest
reader](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/core/src/main/java/org/apache/iceberg/avro/InternalReader.java)
only promotes physical INT when the requested type is LONG, so the set now
contains both Integer and Long. The [UNKNOWN
comparator](https://raw.githubusercontent.com/apache/iceberg/apache-iceberg-1.11.0/api/src/main/java/org/apache/iceberg/types/Comparators.java)
is naturalOrder, and `assignClusters` throws ClassCastException before commit.
This repeats on a freshly loaded table with stab
le metadata, unlike thread 4274739797's concurrent promotion window. Handle
erased historical fields consistently in both clustering and SDK writer inputs
(for example, normalize fields the UNKNOWN writer already encodes as null);
merely widening this comparator can still fail in Iceberg's partition-summary
comparator. Please cover this completed promote/remove/drop sequence.
--
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]
