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 000000000..8b8282625 Binary files /dev/null and b/weed/worker/tasks/iceberg/testdata/plain-dictionary.parquet differ