Skip to content

HIVE-29781: Iceberg: Basic stats fetch optimization - bulk partition stats read with query-scoped cache #6662

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
deniskuzZ wants to merge 3 commits into
base: master
Choose a base branch
from
Open

HIVE-29781: Iceberg: Basic stats fetch optimization - bulk partition stats read with query-scoped cache #6662

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0ff5615
HIVE-29781: Iceberg: Basic stats fetch optimization - bulk partition …
deniskuzZ Jul 30, 2026
8f4212c
refactor with prev unpartitioned spec marker
deniskuzZ Jul 31, 2026
974e1ff
sonar
deniskuzZ Jul 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions common/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -313,6 +313,7 @@ public enum ErrorMsg {
"Not all clauses are supported with mapjoin hint. Please remove mapjoin hint."),

ANALYZE_TABLE_NOSCAN_NON_NATIVE(10228, "ANALYZE TABLE NOSCAN cannot be used for " + "a non-native table"),
ANALYZE_PARTITION_NON_NATIVE(10449, "ANALYZE TABLE with a PARTITION clause cannot be used for a non-native table"),

PARTITION_VALUE_NOT_CONTINUOUS(10234, "Partition values specified are not continuous." +
" A subpartition value is specified without specifying the parent partition's value"),
Expand Down
290 changes: 204 additions & 86 deletions ...g/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
View file
Open in desktop

Large diffs are not rendered by default.

129 changes: 36 additions & 93 deletions iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,11 @@
package org.apache.iceberg.mr.hive;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.time.ZoneId;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Iterator;
import java.util.List;
Expand Down Expand Up @@ -84,7 +86,6 @@
import org.apache.iceberg.Schema;
import org.apache.iceberg.Snapshot;
import org.apache.iceberg.SnapshotRef;
import org.apache.iceberg.SnapshotSummary;
import org.apache.iceberg.StatisticsFile;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.Table;
Expand All @@ -105,7 +106,6 @@
import org.apache.iceberg.puffin.Puffin;
import org.apache.iceberg.puffin.PuffinReader;
import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
Expand Down Expand Up @@ -482,6 +482,15 @@ public static PartitionData toPartitionData(StructLike sourceKey, Types.StructTy
return toPartitionData(projection, targetKeyType);
}

/**
* Returns the partition name of the given tuple, or {@link DummyPartition#UNPARTITIONED} for the empty
* name an unpartitioned spec renders. Statistics and partition pruning join on this name, so both must
* render it the same way.
*/
static String toPartitionName(PartitionSpec spec, PartitionData data) {
return StringUtils.defaultIfEmpty(spec.partitionToPath(data), DummyPartition.UNPARTITIONED);
}

/**
* Builds a filter expression for data table operations (deleteFromRowFilter, FindFiles).
* Only supports identity transforms. Keys are partition field names.
Expand Down Expand Up @@ -527,20 +536,6 @@ private static Expression buildPartitionsFilter(Table table, Map<String, String>
bySourceColumn::containsKey);
}

/**
* Builds a filter expression for the PARTITIONS metadata table from partition field names
* with already-transformed values (e.g. {key_bucket_8: 5}).
*/
private static Expression buildPartitionsFilterFromPath(Table table, Map<String, String> partitionSpec)
throws SemanticException {
Types.StructType partitionType = Partitioning.partitionType(table);

return buildExpression(partitionSpec,
(column, value) ->
buildFieldPredicate(partitionType, column, value, "partition."),
col -> partitionType.field(col) != null);
}

static List<PartitionField> getPartitionFields(Table table, boolean latestSpecOnly) {
if (latestSpecOnly) {
return table.spec().fields();
Expand Down Expand Up @@ -723,7 +718,7 @@ private static List<String> getPartitionNames(Configuration conf,
}
PartitionData data = toPartitionData(
row.get(PART_IDX, StructProjection.class), partitionType, spec.partitionType());
return Maps.immutableEntry(spec.partitionToPath(data), spec.specId());
return Maps.immutableEntry(toPartitionName(spec, data), spec.specId());
})
.filter(Objects::nonNull)
.toSortedList(specIdComparator).stream()
Expand All @@ -736,86 +731,33 @@ private static List<String> getPartitionNames(Configuration conf,
}

/**
* Returns aggregated partition statistics from the PARTITIONS metadata table as a summary map.
* @param table the iceberg table
* @param partitionSpec partition spec map for filtering, or null for all partitions
* @return summary map with record count, file count, file size, and delete counts
* Reads the given snapshot's partition statistics file in a single pass, keyed by partition name;
* empty when the file is missing.
*/
static Map<String, String> getPartitionStats(Table table, Map<String, String> partitionSpec,
Snapshot snapshot) throws IOException {
PartitionsTable partitionsTable = (PartitionsTable) MetadataTableUtils.createMetadataTableInstance(
table, MetadataTableType.PARTITIONS);

Expression filter;
TableScan scan = partitionsTable.newScan();
if (snapshot != null) {
scan = scan.useSnapshot(snapshot.snapshotId());
}
try {
filter = buildPartitionsFilterFromPath(table, partitionSpec);
scan = scan.filter(filter);
} catch (SemanticException e) {
throw new IOException("Failed to build partition filter for " + partitionSpec, e);
}

Evaluator evaluator = new Evaluator(partitionsTable.schema().asStruct(), filter);

try (CloseableIterable<FileScanTask> fileScanTasks = scan.planFiles()) {
return FluentIterable.from(fileScanTasks)
.transformAndConcat(task -> task.asDataTask().rows())
.filter(evaluator::eval)
.transform(BasicPartitionStats::from)
.stream().reduce(BasicPartitionStats::merge)
.orElse(BasicPartitionStats.EMPTY)
.toStatsMap();
}
}

static Map<String, String> toStatsMap(PartitionStatistics stats) {
return ImmutableMap.of(
SnapshotSummary.TOTAL_DATA_FILES_PROP, String.valueOf(stats.dataFileCount()),
SnapshotSummary.TOTAL_RECORDS_PROP, String.valueOf(stats.dataRecordCount()),
SnapshotSummary.TOTAL_EQ_DELETES_PROP, String.valueOf(stats.equalityDeleteRecordCount()),
SnapshotSummary.TOTAL_POS_DELETES_PROP, String.valueOf(stats.positionDeleteRecordCount()),
SnapshotSummary.TOTAL_FILE_SIZE_PROP, String.valueOf(stats.totalDataFileSizeInBytes())
);
}

/** The subset of partition stats counters Hive reports, summed across PARTITIONS metadata table rows. */
private record BasicPartitionStats(
long recordCount, long fileCount, long totalFileSizeInBytes,
long positionDeleteRecordCount, long equalityDeleteRecordCount) {

static final BasicPartitionStats EMPTY = new BasicPartitionStats(0,
0, 0, 0, 0);

static BasicPartitionStats from(StructLike row) {
return new BasicPartitionStats(
row.get(2, Long.class), // record_count
row.get(3, Integer.class), // file_count
row.get(4, Long.class), // total_data_file_size_in_bytes
row.get(5, Long.class), // position_delete_record_count
row.get(7, Long.class)); // equality_delete_record_count
}

BasicPartitionStats merge(BasicPartitionStats other) {
return new BasicPartitionStats(
recordCount + other.recordCount,
fileCount + other.fileCount,
totalFileSizeInBytes + other.totalFileSizeInBytes,
positionDeleteRecordCount + other.positionDeleteRecordCount,
equalityDeleteRecordCount + other.equalityDeleteRecordCount);
static Map<String, PartitionStatistics> readPartitionStats(Table table, Snapshot snapshot) {
PartitionStatisticsFile statsFile = getPartitionStatsFile(table, snapshot.snapshotId());
if (statsFile == null) {
LOG.warn("Partition stats file not found for snapshot: {}", snapshot.snapshotId());
return Map.of();
}
Map<String, PartitionStatistics> result = Maps.newHashMap();
Types.StructType partitionType = Partitioning.partitionType(table);

Map<String, String> toStatsMap() {
return ImmutableMap.of(
SnapshotSummary.TOTAL_DATA_FILES_PROP, String.valueOf(fileCount),
SnapshotSummary.TOTAL_RECORDS_PROP, String.valueOf(recordCount),
SnapshotSummary.TOTAL_EQ_DELETES_PROP, String.valueOf(equalityDeleteRecordCount),
SnapshotSummary.TOTAL_POS_DELETES_PROP, String.valueOf(positionDeleteRecordCount),
SnapshotSummary.TOTAL_FILE_SIZE_PROP, String.valueOf(totalFileSizeInBytes)
);
try (CloseableIterable<PartitionStatistics> records =
table.newPartitionStatisticsScan().useSnapshot(snapshot.snapshotId()).scan()) {
LOG.info("Using partition stats from: {}", statsFile.path());
for (PartitionStatistics partitionStats : records) {
PartitionSpec spec = table.specs().get(partitionStats.specId());
PartitionData data = toPartitionData(partitionStats.partition(), partitionType,
spec.partitionType());
// the scan copies the counters into a fresh object per row, so retaining it is safe;
// only the partition tuple may alias the reader's reused struct - do not use it after this loop
result.put(toPartitionName(spec, data), partitionStats);
}
} catch (IOException e) {
throw new UncheckedIOException(e);
}
return Collections.unmodifiableMap(result);
}

public static PartitionSpec getPartitionSpec(Table icebergTable, String partitionPath)
Expand Down Expand Up @@ -849,6 +791,7 @@ public static <T> List<T> readColStats(Table table, Long snapshotId, Predicate<B

String statsPath = IcebergTableUtil.getColStatsPath(table, snapshotId);
if (statsPath == null) {
LOG.warn("Column stats file not found for snapshot: {}", snapshotId);
return colStats;
}
try (PuffinReader reader = Puffin.read(table.io().newInputFile(statsPath)).build()) {
Expand Down
Loading
Loading