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

@@ -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

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=

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)