[WIP] HIVE-29834: Iceberg: Answer min/max/count aggregates from column statistics - #6716
Open
deniskuzZ wants to merge 8 commits into
Open
[WIP] HIVE-29834: Iceberg: Answer min/max/count aggregates from column statistics#6716deniskuzZ wants to merge 8 commits into
deniskuzZ wants to merge 8 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.
… statistics 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. getPartsFound() replaces the previous size check, so a partition without statistics still stops the optimization. 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 and the canProvideBasicStatistics guard, so other handlers are unchanged, and the Iceberg override defers to the metastore when hive.iceberg.stats.source says the statistics live there.
verifyAndGetPartColumnStats asked a partitioned non-native table whether its column statistics were up to date, which a table keeping them per partition cannot answer: they describe no partition in particular. It now 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 already 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.
deniskuzZ
force-pushed
the
col_stats_stats_opt
branch
from
August 23, 2026 06:26
27fc967 to
d0e8fea
Compare
|
deniskuzZ
changed the title
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?