laskoviymishka commented on code in PR #2006:
URL: https://github.com/apache/iceberg-go/pull/2006#discussion_r4007801875
##########
table/internal/parquet_files.go:
##########
@@ -2019,6 +2044,262 @@ func (w wrapPqArrowReader) GetRecords(ctx
context.Context, cols []int, tester an
return w.GetRecordReader(ctx, cols, rgList)
}
+func groupRowGroupDictionaryPredicates(
+ fieldIDToColIdx map[int]int,
+ preds []RowGroupDictionaryPred,
+) map[int][]int {
+ predsByColumn := make(map[int][]int)
+ for i, pred := range preds {
+ if len(pred.PhysBytes) == 0 {
+ continue
+ }
+
+ colIdx, ok := fieldIDToColIdx[pred.FieldID]
+ if ok {
+ predsByColumn[colIdx] = append(predsByColumn[colIdx], i)
+ }
+ }
+
+ return predsByColumn
+}
+
+// checkRowGroupDictionaries checks each dictionary predicate against the
+// complete dictionary for its column in row group rg. It returns false only
+// when the column metadata proves that every data page is dictionary encoded
+// and none of the predicate values occur in the dictionary. Missing encoding
+// metadata, dictionary pages, unsupported physical types, malformed pages, or
+// reader errors keep the row group so this optimisation cannot drop data.
+func checkRowGroupDictionaries(
+ rdr *file.Reader,
+ fileMeta *metadata.FileMetaData,
+ rg int,
+ predsByColumn map[int][]int,
+ preds []RowGroupDictionaryPred,
+) bool {
+ if len(preds) == 0 {
+ return true
+ }
+
+ for colIdx, predIndexes := range predsByColumn {
+ chunk, err := fileMeta.RowGroup(rg).ColumnChunk(colIdx)
Review Comment:
We call `fileMeta.RowGroup(rg)` and `rdr.RowGroup(rg)` inside the per-column
loop, and both allocate fresh `RowGroupMetaData` (plus a `sortCols` slice and a
`OnceValues` closure) on every iteration. `checkRowGroupBloomFilters` builds
one reader outside the column loop and reuses it.
For K predicate columns over N row groups that's roughly 2*N*K allocations
instead of N, on the row-group-selection hot path. Could we hoist
`rdr.RowGroup(rg)` and `fileMeta.RowGroup(rg)` above the loop and reuse them?
##########
table/internal/parquet_files.go:
##########
@@ -1794,11 +1796,21 @@ type RowGroupBloomPred struct {
PhysBytes [][]byte // one entry for EqualTo; one per value for In
}
-// ParquetRowGroupTester combines stats-based and bloom filter row group
pruning.
+// RowGroupDictionaryPred holds the physical-encoded bytes for each literal in
+// a dictionary-prunable predicate on one field. A row group can be skipped
+// when NONE of the bytes occur in its complete dictionary.
Review Comment:
`RowGroupDictionaryPred` is byte-for-byte identical to `RowGroupBloomPred`,
and the `internal.RowGroupDictionaryPred(pred)` conversion in
`newDictionaryPredicatesFromRewritten` is a no-op reinterpret. Fine to keep
them separate, but a one-line comment on why they aren't a shared type (or an
alias) would save the next reader the double-take.
##########
table/internal/parquet_files.go:
##########
@@ -2019,6 +2044,262 @@ func (w wrapPqArrowReader) GetRecords(ctx
context.Context, cols []int, tester an
return w.GetRecordReader(ctx, cols, rgList)
}
+func groupRowGroupDictionaryPredicates(
+ fieldIDToColIdx map[int]int,
+ preds []RowGroupDictionaryPred,
+) map[int][]int {
+ predsByColumn := make(map[int][]int)
+ for i, pred := range preds {
+ if len(pred.PhysBytes) == 0 {
+ continue
+ }
+
+ colIdx, ok := fieldIDToColIdx[pred.FieldID]
+ if ok {
+ predsByColumn[colIdx] = append(predsByColumn[colIdx], i)
+ }
+ }
+
+ return predsByColumn
+}
+
+// checkRowGroupDictionaries checks each dictionary predicate against the
+// complete dictionary for its column in row group rg. It returns false only
+// when the column metadata proves that every data page is dictionary encoded
+// and none of the predicate values occur in the dictionary. Missing encoding
+// metadata, dictionary pages, unsupported physical types, malformed pages, or
+// reader errors keep the row group so this optimisation cannot drop data.
+func checkRowGroupDictionaries(
+ rdr *file.Reader,
+ fileMeta *metadata.FileMetaData,
+ rg int,
+ predsByColumn map[int][]int,
+ preds []RowGroupDictionaryPred,
+) bool {
+ if len(preds) == 0 {
+ return true
+ }
+
+ for colIdx, predIndexes := range predsByColumn {
+ chunk, err := fileMeta.RowGroup(rg).ColumnChunk(colIdx)
+ if err != nil || !parquetColumnUsesOnlyDictionaryData(chunk) {
+ continue
+ }
+
+ column := fileMeta.Schema.Column(colIdx)
+ pageRdr, err := rdr.RowGroup(rg).GetColumnPageReader(colIdx)
+ if err != nil {
+ continue
+ }
+
+ dictPage, dictErr := pageRdr.GetDictionaryPage()
+ if dictErr != nil || dictPage == nil {
+ if dictPage != nil {
+ dictPage.Release()
+ }
+ _ = pageRdr.Close()
+
+ continue
+ }
+
+ matches, known := dictionaryMatchesPredicates(
+ dictPage, column.PhysicalType(), column.TypeLength(),
preds, predIndexes)
+ dictPage.Release()
+ closeErr := pageRdr.Close()
+ if closeErr != nil {
+ continue
+ }
+ if !known {
+ continue
+ }
+
+ for _, predIndex := range predIndexes {
+ if !matches[predIndex] {
+ return false
+ }
+ }
+ }
+
+ return true
+}
+
+// parquetColumnUsesOnlyDictionaryData verifies that the optional encoding
+// statistics identify a dictionary page and dictionary-encoded data pages,
+// with no PLAIN fallback pages. EncodingStats is deliberately required: the
+// column encoding list cannot distinguish the PLAIN dictionary page itself
+// from a later PLAIN data page in older files.
+func parquetColumnUsesOnlyDictionaryData(chunk *metadata.ColumnChunkMetaData)
bool {
+ if !chunk.HasDictionaryPage() {
+ return false
+ }
+
+ stats := chunk.EncodingStats()
+ if len(stats) == 0 {
+ return false
+ }
+
+ var hasDictionaryPage, hasDictionaryData bool
+ for _, stat := range stats {
+ switch stat.PageType {
+ case file.PageTypeDictionaryPage:
+ if stat.Encoding != parquet.Encodings.Plain &&
stat.Encoding != parquet.Encodings.PlainDict {
+ return false
+ }
+ hasDictionaryPage = true
+ case file.PageTypeDataPage, file.PageTypeDataPageV2:
+ if stat.Encoding != parquet.Encodings.RLEDict &&
stat.Encoding != parquet.Encodings.PlainDict {
+ return false
+ }
+ hasDictionaryData = true
+ default:
+ return false
+ }
+ }
+
+ return hasDictionaryPage && hasDictionaryData
+}
+
+// dictionaryMatchesPredicates decodes a PLAIN dictionary page and records
+// which requested predicates have at least one matching value. The returned
+// known flag is false for unsupported or malformed input, which means the
+// caller must retain the row group.
+func dictionaryMatchesPredicates(
+ page *file.DictionaryPage,
+ physicalType parquet.Type,
+ typeLen int,
+ preds []RowGroupDictionaryPred,
+ predIndexes []int,
+) ([]bool, bool) {
+ pageEncoding := parquet.Encoding(page.Encoding())
+ if pageEncoding != parquet.Encodings.Plain && pageEncoding !=
parquet.Encodings.PlainDict {
+ return nil, false
+ }
+
+ width, variableWidth, ok := parquetDictionaryValueLayout(physicalType,
typeLen)
+ if !ok {
+ return nil, false
+ }
+
+ for _, predIndex := range predIndexes {
+ usable := false
+ for _, candidate := range preds[predIndex].PhysBytes {
+ if variableWidth || len(candidate) == width {
+ usable = true
+
+ break
+ }
+ }
+ if !usable {
+ return nil, false
+ }
+ }
+
+ matches := make([]bool, len(preds))
Review Comment:
`matches` is sized `len(preds)` (the total across all columns) but only the
`predIndexes` positions are ever touched. It's correct because the index is the
global predicate position, which lets the same slot work across column checks,
but that convention isn't obvious from here.
A one-line comment on why it's indexed globally would help. Sizing to
`len(predIndexes)` with a remap is the alternative if we'd rather not carry the
sparse slice. Either's fine.
##########
table/internal/parquet_files.go:
##########
@@ -2019,6 +2044,262 @@ func (w wrapPqArrowReader) GetRecords(ctx
context.Context, cols []int, tester an
return w.GetRecordReader(ctx, cols, rgList)
}
+func groupRowGroupDictionaryPredicates(
+ fieldIDToColIdx map[int]int,
+ preds []RowGroupDictionaryPred,
+) map[int][]int {
+ predsByColumn := make(map[int][]int)
+ for i, pred := range preds {
+ if len(pred.PhysBytes) == 0 {
+ continue
+ }
+
+ colIdx, ok := fieldIDToColIdx[pred.FieldID]
+ if ok {
+ predsByColumn[colIdx] = append(predsByColumn[colIdx], i)
+ }
+ }
+
+ return predsByColumn
+}
+
+// checkRowGroupDictionaries checks each dictionary predicate against the
+// complete dictionary for its column in row group rg. It returns false only
+// when the column metadata proves that every data page is dictionary encoded
+// and none of the predicate values occur in the dictionary. Missing encoding
+// metadata, dictionary pages, unsupported physical types, malformed pages, or
+// reader errors keep the row group so this optimisation cannot drop data.
+func checkRowGroupDictionaries(
+ rdr *file.Reader,
+ fileMeta *metadata.FileMetaData,
+ rg int,
+ predsByColumn map[int][]int,
+ preds []RowGroupDictionaryPred,
+) bool {
+ if len(preds) == 0 {
+ return true
+ }
+
+ for colIdx, predIndexes := range predsByColumn {
+ chunk, err := fileMeta.RowGroup(rg).ColumnChunk(colIdx)
+ if err != nil || !parquetColumnUsesOnlyDictionaryData(chunk) {
+ continue
+ }
+
+ column := fileMeta.Schema.Column(colIdx)
+ pageRdr, err := rdr.RowGroup(rg).GetColumnPageReader(colIdx)
+ if err != nil {
+ continue
+ }
+
+ dictPage, dictErr := pageRdr.GetDictionaryPage()
+ if dictErr != nil || dictPage == nil {
+ if dictPage != nil {
+ dictPage.Release()
+ }
+ _ = pageRdr.Close()
+
+ continue
+ }
+
+ matches, known := dictionaryMatchesPredicates(
+ dictPage, column.PhysicalType(), column.TypeLength(),
preds, predIndexes)
+ dictPage.Release()
+ closeErr := pageRdr.Close()
+ if closeErr != nil {
Review Comment:
Small ordering thing: we check `closeErr` before `!known`, so a non-nil
`Close()` un-prunes a group we'd already proven should be dropped.
`serializedPageReader.Close()` returns nil today so it's latent, but the moment
an implementation returns an error here we'd silently stop pruning even on a
conclusive no-match.
Not blocking as-is. I'd at least add a comment that a close error keeps the
group conservatively, and ideally log it so it isn't invisible.
##########
table/internal/parquet_files.go:
##########
@@ -2019,6 +2044,262 @@ func (w wrapPqArrowReader) GetRecords(ctx
context.Context, cols []int, tester an
return w.GetRecordReader(ctx, cols, rgList)
}
+func groupRowGroupDictionaryPredicates(
+ fieldIDToColIdx map[int]int,
+ preds []RowGroupDictionaryPred,
+) map[int][]int {
+ predsByColumn := make(map[int][]int)
+ for i, pred := range preds {
+ if len(pred.PhysBytes) == 0 {
+ continue
+ }
+
+ colIdx, ok := fieldIDToColIdx[pred.FieldID]
+ if ok {
+ predsByColumn[colIdx] = append(predsByColumn[colIdx], i)
+ }
+ }
+
+ return predsByColumn
+}
+
+// checkRowGroupDictionaries checks each dictionary predicate against the
+// complete dictionary for its column in row group rg. It returns false only
+// when the column metadata proves that every data page is dictionary encoded
+// and none of the predicate values occur in the dictionary. Missing encoding
+// metadata, dictionary pages, unsupported physical types, malformed pages, or
+// reader errors keep the row group so this optimisation cannot drop data.
+func checkRowGroupDictionaries(
+ rdr *file.Reader,
+ fileMeta *metadata.FileMetaData,
+ rg int,
+ predsByColumn map[int][]int,
+ preds []RowGroupDictionaryPred,
+) bool {
+ if len(preds) == 0 {
+ return true
+ }
+
+ for colIdx, predIndexes := range predsByColumn {
+ chunk, err := fileMeta.RowGroup(rg).ColumnChunk(colIdx)
+ if err != nil || !parquetColumnUsesOnlyDictionaryData(chunk) {
+ continue
+ }
+
+ column := fileMeta.Schema.Column(colIdx)
+ pageRdr, err := rdr.RowGroup(rg).GetColumnPageReader(colIdx)
+ if err != nil {
+ continue
+ }
+
+ dictPage, dictErr := pageRdr.GetDictionaryPage()
+ if dictErr != nil || dictPage == nil {
+ if dictPage != nil {
+ dictPage.Release()
+ }
+ _ = pageRdr.Close()
+
+ continue
+ }
+
+ matches, known := dictionaryMatchesPredicates(
+ dictPage, column.PhysicalType(), column.TypeLength(),
preds, predIndexes)
+ dictPage.Release()
+ closeErr := pageRdr.Close()
+ if closeErr != nil {
+ continue
+ }
+ if !known {
+ continue
+ }
+
+ for _, predIndex := range predIndexes {
+ if !matches[predIndex] {
+ return false
+ }
+ }
+ }
+
+ return true
+}
+
+// parquetColumnUsesOnlyDictionaryData verifies that the optional encoding
+// statistics identify a dictionary page and dictionary-encoded data pages,
+// with no PLAIN fallback pages. EncodingStats is deliberately required: the
+// column encoding list cannot distinguish the PLAIN dictionary page itself
+// from a later PLAIN data page in older files.
+func parquetColumnUsesOnlyDictionaryData(chunk *metadata.ColumnChunkMetaData)
bool {
+ if !chunk.HasDictionaryPage() {
+ return false
+ }
+
+ stats := chunk.EncodingStats()
+ if len(stats) == 0 {
+ return false
+ }
+
+ var hasDictionaryPage, hasDictionaryData bool
+ for _, stat := range stats {
+ switch stat.PageType {
+ case file.PageTypeDictionaryPage:
+ if stat.Encoding != parquet.Encodings.Plain &&
stat.Encoding != parquet.Encodings.PlainDict {
+ return false
+ }
+ hasDictionaryPage = true
+ case file.PageTypeDataPage, file.PageTypeDataPageV2:
+ if stat.Encoding != parquet.Encodings.RLEDict &&
stat.Encoding != parquet.Encodings.PlainDict {
+ return false
+ }
+ hasDictionaryData = true
+ default:
+ return false
+ }
+ }
+
+ return hasDictionaryPage && hasDictionaryData
+}
+
+// dictionaryMatchesPredicates decodes a PLAIN dictionary page and records
+// which requested predicates have at least one matching value. The returned
+// known flag is false for unsupported or malformed input, which means the
+// caller must retain the row group.
+func dictionaryMatchesPredicates(
+ page *file.DictionaryPage,
+ physicalType parquet.Type,
+ typeLen int,
+ preds []RowGroupDictionaryPred,
+ predIndexes []int,
+) ([]bool, bool) {
+ pageEncoding := parquet.Encoding(page.Encoding())
+ if pageEncoding != parquet.Encodings.Plain && pageEncoding !=
parquet.Encodings.PlainDict {
+ return nil, false
+ }
+
+ width, variableWidth, ok := parquetDictionaryValueLayout(physicalType,
typeLen)
+ if !ok {
+ return nil, false
+ }
+
+ for _, predIndex := range predIndexes {
+ usable := false
+ for _, candidate := range preds[predIndex].PhysBytes {
+ if variableWidth || len(candidate) == width {
+ usable = true
+
+ break
+ }
+ }
+ if !usable {
+ return nil, false
+ }
+ }
+
+ matches := make([]bool, len(preds))
+ data := page.Data()
+ offset := 0
+ numValues := page.NumValues()
+ if numValues < 0 {
+ return nil, false
+ }
+
+ for range numValues {
+ value, next, ok := nextParquetDictionaryValue(data, offset,
physicalType, width)
+ if !ok {
+ return nil, false
+ }
+ offset = next
+
+ for _, predIndex := range predIndexes {
+ if matches[predIndex] {
+ continue
+ }
+
+ for _, candidate := range preds[predIndex].PhysBytes {
+ if parquetDictionaryValueEqual(physicalType,
value, candidate) {
+ matches[predIndex] = true
+
+ break
+ }
+ }
+ }
+ }
+
+ if offset != len(data) {
+ return nil, false
+ }
+
+ return matches, true
+}
+
+func parquetDictionaryValueLayout(physicalType parquet.Type, typeLen int)
(width int, variableWidth, ok bool) {
+ switch physicalType {
+ case parquet.Types.Int32, parquet.Types.Float:
+ return 4, false, true
+ case parquet.Types.Int64, parquet.Types.Double:
+ return 8, false, true
+ case parquet.Types.Int96:
Review Comment:
This Int96 case looks unreachable for Iceberg types: the spec maps all
timestamps to INT64, and no Iceberg literal produces 12-byte `PhysBytes`, so
`dictionaryMatchesPredicates` bails on the usability check before this branch
matters. I'd drop it, or leave a TODO noting it's for a future legacy-INT96
path so it doesn't read as live support.
##########
table/internal/parquet_files.go:
##########
@@ -2019,6 +2044,262 @@ func (w wrapPqArrowReader) GetRecords(ctx
context.Context, cols []int, tester an
return w.GetRecordReader(ctx, cols, rgList)
}
+func groupRowGroupDictionaryPredicates(
+ fieldIDToColIdx map[int]int,
+ preds []RowGroupDictionaryPred,
+) map[int][]int {
+ predsByColumn := make(map[int][]int)
+ for i, pred := range preds {
+ if len(pred.PhysBytes) == 0 {
+ continue
+ }
+
+ colIdx, ok := fieldIDToColIdx[pred.FieldID]
+ if ok {
+ predsByColumn[colIdx] = append(predsByColumn[colIdx], i)
+ }
+ }
+
+ return predsByColumn
+}
+
+// checkRowGroupDictionaries checks each dictionary predicate against the
+// complete dictionary for its column in row group rg. It returns false only
+// when the column metadata proves that every data page is dictionary encoded
+// and none of the predicate values occur in the dictionary. Missing encoding
+// metadata, dictionary pages, unsupported physical types, malformed pages, or
+// reader errors keep the row group so this optimisation cannot drop data.
+func checkRowGroupDictionaries(
+ rdr *file.Reader,
+ fileMeta *metadata.FileMetaData,
+ rg int,
+ predsByColumn map[int][]int,
+ preds []RowGroupDictionaryPred,
+) bool {
+ if len(preds) == 0 {
+ return true
+ }
+
+ for colIdx, predIndexes := range predsByColumn {
+ chunk, err := fileMeta.RowGroup(rg).ColumnChunk(colIdx)
+ if err != nil || !parquetColumnUsesOnlyDictionaryData(chunk) {
+ continue
+ }
+
+ column := fileMeta.Schema.Column(colIdx)
+ pageRdr, err := rdr.RowGroup(rg).GetColumnPageReader(colIdx)
+ if err != nil {
+ continue
+ }
+
+ dictPage, dictErr := pageRdr.GetDictionaryPage()
+ if dictErr != nil || dictPage == nil {
+ if dictPage != nil {
+ dictPage.Release()
+ }
+ _ = pageRdr.Close()
+
+ continue
+ }
+
+ matches, known := dictionaryMatchesPredicates(
+ dictPage, column.PhysicalType(), column.TypeLength(),
preds, predIndexes)
+ dictPage.Release()
+ closeErr := pageRdr.Close()
+ if closeErr != nil {
+ continue
+ }
+ if !known {
+ continue
+ }
+
+ for _, predIndex := range predIndexes {
+ if !matches[predIndex] {
+ return false
+ }
+ }
+ }
+
+ return true
+}
+
+// parquetColumnUsesOnlyDictionaryData verifies that the optional encoding
+// statistics identify a dictionary page and dictionary-encoded data pages,
+// with no PLAIN fallback pages. EncodingStats is deliberately required: the
+// column encoding list cannot distinguish the PLAIN dictionary page itself
+// from a later PLAIN data page in older files.
+func parquetColumnUsesOnlyDictionaryData(chunk *metadata.ColumnChunkMetaData)
bool {
+ if !chunk.HasDictionaryPage() {
+ return false
+ }
+
+ stats := chunk.EncodingStats()
+ if len(stats) == 0 {
+ return false
+ }
+
+ var hasDictionaryPage, hasDictionaryData bool
+ for _, stat := range stats {
+ switch stat.PageType {
+ case file.PageTypeDictionaryPage:
+ if stat.Encoding != parquet.Encodings.Plain &&
stat.Encoding != parquet.Encodings.PlainDict {
+ return false
+ }
+ hasDictionaryPage = true
+ case file.PageTypeDataPage, file.PageTypeDataPageV2:
+ if stat.Encoding != parquet.Encodings.RLEDict &&
stat.Encoding != parquet.Encodings.PlainDict {
+ return false
+ }
+ hasDictionaryData = true
+ default:
+ return false
+ }
+ }
+
+ return hasDictionaryPage && hasDictionaryData
+}
+
+// dictionaryMatchesPredicates decodes a PLAIN dictionary page and records
+// which requested predicates have at least one matching value. The returned
+// known flag is false for unsupported or malformed input, which means the
+// caller must retain the row group.
+func dictionaryMatchesPredicates(
+ page *file.DictionaryPage,
+ physicalType parquet.Type,
+ typeLen int,
+ preds []RowGroupDictionaryPred,
+ predIndexes []int,
+) ([]bool, bool) {
+ pageEncoding := parquet.Encoding(page.Encoding())
+ if pageEncoding != parquet.Encodings.Plain && pageEncoding !=
parquet.Encodings.PlainDict {
+ return nil, false
+ }
+
+ width, variableWidth, ok := parquetDictionaryValueLayout(physicalType,
typeLen)
+ if !ok {
+ return nil, false
+ }
+
+ for _, predIndex := range predIndexes {
+ usable := false
+ for _, candidate := range preds[predIndex].PhysBytes {
+ if variableWidth || len(candidate) == width {
+ usable = true
+
+ break
+ }
+ }
+ if !usable {
+ return nil, false
+ }
+ }
+
+ matches := make([]bool, len(preds))
+ data := page.Data()
+ offset := 0
+ numValues := page.NumValues()
+ if numValues < 0 {
+ return nil, false
+ }
+
+ for range numValues {
Review Comment:
The outer loop keeps decoding every remaining dictionary entry even after
all predicates for this column are satisfied. The inner `if matches[predIndex]
{ continue }` skips the compare, but we still advance
`nextParquetDictionaryValue` over the rest of the page; the bloom path breaks
on first match.
Low-cardinality dicts make this mostly harmless, but a `remaining` counter
with an outer break would bring it in line with bloom. Minor.
##########
table/internal/parquet_files_test.go:
##########
@@ -2211,6 +2211,267 @@ func TestBloomFilterRowGroupPruning(t *testing.T) {
})
}
+// buildDictionaryTestParquet writes a Parquet file with two low-cardinality
+// dictionary-encoded row groups. The single required INT32 column "id" has
+// Iceberg field_id=1.
+func buildDictionaryTestParquet(t testing.TB, rowGroups ...[]int32) []byte {
+ t.Helper()
+
+ idNode := schema.NewInt32Node("id", parquet.Repetitions.Required, 1)
+ rootNode, err := schema.NewGroupNode("schema",
parquet.Repetitions.Required,
+ schema.FieldList{idNode}, -1)
+ require.NoError(t, err)
+
+ writerProps := parquet.NewWriterProperties(
+ parquet.WithStats(true),
+ parquet.WithDictionaryDefault(true),
+ )
+
+ var buf bytes.Buffer
+ pw := file.NewParquetWriter(&buf, rootNode,
file.WithWriterProps(writerProps))
+ for _, values := range rowGroups {
+ rgw, werr := pw.AppendRowGroupChecked()
+ require.NoError(t, werr)
+ cw, werr := rgw.NextColumn()
+ require.NoError(t, werr)
+ _, werr = cw.(*file.Int32ColumnChunkWriter).WriteBatch(values,
nil, nil)
+ require.NoError(t, werr)
+ require.NoError(t, cw.Close())
+ require.NoError(t, rgw.Close())
+ }
+ require.NoError(t, pw.Close())
+
+ return buf.Bytes()
+}
+
+// TestDictionaryRowGroupPruning verifies that dictionary-only row groups are
+// skipped when an EqualTo/In predicate has no value in the complete
dictionary.
+func TestDictionaryRowGroupPruning(t *testing.T) {
Review Comment:
The per-column loop in `checkRowGroupDictionaries` handles multi-column ANDs
(prune if any column's dict lacks its value), but every integration test here
uses the single-column `id` schema, so that path is unverified.
Could we add a two-column case, say INT32 `id` + BYTE_ARRAY `category`, and
assert `id=X AND category=Y` prunes a group when X is absent from its id-dict,
and separately when Y is absent from its category-dict? That's the common
real-world predicate shape, and a future refactor of
`groupRowGroupDictionaryPredicates` could introduce a false prune with nothing
to catch it.
##########
table/internal/parquet_files.go:
##########
@@ -2019,6 +2044,262 @@ func (w wrapPqArrowReader) GetRecords(ctx
context.Context, cols []int, tester an
return w.GetRecordReader(ctx, cols, rgList)
}
+func groupRowGroupDictionaryPredicates(
+ fieldIDToColIdx map[int]int,
+ preds []RowGroupDictionaryPred,
+) map[int][]int {
+ predsByColumn := make(map[int][]int)
+ for i, pred := range preds {
+ if len(pred.PhysBytes) == 0 {
+ continue
+ }
+
+ colIdx, ok := fieldIDToColIdx[pred.FieldID]
+ if ok {
+ predsByColumn[colIdx] = append(predsByColumn[colIdx], i)
+ }
+ }
+
+ return predsByColumn
+}
+
+// checkRowGroupDictionaries checks each dictionary predicate against the
+// complete dictionary for its column in row group rg. It returns false only
+// when the column metadata proves that every data page is dictionary encoded
+// and none of the predicate values occur in the dictionary. Missing encoding
+// metadata, dictionary pages, unsupported physical types, malformed pages, or
+// reader errors keep the row group so this optimisation cannot drop data.
+func checkRowGroupDictionaries(
+ rdr *file.Reader,
+ fileMeta *metadata.FileMetaData,
+ rg int,
+ predsByColumn map[int][]int,
+ preds []RowGroupDictionaryPred,
+) bool {
+ if len(preds) == 0 {
+ return true
+ }
+
+ for colIdx, predIndexes := range predsByColumn {
+ chunk, err := fileMeta.RowGroup(rg).ColumnChunk(colIdx)
+ if err != nil || !parquetColumnUsesOnlyDictionaryData(chunk) {
+ continue
+ }
+
+ column := fileMeta.Schema.Column(colIdx)
+ pageRdr, err := rdr.RowGroup(rg).GetColumnPageReader(colIdx)
+ if err != nil {
+ continue
+ }
+
+ dictPage, dictErr := pageRdr.GetDictionaryPage()
+ if dictErr != nil || dictPage == nil {
+ if dictPage != nil {
+ dictPage.Release()
+ }
+ _ = pageRdr.Close()
+
+ continue
+ }
+
+ matches, known := dictionaryMatchesPredicates(
+ dictPage, column.PhysicalType(), column.TypeLength(),
preds, predIndexes)
+ dictPage.Release()
+ closeErr := pageRdr.Close()
+ if closeErr != nil {
+ continue
+ }
+ if !known {
+ continue
+ }
+
+ for _, predIndex := range predIndexes {
+ if !matches[predIndex] {
+ return false
+ }
+ }
+ }
+
+ return true
+}
+
+// parquetColumnUsesOnlyDictionaryData verifies that the optional encoding
+// statistics identify a dictionary page and dictionary-encoded data pages,
+// with no PLAIN fallback pages. EncodingStats is deliberately required: the
+// column encoding list cannot distinguish the PLAIN dictionary page itself
+// from a later PLAIN data page in older files.
+func parquetColumnUsesOnlyDictionaryData(chunk *metadata.ColumnChunkMetaData)
bool {
+ if !chunk.HasDictionaryPage() {
+ return false
+ }
+
+ stats := chunk.EncodingStats()
+ if len(stats) == 0 {
+ return false
+ }
+
+ var hasDictionaryPage, hasDictionaryData bool
+ for _, stat := range stats {
+ switch stat.PageType {
+ case file.PageTypeDictionaryPage:
+ if stat.Encoding != parquet.Encodings.Plain &&
stat.Encoding != parquet.Encodings.PlainDict {
+ return false
+ }
+ hasDictionaryPage = true
+ case file.PageTypeDataPage, file.PageTypeDataPageV2:
+ if stat.Encoding != parquet.Encodings.RLEDict &&
stat.Encoding != parquet.Encodings.PlainDict {
+ return false
+ }
+ hasDictionaryData = true
+ default:
+ return false
+ }
+ }
+
+ return hasDictionaryPage && hasDictionaryData
+}
+
+// dictionaryMatchesPredicates decodes a PLAIN dictionary page and records
+// which requested predicates have at least one matching value. The returned
+// known flag is false for unsupported or malformed input, which means the
+// caller must retain the row group.
+func dictionaryMatchesPredicates(
+ page *file.DictionaryPage,
+ physicalType parquet.Type,
+ typeLen int,
+ preds []RowGroupDictionaryPred,
+ predIndexes []int,
+) ([]bool, bool) {
+ pageEncoding := parquet.Encoding(page.Encoding())
+ if pageEncoding != parquet.Encodings.Plain && pageEncoding !=
parquet.Encodings.PlainDict {
+ return nil, false
+ }
+
+ width, variableWidth, ok := parquetDictionaryValueLayout(physicalType,
typeLen)
+ if !ok {
+ return nil, false
+ }
+
+ for _, predIndex := range predIndexes {
+ usable := false
+ for _, candidate := range preds[predIndex].PhysBytes {
+ if variableWidth || len(candidate) == width {
+ usable = true
+
+ break
+ }
+ }
+ if !usable {
+ return nil, false
+ }
+ }
+
+ matches := make([]bool, len(preds))
+ data := page.Data()
+ offset := 0
+ numValues := page.NumValues()
+ if numValues < 0 {
+ return nil, false
+ }
+
+ for range numValues {
+ value, next, ok := nextParquetDictionaryValue(data, offset,
physicalType, width)
+ if !ok {
+ return nil, false
+ }
+ offset = next
+
+ for _, predIndex := range predIndexes {
+ if matches[predIndex] {
+ continue
+ }
+
+ for _, candidate := range preds[predIndex].PhysBytes {
+ if parquetDictionaryValueEqual(physicalType,
value, candidate) {
+ matches[predIndex] = true
+
+ break
+ }
+ }
+ }
+ }
+
+ if offset != len(data) {
+ return nil, false
+ }
+
+ return matches, true
+}
+
+func parquetDictionaryValueLayout(physicalType parquet.Type, typeLen int)
(width int, variableWidth, ok bool) {
+ switch physicalType {
+ case parquet.Types.Int32, parquet.Types.Float:
+ return 4, false, true
+ case parquet.Types.Int64, parquet.Types.Double:
+ return 8, false, true
+ case parquet.Types.Int96:
+ return parquet.Int96SizeBytes, false, true
+ case parquet.Types.ByteArray:
+ return 0, true, true
+ case parquet.Types.FixedLenByteArray:
+ if typeLen > 0 {
+ return typeLen, false, true
+ }
+ }
+
+ return 0, false, false
+}
+
+func nextParquetDictionaryValue(data []byte, offset int, physicalType
parquet.Type, width int) ([]byte, int, bool) {
+ if physicalType == parquet.Types.ByteArray {
+ if offset < 0 || offset > len(data) || len(data)-offset < 4 {
+ return nil, 0, false
+ }
+
+ valueLen := int(binary.LittleEndian.Uint32(data[offset:]))
+ valueStart := offset + 4
+ if valueLen > len(data)-valueStart {
+ return nil, 0, false
+ }
+
+ return data[valueStart : valueStart+valueLen], valueStart +
valueLen, true
+ }
+
+ if offset < 0 || width < 0 || width > len(data)-offset {
+ return nil, 0, false
+ }
+
+ return data[offset : offset+width], offset + width, true
+}
+
+func parquetDictionaryValueEqual(physicalType parquet.Type, value, candidate
[]byte) bool {
+ if physicalType == parquet.Types.Float && len(value) == 4 &&
len(candidate) == 4 {
+ left := math.Float32frombits(binary.LittleEndian.Uint32(value))
+ right :=
math.Float32frombits(binary.LittleEndian.Uint32(candidate))
+ if math.IsNaN(float64(left)) || math.IsNaN(float64(right)) {
Review Comment:
I think this quietly defeats the optimization for any float/double column
that carries a NaN. Because we return `true` whenever either side is NaN, the
first NaN dictionary entry sets `matches[predIndex] = true` for every
candidate, including one that's plainly absent.
Concretely: `score FLOAT` with dict `{NaN, 1.0, 2.0}` and `EqualTo(score,
99.0)`. We hit NaN during iteration, `IsNaN(NaN) || IsNaN(99.0)` is true,
`matches[0]` flips true, and we keep the group even though 99.0 isn't there.
For tables that use NaN as a missing sentinel, that's every row group, and
dictionary pruning silently becomes a no-op. It's fail-open so no data loss,
but the feature stops doing anything.
I'd split the two cases: if the candidate (the predicate literal) is NaN,
keep the conservative `return true`; if only the dictionary value is NaN and
the candidate is a real number, `return false`, since NaN doesn't equal a
number under any reader's semantics. wdyt?
Worth a unit test alongside the fix too: build a Float dict containing
`math.NaN()`, search a non-NaN value, assert the expected keep/prune. And while
we're here, the Double path has no coverage at all right now (the signed-zero
test only exercises Float), so a parallel Double case would be good.
##########
table/internal/parquet_files_test.go:
##########
@@ -2211,6 +2211,267 @@ func TestBloomFilterRowGroupPruning(t *testing.T) {
})
}
+// buildDictionaryTestParquet writes a Parquet file with two low-cardinality
+// dictionary-encoded row groups. The single required INT32 column "id" has
+// Iceberg field_id=1.
+func buildDictionaryTestParquet(t testing.TB, rowGroups ...[]int32) []byte {
+ t.Helper()
+
+ idNode := schema.NewInt32Node("id", parquet.Repetitions.Required, 1)
+ rootNode, err := schema.NewGroupNode("schema",
parquet.Repetitions.Required,
+ schema.FieldList{idNode}, -1)
+ require.NoError(t, err)
+
+ writerProps := parquet.NewWriterProperties(
+ parquet.WithStats(true),
+ parquet.WithDictionaryDefault(true),
+ )
+
+ var buf bytes.Buffer
+ pw := file.NewParquetWriter(&buf, rootNode,
file.WithWriterProps(writerProps))
+ for _, values := range rowGroups {
+ rgw, werr := pw.AppendRowGroupChecked()
+ require.NoError(t, werr)
+ cw, werr := rgw.NextColumn()
+ require.NoError(t, werr)
+ _, werr = cw.(*file.Int32ColumnChunkWriter).WriteBatch(values,
nil, nil)
+ require.NoError(t, werr)
+ require.NoError(t, cw.Close())
+ require.NoError(t, rgw.Close())
+ }
+ require.NoError(t, pw.Close())
+
+ return buf.Bytes()
+}
+
+// TestDictionaryRowGroupPruning verifies that dictionary-only row groups are
+// skipped when an EqualTo/In predicate has no value in the complete
dictionary.
+func TestDictionaryRowGroupPruning(t *testing.T) {
+ const rgSize = 1024
+ rg0 := make([]int32, rgSize)
+ rg1 := make([]int32, rgSize)
+ for i := range rg0 {
+ rg0[i] = 1 + int32(i%2)*2
+ rg1[i] = 5 + int32(i%2)*2
+ }
+
+ data := buildDictionaryTestParquet(t, rg0, rg1)
+ alwaysKeep := func(_ *metadata.RowGroupMetaData, _ []int) (bool, error)
{
+ return true, nil
+ }
+
+ ctx := context.Background()
+ cols := []int{0}
+
+ t.Run("EqualTo absent from both dictionaries", func(t *testing.T) {
+ rdr := openBloomTestReader(t, data)
+ var survivors []internal.RowGroupSpan
+ tester := &internal.ParquetRowGroupTester{
+ DictionaryPreds: []internal.RowGroupDictionaryPred{
+ {FieldID: 1, PhysBytes:
[][]byte{int32PhysBytes(2)}},
+ },
+ Survivors: &survivors,
+ }
+ rr, err := rdr.GetRecords(ctx, cols, tester)
+ require.NoError(t, err)
+
+ assert.Equal(t, int64(0), countRecords(t, rr))
+ assert.Empty(t, survivors)
+ require.NoError(t, rdr.Close())
+ })
+
+ t.Run("EqualTo keeps the matching dictionary", func(t *testing.T) {
+ rdr := openBloomTestReader(t, data)
+ var survivors []internal.RowGroupSpan
+ tester := &internal.ParquetRowGroupTester{
+ DictionaryPreds: []internal.RowGroupDictionaryPred{
+ {FieldID: 1, PhysBytes:
[][]byte{int32PhysBytes(7)}},
+ },
+ StatsFn: alwaysKeep,
+ Survivors: &survivors,
+ }
+ rr, err := rdr.GetRecords(ctx, cols, tester)
+ require.NoError(t, err)
+
+ assert.Equal(t, int64(rgSize), countRecords(t, rr))
+ assert.Equal(t, []internal.RowGroupSpan{{FirstRowPos: rgSize,
NumRows: rgSize}}, survivors)
+ require.NoError(t, rdr.Close())
+ })
+
+ t.Run("In keeps a row group when any value is present", func(t
*testing.T) {
+ rdr := openBloomTestReader(t, data)
+ var survivors []internal.RowGroupSpan
+ tester := &internal.ParquetRowGroupTester{
+ DictionaryPreds: []internal.RowGroupDictionaryPred{
+ {FieldID: 1, PhysBytes:
[][]byte{int32PhysBytes(2), int32PhysBytes(5)}},
+ },
+ StatsFn: alwaysKeep,
+ Survivors: &survivors,
+ }
+ rr, err := rdr.GetRecords(ctx, cols, tester)
+ require.NoError(t, err)
+
+ assert.Equal(t, int64(rgSize), countRecords(t, rr))
+ assert.Equal(t, []internal.RowGroupSpan{{FirstRowPos: rgSize,
NumRows: rgSize}}, survivors)
+ require.NoError(t, rdr.Close())
+ })
+}
+
+func BenchmarkDictionaryRowGroupPruning(b *testing.B) {
+ const (
+ numRowGroups = 16
+ rowsPerGroup = 4096
+ )
+
+ rowGroups := make([][]int32, numRowGroups)
+ for group := range rowGroups {
+ values := make([]int32, rowsPerGroup)
+ for i := range values {
+ values[i] = int32(group*2 + i%2)
+ }
+ rowGroups[group] = values
+ }
+
+ data := buildDictionaryTestParquet(b, rowGroups...)
+ alwaysKeep := func(_ *metadata.RowGroupMetaData, _ []int) (bool, error)
{
+ return true, nil
+ }
+ cols := []int{0}
+ benchmarks := []struct {
+ name string
+ physBytes [][]byte
+ wantRows int64
+ }{
+ {name: "without dictionary", wantRows: int64(numRowGroups *
rowsPerGroup)},
+ {name: "dictionary target absent", physBytes:
[][]byte{int32PhysBytes(-1)}},
+ {
+ name: "dictionary target in one group",
+ physBytes: [][]byte{int32PhysBytes(0)},
+ wantRows: int64(rowsPerGroup),
+ },
+ }
+
+ for _, benchmark := range benchmarks {
+ b.Run(benchmark.name, func(b *testing.B) {
+ b.ReportAllocs()
+ b.SetBytes(int64(len(data)))
+ b.ResetTimer()
+
+ for range b.N {
+ rdr := openBloomTestReader(b, data)
+ tester :=
&internal.ParquetRowGroupTester{StatsFn: alwaysKeep}
+ if len(benchmark.physBytes) > 0 {
+ tester.DictionaryPreds =
[]internal.RowGroupDictionaryPred{
+ {FieldID: 1, PhysBytes:
benchmark.physBytes},
+ }
+ }
+
+ rr, err := rdr.GetRecords(context.Background(),
cols, tester)
+ if err != nil {
+ b.Fatal(err)
+ }
+ rows := countRecords(b, rr)
+ if rows != benchmark.wantRows {
+ b.Fatalf("returned %d rows, want %d",
rows, benchmark.wantRows)
+ }
+ if err := rdr.Close(); err != nil {
+ b.Fatal(err)
+ }
+ }
+ })
+ }
+}
+
+// buildDictionaryFallbackTestParquet writes one row group with dictionary
+// pages followed by plain data pages. The returned target exists only in the
+// plain portion of the column.
+func buildDictionaryFallbackTestParquet(t *testing.T) ([]byte, int) {
+ t.Helper()
+
+ valueWidth := 32
+ lowCardinality := make([]parquet.ByteArray, 2048)
+ for i := range lowCardinality {
+ value := make([]byte, valueWidth)
+ copy(value, fmt.Sprintf("category-%02d", i%16))
+ lowCardinality[i] = parquet.ByteArray(value)
+ }
+
+ target := "plain-only-target"
+ plainValues := make([]parquet.ByteArray, 16)
+ for i := range plainValues {
+ value := fmt.Sprintf("plain-value-%02d", i)
+ if i == len(plainValues)-1 {
+ value = target
+ }
+ plainValues[i] = parquet.ByteArray(value)
+ }
+
+ valueNode := schema.NewByteArrayNode("value",
parquet.Repetitions.Required, 1)
+ rootNode, err := schema.NewGroupNode("schema",
parquet.Repetitions.Required,
+ schema.FieldList{valueNode}, -1)
+ require.NoError(t, err)
+
+ writerProps := parquet.NewWriterProperties(
+ parquet.WithStats(true),
+ parquet.WithDictionaryDefault(true),
+ parquet.WithDataPageSize(1024),
+ )
+
+ var buf bytes.Buffer
+ pw := file.NewParquetWriter(&buf, rootNode,
file.WithWriterProps(writerProps))
+ rgw, err := pw.AppendRowGroupChecked()
+ require.NoError(t, err)
+ cw, err := rgw.NextColumn()
+ require.NoError(t, err)
+ byteWriter := cw.(*file.ByteArrayColumnChunkWriter)
+ _, err = byteWriter.WriteBatch(lowCardinality, nil, nil)
+ require.NoError(t, err)
+ byteWriter.FallbackToPlain()
+ _, err = byteWriter.WriteBatch(plainValues, nil, nil)
+ require.NoError(t, err)
+ require.NoError(t, byteWriter.Close())
+ require.NoError(t, rgw.Close())
+ require.NoError(t, pw.Close())
+
+ return buf.Bytes(), len(lowCardinality) + len(plainValues)
+}
+
+func TestDictionaryRowGroupPruningKeepsFallbackColumns(t *testing.T) {
+ data, numRows := buildDictionaryFallbackTestParquet(t)
+
+ pqReader, err := file.NewParquetReader(bytes.NewReader(data))
+ require.NoError(t, err)
+ defer pqReader.Close()
+ chunk, err := pqReader.MetaData().RowGroup(0).ColumnChunk(0)
+ require.NoError(t, err)
+ assert.True(t, chunk.HasDictionaryPage())
+ assert.Contains(t, chunk.Encodings(), parquet.Encodings.RLEDict)
+ assert.Contains(t, chunk.Encodings(), parquet.Encodings.Plain)
+ var hasPlainData bool
+ for _, stat := range chunk.EncodingStats() {
+ if stat.PageType == file.PageTypeDataPage && stat.Encoding ==
parquet.Encodings.Plain {
+ hasPlainData = true
+ }
+ }
+ require.True(t, hasPlainData, "expected a plain fallback data page")
+
+ rdr := openBloomTestReader(t, data)
+ tester := &internal.ParquetRowGroupTester{
Review Comment:
nit: this `rdr` isn't deferred, so if an earlier `require` fails the reader
leaks. The `pqReader` a few lines up uses `defer`. Could we `defer rdr.Close()`
(or `t.Cleanup`) for consistency?
##########
table/internal/parquet_files_test.go:
##########
@@ -2211,6 +2211,267 @@ func TestBloomFilterRowGroupPruning(t *testing.T) {
})
}
+// buildDictionaryTestParquet writes a Parquet file with two low-cardinality
+// dictionary-encoded row groups. The single required INT32 column "id" has
+// Iceberg field_id=1.
+func buildDictionaryTestParquet(t testing.TB, rowGroups ...[]int32) []byte {
+ t.Helper()
+
+ idNode := schema.NewInt32Node("id", parquet.Repetitions.Required, 1)
+ rootNode, err := schema.NewGroupNode("schema",
parquet.Repetitions.Required,
+ schema.FieldList{idNode}, -1)
+ require.NoError(t, err)
+
+ writerProps := parquet.NewWriterProperties(
+ parquet.WithStats(true),
+ parquet.WithDictionaryDefault(true),
+ )
+
+ var buf bytes.Buffer
+ pw := file.NewParquetWriter(&buf, rootNode,
file.WithWriterProps(writerProps))
+ for _, values := range rowGroups {
+ rgw, werr := pw.AppendRowGroupChecked()
+ require.NoError(t, werr)
+ cw, werr := rgw.NextColumn()
+ require.NoError(t, werr)
+ _, werr = cw.(*file.Int32ColumnChunkWriter).WriteBatch(values,
nil, nil)
+ require.NoError(t, werr)
+ require.NoError(t, cw.Close())
+ require.NoError(t, rgw.Close())
+ }
+ require.NoError(t, pw.Close())
+
+ return buf.Bytes()
+}
+
+// TestDictionaryRowGroupPruning verifies that dictionary-only row groups are
+// skipped when an EqualTo/In predicate has no value in the complete
dictionary.
+func TestDictionaryRowGroupPruning(t *testing.T) {
+ const rgSize = 1024
+ rg0 := make([]int32, rgSize)
+ rg1 := make([]int32, rgSize)
+ for i := range rg0 {
+ rg0[i] = 1 + int32(i%2)*2
+ rg1[i] = 5 + int32(i%2)*2
+ }
+
+ data := buildDictionaryTestParquet(t, rg0, rg1)
+ alwaysKeep := func(_ *metadata.RowGroupMetaData, _ []int) (bool, error)
{
+ return true, nil
+ }
+
+ ctx := context.Background()
+ cols := []int{0}
+
+ t.Run("EqualTo absent from both dictionaries", func(t *testing.T) {
+ rdr := openBloomTestReader(t, data)
+ var survivors []internal.RowGroupSpan
+ tester := &internal.ParquetRowGroupTester{
+ DictionaryPreds: []internal.RowGroupDictionaryPred{
+ {FieldID: 1, PhysBytes:
[][]byte{int32PhysBytes(2)}},
+ },
+ Survivors: &survivors,
+ }
+ rr, err := rdr.GetRecords(ctx, cols, tester)
+ require.NoError(t, err)
+
+ assert.Equal(t, int64(0), countRecords(t, rr))
+ assert.Empty(t, survivors)
+ require.NoError(t, rdr.Close())
+ })
+
+ t.Run("EqualTo keeps the matching dictionary", func(t *testing.T) {
+ rdr := openBloomTestReader(t, data)
+ var survivors []internal.RowGroupSpan
+ tester := &internal.ParquetRowGroupTester{
+ DictionaryPreds: []internal.RowGroupDictionaryPred{
+ {FieldID: 1, PhysBytes:
[][]byte{int32PhysBytes(7)}},
+ },
+ StatsFn: alwaysKeep,
+ Survivors: &survivors,
+ }
+ rr, err := rdr.GetRecords(ctx, cols, tester)
+ require.NoError(t, err)
+
+ assert.Equal(t, int64(rgSize), countRecords(t, rr))
+ assert.Equal(t, []internal.RowGroupSpan{{FirstRowPos: rgSize,
NumRows: rgSize}}, survivors)
+ require.NoError(t, rdr.Close())
+ })
+
+ t.Run("In keeps a row group when any value is present", func(t
*testing.T) {
+ rdr := openBloomTestReader(t, data)
+ var survivors []internal.RowGroupSpan
+ tester := &internal.ParquetRowGroupTester{
+ DictionaryPreds: []internal.RowGroupDictionaryPred{
+ {FieldID: 1, PhysBytes:
[][]byte{int32PhysBytes(2), int32PhysBytes(5)}},
+ },
+ StatsFn: alwaysKeep,
+ Survivors: &survivors,
+ }
+ rr, err := rdr.GetRecords(ctx, cols, tester)
+ require.NoError(t, err)
+
+ assert.Equal(t, int64(rgSize), countRecords(t, rr))
+ assert.Equal(t, []internal.RowGroupSpan{{FirstRowPos: rgSize,
NumRows: rgSize}}, survivors)
+ require.NoError(t, rdr.Close())
+ })
+}
+
+func BenchmarkDictionaryRowGroupPruning(b *testing.B) {
+ const (
+ numRowGroups = 16
+ rowsPerGroup = 4096
+ )
+
+ rowGroups := make([][]int32, numRowGroups)
+ for group := range rowGroups {
+ values := make([]int32, rowsPerGroup)
+ for i := range values {
+ values[i] = int32(group*2 + i%2)
+ }
+ rowGroups[group] = values
+ }
+
+ data := buildDictionaryTestParquet(b, rowGroups...)
+ alwaysKeep := func(_ *metadata.RowGroupMetaData, _ []int) (bool, error)
{
+ return true, nil
+ }
+ cols := []int{0}
+ benchmarks := []struct {
+ name string
+ physBytes [][]byte
+ wantRows int64
+ }{
+ {name: "without dictionary", wantRows: int64(numRowGroups *
rowsPerGroup)},
+ {name: "dictionary target absent", physBytes:
[][]byte{int32PhysBytes(-1)}},
Review Comment:
The benchmark covers absent-from-all and present-in-one, but not the case
that actually costs us: value present in all 16 dictionaries, 0 pruned. That's
the path that pays the full `GetDictionaryPage` + `Close` per row group and
prunes nothing, and against remote storage each of those is a separate
seek+read stacked on top of the bloom pass.
I'd add that fourth case and check it isn't materially slower than "without
dictionary". If it is, that's the signal we need a heuristic (only fire when
stats already pruned some fraction). wdyt?
--
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]
