ajreid21 commented on code in PR #18784:
URL: https://github.com/apache/iceberg/pull/18784#discussion_r4473691368
##########
core/src/main/java/org/apache/iceberg/PartitionStatsHandler.java:
##########
@@ -102,43 +106,55 @@ public static PartitionStatisticsFile
computeAndWriteStatsFile(Table table, long
Snapshot snapshot = table.snapshot(snapshotId);
Preconditions.checkArgument(snapshot != null, "Snapshot not found: %s",
snapshotId);
- Collection<PartitionStatistics> stats;
+ StructType partitionType = Partitioning.partitionType(table);
+ Schema schema = PartitionStatistics.schema(partitionType,
TableUtil.formatVersion(table));
+ Comparator<StructLike> partitionComparator =
Comparators.forType(partitionType);
+ Comparator<PartitionStatistics> statsComparator =
+ Comparator.comparing(PartitionStatistics::partition,
partitionComparator)
+ .thenComparing(PartitionStatistics::specId);
+
PartitionStatisticsFile statisticsFile = latestStatsFile(table,
snapshot.snapshotId());
- if (statisticsFile == null) {
- LOG.info(
- "Using full compute as previous statistics file is not present for
incremental compute.");
- stats =
- computeStats(table, snapshot.allManifests(table.io()), false /*
incremental */).values();
- } else {
+ if (statisticsFile != null) {
if (statisticsFile.snapshotId() == snapshotId) {
// no-op
LOG.info("Returning existing statistics file for snapshot {}",
snapshotId);
return statisticsFile;
}
- try {
- stats = computeAndMergeStatsIncremental(table, snapshot,
statisticsFile.snapshotId());
+ PartitionMap<PartitionStatistics> incrementalStats =
+ computeStatsDiff(table, table.snapshot(statisticsFile.snapshotId()),
snapshot);
+ if (incrementalStats.isEmpty()) {
+ return null;
+ }
+
+ List<PartitionStatistics> sortedIncremental =
+ sortStats(incrementalStats.values(), statsComparator);
+ try (CloseableIterable<PartitionStatistics> baseStats =
+ openBaseStats(table, statisticsFile.snapshotId())) {
+ Iterator<PartitionStatistics> merged =
+ new MergeIterator(
+ new GroupSortedIterator(
Review Comment:
Do we need to be worried about schema changes, particularly dropping
columns, affecting the projections here and the assumptions that rows/output is
sorted by the remaining projected fields due to column ordering in sorting?
Could dropping a partition source column break the ordering of the projected
base statistics? For example, (0, z), (1, a) is sorted, but dropping the first
field projects it to z, a. Since the merge assumes sorted inputs, do we need to
fall back to full computation when projection no longer preserves that order?
--
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]
