This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 849c7a3d fix(table): enable Parquet dictionary encoding by default
(#931)
849c7a3d is described below
commit 849c7a3d969be39475467c1fa0404e851b47153a
Author: Tobias Pütz <[email protected]>
AuthorDate: Thu Aug 6 18:35:36 2026 +0200
fix(table): enable Parquet dictionary encoding by default (#931)
* fix(table): enable Parquet dictionary encoding by default
* add requested test
* fixes
* test(parquet): use AppendRowGroupChecked in dictionary test helpers
Replace deprecated file.ParquetWriter.AppendRowGroup with the checked
variant to satisfy staticcheck SA1019, matching the rest of the file.
Signed-off-by: Tobias Pütz <[email protected]>
* address review
---------
Signed-off-by: Tobias Pütz <[email protected]>
---
table/internal/parquet_files.go | 42 +++++-
table/internal/parquet_files_test.go | 256 ++++++++++++++++++++++++++++++++++-
website/src/configuration.md | 4 +
3 files changed, 294 insertions(+), 8 deletions(-)
diff --git a/table/internal/parquet_files.go b/table/internal/parquet_files.go
index dd301a75..56e9e939 100644
--- a/table/internal/parquet_files.go
+++ b/table/internal/parquet_files.go
@@ -70,6 +70,13 @@ const (
ParquetBloomFilterMaxBytesDefault = 1024 * 1024
ParquetBloomFilterColumnEnabledKeyPrefix =
"write.parquet.bloom-filter-enabled.column"
+ // Deliberately not namespaced under write.parquet: this is the
parquet-mr key
+ // Iceberg Java reads straight from the table properties, so a table
carrying
+ // it gets the same dictionary behavior from either implementation.
+ ParquetDictEnabledKey = "parquet.enable.dictionary"
+ ParquetDictEnabledDefault = true
+ ParquetDictEncodingColumnEnabledKeyPrefix =
"write.parquet.dict-encoding-enabled.column"
+
ParquetShredVariantsKey = "write.parquet.shred-variants"
ParquetShredVariantsDefault = false
// Rows buffered per file to infer shredding (held per open partition
writer).
@@ -260,7 +267,6 @@ func (parquetFormat) GetWriteProperties(props
iceberg.Properties) any {
}
writerProps := []parquet.WriterProperty{
- parquet.WithDictionaryDefault(false),
parquet.WithMaxRowGroupLength(int64(props.GetInt(ParquetRowGroupLimitKey,
ParquetRowGroupLimitDefault))),
parquet.WithDataPageSize(int64(props.GetInt(ParquetPageSizeBytesKey,
@@ -314,16 +320,40 @@ func (parquetFormat) GetWriteProperties(props
iceberg.Properties) any {
if !ok || colName == "" {
continue
}
- // EqualFold matches Java's Boolean.parseBoolean: only "true"
- // (case-insensitive) is true; anything else, including "1" or
- // "yes", is false. strconv.ParseBool behaves differently ("1"
→ true).
- enabled := strings.EqualFold(val, "true")
- writerProps = append(writerProps,
parquet.WithBloomFilterEnabledFor(colName, enabled))
+ writerProps = append(writerProps,
parquet.WithBloomFilterEnabledFor(colName, javaBool(val)))
+ }
+
+ // Dictionary encoding is on unless parquet.enable.dictionary disables
it;
+ // write.parquet.dict-encoding-enabled.column.<col-name> overrides per
column.
+ dictEnabled := ParquetDictEnabledDefault
+ if val, ok := props[ParquetDictEnabledKey]; ok {
+ dictEnabled = javaBool(val)
+ }
+ writerProps = append(writerProps,
parquet.WithDictionaryDefault(dictEnabled))
+
+ // The suffix is matched against the Parquet physical column path, not
the
+ // Iceberg field name. These differ for nested fields (Iceberg
`items.element`
+ // is `items.list.element` in Parquet), so a per-column override on one
of
+ // those has no effect. Same for the bloom filter prefix above.
+ prefix = ParquetDictEncodingColumnEnabledKeyPrefix + "."
+ for key, val := range props {
+ colName, ok := strings.CutPrefix(key, prefix)
+ if !ok || colName == "" {
+ continue
+ }
+ writerProps = append(writerProps,
parquet.WithDictionaryFor(colName, javaBool(val)))
}
return writerProps
}
+// javaBool matches Java's Boolean.parseBoolean, which Iceberg's Java writer
uses
+// for these properties: only "true" (case-insensitive) is true; anything else,
+// including "1" and "yes", is false. strconv.ParseBool differs ("1" → true).
+func javaBool(val string) bool {
+ return strings.EqualFold(val, "true")
+}
+
// ParquetRowGroupTargetSizeBytes returns the configured uncompressed row-group
// size target. Iceberg Java rejects invalid and non-positive values rather
than
// silently falling back to the default.
diff --git a/table/internal/parquet_files_test.go
b/table/internal/parquet_files_test.go
index a7deab5f..b1c8c97f 100644
--- a/table/internal/parquet_files_test.go
+++ b/table/internal/parquet_files_test.go
@@ -1247,12 +1247,264 @@ func TestGetWritePropertiesPageVersion(t *testing.T) {
pageRdr, err := rdr.RowGroup(0).GetColumnPageReader(0)
require.NoError(t, err)
- require.True(t, pageRdr.Next())
- assert.Equal(t, tt.expectedPageType,
pageRdr.Page().Type())
+ var dataPageType file.PageType
+ var foundDataPage bool
+ for pageRdr.Next() {
+ pt := pageRdr.Page().Type()
+ if pt == file.PageTypeDictionaryPage {
+ continue
+ }
+ dataPageType = pt
+ foundDataPage = true
+
+ break
+ }
+ require.True(t, foundDataPage, "expected a data page in
row group")
+ assert.Equal(t, tt.expectedPageType, dataPageType)
})
}
}
+func TestGetWritePropertiesDictionaryEnabledByDefault(t *testing.T) {
+ format := internal.GetFileFormat(iceberg.ParquetFile)
+ writeProps :=
format.GetWriteProperties(iceberg.Properties{}).([]parquet.WriterProperty)
+
+ root, err := schema.NewGroupNode("schema",
parquet.Repetitions.Required, schema.FieldList{
+ schema.NewInt32Node("col", parquet.Repetitions.Required, -1),
+ }, -1)
+ require.NoError(t, err)
+
+ var buf bytes.Buffer
+ pw := file.NewParquetWriter(&buf, root, file.WithWriterProps(
+ parquet.NewWriterProperties(writeProps...),
+ ))
+
+ rgw, err := pw.AppendRowGroupChecked()
+ require.NoError(t, err)
+ cw, err := rgw.NextColumn()
+ require.NoError(t, err)
+
+ // Low-cardinality column stays dictionary-encoded (well under the
dict-size limit).
+ values := make([]int32, 1024)
+ for i := range values {
+ values[i] = int32(i % 8)
+ }
+ _, err = cw.(*file.Int32ColumnChunkWriter).WriteBatch(values, nil, nil)
+ require.NoError(t, err)
+ require.NoError(t, cw.Close())
+ require.NoError(t, rgw.Close())
+ require.NoError(t, pw.Close())
+
+ rdr, err := file.NewParquetReader(bytes.NewReader(buf.Bytes()))
+ require.NoError(t, err)
+ defer rdr.Close()
+
+ colMeta, err := rdr.MetaData().RowGroup(0).ColumnChunk(0)
+ require.NoError(t, err)
+
+ assert.True(t, colMeta.HasDictionaryPage(),
+ "expected a dictionary page with default write properties")
+ assert.Contains(t, colMeta.Encodings(), parquet.Encodings.RLEDict,
+ "expected RLE_DICTIONARY encoding with default write
properties")
+}
+
+func TestGetWritePropertiesDictionaryEncoding(t *testing.T) {
+ format := internal.GetFileFormat(iceberg.ParquetFile)
+ writerProps := func(props iceberg.Properties) *parquet.WriterProperties
{
+ return
parquet.NewWriterProperties(format.GetWriteProperties(props).([]parquet.WriterProperty)...)
+ }
+
+ t.Run("enabled by default", func(t *testing.T) {
+ wp := writerProps(iceberg.Properties{})
+ assert.True(t, wp.DictionaryEnabled())
+ assert.True(t, wp.DictionaryEnabledFor("any_col"))
+ })
+
+ t.Run("global opt-out", func(t *testing.T) {
+ wp :=
writerProps(iceberg.Properties{internal.ParquetDictEnabledKey: "false"})
+ assert.False(t, wp.DictionaryEnabled())
+ assert.False(t, wp.DictionaryEnabledFor("any_col"))
+ })
+
+ t.Run("per-column opt-out under the default", func(t *testing.T) {
+ wp := writerProps(iceberg.Properties{
+ internal.ParquetDictEncodingColumnEnabledKeyPrefix +
".name": "false",
+ })
+ assert.False(t, wp.DictionaryEnabledFor("name"))
+ assert.True(t, wp.DictionaryEnabledFor("id"), "unmentioned
columns follow the global setting")
+ })
+
+ t.Run("per-column opt-in under a global opt-out", func(t *testing.T) {
+ wp := writerProps(iceberg.Properties{
+ internal.ParquetDictEnabledKey:
"false",
+ internal.ParquetDictEncodingColumnEnabledKeyPrefix +
".id": "true",
+ internal.ParquetDictEncodingColumnEnabledKeyPrefix +
".name": "false",
+ })
+ assert.True(t, wp.DictionaryEnabledFor("id"))
+ assert.False(t, wp.DictionaryEnabledFor("name"))
+ assert.False(t, wp.DictionaryEnabledFor("unmentioned_col"),
"unmentioned columns follow the global setting")
+ })
+
+ // Java reads both keys with Boolean.parseBoolean, so only "true"
enables.
+ t.Run("java boolean semantics", func(t *testing.T) {
+ for _, val := range []string{"true", "TRUE", "True"} {
+ assert.True(t,
writerProps(iceberg.Properties{internal.ParquetDictEnabledKey:
val}).DictionaryEnabled(), val)
+ }
+ for _, val := range []string{"false", "FALSE", "1", "yes", "",
"bogus"} {
+ assert.False(t,
writerProps(iceberg.Properties{internal.ParquetDictEnabledKey:
val}).DictionaryEnabled(), val)
+ }
+ })
+}
+
+func TestGetWritePropertiesDictionaryEncodingRoundTrip(t *testing.T) {
+ values := make([]parquet.ByteArray, 4096)
+ for i := range values {
+ values[i] = parquet.ByteArray(fmt.Sprintf("cat_%02d", i%8))
+ }
+
+ t.Run("default writes a dictionary", func(t *testing.T) {
+ chunk := writeDictTestColumn(t, iceberg.Properties{}, values)
+ assert.True(t, chunk.hasDictPage)
+ assert.Positive(t, chunk.dictDataPages)
+ assert.Zero(t, chunk.plainDataPages)
+ })
+
+ t.Run("global opt-out writes no dictionary", func(t *testing.T) {
+ chunk := writeDictTestColumn(t, iceberg.Properties{
+ internal.ParquetDictEnabledKey: "false",
+ }, values)
+ assert.False(t, chunk.hasDictPage)
+ assert.Zero(t, chunk.dictDataPages)
+ assert.Positive(t, chunk.plainDataPages)
+ })
+
+ t.Run("per-column opt-out writes no dictionary", func(t *testing.T) {
+ chunk := writeDictTestColumn(t, iceberg.Properties{
+ internal.ParquetDictEncodingColumnEnabledKeyPrefix +
"." + dictTestColumn: "false",
+ }, values)
+ assert.False(t, chunk.hasDictPage)
+ assert.Zero(t, chunk.dictDataPages)
+ assert.Positive(t, chunk.plainDataPages)
+ })
+
+ t.Run("per-column opt-in under a global opt-out writes a dictionary",
func(t *testing.T) {
+ chunk := writeDictTestColumn(t, iceberg.Properties{
+ internal.ParquetDictEnabledKey:
"false",
+ internal.ParquetDictEncodingColumnEnabledKeyPrefix +
"." + dictTestColumn: "true",
+ }, values)
+ assert.True(t, chunk.hasDictPage)
+ assert.Positive(t, chunk.dictDataPages)
+ assert.Zero(t, chunk.plainDataPages)
+ })
+}
+
+// TestGetWritePropertiesDictionaryHighCardinalityFallback guards the risk that
+// kept dictionary encoding disabled until now: with the default write
properties
+// a column too distinct to dictionary encode must fall back to PLAIN, strand
no
+// dictionary page, and not grow the chunk against a dictionary-disabled write.
+func TestGetWritePropertiesDictionaryHighCardinalityFallback(t *testing.T) {
+ const valueWidth = 32
+
+ // Enough distinct values to overflow the default 2 MB dictionary
budget.
+ values := make([]parquet.ByteArray, 100000)
+ for i := range values {
+ b := make([]byte, valueWidth)
+ for j := range b {
+ b[j] = byte('A' + (i+j*31)%26)
+ }
+ tail := fmt.Appendf(nil, "-%07d", i)
+ copy(b[valueWidth-len(tail):], tail)
+ values[i] = parquet.ByteArray(b)
+ }
+
+ dictOn := writeDictTestColumn(t, iceberg.Properties{}, values)
+ dictOff := writeDictTestColumn(t, iceberg.Properties{
+ internal.ParquetDictEnabledKey: "false",
+ }, values)
+
+ t.Logf("dict=on compressed=%d hasDictPage=%v dictPages=%d
plainPages=%d; dict=off compressed=%d",
+ dictOn.compressedSize, dictOn.hasDictPage,
dictOn.dictDataPages, dictOn.plainDataPages,
+ dictOff.compressedSize)
+
+ assert.False(t, dictOn.hasDictPage, "overflowed dictionary must be
discarded, not stranded in the chunk")
+ assert.Zero(t, dictOn.dictDataPages)
+ assert.Positive(t, dictOn.plainDataPages, "high-cardinality column must
fall back to PLAIN")
+
+ // Discarding re-encodes the buffered values as PLAIN, so the chunk
matches a
+ // dictionary-disabled write up to page-boundary rounding.
+ assert.LessOrEqual(t, dictOn.compressedSize,
dictOff.compressedSize+dictOff.compressedSize/50,
+ "dictionary fallback must not grow the column chunk")
+}
+
+const dictTestColumn = "v"
+
+type dictChunkSummary struct {
+ compressedSize int64
+ hasDictPage bool
+ dictDataPages int
+ plainDataPages int
+}
+
+// writeDictTestColumn writes values as a single required BYTE_ARRAY column
using
+// the write properties derived from tableProps, then summarizes the encodings
of
+// the resulting column chunk.
+func writeDictTestColumn(t *testing.T, tableProps iceberg.Properties, values
[]parquet.ByteArray) dictChunkSummary {
+ t.Helper()
+
+ format := internal.GetFileFormat(iceberg.ParquetFile)
+ writeProps :=
format.GetWriteProperties(tableProps).([]parquet.WriterProperty)
+
+ root, err := schema.NewGroupNode("schema",
parquet.Repetitions.Required, schema.FieldList{
+ schema.NewByteArrayNode(dictTestColumn,
parquet.Repetitions.Required, -1),
+ }, -1)
+ require.NoError(t, err)
+
+ var buf bytes.Buffer
+ pw := file.NewParquetWriter(&buf, root, file.WithWriterProps(
+ parquet.NewWriterProperties(writeProps...),
+ ))
+
+ rgw, err := pw.AppendRowGroupChecked()
+ require.NoError(t, err)
+ cw, err := rgw.NextColumn()
+ require.NoError(t, err)
+ _, err = cw.(*file.ByteArrayColumnChunkWriter).WriteBatch(values, nil,
nil)
+ require.NoError(t, err)
+ require.NoError(t, cw.Close())
+ require.NoError(t, rgw.Close())
+ require.NoError(t, pw.Close())
+
+ rdr, err := file.NewParquetReader(bytes.NewReader(buf.Bytes()))
+ require.NoError(t, err)
+ defer rdr.Close()
+
+ colMeta, err := rdr.MetaData().RowGroup(0).ColumnChunk(0)
+ require.NoError(t, err)
+
+ summary := dictChunkSummary{
+ compressedSize: colMeta.TotalCompressedSize(),
+ hasDictPage: colMeta.HasDictionaryPage(),
+ }
+
+ pageRdr, err := rdr.RowGroup(0).GetColumnPageReader(0)
+ require.NoError(t, err)
+ for pageRdr.Next() {
+ page := pageRdr.Page()
+ if page.Type() != file.PageTypeDataPage && page.Type() !=
file.PageTypeDataPageV2 {
+ continue
+ }
+ switch parquet.Encoding(page.Encoding()) {
+ case parquet.Encodings.RLEDict, parquet.Encodings.PlainDict:
+ summary.dictDataPages++
+ case parquet.Encodings.Plain:
+ summary.plainDataPages++
+ }
+ }
+ require.NoError(t, pageRdr.Err())
+
+ return summary
+}
+
func TestGetWritePropertiesBloomFilter(t *testing.T) {
format := internal.GetFileFormat(iceberg.ParquetFile)
diff --git a/website/src/configuration.md b/website/src/configuration.md
index 827c24ca..277984cc 100644
--- a/website/src/configuration.md
+++ b/website/src/configuration.md
@@ -331,9 +331,13 @@ All defined in `table/internal/parquet_files.go` and read
by the Parquet writer:
| `write.parquet.compression-level` | `-1` (codec default) |
| `write.parquet.bloom-filter-max-bytes` | 1 MB |
| `write.parquet.bloom-filter-enabled.column.<name>` | (per-column toggle,
prefix-matched) |
+| `parquet.enable.dictionary` | `true` (global dictionary encoding switch;
same key as Iceberg Java) |
+| `write.parquet.dict-encoding-enabled.column.<name>` | (per-column toggle,
prefix-matched; overrides the global switch) |
| `write.parquet.shred-variants` | `false` (opt-in shredding of top-level
variant columns) |
| `write.parquet.variant-inference-buffer-size` | 100 rows (buffered per file
to infer the shredding schema) |
+The bloom filter and dictionary toggles are parsed the way Iceberg Java parses
them: only `true` (case-insensitive) enables, so any other value — including
`1` — reads as `false`.
+
### Parquet reader
| Key | Description |
