This is an automated email from the ASF dual-hosted git repository.
deniskuzZ pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new c021ce567a0 HIVE-29781: Iceberg: Basic stats fetch optimization - bulk
partition stats read with query-scoped cache (#6662)
c021ce567a0 is described below
commit c021ce567a0dfcebba720286d762a12b0f21b63e
Author: Denys Kuzmenko <[email protected]>
AuthorDate: Tue Aug 4 19:41:10 2026 +0300
HIVE-29781: Iceberg: Basic stats fetch optimization - bulk partition stats
read with query-scoped cache (#6662)
---
.../java/org/apache/hadoop/hive/ql/ErrorMsg.java | 1 +
.../iceberg/mr/hive/HiveIcebergStorageHandler.java | 278 +++++++++++-----
.../apache/iceberg/mr/hive/IcebergTableUtil.java | 129 ++------
.../iceberg/mr/hive/TestHiveIcebergStatistics.java | 361 +++++++++++++++++++++
...iceberg_truncate_partition_with_evolution.q.out | 151 +++++----
.../src/test/results/positive/row_count.q.out | 22 +-
.../truncate_partitioned_iceberg_table.q.out | 2 +
.../ql/ddl/table/info/desc/DescTableOperation.java | 4 +-
.../hadoop/hive/ql/metadata/DummyPartition.java | 10 +
.../hive/ql/metadata/HiveStorageHandler.java | 56 ++--
.../hadoop/hive/ql/optimizer/StatsOptimizer.java | 56 +++-
.../hadoop/hive/ql/parse/BaseSemanticAnalyzer.java | 17 +
.../hive/ql/parse/ColumnStatsSemanticAnalyzer.java | 5 +
.../hadoop/hive/ql/parse/PrunedPartitionList.java | 5 +-
.../hadoop/hive/ql/parse/SemanticAnalyzer.java | 4 +
.../apache/hadoop/hive/ql/stats/BasicStats.java | 82 ++---
.../hadoop/hive/ql/stats/BasicStatsNoJobTask.java | 23 +-
.../hadoop/hive/ql/stats/BasicStatsTask.java | 9 +-
.../apache/hadoop/hive/ql/stats/StatsUtils.java | 120 ++++---
.../hadoop/hive/ql/stats/TestBasicStats.java | 28 +-
20 files changed, 972 insertions(+), 391 deletions(-)
diff --git a/common/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java
b/common/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java
index 156d7c7ebe7..a900e3eb60b 100644
--- a/common/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java
+++ b/common/src/java/org/apache/hadoop/hive/ql/ErrorMsg.java
@@ -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"),
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
index 41261ac4228..070bd9b70ae 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java
@@ -29,7 +29,6 @@
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
@@ -39,6 +38,7 @@
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
+import java.util.function.BiConsumer;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -46,6 +46,7 @@
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.SerializationUtils;
import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.math.NumberUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
@@ -119,7 +120,6 @@
import
org.apache.hadoop.hive.ql.security.authorization.HiveCustomStorageHandlerUtils;
import org.apache.hadoop.hive.ql.session.SessionState;
import org.apache.hadoop.hive.ql.session.SessionStateUtil;
-import org.apache.hadoop.hive.ql.stats.Partish;
import org.apache.hadoop.hive.ql.util.NullOrdering;
import org.apache.hadoop.hive.serde2.AbstractSerDe;
import org.apache.hadoop.hive.serde2.Deserializer;
@@ -153,7 +153,6 @@
import org.apache.iceberg.PartitionStatistics;
import org.apache.iceberg.PartitionStatisticsFile;
import org.apache.iceberg.PartitionStatsHandler;
-import org.apache.iceberg.Partitioning;
import org.apache.iceberg.RowLevelOperationMode;
import org.apache.iceberg.Schema;
import org.apache.iceberg.SchemaParser;
@@ -199,6 +198,7 @@
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
import org.apache.iceberg.relocated.com.google.common.collect.FluentIterable;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+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;
@@ -237,6 +237,7 @@ public class HiveIcebergStorageHandler extends
DefaultStorageHandler implements
public static final String TABLE_DEFAULT_LOCATION = "TABLE_DEFAULT_LOCATION";
private static final String PARTITION = "partition";
+ private static final String PARTITION_STATS_PREFIX = "partitionStats.";
public static final String STATS = "/stats/snap-";
public static final String COPY_ON_WRITE =
RowLevelOperationMode.COPY_ON_WRITE.modeName();
@@ -504,68 +505,108 @@ public void
appendFiles(org.apache.hadoop.hive.metastore.api.Table table, URI fr
}
@Override
- public Map<String, String> getBasicStatistics(Partish partish) {
- return getBasicStatistics(partish, false);
+ public Map<String, String>
getBasicStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ return getBasicStatistics(hmsTable, false);
}
- @SuppressWarnings("checkstyle:CyclomaticComplexity")
- private Map<String, String> getBasicStatistics(Partish partish, boolean
quickStats) {
- Map<String, String> stats = Maps.newHashMap();
+ private Map<String, String>
getBasicStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
+ boolean quickStats) {
+ Map<String, String> stats;
- org.apache.hadoop.hive.ql.metadata.Table hmsTable = partish.getTable();
// For write queries where rows got modified, don't fetch from cache as
values could have changed.
Table table = getTable(hmsTable);
Snapshot snapshot = IcebergTableUtil.getTableSnapshot(table, hmsTable);
if (snapshot == null) {
- stats.put(StatsSetupConst.NUM_FILES, "0");
- stats.put(StatsSetupConst.ROW_COUNT, "0");
- stats.put(StatsSetupConst.TOTAL_SIZE, "0");
+ stats = emptyStatsMap();
} else if (!HiveMetaHook.ICEBERG.equals(getStatsSource()) && !quickStats) {
- stats = partish.getPartParameters();
+ stats = hmsTable.getParameters();
} else {
- Map<String, String> summary = getPartishSummary(partish, table,
snapshot);
- if (summary != null) {
- if (summary.containsKey(TOTAL_DATA_FILES_PROP)) {
- stats.put(StatsSetupConst.NUM_FILES,
summary.get(TOTAL_DATA_FILES_PROP));
- }
- if (summary.containsKey(TOTAL_RECORDS_PROP) && !quickStats) {
- long totalRecords = Long.parseLong(summary.get(TOTAL_RECORDS_PROP));
- if (summary.containsKey(TOTAL_EQ_DELETES_PROP) &&
- summary.containsKey(TOTAL_POS_DELETES_PROP)) {
+ Map<String, String> summary = snapshot.summary();
+ stats = summary != null ? toStatsMap(summary, quickStats) : Map.of();
+ }
+ return stats;
+ }
- long totalEqDeletes =
Long.parseLong(summary.get(TOTAL_EQ_DELETES_PROP));
- long totalPosDeletes =
Long.parseLong(summary.get(TOTAL_POS_DELETES_PROP));
+ private static Map<String, String> emptyStatsMap() {
+ return Map.of(
+ StatsSetupConst.NUM_FILES, "0",
+ StatsSetupConst.ROW_COUNT, "0",
+ StatsSetupConst.TOTAL_SIZE, "0");
+ }
- long actualRecords = totalRecords - (totalEqDeletes > 0 ? 0 :
totalPosDeletes);
- totalRecords = actualRecords > 0 ? actualRecords : totalRecords;
- // actualRecords maybe -ve in edge cases
- }
- stats.put(StatsSetupConst.ROW_COUNT, String.valueOf(totalRecords));
- }
- if (summary.containsKey(TOTAL_FILE_SIZE_PROP)) {
- stats.put(StatsSetupConst.TOTAL_SIZE,
summary.get(TOTAL_FILE_SIZE_PROP));
- }
- }
+ /**
+ * Converts the Iceberg counters into the basic statistics Hive reports:
numFiles, numRows and totalSize.
+ *
+ * @param quickStats when true the record count is left out
+ */
+ private static Map<String, String> toStatsMap(Map<String, String> summary,
boolean quickStats) {
+ Map<String, String> stats = Maps.newHashMap();
+ if (summary.containsKey(TOTAL_DATA_FILES_PROP)) {
+ stats.put(StatsSetupConst.NUM_FILES, summary.get(TOTAL_DATA_FILES_PROP));
+ }
+ if (summary.containsKey(TOTAL_RECORDS_PROP) && !quickStats) {
+ stats.put(StatsSetupConst.ROW_COUNT,
String.valueOf(estimateNumRows(summary)));
+ }
+ if (summary.containsKey(TOTAL_FILE_SIZE_PROP)) {
+ stats.put(StatsSetupConst.TOTAL_SIZE, summary.get(TOTAL_FILE_SIZE_PROP));
}
return stats;
}
+ private static Map<String, String> toStatsMap(PartitionStatistics stats) {
+ Map<String, Long> counters = Map.of(
+ StatsSetupConst.NUM_FILES, (long) stats.dataFileCount(),
+ StatsSetupConst.ROW_COUNT, estimateNumRows(stats.dataRecordCount(),
+ stats.equalityDeleteRecordCount(),
stats.positionDeleteRecordCount()),
+ StatsSetupConst.TOTAL_SIZE, stats.totalDataFileSizeInBytes());
+ return ImmutableMap.copyOf(
+ Maps.transformValues(counters, String::valueOf));
+ }
+
+ private static long estimateNumRows(Map<String, String> summary) {
+ long totalRecords = Long.parseLong(summary.get(TOTAL_RECORDS_PROP));
+ if (!summary.containsKey(TOTAL_EQ_DELETES_PROP) ||
!summary.containsKey(TOTAL_POS_DELETES_PROP)) {
+ return totalRecords;
+ }
+ return estimateNumRows(totalRecords,
+ Long.parseLong(summary.get(TOTAL_EQ_DELETES_PROP)),
+ Long.parseLong(summary.get(TOTAL_POS_DELETES_PROP)));
+ }
+
+ /**
+ * Estimates the row count by subtracting position deletes from the record
count. Falls back to the raw
+ * record count when equality deletes are present or the deletes
over-subtract, as their exact effect is
+ * unknown from the counts alone.
+ */
+ private static long estimateNumRows(long totalRecords, long eqDeletes, long
posDeletes) {
+ long numRows = totalRecords - (eqDeletes > 0 ? 0 : posDeletes);
+ return numRows > 0 ? numRows : totalRecords;
+ }
+
+ private static boolean hasNoDeletes(Map<String, String> summary) {
+ if (summary == null || !summary.containsKey(TOTAL_EQ_DELETES_PROP) ||
+ !summary.containsKey(TOTAL_POS_DELETES_PROP)) {
+ return false;
+ }
+ return Long.parseLong(summary.get(TOTAL_EQ_DELETES_PROP)) +
+ Long.parseLong(summary.get(TOTAL_POS_DELETES_PROP)) == 0;
+ }
+
@Override
- public Map<String, String> computeBasicStatistics(Partish partish) {
+ public Map<String, String>
computeBasicStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
Map<String, String> stats;
if (!getStatsSource().equals(HiveMetaHook.ICEBERG)) {
- stats = partish.getPartParameters();
+ stats = hmsTable.getParameters();
if (!StatsSetupConst.areBasicStatsUptoDate(stats)) {
// populate quick-stats
- stats = getBasicStatistics(partish, true);
+ stats = getBasicStatistics(hmsTable, true);
}
return stats;
}
- org.apache.hadoop.hive.ql.metadata.Table hmsTable = partish.getTable();
// For write queries where rows got modified, don't fetch from cache as
values could have changed.
Table table = getTable(hmsTable);
Snapshot snapshot = IcebergTableUtil.getTableSnapshot(table, hmsTable);
@@ -596,37 +637,78 @@ public Map<String, String> computeBasicStatistics(Partish
partish) {
.commit();
}
}
- return getBasicStatistics(partish);
+ return getBasicStatistics(hmsTable);
}
- private static Map<String, String> getPartishSummary(Partish partish, Table
table, Snapshot snapshot) {
- if (partish.getPartition() != null) {
- Types.StructType partitionType = Partitioning.partitionType(table);
- try (CloseableIterable<PartitionStatistics> records =
-
table.newPartitionStatisticsScan().useSnapshot(snapshot.snapshotId()).scan()) {
- Iterator<PartitionStatistics> recordsIterator = records.iterator();
- if (!recordsIterator.hasNext()) {
- LOG.warn("Partition stats file not found for snapshot: {}",
snapshot.snapshotId());
- return null;
- }
- String partName = partish.getPartition().getName();
- while (recordsIterator.hasNext()) {
- PartitionStatistics stats = recordsIterator.next();
- PartitionSpec spec = table.specs().get(stats.specId());
- PartitionData data =
IcebergTableUtil.toPartitionData(stats.partition(), partitionType,
- spec.partitionType());
- if (spec.partitionToPath(data).equals(partName)) {
- return IcebergTableUtil.toStatsMap(stats);
- }
- }
+ @Override
+ public Map<String, Map<String, String>>
getAggrBasicStatsFor(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
+ List<String> partNames) {
+ if (!HiveMetaHook.ICEBERG.equals(getStatsSource())) {
+ return Map.of();
+ }
+ Table table = getTable(hmsTable);
+ Snapshot snapshot = IcebergTableUtil.getTableSnapshot(table, hmsTable);
- LOG.warn("Partition {} not found in partition stats, falling back to
metadata scan", partName);
- return IcebergTableUtil.getPartitionStats(table,
partish.getPartition().getSpec(), snapshot);
- } catch (IOException e) {
- throw new UncheckedIOException(e);
+ Map<String, Map<String, String>> result =
Maps.newHashMapWithExpectedSize(partNames.size());
+ if (snapshot == null) {
+ partNames.forEach(partName -> result.put(partName, emptyStatsMap()));
+ return result;
+ }
+ collectPartitionStatsFor(table, snapshot, partNames,
+ (partName, stats) -> result.put(partName, toStatsMap(stats)));
+ return result;
+ }
+
+ private PartitionStatistics
getPartitionStatsFor(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
+ String partName) {
+ if (!HiveMetaHook.ICEBERG.equals(getStatsSource())) {
+ return null;
+ }
+ Table table = getTable(hmsTable);
+ Snapshot snapshot = IcebergTableUtil.getTableSnapshot(table, hmsTable);
+ return snapshot != null ?
+ getOrCachePartitionStats(table, snapshot).get(partName) : null;
+ }
+
+ /**
+ * Passes each named partition's statistics to {@code statsConsumer}, served
from the snapshot's partition
+ * statistics file. Partitions missing from the file are skipped and logged.
+ */
+ private void collectPartitionStatsFor(Table table, Snapshot snapshot,
List<String> partNames,
+ BiConsumer<String, PartitionStatistics> statsConsumer) {
+ Map<String, PartitionStatistics> partitionStats =
getOrCachePartitionStats(table, snapshot);
+ int missing = 0;
+
+ for (String partName : partNames) {
+ PartitionStatistics stats = partitionStats.get(partName);
+ if (stats != null) {
+ statsConsumer.accept(partName, stats);
+ } else {
+ missing++;
}
}
- return snapshot.summary();
+ if (missing > 0) {
+ LOG.warn("{} of {} partitions not found in the partition stats file of
snapshot {}",
+ missing, partNames.size(), snapshot.snapshotId());
+ }
+ }
+
+ /**
+ * Returns the snapshot's partition statistics keyed by partition name,
empty if the snapshot has no
+ * partition statistics file. The result is cached for the rest of the query
under a key that includes
+ * the snapshot id, so repeated lookups share a single read and a new
snapshot never sees an older
+ * snapshot's entry.
+ */
+ @SuppressWarnings("unchecked")
+ private Map<String, PartitionStatistics> getOrCachePartitionStats(Table
table, Snapshot snapshot) {
+ String cacheKey = PARTITION_STATS_PREFIX + table.name() + '.' +
snapshot.snapshotId();
+ Optional<Object> cached = SessionStateUtil.getResource(conf, cacheKey);
+ if (cached.isPresent()) {
+ return (Map<String, PartitionStatistics>) cached.get();
+ }
+ Map<String, PartitionStatistics> partitionStats =
IcebergTableUtil.readPartitionStats(table, snapshot);
+ SessionStateUtil.addResource(conf, cacheKey, partitionStats);
+ return partitionStats;
}
private Table getTable(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
@@ -816,28 +898,55 @@ public AggrStats
getAggrColStatsFor(org.apache.hadoop.hive.ql.metadata.Table hms
}
@Override
- public boolean canComputeQueryUsingStats(Partish partish) {
- org.apache.hadoop.hive.ql.metadata.Table hmsTable = partish.getTable();
+ public Long getRowCount(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
if (hmsTable.getMetaTable() != null) {
- return false;
- }
- if (!getStatsSource().equals(HiveMetaHook.ICEBERG) &&
StatsSetupConst.areBasicStatsUptoDate(
- partish.getPartParameters())) {
- return true;
+ return null;
}
+ return getStatsSource().equals(HiveMetaHook.ICEBERG) ?
+ snapshotRowCount(hmsTable) : metastoreRowCount(hmsTable);
+ }
+
+ /** Row count from the HMS table parameters */
+ private static Long
metastoreRowCount(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ Map<String, String> parameters = hmsTable.getParameters();
+ return StatsSetupConst.areBasicStatsUptoDate(parameters) ?
+ NumberUtils.createLong(parameters.get(StatsSetupConst.ROW_COUNT)) :
null;
+ }
+
+ /** Row count from the snapshot summary */
+ private Long snapshotRowCount(org.apache.hadoop.hive.ql.metadata.Table
hmsTable) {
Table table = getTable(hmsTable);
Snapshot snapshot = IcebergTableUtil.getTableSnapshot(table, hmsTable);
- if (snapshot != null) {
- Map<String, String> summary = getPartishSummary(partish, table,
snapshot);
- if (summary != null && summary.containsKey(TOTAL_EQ_DELETES_PROP) &&
- summary.containsKey(TOTAL_POS_DELETES_PROP)) {
+ if (snapshot == null) {
+ return null;
+ }
+ Map<String, String> summary = snapshot.summary();
+ return hasNoDeletes(summary) ?
NumberUtils.createLong(summary.get(TOTAL_RECORDS_PROP)) : null;
+ }
- long totalEqDeletes =
Long.parseLong(summary.get(TOTAL_EQ_DELETES_PROP));
- long totalPosDeletes =
Long.parseLong(summary.get(TOTAL_POS_DELETES_PROP));
- return totalEqDeletes + totalPosDeletes == 0;
- }
+ @Override
+ public Map<String, Long>
getRowCount(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
+ List<String> partNames) {
+ if (!HiveMetaHook.ICEBERG.equals(getStatsSource())) {
+ return Map.of();
}
- return false;
+ Table table = getTable(hmsTable);
+ Snapshot snapshot = IcebergTableUtil.getTableSnapshot(table, hmsTable);
+ if (snapshot == null) {
+ return Map.of();
+ }
+ if (partNames.stream().anyMatch(DummyPartition::isVoid)) {
+ // rows that belong to no partition are never pruned, so their count may
include rows the predicate
+ // does not select
+ return Map.of();
+ }
+ Map<String, Long> rowCounts =
Maps.newHashMapWithExpectedSize(partNames.size());
+ collectPartitionStatsFor(table, snapshot, partNames, (partName, stats) -> {
+ if (stats.equalityDeleteRecordCount() == 0 &&
stats.positionDeleteRecordCount() == 0) {
+ rowCounts.put(partName, stats.dataRecordCount());
+ }
+ });
+ return rowCounts;
}
private String getStatsSource() {
@@ -2216,9 +2325,9 @@ public Partition
getPartition(org.apache.hadoop.hive.ql.metadata.Table table,
// Populate basic statistics
if (partition != null) {
- Map<String, String> stats = getBasicStatistics(Partish.buildFor(table,
partition));
- if (stats != null && !stats.isEmpty()) {
- partition.getTPartition().setParameters(stats);
+ PartitionStatistics stats = getPartitionStatsFor(table,
partition.getName());
+ if (stats != null) {
+ partition.getTPartition().setParameters(toStatsMap(stats));
}
}
@@ -2367,12 +2476,11 @@ public List<Partition>
getPartitionsByExpr(org.apache.hadoop.hive.ql.metadata.Ta
try (CloseableIterable<FileScanTask> tasks = scan.planFiles()) {
FluentIterable.from(tasks)
- .filter(task -> task.spec().isPartitioned())
.filter(task -> specFilter.test(task.file().specId()))
.forEach(task -> {
PartitionSpec spec = task.spec();
PartitionData partitionData =
IcebergTableUtil.toPartitionData(task.partition(), spec.partitionType());
- String partName = spec.partitionToPath(partitionData);
+ String partName = IcebergTableUtil.toPartitionName(spec,
partitionData);
Map<String, String> partSpecMap =
IcebergTableUtil.makeSpecFromName(partName, spec,
partitionData, defaultPartitionName);
diff --git
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java
index 6e4f50ff3e2..ad2ff8e4c2f 100644
---
a/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java
+++
b/iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/IcebergTableUtil.java
@@ -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;
@@ -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;
@@ -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;
@@ -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#VOID} 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.VOID);
+ }
+
/**
* Builds a filter expression for data table operations
(deleteFromRowFilter, FindFiles).
* Only supports identity transforms. Keys are partition field names.
@@ -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();
@@ -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()
@@ -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)
@@ -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()) {
diff --git
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStatistics.java
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStatistics.java
index fc5a2b981f9..df57578b705 100644
---
a/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStatistics.java
+++
b/iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergStatistics.java
@@ -21,13 +21,24 @@
import java.io.IOException;
import java.util.Collection;
+import java.util.Collections;
import java.util.List;
+import java.util.Map;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.StatsSetupConst;
import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.api.AggrStats;
+import org.apache.hadoop.hive.metastore.api.ColumnStatistics;
+import org.apache.hadoop.hive.metastore.api.ColumnStatisticsObj;
+import org.apache.hadoop.hive.ql.ErrorMsg;
+import org.apache.hadoop.hive.ql.metadata.DummyPartition;
+import org.apache.hadoop.hive.ql.metadata.Partition;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.iceberg.AssertHelpers;
+import org.apache.iceberg.FileFormat;
import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.PartitionStatistics;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.catalog.TableIdentifier;
@@ -358,6 +369,356 @@ public void testIcebergColStatsPath() throws IOException {
HiveIcebergTestUtils.valueForRow(HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA,
result));
}
+ @Test
+ public void testGetAggrBasicStatsForPartitioned() {
+ assumeParquetHiveCatalogIceberg();
+
+ TableIdentifier identifier = TableIdentifier.of("default", "customers");
+ createPartitionedCustomers(identifier);
+
+ org.apache.hadoop.hive.ql.metadata.Table hmsTable = hmsTable(identifier);
+ HiveIcebergStorageHandler handler = storageHandler();
+
+ List<String> partNames = partitionNames(handler, hmsTable);
+ Assert.assertEquals(3, partNames.size());
+
+ Map<String, Map<String, String>> aggr =
handler.getAggrBasicStatsFor(hmsTable, partNames);
+ Assert.assertEquals(3, aggr.size());
+
+ // each customer record lands in its own last_name partition
+ for (String partName : partNames) {
+ Map<String, String> basicStats = aggr.get(partName);
+ Assert.assertEquals("1", basicStats.get(StatsSetupConst.ROW_COUNT));
+ Assert.assertEquals("1", basicStats.get(StatsSetupConst.NUM_FILES));
+
Assert.assertTrue(Long.parseLong(basicStats.get(StatsSetupConst.TOTAL_SIZE)) >
0);
+ }
+ // no deletes: every partition's row count is answered exactly
+ Map<String, Long> rowCounts = handler.getRowCount(hmsTable, partNames);
+ Assert.assertEquals(partNames.size(), rowCounts.size());
+ rowCounts.values().forEach(rowCount ->
Assert.assertEquals(Long.valueOf(1), rowCount));
+ }
+
+ @Test
+ public void testAnalyzePartitionSpecRejected() {
+ assumeParquetHiveCatalogIceberg();
+
+ TableIdentifier identifier = TableIdentifier.of("default", "customers");
+ createPartitionedCustomers(identifier);
+ String expected = ErrorMsg.ANALYZE_PARTITION_NON_NATIVE.getMsg();
+
+ // basic statistics are maintained incrementally for all partitions as a
whole:
+ // a partition-scoped basic-stats ANALYZE cannot be honored and must be
rejected
+ AssertHelpers.assertThrows(
+ "Should reject partition-scoped basic-stats ANALYZE for non-native
partitioned tables",
+ IllegalArgumentException.class, expected,
+ () -> shell.executeStatement(
+ "ANALYZE TABLE " + identifier + " PARTITION (last_name='Brown')
COMPUTE STATISTICS")
+ );
+ // same for column statistics: the rewrite would drop every other
partition's column stats
+ AssertHelpers.assertThrows(
+ "Should reject partition-scoped column-stats ANALYZE for non-native
partitioned tables",
+ IllegalArgumentException.class, expected,
+ () -> shell.executeStatement(
+ "ANALYZE TABLE " + identifier + " PARTITION (last_name='Brown')
COMPUTE STATISTICS FOR COLUMNS")
+ );
+ }
+
+ @Test
+ public void testAnalyzeCatchesUpPartitionStats() {
+ assumeParquetHiveCatalogIceberg();
+
+ TableIdentifier identifier = TableIdentifier.of("default", "customers");
+ createPartitionedCustomers(identifier);
+
+ // a write that skips the stats auto-gathering leaves the new snapshot
without a partition stats file
+ shell.setHiveSessionValue(HiveConf.ConfVars.HIVE_STATS_AUTOGATHER.varname,
false);
+ shell.executeStatement(testTables.getInsertQuery(
+ HiveIcebergStorageHandlerTestUtils.OTHER_CUSTOMER_RECORDS_2,
identifier, false));
+
+ org.apache.hadoop.hive.ql.metadata.Table hmsTable = hmsTable(identifier);
+ HiveIcebergStorageHandler handler = storageHandler();
+ List<String> partNames = partitionNames(handler, hmsTable);
+
+ // the partition stats file is missing: every partition is reported
missing (estimated by the planner)
+ // and exact query answering is refused
+ Assert.assertEquals(0, handler.getAggrBasicStatsFor(hmsTable,
partNames).size());
+ Assert.assertTrue(handler.getRowCount(hmsTable, partNames).isEmpty());
+
+ // ANALYZE catches up: the incremental computation publishes the partition
stats file
+ shell.executeStatement("ANALYZE TABLE " + identifier + " COMPUTE
STATISTICS");
+
+ Map<String, Map<String, String>> refreshed =
handler.getAggrBasicStatsFor(hmsTable, partNames);
+ Assert.assertEquals(partNames.size(), refreshed.size());
+ refreshed.values().forEach(basicStats ->
+
Assert.assertTrue(Long.parseLong(basicStats.get(StatsSetupConst.ROW_COUNT)) >
0));
+ Assert.assertEquals(partNames.size(), handler.getRowCount(hmsTable,
partNames).size());
+ }
+
+ @Test
+ public void testGetAggrColStatsForPartitioned() throws Exception {
+ // bulk partition column statistics: a single puffin read aggregated
across the requested
+ // partitions, mirroring testGetAggrBasicStatsForPartitioned
+ assumeParquetHiveCatalogIceberg();
+
+ TableIdentifier identifier = TableIdentifier.of("default", "customers");
+ createPartitionedCustomers(identifier);
+
+ org.apache.hadoop.hive.ql.metadata.Table hmsTable = hmsTable(identifier);
+ HiveIcebergStorageHandler handler = storageHandler();
+
+ List<String> partNames = partitionNames(handler, hmsTable);
+ Assert.assertEquals(3, partNames.size());
+
+ AggrStats aggrStats = handler.getAggrColStatsFor(hmsTable,
ImmutableList.of("customer_id"), partNames);
+ Assert.assertEquals(3, aggrStats.getPartsFound());
+ Assert.assertEquals(1, aggrStats.getColStatsSize());
+ ColumnStatisticsObj statsObj = aggrStats.getColStats().get(0);
+ Assert.assertEquals("customer_id", statsObj.getColName());
+ // customer ids 0..2, one per last_name partition, merged across the three
partitions
+ Assert.assertEquals(0,
statsObj.getStatsData().getLongStats().getLowValue());
+ Assert.assertEquals(2,
statsObj.getStatsData().getLongStats().getHighValue());
+ }
+
+ @Test
+ public void testCountStarPartitioned() {
+ // end-to-end count(*) correctness on a partitioned table, for both stats
sources:
+ // iceberg source answers from the per-partition stats (exactness gate),
metastore source must
+ // not fail (regression: NumberFormatException on the missing
per-partition ROW_COUNT) and a
+ // partition-pruned count(*) must not be answered from the table-level row
count
+ assumeParquetHiveCatalog();
+
+ TableIdentifier identifier = TableIdentifier.of("default", "customers");
+ createPartitionedCustomers(identifier);
+
+ List<Object[]> result = shell.executeStatement("SELECT count(*) FROM " +
identifier);
+ Assert.assertEquals(3L, result.get(0)[0]);
+
+ result = shell.executeStatement("SELECT count(*) FROM " + identifier + "
WHERE last_name = 'Brown'");
+ Assert.assertEquals(1L, result.get(0)[0]);
+
+ if (statsSource.equals("iceberg")) {
+ // the counts must be answered from statistics: the rewritten plans
contain no table scan
+ String plan = shell.executeAndStringify("EXPLAIN SELECT count(*) FROM "
+ identifier);
+ Assert.assertFalse(plan, plan.contains("TableScan"));
+ plan = shell.executeAndStringify("EXPLAIN SELECT count(*) FROM " +
identifier + " WHERE last_name = 'Brown'");
+ Assert.assertFalse(plan, plan.contains("TableScan"));
+ }
+ }
+
+ @Test
+ public void testCountStarWithoutPartitionStatsFile() {
+ assumeParquetHiveCatalogIceberg();
+
+ TableIdentifier identifier = TableIdentifier.of("default", "customers");
+ createPartitionedCustomers(identifier, false);
+
+ Table icebergTable = testTables.loadTable(identifier);
+ Assert.assertNull(IcebergTableUtil.getPartitionStatsFile(
+ icebergTable, icebergTable.currentSnapshot().snapshotId()));
+
+ // the per-partition path needs that file, so a pruned count is answered
by a scan, while the
+ // unpruned one still comes from the table-level statistics
+ Assert.assertEquals(1L, shell.executeStatement(
+ "SELECT count(*) FROM " + identifier + " WHERE last_name =
'Brown'").get(0)[0]);
+ Assert.assertEquals(3L, shell.executeStatement("SELECT count(*) FROM " +
identifier).get(0)[0]);
+ String plan = shell.executeAndStringify("EXPLAIN SELECT count(*) FROM " +
identifier);
+ Assert.assertFalse(plan, plan.contains("TableScan"));
+ }
+
+ @Test
+ public void testRowCountWithDeletes() {
+ assumeParquetHiveCatalogIceberg();
+
+ TableIdentifier identifier = TableIdentifier.of("default", "customers");
+ createPartitionedCustomers(identifier);
+ shell.executeStatement(testTables.getInsertQuery(
+ HiveIcebergStorageHandlerTestUtils.OTHER_CUSTOMER_RECORDS_2,
identifier, false));
+ // position-delete one of the two Silver rows (the default delete mode is
copy-on-write, which would
+ // rewrite the file and leave no delete file behind)
+ shell.executeStatement("ALTER TABLE " + identifier + " SET TBLPROPERTIES
('write.delete.mode'='merge-on-read')");
+ shell.executeStatement("DELETE FROM " + identifier + " WHERE customer_id =
2 AND first_name = 'Joanna'");
+
+ org.apache.hadoop.hive.ql.metadata.Table hmsTable = hmsTable(identifier);
+ HiveIcebergStorageHandler handler = storageHandler();
+ List<String> partNames = partitionNames(handler, hmsTable);
+ Assert.assertEquals(11, partNames.size());
+
+ // the delete-covered partition's count is inexact and must be omitted;
all others stay exact
+ Map<String, Long> rowCounts = handler.getRowCount(hmsTable, partNames);
+ Assert.assertEquals(10, rowCounts.size());
+ Assert.assertFalse(rowCounts.containsKey("last_name=Silver"));
+
+ // end-to-end: the rewrite is refused for the delete-covered scope and the
counts stay correct
+ Assert.assertEquals(11L, shell.executeStatement("SELECT count(*) FROM " +
identifier).get(0)[0]);
+ Assert.assertEquals(1L,
+ shell.executeStatement("SELECT count(*) FROM " + identifier + " WHERE
last_name = 'Silver'").get(0)[0]);
+ }
+
+ @Test
+ public void testStatsAfterEvolutionFromUnpartitioned() throws Exception {
+ assumeParquetHiveCatalogIceberg();
+
+ TableIdentifier identifier = TableIdentifier.of("default", "customers");
+ createEvolvedCustomers(identifier);
+
+ // ANALYZE computes both stats families on the evolved table
+ shell.executeStatement("ANALYZE TABLE " + identifier + " COMPUTE
STATISTICS FOR COLUMNS");
+ checkColStat(identifier.name(), "customer_id", true);
+ Table icebergTable = testTables.loadTable(identifier);
+ validateBasicStats(icebergTable, "default", identifier.name());
+
+ // the partition stats file accounts for the legacy rows too, under the
empty partition tuple
+ Map<String, PartitionStatistics> fileStats =
+ IcebergTableUtil.readPartitionStats(icebergTable,
icebergTable.currentSnapshot());
+ Assert.assertEquals(3L,
fileStats.get(DummyPartition.VOID).dataRecordCount().longValue());
+
+ // column stats blobs are written for the physical partitions only: values
existing solely among the
+ // legacy unpartitioned rows (Green, Pink) get no blob
+ List<ColumnStatistics> colStats =
+ IcebergTableUtil.readColStats(icebergTable,
icebergTable.currentSnapshot().snapshotId(), null);
+ Assert.assertEquals(
+ List.of("last_name=Barna", "last_name=Brown", "last_name=Rozsaszin",
"last_name=Zold"),
+ colStats.stream().map(stats ->
stats.getStatsDesc().getPartName()).sorted().toList());
+
+ // a blob describes the physical partition's files only: the legacy Brown
row (customer_id 0) does not
+ // merge into the Brown partition's blob, which covers just the new-spec
row (customer_id 3)
+ ColumnStatisticsObj brownCustomerId = colStats.stream()
+ .filter(stats ->
"last_name=Brown".equals(stats.getStatsDesc().getPartName()))
+ .flatMap(stats -> stats.getStatsObj().stream())
+ .filter(obj -> "customer_id".equals(obj.getColName()))
+ .findFirst().orElseThrow();
+ Assert.assertEquals(3L,
brownCustomerId.getStatsData().getLongStats().getLowValue());
+ Assert.assertEquals(3L,
brownCustomerId.getStatsData().getLongStats().getHighValue());
+
+ org.apache.hadoop.hive.ql.metadata.Table hmsTable = hmsTable(identifier);
+ HiveIcebergStorageHandler handler = storageHandler();
+ List<String> partNames = partitionNames(handler, hmsTable);
+ Assert.assertEquals(4, partNames.size());
+
+ // planner basic stats answer the requested partitions and carry the
legacy rows as an extra entry
+ List<String> statNames = Lists.newArrayList(partNames);
+ statNames.add(DummyPartition.VOID);
+ Map<String, Map<String, String>> aggr =
handler.getAggrBasicStatsFor(hmsTable, statNames);
+ Assert.assertEquals(statNames.size(), aggr.size());
+ partNames.forEach(name -> Assert.assertEquals("1",
aggr.get(name).get(StatsSetupConst.ROW_COUNT)));
+ Assert.assertEquals("3",
+ aggr.get(DummyPartition.VOID).get(StatsSetupConst.ROW_COUNT));
+
+ // an unpruned scan plans with the table-level statistics; a pruned scan
sums the matched partitions
+ // and the synthetic partition, which the pruner keeps whenever a legacy
file survives file-level pruning
+ String plan = shell.executeAndStringify("EXPLAIN SELECT * FROM " +
identifier);
+ Assert.assertTrue(plan, plan.contains("rows=7"));
+ // the legacy values span Brown..Pink, so the legacy file is pruned away
for Barna
+ plan = shell.executeAndStringify("EXPLAIN SELECT * FROM " + identifier + "
WHERE last_name = 'Barna'");
+ Assert.assertTrue(plan, plan.contains("rows=1"));
+ // Brown matches both the partition and the legacy file: 1 partition row +
3 legacy rows are read
+ plan = shell.executeAndStringify("EXPLAIN SELECT * FROM " + identifier + "
WHERE last_name = 'Brown'");
+ Assert.assertTrue(plan, plan.contains("rows=4"));
+ // Green has no partition of its own, only legacy rows: the synthetic
partition alone answers
+ plan = shell.executeAndStringify("EXPLAIN SELECT * FROM " + identifier + "
WHERE last_name = 'Green'");
+ Assert.assertTrue(plan, plan.contains("rows=3"));
+ }
+
+ @Test
+ public void testRowCountAfterEvolutionFromUnpartitioned() {
+ assumeParquetHiveCatalogIceberg();
+
+ TableIdentifier identifier = TableIdentifier.of("default", "customers");
+ createEvolvedCustomers(identifier);
+
+ org.apache.hadoop.hive.ql.metadata.Table hmsTable = hmsTable(identifier);
+ HiveIcebergStorageHandler handler = storageHandler();
+ List<String> partNames = partitionNames(handler, hmsTable);
+
+ // the synthetic partition cannot answer a predicate exactly, so a pruned
list holding it is refused
+ List<String> withPseudo = Lists.newArrayList(partNames);
+ withPseudo.add(DummyPartition.VOID);
+ Assert.assertTrue(handler.getRowCount(hmsTable, withPseudo).isEmpty());
+ // without it the partition counts are exact
+ Assert.assertEquals(partNames.size(), handler.getRowCount(hmsTable,
partNames).size());
+
+ // counts stay correct via a real scan, pruned and unpruned alike; Green
exists only in the legacy
+ // rows so its pruned list is empty (partition statistics alone would have
answered 0)
+ Assert.assertEquals(7L, shell.executeStatement("SELECT count(*) FROM " +
identifier).get(0)[0]);
+ Assert.assertEquals(1L,
+ shell.executeStatement("SELECT count(*) FROM " + identifier + " WHERE
last_name = 'Green'").get(0)[0]);
+ Assert.assertEquals(1L,
+ shell.executeStatement("SELECT count(*) FROM " + identifier + " WHERE
last_name = 'Barna'").get(0)[0]);
+ // Brown is both a partition of its own and a value among the legacy rows:
the synthetic partition
+ // keeps the filter in the plan, so only the two genuine Brown rows are
returned
+ Assert.assertEquals(2L,
+ shell.executeStatement("SELECT count(*) FROM " + identifier + " WHERE
last_name = 'Brown'").get(0)[0]);
+ List<Object[]> brown = shell.executeStatement(
+ "SELECT customer_id FROM " + identifier + " WHERE last_name = 'Brown'
ORDER BY customer_id");
+ Assert.assertEquals(2, brown.size());
+ Assert.assertEquals(0L, brown.get(0)[0]);
+ Assert.assertEquals(3L, brown.get(1)[0]);
+ }
+
+ private void createEvolvedCustomers(TableIdentifier identifier) {
+ shell.setHiveSessionValue(HiveConf.ConfVars.HIVE_STATS_AUTOGATHER.varname,
true);
+ testTables.createTable(shell, identifier.name(),
HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA,
+ PartitionSpec.unpartitioned(), fileFormat, ImmutableList.of(),
formatVersion);
+ shell.executeStatement(testTables.getInsertQuery(
+ HiveIcebergStorageHandlerTestUtils.CUSTOMER_RECORDS, identifier,
false));
+ shell.executeStatement("ALTER TABLE " + identifier + " SET PARTITION SPEC
(last_name)");
+ shell.executeStatement(testTables.getInsertQuery(
+ HiveIcebergStorageHandlerTestUtils.OTHER_CUSTOMER_RECORDS_1,
identifier, false));
+ // a Brown row in the new spec: its partition's value is also present
among the legacy rows
+ shell.executeStatement("INSERT INTO " + identifier + " VALUES (3, 'Alice',
'Brown')");
+ }
+
+ /** The bulk-stats scenarios are independent of the file format and catalog;
pin one combination. */
+ private void assumeParquetHiveCatalog() {
+ Assume.assumeTrue(fileFormat == FileFormat.PARQUET && testTableType ==
TestTableType.HIVE_CATALOG);
+ }
+
+ private void assumeParquetHiveCatalogIceberg() {
+ assumeParquetHiveCatalog();
+ Assume.assumeTrue(statsSource.equals("iceberg"));
+ }
+
+ private void createPartitionedCustomers(TableIdentifier identifier) {
+ createPartitionedCustomers(identifier, true);
+ }
+
+ /** Auto-gathering on write triggers computeBasicStatistics, which publishes
the partition stats file. */
+ private void createPartitionedCustomers(TableIdentifier identifier, boolean
autoGather) {
+ PartitionSpec spec =
PartitionSpec.builderFor(HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA)
+ .identity("last_name").build();
+ shell.setHiveSessionValue(HiveConf.ConfVars.HIVE_STATS_AUTOGATHER.varname,
autoGather);
+ testTables.createTable(shell, identifier.name(),
HiveIcebergStorageHandlerTestUtils.CUSTOMER_SCHEMA, spec,
+ fileFormat, ImmutableList.of(), formatVersion);
+ shell.executeStatement(testTables.getInsertQuery(
+ HiveIcebergStorageHandlerTestUtils.CUSTOMER_RECORDS, identifier,
false));
+ }
+
+ private org.apache.hadoop.hive.ql.metadata.Table hmsTable(TableIdentifier
identifier) {
+ try {
+ return new org.apache.hadoop.hive.ql.metadata.Table(
+ shell.metastore().getTable("default", identifier.name()));
+ } catch (TException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private HiveIcebergStorageHandler storageHandler() {
+ HiveIcebergStorageHandler handler = new HiveIcebergStorageHandler();
+ handler.setConf(shell.getHiveConf());
+ return handler;
+ }
+
+ private static List<String> partitionNames(HiveIcebergStorageHandler handler,
+ org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ try {
+ return handler.getPartitions(hmsTable, Collections.emptyMap(),
true).stream()
+ .map(Partition::getName)
+ .toList();
+ } catch (SemanticException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
private void checkColStat(String tableName, String colName, boolean
accurate) {
List<Object[]> rows = shell.executeStatement("DESCRIBE " + tableName + " "
+ colName);
diff --git
a/iceberg/iceberg-handler/src/test/results/positive/iceberg_truncate_partition_with_evolution.q.out
b/iceberg/iceberg-handler/src/test/results/positive/iceberg_truncate_partition_with_evolution.q.out
index 9218f556ab1..ea652fe65e9 100644
---
a/iceberg/iceberg-handler/src/test/results/positive/iceberg_truncate_partition_with_evolution.q.out
+++
b/iceberg/iceberg-handler/src/test/results/positive/iceberg_truncate_partition_with_evolution.q.out
@@ -85,19 +85,22 @@ STAGE PLANS:
TableScan
alias: test_ice_int
filterExpr: (a = 22) (type: boolean)
- Statistics: Num rows: 1 Data size: 88 Basic stats: COMPLETE
Column stats: PARTIAL
- Select Operator
- expressions: PARTITION__SPEC__ID (type: int),
PARTITION__HASH (type: bigint), FILE__PATH (type: string), ROW__POSITION (type:
bigint), PARTITION__PROJECTION (type: string), 22 (type: int), b (type: string)
- outputColumnNames: _col0, _col1, _col2, _col3, _col4,
_col5, _col6
- Statistics: Num rows: 1 Data size: 480 Basic stats:
COMPLETE Column stats: PARTIAL
- File Output Operator
- compressed: false
- Statistics: Num rows: 1 Data size: 480 Basic stats:
COMPLETE Column stats: PARTIAL
- table:
- input format:
org.apache.iceberg.mr.hive.HiveIcebergInputFormat
- output format:
org.apache.iceberg.mr.hive.HiveIcebergOutputFormat
- serde: org.apache.iceberg.mr.hive.HiveIcebergSerDe
- name: default.test_ice_int
+ Statistics: Num rows: 3 Data size: 276 Basic stats: COMPLETE
Column stats: PARTIAL
+ Filter Operator
+ predicate: (a = 22) (type: boolean)
+ Statistics: Num rows: 3 Data size: 276 Basic stats:
COMPLETE Column stats: PARTIAL
+ Select Operator
+ expressions: PARTITION__SPEC__ID (type: int),
PARTITION__HASH (type: bigint), FILE__PATH (type: string), ROW__POSITION (type:
bigint), PARTITION__PROJECTION (type: string), 22 (type: int), b (type: string)
+ outputColumnNames: _col0, _col1, _col2, _col3, _col4,
_col5, _col6
+ Statistics: Num rows: 3 Data size: 1440 Basic stats:
COMPLETE Column stats: PARTIAL
+ File Output Operator
+ compressed: false
+ Statistics: Num rows: 3 Data size: 1440 Basic stats:
COMPLETE Column stats: PARTIAL
+ table:
+ input format:
org.apache.iceberg.mr.hive.HiveIcebergInputFormat
+ output format:
org.apache.iceberg.mr.hive.HiveIcebergOutputFormat
+ serde: org.apache.iceberg.mr.hive.HiveIcebergSerDe
+ name: default.test_ice_int
Execution mode: vectorized
Stage: Stage-3
@@ -132,6 +135,7 @@ POSTHOOK: type: QUERY
POSTHOOK: Input: default@test_ice_int
POSTHOOK: Output: hdfs://### HDFS PATH ###
11 ddd
+11 ddd
33 rrfdfdf
33 rrfdfdf
PREHOOK: query: select `partition` from default.test_ice_int.partitions
@@ -177,6 +181,7 @@ POSTHOOK: type: QUERY
POSTHOOK: Input: default@test_ice_int
POSTHOOK: Output: hdfs://### HDFS PATH ###
11 ddd
+11 ddd
PREHOOK: query: select `partition` from default.test_ice_int.partitions
PREHOOK: type: QUERY
PREHOOK: Input: default@test_ice_int
@@ -275,19 +280,22 @@ STAGE PLANS:
TableScan
alias: test_ice_bigint
filterExpr: (a = 226784902765739L) (type: boolean)
- Statistics: Num rows: 1 Data size: 88 Basic stats: COMPLETE
Column stats: PARTIAL
- Select Operator
- expressions: PARTITION__SPEC__ID (type: int),
PARTITION__HASH (type: bigint), FILE__PATH (type: string), ROW__POSITION (type:
bigint), PARTITION__PROJECTION (type: string), 226784902765739L (type: bigint),
b (type: string)
- outputColumnNames: _col0, _col1, _col2, _col3, _col4,
_col5, _col6
- Statistics: Num rows: 1 Data size: 484 Basic stats:
COMPLETE Column stats: PARTIAL
- File Output Operator
- compressed: false
- Statistics: Num rows: 1 Data size: 484 Basic stats:
COMPLETE Column stats: PARTIAL
- table:
- input format:
org.apache.iceberg.mr.hive.HiveIcebergInputFormat
- output format:
org.apache.iceberg.mr.hive.HiveIcebergOutputFormat
- serde: org.apache.iceberg.mr.hive.HiveIcebergSerDe
- name: default.test_ice_bigint
+ Statistics: Num rows: 3 Data size: 288 Basic stats: COMPLETE
Column stats: PARTIAL
+ Filter Operator
+ predicate: (a = 226784902765739L) (type: boolean)
+ Statistics: Num rows: 3 Data size: 288 Basic stats:
COMPLETE Column stats: PARTIAL
+ Select Operator
+ expressions: PARTITION__SPEC__ID (type: int),
PARTITION__HASH (type: bigint), FILE__PATH (type: string), ROW__POSITION (type:
bigint), PARTITION__PROJECTION (type: string), 226784902765739L (type: bigint),
b (type: string)
+ outputColumnNames: _col0, _col1, _col2, _col3, _col4,
_col5, _col6
+ Statistics: Num rows: 3 Data size: 1452 Basic stats:
COMPLETE Column stats: PARTIAL
+ File Output Operator
+ compressed: false
+ Statistics: Num rows: 3 Data size: 1452 Basic stats:
COMPLETE Column stats: PARTIAL
+ table:
+ input format:
org.apache.iceberg.mr.hive.HiveIcebergInputFormat
+ output format:
org.apache.iceberg.mr.hive.HiveIcebergOutputFormat
+ serde: org.apache.iceberg.mr.hive.HiveIcebergSerDe
+ name: default.test_ice_bigint
Execution mode: vectorized
Stage: Stage-3
@@ -322,6 +330,7 @@ POSTHOOK: type: QUERY
POSTHOOK: Input: default@test_ice_bigint
POSTHOOK: Output: hdfs://### HDFS PATH ###
115674892756 ddd
+115674892756 ddd
3367849937755673 rrfdfdf
3367849937755673 rrfdfdf
PREHOOK: query: select `partition` from default.test_ice_bigint.partitions
@@ -367,6 +376,7 @@ POSTHOOK: type: QUERY
POSTHOOK: Input: default@test_ice_bigint
POSTHOOK: Output: hdfs://### HDFS PATH ###
115674892756 ddd
+115674892756 ddd
PREHOOK: query: select `partition` from default.test_ice_bigint.partitions
PREHOOK: type: QUERY
PREHOOK: Input: default@test_ice_bigint
@@ -491,19 +501,22 @@ STAGE PLANS:
TableScan
alias: test_ice_str
filterExpr: (b = 'ddd') (type: boolean)
- Statistics: Num rows: 1 Data size: 8 Basic stats: COMPLETE
Column stats: PARTIAL
- Select Operator
- expressions: PARTITION__SPEC__ID (type: int),
PARTITION__HASH (type: bigint), FILE__PATH (type: string), ROW__POSITION (type:
bigint), PARTITION__PROJECTION (type: string), a (type: bigint), 'ddd' (type:
string)
- outputColumnNames: _col0, _col1, _col2, _col3, _col4,
_col5, _col6
- Statistics: Num rows: 1 Data size: 483 Basic stats:
COMPLETE Column stats: PARTIAL
- File Output Operator
- compressed: false
- Statistics: Num rows: 1 Data size: 483 Basic stats:
COMPLETE Column stats: PARTIAL
- table:
- input format:
org.apache.iceberg.mr.hive.HiveIcebergInputFormat
- output format:
org.apache.iceberg.mr.hive.HiveIcebergOutputFormat
- serde: org.apache.iceberg.mr.hive.HiveIcebergSerDe
- name: default.test_ice_str
+ Statistics: Num rows: 7 Data size: 665 Basic stats: COMPLETE
Column stats: PARTIAL
+ Filter Operator
+ predicate: (b = 'ddd') (type: boolean)
+ Statistics: Num rows: 7 Data size: 665 Basic stats:
COMPLETE Column stats: PARTIAL
+ Select Operator
+ expressions: PARTITION__SPEC__ID (type: int),
PARTITION__HASH (type: bigint), FILE__PATH (type: string), ROW__POSITION (type:
bigint), PARTITION__PROJECTION (type: string), a (type: bigint), 'ddd' (type:
string)
+ outputColumnNames: _col0, _col1, _col2, _col3, _col4,
_col5, _col6
+ Statistics: Num rows: 7 Data size: 3381 Basic stats:
COMPLETE Column stats: PARTIAL
+ File Output Operator
+ compressed: false
+ Statistics: Num rows: 7 Data size: 3381 Basic stats:
COMPLETE Column stats: PARTIAL
+ table:
+ input format:
org.apache.iceberg.mr.hive.HiveIcebergInputFormat
+ output format:
org.apache.iceberg.mr.hive.HiveIcebergOutputFormat
+ serde: org.apache.iceberg.mr.hive.HiveIcebergSerDe
+ name: default.test_ice_str
Execution mode: vectorized
Stage: Stage-3
@@ -538,7 +551,12 @@ POSTHOOK: type: QUERY
POSTHOOK: Input: default@test_ice_str
POSTHOOK: Output: hdfs://### HDFS PATH ###
115674892756 a"ab
+115674892756 a"ab
115674892756 a'ab
+115674892756 a'ab
+2267849027657399057 eefe
+2267849027657399057 eefe
+2267849027657399057 eefe
2267849027657399057 eefe
2267849027657399057 eefe
2267849027657399057 eefe
@@ -589,7 +607,12 @@ POSTHOOK: type: QUERY
POSTHOOK: Input: default@test_ice_str
POSTHOOK: Output: hdfs://### HDFS PATH ###
115674892756 a"ab
+115674892756 a"ab
115674892756 a'ab
+115674892756 a'ab
+2267849027657399057 eefe
+2267849027657399057 eefe
+2267849027657399057 eefe
2267849027657399057 eefe
2267849027657399057 eefe
2267849027657399057 eefe
@@ -629,6 +652,9 @@ POSTHOOK: Output: hdfs://### HDFS PATH ###
2267849027657399057 eefe
2267849027657399057 eefe
2267849027657399057 eefe
+2267849027657399057 eefe
+2267849027657399057 eefe
+2267849027657399057 eefe
PREHOOK: query: select `partition` from default.test_ice_str.partitions
PREHOOK: type: QUERY
PREHOOK: Input: default@test_ice_str
@@ -716,17 +742,17 @@ STAGE PLANS:
TableScan
alias: test_ice_date
filterExpr: (b = DATE'2022-02-07') (type: boolean)
- Statistics: Num rows: 1 Data size: 56 Basic stats: COMPLETE
Column stats: PARTIAL
+ Statistics: Num rows: 2 Data size: 128 Basic stats: COMPLETE
Column stats: NONE
Filter Operator
predicate: (b = DATE'2022-02-07') (type: boolean)
- Statistics: Num rows: 1 Data size: 56 Basic stats:
COMPLETE Column stats: PARTIAL
+ Statistics: Num rows: 2 Data size: 128 Basic stats:
COMPLETE Column stats: NONE
Select Operator
expressions: PARTITION__SPEC__ID (type: int),
PARTITION__HASH (type: bigint), FILE__PATH (type: string), ROW__POSITION (type:
bigint), PARTITION__PROJECTION (type: string), a (type: bigint),
DATE'2022-02-07' (type: date)
outputColumnNames: _col0, _col1, _col2, _col3, _col4,
_col5, _col6
- Statistics: Num rows: 1 Data size: 444 Basic stats:
COMPLETE Column stats: PARTIAL
+ Statistics: Num rows: 2 Data size: 128 Basic stats:
COMPLETE Column stats: NONE
File Output Operator
compressed: false
- Statistics: Num rows: 1 Data size: 444 Basic stats:
COMPLETE Column stats: PARTIAL
+ Statistics: Num rows: 2 Data size: 128 Basic stats:
COMPLETE Column stats: NONE
table:
input format:
org.apache.iceberg.mr.hive.HiveIcebergInputFormat
output format:
org.apache.iceberg.mr.hive.HiveIcebergOutputFormat
@@ -904,19 +930,22 @@ STAGE PLANS:
TableScan
alias: test_ice_double
filterExpr: (a = 1.156748927566759E11D) (type: boolean)
- Statistics: Num rows: 1 Data size: 56 Basic stats: COMPLETE
Column stats: PARTIAL
- Select Operator
- expressions: PARTITION__SPEC__ID (type: int),
PARTITION__HASH (type: bigint), FILE__PATH (type: string), ROW__POSITION (type:
bigint), PARTITION__PROJECTION (type: string), 1.156748927566759E11D (type:
double), b (type: date)
- outputColumnNames: _col0, _col1, _col2, _col3, _col4,
_col5, _col6
- Statistics: Num rows: 1 Data size: 452 Basic stats:
COMPLETE Column stats: PARTIAL
- File Output Operator
- compressed: false
- Statistics: Num rows: 1 Data size: 452 Basic stats:
COMPLETE Column stats: PARTIAL
- table:
- input format:
org.apache.iceberg.mr.hive.HiveIcebergInputFormat
- output format:
org.apache.iceberg.mr.hive.HiveIcebergOutputFormat
- serde: org.apache.iceberg.mr.hive.HiveIcebergSerDe
- name: default.test_ice_double
+ Statistics: Num rows: 3 Data size: 192 Basic stats: COMPLETE
Column stats: PARTIAL
+ Filter Operator
+ predicate: (a = 1.156748927566759E11D) (type: boolean)
+ Statistics: Num rows: 3 Data size: 192 Basic stats:
COMPLETE Column stats: PARTIAL
+ Select Operator
+ expressions: PARTITION__SPEC__ID (type: int),
PARTITION__HASH (type: bigint), FILE__PATH (type: string), ROW__POSITION (type:
bigint), PARTITION__PROJECTION (type: string), 1.156748927566759E11D (type:
double), b (type: date)
+ outputColumnNames: _col0, _col1, _col2, _col3, _col4,
_col5, _col6
+ Statistics: Num rows: 3 Data size: 1356 Basic stats:
COMPLETE Column stats: PARTIAL
+ File Output Operator
+ compressed: false
+ Statistics: Num rows: 3 Data size: 1356 Basic stats:
COMPLETE Column stats: PARTIAL
+ table:
+ input format:
org.apache.iceberg.mr.hive.HiveIcebergInputFormat
+ output format:
org.apache.iceberg.mr.hive.HiveIcebergOutputFormat
+ serde: org.apache.iceberg.mr.hive.HiveIcebergSerDe
+ name: default.test_ice_double
Execution mode: vectorized
Stage: Stage-3
@@ -951,6 +980,7 @@ POSTHOOK: type: QUERY
POSTHOOK: Input: default@test_ice_double
POSTHOOK: Output: hdfs://### HDFS PATH ###
2.267849027657399E9 2022-08-07
+2.267849027657399E9 2022-08-07
3.367849937755674E7 2022-08-09
3.367849937755674E7 2022-08-09
PREHOOK: query: select `partition` from default.test_ice_double.partitions
@@ -996,6 +1026,7 @@ POSTHOOK: type: QUERY
POSTHOOK: Input: default@test_ice_double
POSTHOOK: Output: hdfs://### HDFS PATH ###
2.267849027657399E9 2022-08-07
+2.267849027657399E9 2022-08-07
PREHOOK: query: select `partition` from default.test_ice_double.partitions
PREHOOK: type: QUERY
PREHOOK: Input: default@test_ice_double
@@ -1081,17 +1112,17 @@ STAGE PLANS:
TableScan
alias: test_ice_double_date
filterExpr: ((a = 1.156748927566759E11D) and (b =
DATE'2022-02-07')) (type: boolean)
- Statistics: Num rows: 1 Data size: 56 Basic stats: COMPLETE
Column stats: PARTIAL
+ Statistics: Num rows: 2 Data size: 128 Basic stats: COMPLETE
Column stats: NONE
Filter Operator
predicate: ((a = 1.156748927566759E11D) and (b =
DATE'2022-02-07')) (type: boolean)
- Statistics: Num rows: 1 Data size: 56 Basic stats:
COMPLETE Column stats: PARTIAL
+ Statistics: Num rows: 2 Data size: 128 Basic stats:
COMPLETE Column stats: NONE
Select Operator
expressions: PARTITION__SPEC__ID (type: int),
PARTITION__HASH (type: bigint), FILE__PATH (type: string), ROW__POSITION (type:
bigint), PARTITION__PROJECTION (type: string), 1.156748927566759E11D (type:
double), DATE'2022-02-07' (type: date)
outputColumnNames: _col0, _col1, _col2, _col3, _col4,
_col5, _col6
- Statistics: Num rows: 1 Data size: 452 Basic stats:
COMPLETE Column stats: PARTIAL
+ Statistics: Num rows: 2 Data size: 128 Basic stats:
COMPLETE Column stats: NONE
File Output Operator
compressed: false
- Statistics: Num rows: 1 Data size: 452 Basic stats:
COMPLETE Column stats: PARTIAL
+ Statistics: Num rows: 2 Data size: 128 Basic stats:
COMPLETE Column stats: NONE
table:
input format:
org.apache.iceberg.mr.hive.HiveIcebergInputFormat
output format:
org.apache.iceberg.mr.hive.HiveIcebergOutputFormat
diff --git a/iceberg/iceberg-handler/src/test/results/positive/row_count.q.out
b/iceberg/iceberg-handler/src/test/results/positive/row_count.q.out
index 4303ea510c7..0ff1c86f210 100644
--- a/iceberg/iceberg-handler/src/test/results/positive/row_count.q.out
+++ b/iceberg/iceberg-handler/src/test/results/positive/row_count.q.out
@@ -249,8 +249,8 @@ Stage-0
PARTITION_ONLY_SHUFFLE [RS_9]
Group By Operator [GBY_8] (rows=1 width=8)
Output:["_col0"],aggregations:["count()"]
- Select Operator [SEL_7] (rows=14 width=5###)
- TableScan [TS_0] (rows=14 width=5###)
+ Select Operator [SEL_7] (rows=11 width=1###)
+ TableScan [TS_0] (rows=11 width=1###)
default@llap_orders,llap_orders,Tbl:COMPLETE,Col:COMPLETE
PREHOOK: query: explain insert into llap_orders select * from llap_orders
limit 100000
@@ -284,30 +284,30 @@ Stage-3
<-Reducer 2 [SIMPLE_EDGE] vectorized
SHUFFLE [RS_21]
PartitionCols:_col4, _col5
- Select Operator [SEL_20] (rows=14 width=420)
+ Select Operator [SEL_20] (rows=11 width=420)
Output:["_col0","_col1","_col2","_col3","_col4","_col5"]
- Limit [LIM_19] (rows=14 width=420)
+ Limit [LIM_19] (rows=11 width=420)
Number of rows:100000
<-Map 1 [CUSTOM_SIMPLE_EDGE] vectorized
PARTITION_ONLY_SHUFFLE [RS_18]
- Select Operator [SEL_17] (rows=14 width=420)
+ Select Operator [SEL_17] (rows=11 width=420)
Output:["_col0","_col1","_col2","_col3","_col4","_col5"]
- Limit [LIM_16] (rows=14 width=420)
+ Limit [LIM_16] (rows=11 width=420)
Number of rows:100000
- TableScan [TS_0] (rows=14 width=420)
+ TableScan [TS_0] (rows=11 width=420)
default@llap_orders,llap_orders,Tbl:COMPLETE,Col:NONE,Output:["orderid","quantity","itemid","tradets","p1","p2"]
Reducer 4 vectorized
File Output Operator [FS_29]
- Select Operator [SEL_28] (rows=7 width=420)
+ Select Operator [SEL_28] (rows=5 width=420)
Output:["_col0","_col1","_col2","_col3","_col4","_col5","_col6","_col7","_col8","_col9","_col10","_col11","_col12","_col13","_col14","_col15","_col16","_col17","_col18","_col19","_col20","_col21","_col22","_col23","_col24","_col25","_col26","_col27","_col28","_col29","_col30","_col31","_col32","_col33","_col34","_col35","_col36"]
- Group By Operator [GBY_27] (rows=7 width=420)
+ Group By Operator [GBY_27] (rows=5 width=420)
Output:["_col0","_col1","_col2","_col3","_col4","_col5","_col6","_col7","_col8","_col9","_col10","_col11","_col12","_col13","_col14","_col15","_col16","_col17","_col18","_col19","_col20","_col21","_col22","_col23","_col24","_col25","_col26"],aggregations:["min(VALUE._col0)","max(VALUE._col1)","count(VALUE._col2)","count(VALUE._col3)","compute_bit_vector_hll(VALUE._col4)","min(VALUE._col5)","max(VALUE._col6)","count(VALUE._col7)","compute_bit_vector_hll(VALUE._col8)","
[...]
<-Reducer 2 [SIMPLE_EDGE] vectorized
SHUFFLE [RS_24]
PartitionCols:_col0, _col1
- Group By Operator [GBY_23] (rows=14 width=420)
+ Group By Operator [GBY_23] (rows=11 width=420)
Output:["_col0","_col1","_col2","_col3","_col4","_col5","_col6","_col7","_col8","_col9","_col10","_col11","_col12","_col13","_col14","_col15","_col16","_col17","_col18","_col19","_col20","_col21","_col22","_col23","_col24","_col25","_col26"],aggregations:["min(orderid)","max(orderid)","count(1)","count(orderid)","compute_bit_vector_hll(orderid)","min(quantity)","max(quantity)","count(quantity)","compute_bit_vector_hll(quantity)","min(itemid)","max(itemid)","count(
[...]
- Select Operator [SEL_22] (rows=14 width=420)
+ Select Operator [SEL_22] (rows=11 width=420)
Output:["orderid","quantity","itemid","tradets","p1","p2"]
Please refer to the previous Select Operator
[SEL_20]
diff --git
a/iceberg/iceberg-handler/src/test/results/positive/truncate_partitioned_iceberg_table.q.out
b/iceberg/iceberg-handler/src/test/results/positive/truncate_partitioned_iceberg_table.q.out
index 36456ba85fc..526707f5ada 100644
---
a/iceberg/iceberg-handler/src/test/results/positive/truncate_partitioned_iceberg_table.q.out
+++
b/iceberg/iceberg-handler/src/test/results/positive/truncate_partitioned_iceberg_table.q.out
@@ -109,9 +109,11 @@ Table Parameters:
#### A masked pattern was here ####
metadata_location hdfs://### HDFS PATH ###
numFiles 4
+ numPartitions 4
numRows 9
parquet.compression zstd
previous_metadata_location hdfs://### HDFS PATH ###
+ rawDataSize 0
schema.name-mapping.default [ {
\"field-id\" : 1,
\"names\" : [ \"a\" ]
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/ddl/table/info/desc/DescTableOperation.java
b/ql/src/java/org/apache/hadoop/hive/ql/ddl/table/info/desc/DescTableOperation.java
index 8d45892e17f..c9d89a2f427 100644
---
a/ql/src/java/org/apache/hadoop/hive/ql/ddl/table/info/desc/DescTableOperation.java
+++
b/ql/src/java/org/apache/hadoop/hive/ql/ddl/table/info/desc/DescTableOperation.java
@@ -54,7 +54,6 @@
import org.apache.hadoop.hive.ql.metadata.TableConstraintsInfo;
import org.apache.hadoop.hive.ql.plan.ColStatistics;
import org.apache.hadoop.hive.ql.session.SessionState;
-import org.apache.hadoop.hive.ql.stats.Partish;
import org.apache.hadoop.hive.ql.stats.StatsUtils;
import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.hadoop.hive.serde2.Deserializer;
@@ -152,8 +151,7 @@ private void getColumnsNoColumnPath(Table table, Partition
partition, List<Field
if (table.isNonNative()) {
HiveStorageHandler sh = table.getStorageHandler();
- sh.getBasicStatistics(Partish.buildFor(table)).forEach((k, v) ->
- valueMap.put(k, Longs.tryParse(v)));
+ sh.getBasicStatistics(table).forEach((k, v) -> valueMap.put(k,
Longs.tryParse(v)));
numParts = sh.getPartitionNames(table).size();
} else {
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/DummyPartition.java
b/ql/src/java/org/apache/hadoop/hive/ql/metadata/DummyPartition.java
index 9f871d05feb..5209879e66a 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/DummyPartition.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/DummyPartition.java
@@ -36,6 +36,16 @@
*/
public class DummyPartition extends Partition {
+ /**
+ * Name of the synthetic partition holding the rows that belong to no
partition, e.g. data written
+ * before the table was partitioned. Its partition spec is empty, so no
predicate can exclude it.
+ */
+ public static final String VOID = "__NO_PARTITION__";
+
+ public static boolean isVoid(String partName) {
+ return VOID.equals(partName);
+ }
+
private String name;
private LinkedHashMap<String, String> partSpec;
private List<String> values;
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveStorageHandler.java
b/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveStorageHandler.java
index c50f0b33b05..ae0528bc207 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveStorageHandler.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/HiveStorageHandler.java
@@ -70,7 +70,6 @@
import org.apache.hadoop.hive.ql.plan.TableDesc;
import
org.apache.hadoop.hive.ql.security.authorization.HiveAuthorizationProvider;
import
org.apache.hadoop.hive.ql.security.authorization.HiveCustomStorageHandlerUtils;
-import org.apache.hadoop.hive.ql.stats.Partish;
import org.apache.hadoop.hive.serde2.AbstractSerDe;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.JobConf;
@@ -248,19 +247,19 @@ default Map<String, String>
getOperatorDescProperties(OperatorDesc operatorDesc,
/**
* Returns basic statistics (numRows, numFiles, totalSize) calculated by the
underlying storage handler
* implementation.
- * @param partish table/partition wrapper object
+ * @param hmsTable table object
* @return map of basic statistics
*/
- default Map<String, String> getBasicStatistics(Partish partish) {
+ default Map<String, String>
getBasicStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return null;
}
/**
- * Compute basic statistics (numRows, numFiles, totalSize) for the given
table/partition.
- * @param partish table/partition wrapper object
+ * Compute basic statistics (numRows, numFiles, totalSize) for the given
table.
+ * @param hmsTable table object
* @return map of basic statistics
*/
- default Map<String, String> computeBasicStatistics(Partish partish) {
+ default Map<String, String>
computeBasicStatistics(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
return null;
}
@@ -309,12 +308,24 @@ default List<ColumnStatisticsObj>
getColStatistics(org.apache.hadoop.hive.ql.met
* @param colNames list of column names
* @param partNames list of partition names
* @return AggrStats object
- */
- default AggrStats
getAggrColStatsFor(org.apache.hadoop.hive.ql.metadata.Table table, List<String>
colNames,
+ */
+ default AggrStats
getAggrColStatsFor(org.apache.hadoop.hive.ql.metadata.Table table, List<String>
colNames,
List<String> partNames) throws MetaException {
return null;
}
+ /**
+ * Fetches basic statistics (numRows, numFiles, totalSize) for the supplied
partitions in a single batch.
+ * @param hmsTable table object
+ * @param partNames list of partition names
+ * @return map of partition name to basic statistics, omitting partitions
that have none;
+ * empty if the handler does not provide them
+ */
+ default Map<String, Map<String, String>>
getAggrBasicStatsFor(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
+ List<String> partNames) {
+ return Map.of();
+ }
+
/**
* Set column stats for non-native tables
* @param table table object
@@ -344,26 +355,27 @@ default boolean
canSetColStatistics(org.apache.hadoop.hive.ql.metadata.Table tab
}
/**
- * Check if the storage handler can answer a few queries like count(1)
purely using statistics.
- * @param partish table/partition wrapper object
- * @return true if the storage handler can answer query using statistics
+ * Returns the row count of the table, letting queries like count(1) be
answered from statistics.
+ * @param hmsTable table object
+ * @return the row count, or null if it cannot be determined exactly
*/
- default boolean canComputeQueryUsingStats(Partish partish) {
- return false;
+ default Long getRowCount(org.apache.hadoop.hive.ql.metadata.Table hmsTable) {
+ return null;
}
/**
- * Check if the storage handler can answer a few queries like count(1)
purely using statistics.
- * @param table table wrapper object
- *
- * @deprecated since 4.0.1, will be removed in 5.0.0,
- * use {@link #canComputeQueryUsingStats(Partish)} instead.
+ * Returns the row count of the given partitions, omitting the ones whose
count cannot be determined
+ * exactly, e.g. because they are covered by delete files.
+ * @param hmsTable table object
+ * @param partNames list of partition names
+ * @return map of partition name to row count, empty if the handler cannot
determine partition
+ * row counts at all, e.g. when some rows do not belong to any
partition
*/
- @Deprecated
- default boolean
canComputeQueryUsingStats(org.apache.hadoop.hive.ql.metadata.Table table) {
- return canComputeQueryUsingStats(Partish.buildFor(table));
+ default Map<String, Long>
getRowCount(org.apache.hadoop.hive.ql.metadata.Table hmsTable,
+ List<String> partNames) {
+ return Map.of();
}
-
+
/**
*
* Gets the storage format descriptor to be used for temp table for LOAD
data.
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/StatsOptimizer.java
b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/StatsOptimizer.java
index c3eacc59bd2..35cb5fd774f 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/StatsOptimizer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/StatsOptimizer.java
@@ -51,9 +51,11 @@
import org.apache.hadoop.hive.ql.lockmgr.LockException;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.Table;
import org.apache.hadoop.hive.ql.parse.ParseContext;
+import org.apache.hadoop.hive.ql.parse.PrunedPartitionList;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.plan.AggregationDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
@@ -88,7 +90,6 @@
import java.util.Map;
import java.util.Set;
import java.util.Stack;
-import java.util.stream.Collectors;
/** There is a set of queries which can be answered entirely from statistics
stored in metastore.
@@ -933,29 +934,60 @@ private Collection<List<ColumnStatisticsObj>>
verifyAndGetPartColumnStats(
}
private Long getRowCnt(TableScanOperator tsOp, Table tbl) throws
HiveException {
- long rowCnt = 0L;
+ if (tbl.isNonNative()) {
+ return getRowCntFromStorageHandler(tsOp, tbl);
+ }
final List<Partish> partishList;
- if (tbl.isPartitioned() &&
StatsUtils.checkCanProvidePartitionStats(tbl)) {
+ if (tbl.isPartitioned()) {
partishList = pctx.getPrunedPartitions(tsOp.getConf().getAlias(),
tsOp).getPartitions().stream()
.map(Partish::buildFor)
- .collect(Collectors.toList());
+ .toList();
} else {
partishList = Lists.newArrayList(Partish.buildFor(tbl));
}
+ long rowCnt = 0L;
for (Partish partish : partishList) {
Map<String, String> basicStats = partish.getPartParameters();
- if (tbl.isNonNative()) {
- if (!tbl.getStorageHandler().canComputeQueryUsingStats(partish)) {
- return null;
- }
- basicStats = tbl.getStorageHandler().getBasicStatistics(partish);
- } else if
(!StatsUtils.areBasicStatsUptoDateForQueryAnswering(partish.getTable(),
partish.getPartParameters())) {
+ if
(!StatsUtils.areBasicStatsUptoDateForQueryAnswering(partish.getTable(),
basicStats)) {
return null;
}
- long partRowCnt =
Long.parseLong(basicStats.get(StatsSetupConst.ROW_COUNT));
- rowCnt += partRowCnt;
+ rowCnt += Long.parseLong(basicStats.get(StatsSetupConst.ROW_COUNT));
}
return rowCnt;
}
+
+ private Long getRowCntFromStorageHandler(TableScanOperator tsOp, Table
tbl) throws HiveException {
+ if (tbl.getMetaTable() != null) {
+ // metadata table scans cannot be answered from the data table's
statistics
+ return null;
+ }
+ HiveStorageHandler storageHandler = tbl.getStorageHandler();
+ if (tbl.isPartitioned()) {
+ PrunedPartitionList prunedList =
pctx.getPrunedPartitions(tsOp.getConf().getAlias(), tsOp);
+ if (!prunedList.getReferredPartCols().isEmpty()) {
+ // a partition predicate may have pruned the list to a subset, so
only the
+ // per-partition statistics can answer exactly
+ if (!StatsUtils.checkCanProvidePartitionStats(tbl)) {
+ return null;
+ }
+ List<String> partNames = prunedList.getPartitions().stream()
+ .map(Partition::getName)
+ .toList();
+ if (partNames.isEmpty()) {
+ // the predicate matched nothing the scan would read
+ return 0L;
+ }
+ Map<String, Long> rowCounts = storageHandler.getRowCount(tbl,
partNames);
+ if (rowCounts.size() != partNames.size()) {
+ // the rewrite substitutes a constant, so it needs an exact count
for every partition
+ return null;
+ }
+ return rowCounts.values().stream().mapToLong(Long::longValue).sum();
+ }
+ }
+ // unpruned or unpartitioned: the table-level statistics are exactly the
aggregate of every
+ // partition - answered without reading the partition stats file
+ return storageHandler.getRowCount(tbl);
+ }
}
}
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java
b/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java
index 434e131b4d0..7d83425aff8 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/BaseSemanticAnalyzer.java
@@ -1726,6 +1726,23 @@ private static boolean getPartExprNodeDesc(ASTNode
astNode, HiveConf conf,
return result;
}
+ /**
+ * Rejects a query that has a partition clause but targets a table that is
never partitioned on the HMS-level.
+ * Even though the table is not partitioned from the HMS's point of view, it
might have some other notion of
+ * partitioning under the hood (e.g. Iceberg tables). In these cases, we
might decide to proactively throw a more
+ * descriptive, unified error message instead of failing on some other
semantic analysis validation step, which
+ * could provide a more counter-intuitive exception message.
+ *
+ * @param tbl The table object, should not be null.
+ * @param partitionClausePresent Whether a partition clause is present in
the query (e.g. PARTITION(p='x'))
+ */
+ protected static void validateUnsupportedPartitionClause(Table tbl, boolean
partitionClausePresent)
+ throws SemanticException {
+ if (partitionClausePresent && tbl.hasNonNativePartitionSupport()) {
+ throw new
SemanticException(ErrorMsg.ANALYZE_PARTITION_NON_NATIVE.getMsg());
+ }
+ }
+
/**
* Get the partition specs from the tree.
*
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/parse/ColumnStatsSemanticAnalyzer.java
b/ql/src/java/org/apache/hadoop/hive/ql/parse/ColumnStatsSemanticAnalyzer.java
index 88b4a4f0e9f..e952632d73b 100644
---
a/ql/src/java/org/apache/hadoop/hive/ql/parse/ColumnStatsSemanticAnalyzer.java
+++
b/ql/src/java/org/apache/hadoop/hive/ql/parse/ColumnStatsSemanticAnalyzer.java
@@ -629,6 +629,11 @@ public void analyze(ASTNode ast, Context origCtx) throws
SemanticException {
originalTree = ast;
boolean isPartitionStats =
AnalyzeCommandUtils.isPartitionLevelStats(ast)
|| StatsUtils.isPartitionStats(tbl, conf);
+ // a partition-scoped column statistics ANALYZE cannot be honored for
non-native tables: the stats
+ // of all partitions are rewritten as a whole, so it would drop every
other partition's statistics
+ // (table-level ANALYZE of a partitioned table still computes all
partitions - only the explicit
+ // partition spec is rejected; the auto-gather path merges instead and
stays unaffected)
+ validateUnsupportedPartitionClause(tbl,
AnalyzeCommandUtils.isPartitionLevelStats(ast));
Map<Integer, List<TransformSpec>> partTransformSpecs =
Collections.singletonMap(-1, null);
Map<String, String> partSpec = (isPartitionStats) ?
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/parse/PrunedPartitionList.java
b/ql/src/java/org/apache/hadoop/hive/ql/parse/PrunedPartitionList.java
index 3329d4ae749..0a4caa08c38 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/PrunedPartitionList.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/PrunedPartitionList.java
@@ -40,7 +40,10 @@ public class PrunedPartitionList {
/** Partitions that either satisfy the partition criteria, or may satisfy
it. */
private final Set<Partition> partitions;
- /** partition columns referred by pruner expr */
+ /**
+ * Partition columns referred by the pruner expr; empty when no partition
predicate narrowed the list,
+ * or when the predicate was constant false - the list is empty in that case
as well.
+ */
private final List<String> referred;
/** Whether there are partitions in the list that may or may not satisfy the
criteria. */
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
index b728bd85402..7042cff5387 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
@@ -2472,6 +2472,10 @@ private void getMetaData(QB qb, ReadEntity parentInput)
if (qb.getParseInfo().isAnalyzeCommand()) {
// allow partial partition specification for nonscan since noscan is
fast.
TableSpec ts = new TableSpec(db, conf, (ASTNode) ast.getChild(0),
true, this.noscan);
+ // basic statistics of non-native partitioned tables are maintained
incrementally for all
+ // partitions as a whole, so a partition-scoped ANALYZE cannot be
honored
+ validateUnsupportedPartitionClause(tab, ts.specType !=
SpecType.TABLE_ONLY);
+
if (ts.specType == SpecType.DYNAMIC_PARTITION) { // dynamic partitions
try {
ts.partitions = db.getPartitionsByNames(ts.tableHandle,
ts.partSpec);
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStats.java
b/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStats.java
index 507d5607aa7..780089310d1 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStats.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStats.java
@@ -25,7 +25,6 @@
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -65,7 +64,11 @@ public void addEnhancer(IStatsEnhancer enhancer) {
}
public BasicStats build(Partish p) {
- BasicStats ret = new BasicStats(p);
+ return build(p, null);
+ }
+
+ public BasicStats build(Partish p, Map<String, String> providedStats) {
+ BasicStats ret = new BasicStats(p, providedStats);
for (IStatsEnhancer enhancer : enhancers) {
ret.apply(enhancer);
}
@@ -73,39 +76,27 @@ public BasicStats build(Partish p) {
}
public List<BasicStats> buildAll(HiveConf conf, Collection<Partish> parts)
{
- LOG.info("Number of partishes : " + parts.size());
+ LOG.info("Number of partishes : {}", parts.size());
- final List<BasicStats> ret = new ArrayList<>(parts.size());
if (parts.size() <= 1) {
- for (Partish partish : parts) {
- ret.add(build(partish));
- }
- return ret;
+ return parts.stream().map(this::build).toList();
}
-
- List<Future<BasicStats>> futures = new ArrayList<>();
-
int threads =
conf.getIntVar(ConfVars.METASTORE_FS_HANDLER_THREADS_COUNT);
-
- final ExecutorService pool;
- if (threads <= 1) {
- pool = MoreExecutors.newDirectExecutorService();
- } else {
- pool = Executors.newFixedThreadPool(threads, new
ThreadFactoryBuilder().setDaemon(true).setNameFormat("Get-Partitions-Size-%d").build());
- }
-
- for (final Partish part : parts) {
- futures.add(pool.submit(new Callable<BasicStats>() {
- @Override
- public BasicStats call() throws Exception {
- return build(part);
- }
- }));
- }
-
+ ExecutorService pool = threads <= 1 ?
MoreExecutors.newDirectExecutorService() :
+ Executors.newFixedThreadPool(threads,
+ new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat("Get-Partitions-Size-%d")
+ .build());
+
+ List<Future<BasicStats>> futures = parts.stream()
+ .map(part -> pool.submit(() -> build(part)))
+ .toList();
+
+ List<BasicStats> ret = new ArrayList<>(parts.size());
try {
- for (int i = 0; i < futures.size(); i++) {
- ret.add(i, futures.get(i).get());
+ for (Future<BasicStats> future : futures) {
+ ret.add(future.get());
}
} catch (InterruptedException | ExecutionException e) {
LOG.warn("Exception in processing files ", e);
@@ -219,6 +210,7 @@ private long getFileSizeForPath(Path path) throws
IOException {
}
private Partish partish;
+ private Map<String, String> params;
private long rowCount;
private long totalSize;
@@ -230,8 +222,20 @@ private long getFileSizeForPath(Path path) throws
IOException {
private Statistics.State state;
public BasicStats(Partish p) {
+ this(p, null);
+ }
+
+ /**
+ * Creates the basic stats for the given partish.
+ *
+ * @param p the table/partition wrapper object
+ * @param providedStats pre-fetched basic statistics, e.g. obtained from the
storage handler in batch via
+ * {@link
org.apache.hadoop.hive.ql.metadata.HiveStorageHandler#getAggrBasicStatsFor};
+ * when null, the stats are read from the partish parameters
+ */
+ public BasicStats(Partish p, Map<String, String> providedStats) {
partish = p;
- tryGetBasicStatsFromStorageHandler();
+ params = providedStats != null ? providedStats :
partish.getPartParameters();
rowCount = parseLong(StatsSetupConst.ROW_COUNT);
rawDataSize = parseLong(StatsSetupConst.RAW_DATA_SIZE);
@@ -272,18 +276,17 @@ public BasicStats(List<BasicStats> partStats) {
}
- private void tryGetBasicStatsFromStorageHandler() {
- if (partish.getTable() != null && partish.getTable().isNonNative() &&
- partish.getTable().getStorageHandler().canProvideBasicStatistics()) {
- partish.getPartParameters().putAll(
- partish.getTable().getStorageHandler().getBasicStatistics(partish));
- }
- }
-
public long getNumRows() {
return currentNumRows;
}
+ /**
+ * @return the row count parsed from the stats source, before any
enhancer-based estimation
+ */
+ public long getRawNumRows() {
+ return rowCount;
+ }
+
public long getDataSize() {
return currentDataSize;
}
@@ -317,7 +320,6 @@ protected long getRawDataSize() {
}
private long parseLong(String fieldName) {
- Map<String, String> params = partish.getPartParameters();
long result = -1;
if (params != null) {
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsNoJobTask.java
b/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsNoJobTask.java
index defcd586098..24305b1ffb7 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsNoJobTask.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsNoJobTask.java
@@ -174,17 +174,22 @@ public HiveStorageHandlerStatCollector(Partish partish) {
public void run() {
try {
Table table = partish.getTable();
- Map<String, String> parameters;
- Map<String, String> basicStatistics =
table.getStorageHandler().getBasicStatistics(partish);
if (partish.getPartition() != null) {
- parameters = partish.getPartParameters();
- result = new Partition(table,
partish.getPartition().getTPartition());
- } else {
- parameters = table.getParameters();
- result = new Table(table.getTTable());
+ // the storage handler maintains the statistics of all partitions as
a whole; a partition-scoped
+ // ANALYZE is rejected at compile time
(ErrorMsg.ANALYZE_PARTITION_NON_NATIVE)
+ throw new IllegalStateException("Partition-scoped statistics
collection is not supported for "
+ + table.getFullyQualifiedName());
+ }
+ Map<String, String> parameters = table.getParameters();
+ // computes and publishes the statistics the storage handler maintains
(e.g. the Iceberg
+ // partition stats file) before reading them back: ANALYZE catches up
on writes that
+ // skipped the auto-gathering (a no-op when the statistics are already
current)
+ Map<String, String> basicStatistics =
table.getStorageHandler().computeBasicStatistics(table);
+ result = new Table(table.getTTable());
+ if (basicStatistics != null) {
+ parameters.putAll(basicStatistics);
+ StatsSetupConst.setBasicStatsState(parameters, StatsSetupConst.TRUE);
}
- parameters.putAll(basicStatistics);
- StatsSetupConst.setBasicStatsState(parameters, StatsSetupConst.TRUE);
String who = (partish.getPartition() == null) ? ("table " +
partish.getTable().getFullyQualifiedName())
: ("partition " + partish.getPartition().getName());
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsTask.java
b/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsTask.java
index 722b2f70119..41d152a49e8 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsTask.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/stats/BasicStatsTask.java
@@ -138,8 +138,11 @@ public BasicStatsProcessor(Partish partish, BasicStatsWork
work, boolean followe
this.followedColStats1 = followedColStats2;
Table table = partish.getTable();
- if (table.isNonNative() &&
table.getStorageHandler().canProvideBasicStatistics()) {
- this.providedBasicStats =
table.getStorageHandler().computeBasicStatistics(partish);
+ // table scope only: a partition-scoped ANALYZE is rejected at compile
time for tables with
+ // non-native partitioning (ErrorMsg.ANALYZE_PARTITION_NON_NATIVE)
+ if (table.isNonNative() &&
table.getStorageHandler().canProvideBasicStatistics()
+ && partish.getPartition() == null) {
+ this.providedBasicStats =
table.getStorageHandler().computeBasicStatistics(table);
this.skipStatsUpdate = StatsSetupConst.STATS_REQUIRE_COMPUTE.stream()
.anyMatch(providedBasicStats::containsKey);
}
@@ -515,7 +518,7 @@ private List<Partition> getPartitionsList(Hive db) throws
HiveException {
// ANALYZE command
TableSpec tblSpec = work.getTableSpecs();
table = tblSpec.tableHandle;
- if (!table.isPartitioned()) {
+ if (!table.isPartitioned() || table.hasNonNativePartitionSupport()) {
return null;
}
// get all partitions that match with the partition spec
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsUtils.java
b/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsUtils.java
index e539f0ab9c4..dbb601d4c99 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsUtils.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/stats/StatsUtils.java
@@ -35,9 +35,7 @@
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
-import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.hadoop.fs.FileSystem;
@@ -63,6 +61,7 @@
import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.metadata.Hive;
import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.metadata.PartitionIterable;
import org.apache.hadoop.hive.ql.metadata.Table;
@@ -89,7 +88,6 @@
import org.apache.hadoop.hive.ql.udf.generic.GenericUDFBridge;
import org.apache.hadoop.hive.ql.udf.generic.NDV;
import org.apache.hadoop.hive.ql.util.JavaDataModel;
-import org.apache.hadoop.hive.ql.util.NamedForkJoinWorkerThreadFactory;
import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.hadoop.hive.serde2.objectinspector.ConstantObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
@@ -151,17 +149,6 @@ public class StatsUtils {
// Range upper limit for timestamp type when not defined (seconds,
heuristic): '2030-12-31 23:59:59'
private static final long TIMESTAMP_RANGE_UPPER_LIMIT = 1924991999L;
- private static final ForkJoinPool statsForkJoinPool = new ForkJoinPool(
- Runtime.getRuntime().availableProcessors(),
- new NamedForkJoinWorkerThreadFactory("basic-stats-"),
- getUncaughtExceptionHandler(),
- false
- );
-
- private static Thread.UncaughtExceptionHandler getUncaughtExceptionHandler()
{
- return (t, e) -> LOG.error(String.format("Thread %s exited with error",
t.getName()), e);
- }
-
/**
* Collect table, partition and column level statistics
* @param conf
@@ -175,7 +162,8 @@ private static Thread.UncaughtExceptionHandler
getUncaughtExceptionHandler() {
* @return statistics object
* @throws HiveException
*/
- public static Statistics collectStatistics(HiveConf conf,
PrunedPartitionList partList, ColumnStatsList colStatsCache,
+ public static Statistics collectStatistics(HiveConf conf,
PrunedPartitionList partList,
+ ColumnStatsList colStatsCache,
Table table, TableScanOperator tableScanOperator) throws HiveException {
// column level statistics are required only for the columns that are
needed
@@ -183,11 +171,13 @@ public static Statistics collectStatistics(HiveConf conf,
PrunedPartitionList pa
List<String> neededColumns = tableScanOperator.getNeededColumns();
List<String> referencedColumns = tableScanOperator.getReferencedColumns();
- return collectStatistics(conf, partList, table, schema, neededColumns,
colStatsCache, referencedColumns);
+ return collectStatistics(conf, partList, table, schema, neededColumns,
colStatsCache,
+ referencedColumns);
}
private static Statistics collectStatistics(HiveConf conf,
PrunedPartitionList partList,
- Table table, List<ColumnInfo> schema, List<String> neededColumns,
ColumnStatsList colStatsCache,
+ Table table, List<ColumnInfo> schema, List<String> neededColumns,
+ ColumnStatsList colStatsCache,
List<String> referencedColumns) throws HiveException {
boolean fetchColStats =
@@ -195,8 +185,8 @@ private static Statistics collectStatistics(HiveConf conf,
PrunedPartitionList p
boolean testMode =
HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_IN_TEST);
- return collectStatistics(conf, partList, table, schema, neededColumns,
colStatsCache, referencedColumns,
- fetchColStats, testMode);
+ return collectStatistics(conf, partList, table, schema, neededColumns,
colStatsCache,
+ referencedColumns, fetchColStats, testMode);
}
/**
@@ -207,38 +197,73 @@ private static Statistics collectStatistics(HiveConf
conf, PrunedPartitionList p
public static long getNumRows(HiveConf conf, List<ColumnInfo> schema, Table
table, PrunedPartitionList partitionList,
AtomicInteger noColsMissingStats) {
- List<Partish> inputs = new ArrayList<>();
- if (table.isPartitioned()) {
- for (Partition part : partitionList.getNotDeniedPartns()) {
- inputs.add(Partish.buildFor(table, part));
- }
- } else {
- inputs.add(Partish.buildFor(table));
- }
-
Factory basicStatsFactory = new BasicStats.Factory();
if (HiveConf.getBoolVar(conf, ConfVars.HIVE_STATS_ESTIMATE_STATS)) {
basicStatsFactory.addEnhancer(new BasicStats.DataSizeEstimator(conf));
basicStatsFactory.addEnhancer(new
BasicStats.RowNumEstimator(estimateRowSizeFromSchema(conf, schema)));
}
-
- for (Partish pi : inputs) {
- BasicStats bStats = new BasicStats(pi);
- long nr = bStats.getNumRows();
- // FIXME: this point will be lost after the factory; check that it's
really a warning....cleanup/etc
- if (nr <= 0) {
- // log warning if row count is missing
- noColsMissingStats.getAndIncrement();
- }
+
+ // when partition-level statistics are unavailable (e.g. non-native table
with an external stats
+ // source) fall back to the table-level statistics rather than
per-partition minimums
+ List<BasicStats> results;
+ if (table.isPartitioned() && checkCanProvidePartitionStats(table)) {
+ results = buildPartitionStats(conf, table, partitionList,
basicStatsFactory);
+ } else {
+ results = List.of(buildTableStats(table, basicStatsFactory));
}
- List<BasicStats> results = basicStatsFactory.buildAll(conf, inputs);
+ // count the entries with missing row counts (estimated rows do not count
as provided)
+ noColsMissingStats.addAndGet((int) results.stream()
+ .filter(bStats -> bStats.getRawNumRows() <= 0)
+ .count());
BasicStats aggregateStat = BasicStats.buildFrom(results);
aggregateStat.apply(new BasicStats.SetMinRowNumber());
return aggregateStat.getNumRows();
}
+ /**
+ * Builds the per-partition basic stats. When the storage handler provides
them, one batched read
+ * (see {@link HiveStorageHandler#getAggrBasicStatsFor}) serves the whole
partition list; otherwise each
+ * partition is read individually (see {@link BasicStats.Factory#buildAll}).
+ */
+ private static List<BasicStats> buildPartitionStats(HiveConf conf, Table
table, PrunedPartitionList partList,
+ BasicStats.Factory factory) {
+ List<Partish> inputs = partList.getNotDeniedPartns().stream()
+ .map(part -> Partish.buildFor(table, part))
+ .toList();
+ HiveStorageHandler storageHandler = table.isNonNative() ?
table.getStorageHandler() : null;
+ if (storageHandler != null && storageHandler.canProvideBasicStatistics()) {
+ if (partList.getReferredPartCols().isEmpty() && !inputs.isEmpty()) {
+ // no partition predicate: a non-empty list covers every partition, so
the table-level statistics
+ // are exactly their aggregate - skip the per-partition read
+ return List.of(buildTableStats(table, factory));
+ }
+ List<String> partNames = inputs.stream()
+ .map(partish -> partish.getPartition().getName())
+ .toList();
+ Map<String, Map<String, String>> aggrBasicStats = partNames.isEmpty() ?
Map.of() :
+ storageHandler.getAggrBasicStatsFor(table, partNames);
+ if (!aggrBasicStats.isEmpty()) {
+ return inputs.stream()
+ .map(pi -> factory.build(pi,
+ aggrBasicStats.getOrDefault(pi.getPartition().getName(),
Map.of())))
+ .toList();
+ }
+ }
+ return factory.buildAll(conf, inputs);
+ }
+
+ /**
+ * Builds the table-level basic stats, sourced from the storage handler when
it provides them.
+ */
+ private static BasicStats buildTableStats(Table table, BasicStats.Factory
factory) {
+ HiveStorageHandler storageHandler = table.isNonNative() ?
table.getStorageHandler() : null;
+ Map<String, String> providedStats = storageHandler != null &&
storageHandler.canProvideBasicStatistics() ?
+ storageHandler.getBasicStatistics(table) : null;
+ return factory.build(Partish.buildFor(table), providedStats);
+ }
+
/**
* Estimates column statistics for columns specified in {@code
neededColumnNames}
* that do not already have statistics in the {@code existingColStats} list.
@@ -269,7 +294,8 @@ static List<ColStatistics> estimateStatsForMissingCols(
}
public static Statistics collectStatistics(HiveConf conf,
PrunedPartitionList partList,
- Table table, List<ColumnInfo> schema, List<String> neededColumns,
ColumnStatsList colStatsCache,
+ Table table, List<ColumnInfo> schema, List<String> neededColumns,
+ ColumnStatsList colStatsCache,
List<String> referencedColumns, boolean needColStats)
throws HiveException {
return collectStatistics(conf, partList, table, schema, neededColumns,
colStatsCache,
@@ -277,7 +303,8 @@ public static Statistics collectStatistics(HiveConf conf,
PrunedPartitionList pa
}
private static Statistics collectStatistics(HiveConf conf,
PrunedPartitionList partList, Table table,
- List<ColumnInfo> schema, List<String> neededColumns, ColumnStatsList
colStatsCache,
+ List<ColumnInfo> schema, List<String> neededColumns,
+ ColumnStatsList colStatsCache,
List<String> referencedColumns, boolean needColStats, boolean
failIfCacheMiss) throws HiveException {
Statistics stats = null;
@@ -298,7 +325,7 @@ private static Statistics collectStatistics(HiveConf conf,
PrunedPartitionList p
basicStatsFactory.addEnhancer(new
BasicStats.RowNumEstimator(estimateRowSizeFromSchema(conf, schema)));
basicStatsFactory.addEnhancer(new BasicStats.SetMinRowNumber());
- BasicStats basicStats = basicStatsFactory.build(Partish.buildFor(table));
+ BasicStats basicStats = buildTableStats(table, basicStatsFactory);
// long nr = getNumRows(conf, schema, neededColumns, table, ds);
long ds = basicStats.getDataSize();
@@ -339,16 +366,7 @@ private static Statistics collectStatistics(HiveConf conf,
PrunedPartitionList p
basicStatsFactory.addEnhancer(new
BasicStats.RowNumEstimator(estimateRowSizeFromSchema(conf, schema)));
- List<BasicStats> partStats = null;
- try {
- partStats = statsForkJoinPool.submit(() ->
- partList.getNotDeniedPartns().parallelStream().
- map(p -> basicStatsFactory.build(Partish.buildFor(table,
p))).
- collect(Collectors.toList())
- ).get();
- } catch (Exception e) {
- throw new HiveException(e);
- }
+ List<BasicStats> partStats = buildPartitionStats(conf, table, partList,
basicStatsFactory);
BasicStats bbs = BasicStats.buildFrom(partStats);
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/stats/TestBasicStats.java
b/ql/src/test/org/apache/hadoop/hive/ql/stats/TestBasicStats.java
index cb98259dbce..9fba3eeaac8 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/stats/TestBasicStats.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/stats/TestBasicStats.java
@@ -29,7 +29,6 @@
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.ql.metadata.Partition;
import org.apache.hadoop.hive.ql.plan.Statistics.State;
-import org.apache.hadoop.hive.ql.stats.BasicStats;
import org.junit.Ignore;
import org.junit.Test;
import org.mockito.Mockito;
@@ -65,6 +64,33 @@ public Partish buildPartition() {
}
}
+ @Test
+ public void testProvidedStatsTakePrecedenceOverPartitionParameters() {
+ Map<String, String> partParams = new HashMap<>();
+ partParams.put(StatsSetupConst.ROW_COUNT, "7");
+ Partish partish = partitionPartish(partParams);
+
+ // Stats provided explicitly (e.g. batched via
HiveStorageHandler#getAggrBasicStatsFor):
+ // used as-is, no partition parameter reads or mutation.
+ Map<String, String> providedStats = new HashMap<>();
+ providedStats.put(StatsSetupConst.ROW_COUNT, "42");
+
+ assertEquals(42, new BasicStats(partish, providedStats).getNumRows());
+ assertEquals("7", partParams.get(StatsSetupConst.ROW_COUNT));
+
+ // Without provided stats, the partition parameters are read.
+ assertEquals(7, new BasicStats(partish).getNumRows());
+ }
+
+ private static Partish partitionPartish(Map<String, String> params) {
+ Partition partition = Mockito.mock(Partition.class);
+ org.apache.hadoop.hive.metastore.api.Partition tPartition =
+ Mockito.mock(org.apache.hadoop.hive.metastore.api.Partition.class);
+ doReturn(tPartition).when(partition).getTPartition();
+ doReturn(params).when(tPartition).getParameters();
+ return Partish.buildFor(null, partition);
+ }
+
@Test
public void testDataSizeEstimator() {
Partish p1 = new LocalPartishBuilder().totalSize(10).buildPartition();
