Copilot commented on code in PR #6716:
URL: https://github.com/apache/hive/pull/6716#discussion_r4010554387
##########
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergStorageHandler.java:
##########
@@ -511,8 +511,9 @@ private Map<String, String>
getBasicStatistics(org.apache.hadoop.hive.ql.metadat
stats = emptyStatsMap();
} else if (!HiveMetaHook.ICEBERG.equals(getStatsSource()) && !quickStats &&
- hmsTable.getSnapshotRef() == null) {
- // the metastore parameters describe the table, not a branch: use the
snapshot's counters
+ hmsTable.getQualifier().isEmpty()) {
Review Comment:
This new qualifier check sends metadata-table scans into the
snapshot-summary branch, but `getRowCount` explicitly refuses to answer
metadata tables because their snapshots describe the base data table. The
`TOTAL_RECORDS` copied here therefore reports base-table rows for
`.files`/`.entries`/`.manifests`, not rows of the metadata relation, which
gives callers and the optimizer incorrect basic statistics. Return no
statistics for metadata tables before converting the snapshot summary.
##########
ql/src/java/org/apache/hadoop/hive/ql/optimizer/StatsOptimizer.java:
##########
@@ -899,39 +590,198 @@ else if (udaf instanceof GenericUDAFCount) {
}
}
- private ColumnStatisticsData
validateSingleColStat(List<ColumnStatisticsObj> statObj) {
- if (statObj.size() > 1) {
- Logger.error("More than one stat for a single column!");
- return null;
- } else if (statObj.isEmpty()) {
- Logger.debug("No stats for some partition and column");
- return null;
- }
- return statObj.get(0).getStatsData();
+ /** The columns the aggregates read, which are the ones statistics have to
be fetched for. */
+ private static List<String> aggregateColumns(GroupByOperator pgbyOp,
Map<String, ExprNodeDesc> exprMap) {
+ return pgbyOp.getConf().getAggregators().stream()
+ .filter(aggr -> !aggr.getParameters().isEmpty())
+ .map(aggr -> aggr.getParameters().get(0))
+ .filter(ExprNodeColumnDesc.class::isInstance)
+ .map(desc -> exprMap.get(((ExprNodeColumnDesc) desc).getColumn()))
+ .filter(ExprNodeColumnDesc.class::isInstance)
+ .map(desc -> ((ExprNodeColumnDesc) desc).getColumn())
+ .distinct()
+ .collect(Collectors.toList());
}
- private Collection<List<ColumnStatisticsObj>> verifyAndGetPartColumnStats(
- Hive hive, Table tbl, String colName, Set<Partition> parts) throws
TException, LockException {
- List<String> partNames = new ArrayList<String>(parts.size());
- for (Partition part : parts) {
- if
(!StatsUtils.areColumnStatsUptoDateForQueryAnswering(part.getTable(),
part.getParameters(), colName)) {
- Logger.debug("Stats for part : " + part.getSpec() + " column " +
colName
+ /**
+ * The statistics of the columns a scan's aggregates read, fetched once
when the first
+ * aggregate needs them and shared by the rest. An aggregate this rewrite
cannot answer leaves
+ * the query for execution, whole or not at all. Answers for a scan of a
partitioned table.
+ */
+ private static final class ScanColStats {
+ private final Hive hive;
+ private final Table tbl;
+ private final List<String> colNames;
+ private final PrunedPartitionList prunedList;
+ private Map<String, ColumnStatisticsObj> colStatsByName;
+ private boolean fetched;
+
+ ScanColStats(Hive hive, Table tbl, List<String> colNames,
PrunedPartitionList prunedList) {
+ this.hive = hive;
+ this.tbl = tbl;
+ this.colNames = colNames;
+ this.prunedList = prunedList;
+ }
+
+ /**
+ * One column's statistics. A scan pruned to no partitions reads no
rows, and the statistics
+ * of no rows are the empty ones: nothing counted, and no least or
greatest to name.
+ */
+ ColumnStatisticsData statsFor(String colName, StatType type) throws
HiveException {
+ if (prunedList != null && prunedList.getPartitions().isEmpty()) {
+ return emptyColStats(type);
+ }
+ if (!fetched) {
+ fetched = true;
+ colStatsByName = prunedList == null ? tableColStats() :
partitionColStats();
+ }
+ ColumnStatisticsObj stat = colStatsByName == null ? null :
colStatsByName.get(colName);
+ if (stat == null) {
+ Logger.debug("No stats for " + tbl.getTableName() + " column " +
colName);
+ return null;
+ }
+ return stat.getStatsData();
+ }
+
+ /**
+ * Whether the table's own statistics answer for this scan: it keeps
them for the table as
+ * a whole, and the scan reads every partition. They then describe
exactly the rows read.
+ */
+ private boolean answeredByTableStats() {
+ return !StatsUtils.isPartitionStats(tbl, hive.getConf()) &&
+ prunedList.getReferredPartCols().isEmpty() &&
!prunedList.hasUnknownPartitions();
+ }
+
+ /** The table's own statistics, taken only while they still describe it.
*/
+ private Map<String, ColumnStatisticsObj> tableColStats() throws
HiveException {
+ if (!StatsUtils.areColumnStatsUptoDateForQueryAnswering(tbl,
tbl.getParameters(), colNames)) {
+ Logger.debug("Stats for table : " + tbl.getTableName() + " columns "
+ colNames
+ " are not up to date.");
return null;
}
- partNames.add(part.getName());
+ return indexByColumnName(hive.getTableColumnStatistics(tbl, colNames,
true));
+ }
+
+ /** What the scan's partitions hold for every column asked about, or
null to decline. */
+ private Map<String, ColumnStatisticsObj> partitionColStats() throws
HiveException {
+ Set<Partition> parts = prunedList.getPartitions();
+ List<String> partNames = new ArrayList<>(parts.size());
+ // a storage handler holds no partition parameters, and one kept per
partition describes no
+ // partition in particular: whether each still describes itself is
answered by the aggregate
+ // below, which is told the partitions this query pruned to
+ if (tbl.isNonNative()) {
+ if (!StatsUtils.checkCanProvideColumnStats(tbl)) {
+ Logger.debug("Table : " + tbl.getTableName() + " provides no
column statistics.");
+ return null;
+ }
+ if (answeredByTableStats()) {
+ return tableColStats();
+ }
+ parts.forEach(part -> partNames.add(part.getName()));
+ } else {
+ for (Partition part : parts) {
+ if (!StatsUtils.areColumnStatsUptoDateForQueryAnswering(
+ part.getTable(), part.getParameters(), colNames)) {
+ Logger.debug("Stats for part : " + part.getSpec() + " columns "
+ colNames
+ + " are not up to date.");
+ return null;
+ }
+ partNames.add(part.getName());
+ }
+ }
+ // Aggregated rather than per partition: the callers fold these with
min, max or a sum, so
+ // merging first gives the same answer. A handler aggregates its own
statistics, which
+ // the metastore cannot hold: PART_COL_STATS rows need a partition
Iceberg never creates.
+ AggrStats aggrStats;
+ try {
+ aggrStats = tbl.isNonNative()
+ ? tbl.getStorageHandler().getAggrColStatsFor(tbl, colNames,
partNames)
+ : exactAggrColStats(partNames);
+ } catch (MetaException e) {
+ throw new HiveException(e);
+ }
+ if (aggrStats == null || aggrStats.getColStats() == null) {
+ Logger.debug("No stats for " + tbl.getTableName() + " columns " +
colNames);
+ return null;
+ }
+ if (aggrStats.getPartsFound() != parts.size()) {
+ // a partition whose statistics are missing would leave the answer
describing a subset
+ Logger.debug("Received " + aggrStats.getPartsFound() + " stats for "
+ parts.size() + " partitions");
+ return null;
+ }
+ return indexByColumnName(aggrStats.getColStats());
+ }
+
+ /**
+ * Each partition fetched and folded the way a storage handler folds its
own: the
+ * metastore's aggregate endpoint may serve a cached aggregate of a
different partition
+ * set within its variance, which estimates a plan fine but must not
answer a query.
+ */
+ private AggrStats exactAggrColStats(List<String> partNames) throws
HiveException, MetaException {
+ Map<String, List<ColumnStatisticsObj>> statsByPart =
hive.getPartitionColumnStatistics(
+ tbl.getDbName(), tbl.getTableName(), partNames, colNames, true);
+ List<ColumnStatistics> partStats = new ArrayList<>();
+ statsByPart.forEach((partitionName, statsObjs) -> {
+ // a partition counts as found only when it holds every column asked
about
+ if (statsObjs.size() == colNames.size()) {
+ ColumnStatisticsDesc statsDesc = new ColumnStatisticsDesc(false,
tbl.getDbName(), tbl.getTableName());
+ statsDesc.setPartName(partitionName);
+ partStats.add(new ColumnStatistics(statsDesc, statsObjs));
+ }
+ });
+ HiveConf conf = hive.getConf();
+ List<ColumnStatisticsObj> aggregated =
MetaStoreServerUtils.aggrPartitionStats(partStats,
+ MetaStoreUtils.getDefaultCatalog(conf), tbl.getDbName(),
tbl.getTableName(),
+ partNames, colNames,
+ partStats.size() == partNames.size(),
+ MetastoreConf.getBoolVar(conf,
MetastoreConf.ConfVars.STATS_NDV_DENSITY_FUNCTION),
+ MetastoreConf.getDoubleVar(conf,
MetastoreConf.ConfVars.STATS_NDV_TUNER));
+ return new AggrStats(aggregated, partStats.size());
+ }
+
+ /**
+ * The statistics by the column they describe. A source naming one
column twice disagrees with
+ * itself: collecting without a merge function throws, and the query
leaves for execution
+ * rather than an arbitrary one of them standing as an exact answer.
+ */
+ private static Map<String, ColumnStatisticsObj> indexByColumnName(
+ List<ColumnStatisticsObj> colStats) {
+ return colStats.stream().collect(
+ Collectors.toMap(ColumnStatisticsObj::getColName,
Function.identity()));
+ }
+ }
+
+ /** The rows a COUNT reads, or null to decline - logged. */
+ private Long countFor(AggregationDesc aggr, Map<String, ExprNodeDesc>
exprMap, long rowCnt,
+ ScanColStats scanColStats) throws HiveException {
+ if (aggr.getParameters().isEmpty()) {
+ // count(*) or count()
+ return rowCnt;
+ }
+ ExprNodeDesc param = aggr.getParameters().get(0);
+ if (param instanceof ExprNodeColumnDesc column) {
+ param = exprMap.get(column.getColumn());
+ }
+ if (param instanceof ExprNodeConstantDesc constant) {
+ // count(1) reads every row, count(null) none
+ return constant.getValue() == null ? 0L : rowCnt;
+ }
+ // count(col): the rows where it is set
+ ExprNodeColumnDesc desc = (ExprNodeColumnDesc) param;
+ String colName = desc.getColumn();
+ StatType type = getType(desc.getTypeString());
+
+ ColumnStatisticsData statData = scanColStats.statsFor(colName, type);
+ if (statData == null) {
+ return null; // logging inside
}
- AcidUtils.TableSnapshot tableSnapshot =
- AcidUtils.getTableSnapshot(hive.getConf(), tbl);
-
- Map<String, List<ColumnStatisticsObj>> result =
hive.getMSC().getPartitionColumnStatistics(
- tbl.getDbName(), tbl.getTableName(), partNames,
Lists.newArrayList(colName),
- Constants.HIVE_ENGINE, tableSnapshot != null ?
tableSnapshot.getValidWriteIdList() : null);
- if (result.size() != parts.size()) {
- Logger.debug("Received " + result.size() + " stats for " +
parts.size() + " partitions");
+ Long nullCnt = getNullCountFor(type, statData);
+ if (nullCnt == null) {
Review Comment:
`ColumnStatisticsData` uses a negative `numNulls` (for example `-1`) to mean
that the null count is unknown, but this branch treats it as a real count.
`count(col)` would therefore return `rowCnt + 1` instead of declining the
rewrite (e.g. for a 100-row table with unknown nulls it returns 101). Reject
negative null counts here before subtracting.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
