From 6faa9d20e804157e65e7e0746893a6aa39163e06 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 21 Aug 2026 10:12:01 -0700 Subject: [PATCH] iceberg: stop compaction from corrupting dictionary-encoded columns (#10857) * deps: upgrade parquet-go to v0.32.0 Iceberg compaction writes the merged file with the schema of its first input, encodings included. parquet-go before v0.31.0 took the deprecated PLAIN_DICTIONARY encoding that DuckDB writes at face value and encoded those pages as plain int32 indices, but the spec gives PLAIN_DICTIONARY the same bit-width-prefixed RLE layout as RLE_DICTIONARY. Every dictionary-encoded column in a compacted file then decoded onto a single dictionary entry, and anything past one page failed to decode at all. * iceberg: cover compaction of dictionary-encoded input The fixture is a DuckDB-written file, so it carries the PLAIN_DICTIONARY encoding a Go writer will not produce. * iceberg: tally whole rows in the dictionary merge test Counting each column on its own passes a merge that remaps names while leaving their cardinality intact. --- go.mod | 2 +- go.sum | 4 +- weed/worker/tasks/iceberg/handler_test.go | 73 ++++++++++++++++++ .../iceberg/testdata/plain-dictionary.parquet | Bin 0 -> 613 bytes 4 files changed, 76 insertions(+), 3 deletions(-) create mode 100644 weed/worker/tasks/iceberg/testdata/plain-dictionary.parquet diff --git a/go.mod b/go.mod index c652f977e..03866bf01 100644 --- a/go.mod +++ b/go.mod @@ -141,7 +141,7 @@ require ( github.com/linkedin/goavro/v2 v2.15.0 github.com/minio/crc64nvme v1.1.1 github.com/orcaman/concurrent-map/v2 v2.0.1 - github.com/parquet-go/parquet-go v0.30.1 + github.com/parquet-go/parquet-go v0.32.0 github.com/pkg/sftp v1.13.11 github.com/rabbitmq/amqp091-go v1.13.0 github.com/rclone/rclone v1.75.0 diff --git a/go.sum b/go.sum index 83731e42a..b455c7c7f 100644 --- a/go.sum +++ b/go.sum @@ -1636,8 +1636,8 @@ github.com/parquet-go/bitpack v1.0.0 h1:AUqzlKzPPXf2bCdjfj4sTeacrUwsT7NlcYDMUQxP github.com/parquet-go/bitpack v1.0.0/go.mod h1:XnVk9TH+O40eOOmvpAVZ7K2ocQFrQwysLMnc6M/8lgs= github.com/parquet-go/jsonlite v1.0.0 h1:87QNdi56wOfsE5bdgas0vRzHPxfJgzrXGml1zZdd7VU= github.com/parquet-go/jsonlite v1.0.0/go.mod h1:nDjpkpL4EOtqs6NQugUsi0Rleq9sW/OtC1NnZEnxzF0= -github.com/parquet-go/parquet-go v0.30.1 h1:Oy6ganNrAdFiVwy7wNmWagfPTWA2X9Z3tVHBc7JtuX8= -github.com/parquet-go/parquet-go v0.30.1/go.mod h1:navtkAYr2LGoJVp141oXPlO/sxLvaOe3la2JEoD8+rg= +github.com/parquet-go/parquet-go v0.32.0 h1:NWDqTUHfrCS4cJP/Fj2HlxvqsrVedWG3sayMkf+znzM= +github.com/parquet-go/parquet-go v0.32.0/go.mod h1:navtkAYr2LGoJVp141oXPlO/sxLvaOe3la2JEoD8+rg= github.com/pascaldekloe/goe v0.1.0 h1:cBOtyMzM9HTpWjXfbbunk26uA6nG3a8n06Wieeh0MwY= github.com/pascaldekloe/goe v0.1.0/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc= github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc= diff --git a/weed/worker/tasks/iceberg/handler_test.go b/weed/worker/tasks/iceberg/handler_test.go index e25bfc1aa..598c65296 100644 --- a/weed/worker/tasks/iceberg/handler_test.go +++ b/weed/worker/tasks/iceberg/handler_test.go @@ -6,7 +6,9 @@ import ( "encoding/json" "fmt" "io" + "os" "path" + "path/filepath" "strconv" "testing" "time" @@ -1449,6 +1451,77 @@ func TestMergeParquetFilesWithEqualityDeletes(t *testing.T) { } } +// testdata/plain-dictionary.parquet was written by DuckDB, which labels its +// dictionary pages with the deprecated PLAIN_DICTIONARY encoding. The merge +// writer inherits that encoding from the input schema, and encoding those +// pages as plain int32 indices instead of RLE collapses every row of the +// column onto one dictionary entry. +func TestMergeParquetFilesDictionaryEncodedInput(t *testing.T) { + fs, client := startFakeFiler(t) + + content, err := os.ReadFile(filepath.Join("testdata", "plain-dictionary.parquet")) + if err != nil { + t.Fatalf("read fixture: %v", err) + } + + dataDir := "/buckets/test-bucket/ns/tbl/data" + spec := *iceberg.UnpartitionedSpec + var entries []iceberg.ManifestEntry + for _, name := range []string{"dict1.parquet", "dict2.parquet"} { + fs.putEntry(dataDir, name, &filer_pb.Entry{Name: name, Content: content}) + dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, "data/"+name, iceberg.ParquetFile, map[int]any{}, nil, nil, 200, int64(len(content))) + if err != nil { + t.Fatalf("build data file: %v", err) + } + snapID := int64(1) + entries = append(entries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build())) + } + + merged, count, err := mergeParquetFiles( + context.Background(), client, "test-bucket", "ns/tbl", + entries, nil, nil, nil, + ) + if err != nil { + t.Fatalf("mergeParquetFiles: %v", err) + } + if count != 400 { + t.Fatalf("expected 400 merged rows, got %d", count) + } + + type dictRow struct { + ID int64 `parquet:"id"` + Name string `parquet:"name"` + } + tally := func(data []byte, what string) map[dictRow]int { + counts := map[dictRow]int{} + reader := parquet.NewReader(bytes.NewReader(data)) + defer reader.Close() + for { + var r dictRow + err := reader.Read(&r) + if err == io.EOF { + break + } + if err != nil { + t.Fatalf("read %s row: %v", what, err) + } + counts[r]++ + } + return counts + } + + source := tally(content, "fixture") + got := tally(merged, "merged") + if len(got) != len(source) { + t.Fatalf("expected %d distinct rows, got %d", len(source), len(got)) + } + for row, n := range source { + if got[row] != 2*n { + t.Errorf("row %+v appears %d times, expected %d", row, got[row], 2*n) + } + } +} + func TestDetectNilRequest(t *testing.T) { handler := NewHandler(nil) err := handler.Detect(nil, nil, nil) diff --git a/weed/worker/tasks/iceberg/testdata/plain-dictionary.parquet b/weed/worker/tasks/iceberg/testdata/plain-dictionary.parquet new file mode 100644 index 0000000000000000000000000000000000000000..8b8282625ef21a9cd7dd481d76152efb3d20d2e3 GIT binary patch literal 613 zcmbtSJ8K(35T1SHyb_YI40G&>D=y@8oIx>`<%?aoI0s3%2;m4RTqTio3Kq6x<5w-) z5LCpV{Da_HMe-L?xC|tPNSVsyXJk%~K!Uq0`!UNrzS;S9qwN=Elz4>EGt40XY)TF( z1|qc}L1q;^AQ)M-wLr0Cv8e$B{LU~c;ZRV$fd)Y5R8VjznAjfN7QXF1D(qKV*L$P< zb#U{!zCIo&{LdLs{(FWmXASEvl|rx4Z9iVaG*k%dgn;k_)BeVvX!(CyhPDkb2k5?5 z@08@c)=vZXcA6}dEc`ME>)`lS2>5q8gdaXVbU9a7ayU(A`gZEY7D}aPDGohUHnv#AL<{k5XaPJqhwJ}k8V`j?}%71(b z-AE>M=8ZwO@|1Zi!1t>(UiGPS-abC*9QK^^@=CB0IDYf2a}Yc6W-D$qDwVhKvQ