[WIP] HIVE-29834: Iceberg: Answer min/max/count aggregates from column statistics - #6716
Open
deniskuzZ wants to merge 11 commits into
Open
[WIP] HIVE-29834: Iceberg: Answer min/max/count aggregates from column statistics#6716deniskuzZ wants to merge 11 commits into
deniskuzZ wants to merge 11 commits into
Conversation
…berg virtual column plumbing Adds VirtualColumn.PARTITION_NAME, materialized by the Iceberg readers: vectorized via HiveBatchIterator, row mode via PartitionInfo -> IOContext -> MapOperator. Refactors the read path to take everything the file already carries from the scan task instead of the per-record constants map: spec id, partition hash, file path, partition name, first row id and file sequence number are computed once per task in both VirtualColumnAwareIterator and HiveBatchIterator. Drops the now unread _partition column from the ACID read schema, along with the write-only VIRTUAL_COLS_TO_META_COLS map and the per-record helpers it fed. Fixes VectorizedParquetRecordReader taking a column type from the job column list while indexing it with a batch column number: the two orders diverge past the data columns because the batch drops virtual columns the readers cannot fill.
…ead of a virtual column The partition key a merge task carries is read back only by HiveIcebergDeleteWriter under isMergeTask: an ordinary delete derives the key from the row data, and a merge task cannot because it reads delete files, whose records hold only a path and a position. Declaring it as a virtual column put an always empty string into every DELETE, UPDATE and MERGE record and into every Iceberg plan. It now lives in a merge specific serde layout that IcebergMergeRecordReader and HiveIcebergSerDe select on the merge task flag, so the ordinary delete record is one column shorter and the plans no longer estimate it.
…tatistics Adds hive.iceberg.stats.collect.partlevel (default false), scoping column statistics granularity for Iceberg tables independently of the generic hive.analyze.stmt.collect.partlevel.stats. Partition-level statistics are produced by ANALYZE; every other write maintains table-level ones. Keys the column-statistics read path on the calling API rather than on session config, so a granularity mismatch cannot surface a wrongly-shaped blob. Stores branch-scoped statistics on the branch's snapshot and stops branch writes from overwriting the table-scoped metastore row, including the footer scan path used by ANALYZE ... COMPUTE STATISTICS. Lets a whole-table INSERT OVERWRITE of a partitioned table replace its column statistics, which previously skipped because whole-table and partition-scoped overwrites are indistinguishable in the snapshot summary.
Derive partition names through the Iceberg spec so a write keys statistics the way a read looks them up; substitute per partition for a scoped write or ANALYZE; gather partition-level statistics for CTAS.
A statistics file records the snapshot it was written for, and every blob in it was stamped with that same snapshot. An entry the merge carried forward from an earlier file therefore looked measured by the write that stored it: after a full ANALYZE, a write reaching two partitions, and an ANALYZE naming one of them, the other's numbers still came from before that write yet claimed the snapshot after it. Stamp each blob with the snapshot its numbers were computed at. Puffin carries the field per blob already, so nothing about the format changes and existing files stay readable. checkAndMergeColStats tells writeColStats which entries it carried, and those keep the anchor they arrived with, while what this write measured is anchored at it. A read then judges each partition on its own anchor. partitionsChangedSince walks back from the current snapshot as far as the oldest anchor among the partitions asked about, collecting the ones the writes in between reached and skipping the ones that preserve rows, and getAggrColStatsFor leaves out the partitions those writes left behind. The walk reads manifests, and every column of a query asks the same of them, so its answer is cached for the query. It reports every partition changed when it cannot trace them - an expired snapshot breaks the ancestry, or a file names a partition of a spec the statistics never described - which is the verdict the whole table had before. A compaction changes no rows, so its statistics stand only for what it read whole: the whole table, or one named partition of a table that keeps them per partition. The rewrite policy alone does not say which - clearing the partitions an older spec left behind carries PARTITION too, while rewriting into partitions that hold rows it never read - so the decision reads four facts: whether it is a compaction, whether it skipped files by size, and which of the two scopes it was pointed at. What it may not store it no longer measures.
deniskuzZ
force-pushed
the
col_stats_stats_opt
branch
from
August 23, 2026 06:26
27fc967 to
d0e8fea
Compare
asf-ci-hive
added
tests pending
tests unstable
and removed
tests unstable
tests pending
labels
deniskuzZ
changed the title
Statistics stored for the table as a whole were served whatever had happened to it since, because the read walked to the nearest ancestor that had any. A query planned on them was told they were complete, so after an UPDATE or a DELETE the planner folded and estimated on the numbers of a table that no longer existed. The metastore withholds its own on the same terms: get_table_statistics_req returns nothing for statistics that do not describe the write id asked about, and get_partitions_statistics_req skips such a partition rather than passing it off as current. An Iceberg table now answers the same way - what an intervening write left behind is not served, and a read of an unpartitioned table after a DML falls back to estimates, as it does for a transactional one. An aggregate of the partitions that were asked about is a different case: the ones it found still describe themselves, and only their number tells the planner it covers a subset. Statistics assembled from fewer partitions than the scan reads are marked PARTIAL, which is the state a partitioned table already carries when only some of its partitions have been analyzed.
…them An entry carried across writes kept the snapshot it was measured at, so a read had to walk the manifests back to it to learn whether anything had reached its partition since. After an ANALYZE of one partition the file sits on the current snapshot while the entries beside it are anchored wherever they were computed, which could be hundreds of commits back, and every query paid for that walk. The merge is the one place that can settle it once: it already reads what it carries, so it now asks whether each entry still holds and anchors the ones that do at the write it is making. A read then walks only what happened after the file it serves, which for a table analyzed at all recently is nothing. What no longer holds keeps the snapshot it was measured at, which is what marks it behind, and the walk is bounded for reads by hive.iceberg.stats.max.snapshot.lookback: beyond it, entries measured further back are treated as describing their partition no longer, while ones measured within it are still placed. A write is not bounded - what it records stands for every read after it, so giving up there would leave those entries unplaceable for good.
…istics StatsOptimizer fetched partition column statistics straight from the metastore, which cannot hold them for an Iceberg table: PART_COL_STATS rows require a PARTITIONS row that Iceberg never creates. Aggregates over a partitioned Iceberg table therefore always executed, even with statistics available in Puffin files. Fetch through Hive.getAggrColStatsFor, which consults the storage handler, mirroring what getRowCnt already does for row counts. The statistics arrive aggregated over the pruned partitions rather than one entry per partition; the callers folded them with min, max or a sum, so merging first gives the same answer and the folds collapse. Accuracy is asked of the component that owns it. The metastore's marker only records Hive's own writes, while an Iceberg table may be written by other engines, so areColumnStatsUptoDate is added to HiveStorageHandler and answered from table metadata by the Iceberg handler. The default keeps the metastore marker, so other handlers are unchanged, and the Iceberg override defers to the metastore when hive.iceberg.stats.source says the statistics live there. A table that keeps statistics per partition has no table-wide verdict to give: they describe no partition in particular. The partitioned non-native path therefore asks only whether the table provides column statistics at all, and the aggregate - the one place told the partitions the scan pruned to - reports how many of them it found, which is what stops the rewrite from describing a subset. count(col) needs a row count as well as the column's null count, and read it from partition parameters, which a storage handler does not keep, so it gave up on every partitioned non-native table before reaching the column statistics at all. It now asks getRowCnt, which consults the handler and is given the pruned partitions; a count no handler can determine exactly, such as one covering delete files, still stops the rewrite.
…ler's statistics An unpartitioned table's aggregates were fetched with the metastore client directly, which for a table whose statistics a handler keeps returns nothing: ColStatsProcessor writes the values to the handler and only the accuracy marker to the metastore. The accuracy check above the fetch therefore guarded a fetch that could not have served them, and min, max and count over an unpartitioned Iceberg table always read the data. Fetch through Hive.getTableColumnStatistics, which asks the handler when it can provide, mirroring what the partitioned path already does. A native table takes the same call it took before, resolving its snapshot and write id list the same way. The accuracy check now guards statistics that exist, which is what stops a write another engine made from being answered from numbers that predate it. The third of the three fetches took the first element without checking that there was one; a handler may legitimately return none.
The statistics of one partition carry every column of it, and the fetch was made once per aggregate, so a query over a thousand columns opened the statistics file a thousand times and discarded all but one column of each partition it had just deserialised. Collect the columns the aggregates read before walking them and fetch once for all of them, leaving each aggregate a lookup. A native table keeps its check of each column against each partition's own marker; only the fetch is shared.
deniskuzZ
force-pushed
the
col_stats_stats_opt
branch
from
August 23, 2026 11:05
d0e8fea to
be100ec
Compare
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Why are the changes needed?
Does this PR introduce any user-facing change?
How was this patch tested?
depends on #6707