[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.
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
deniskuzZ
force-pushed
the
col_stats_stats_opt
branch
from
August 23, 2026 11:05
d0e8fea to
be100ec
Compare
asf-ci-hive
added
tests pending
tests unstable
and removed
tests unstable
tests pending
labels
…each write in one place A partition scoped gather ran one query per partition spec, each grouped by that spec's own transforms. The rows of every spec already carry the name of the partition they belong to, so one scan grouped by that name gathers all of them, and the map of specs it took to drive the old shape is gone. What a write then does to the stored statistics - replace them, merge into them, or leave them alone - was settled across writeColStats, shouldRewriteColStats, canProvideColStats and the merge itself. ColStatsWritePolicy now states every case of that decision over the facts of the write alone, so a test can cover it without a table. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… keep a branch's on the branch hive.iceberg.stats.collect.partlevel (default false) chooses the granularity for Iceberg tables, apart from the generic hive.analyze.stmt.collect.partlevel.stats. ANALYZE produces partition-level statistics; every other write maintains table-level ones. The read path keys on the calling API rather than on session config, so a granularity that changed between a write and a read cannot surface a blob of the wrong shape. A write to a branch stores its statistics on that branch's snapshot and leaves the table-scoped metastore row alone, including on the footer scan that ANALYZE ... COMPUTE STATISTICS uses. A whole-table INSERT OVERWRITE of a partitioned table now replaces its column statistics. It skipped before, because the snapshot summary does not distinguish a whole-table overwrite from a partition-scoped one. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…em up under A write derived partition names its own way, so the statistics it stored were keyed differently from the way a read asks for them, and a read found nothing for a partition a write had just described. Both sides now take the name from the Iceberg spec. A write reaching some partitions, and an ANALYZE naming them, stand in for those partitions alone rather than for the whole table. A CTAS gathers partition-level statistics for the table it creates. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… was written for A statistics file holds only what was valid for the snapshot it was written for, so statsFile.snapshotId() is the only anchor a read needs. A merge carries an entry forward only when no write since changed its partition, and a read serves what nothing changed after the file; partsFound reports how much of the scan is covered, so a caller needing an exact answer refuses and one estimating reports PARTIAL. Ordering by position on the ancestry rather than by sequence number fixes format version 1 tables, which number every snapshot 0 and so judged every partition fresh no matter what had been written since. A compaction reads the settings its table carries for it, and did so only after the query it runs and the statistics it gathers had already been settled from the settings it did not have. They are in hand first now, so a table asking to keep statistics per partition is heard by the compaction of one of its partitions, and one that asks for nothing of the sort still has that compaction leave its statistics alone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…tition of no value as Hive does A partition scoped ANALYZE scoped its scan by a value predicate, which names rows rather than partitions. On a table whose partitioning had changed, that predicate reached rows an older spec still held in a partition of its own, and the group that formed described the whole of that partition by the few rows it saw. A table evolved from unpartitioned answered max(id) as 7 where the truth was 100. The scan is now scoped by partition identity, as compaction scopes its rewrite: the values name partitions of the current spec, and the spec's own transforms applied to those values keep every row of each and still prune what the scan reads. A partition of an older spec keeps the statistics it has, which no write since could have changed, because a write lands in the current spec. Rendering a partition of no value as the text "null" also gave it the name a value of that text takes, so the two shared a name and a statistics blob. Hive names it __HIVE_DEFAULT_PARTITION__ and Iceberg's own Conversions reads that name back as no value, so both now say so, which lets a statement name that partition too. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…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.
…ne fetch The rewrite asks for every aggregate's column at once instead of once per aggregate, and only when an aggregate it can answer actually asks, so a query holding a sum or a count(distinct) pays nothing. A scan of an unpartitioned table is answered the same way, and one pruned to no partitions folds to the identity of each aggregate rather than declining. An unpruned scan of a table whose handler keeps whole-table statistics is answered from those directly: they describe exactly the rows read, so there is nothing to aggregate. The row count of an unpartitioned scan is asked of the table rather than read off its parameters, which describe the current snapshot however the scan resolved a branch, a tag or a version. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
deniskuzZ
force-pushed
the
col_stats_stats_opt
branch
from
August 24, 2026 13:09
be100ec to
efa65c1
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