diff --git a/weed/worker/tasks/iceberg/compact.go b/weed/worker/tasks/iceberg/compact.go index a6ec51bec..eb99e68ca 100644 --- a/weed/worker/tasks/iceberg/compact.go +++ b/weed/worker/tasks/iceberg/compact.go @@ -189,9 +189,11 @@ func (h *Handler) compactDataFiles( return "", nil, err } - // Build compaction bins: group small data files by partition. + // Build compaction bins: group small data files by partition, each bin's + // files in the order of their bounds so the merged file keeps the + // ordering its inputs had. targetSize := compactionTargetSizeForPlan(config, rewritePlan) - bins := buildCompactionBins(candidateEntries, targetSize, minInputFiles) + bins := buildCompactionBins(candidateEntries, targetSize, minInputFiles, meta) initialBinCount := len(bins) bins = filterCompactionBinsByPlan(bins, config, rewritePlan) if len(bins) == 0 { @@ -269,10 +271,11 @@ func (h *Handler) compactDataFiles( var mergedData []byte var recordCount int64 + rowGroupRows := rowsPerRowGroup(bin, config) if rewritePlan != nil && rewritePlan.strategy == "sort" { - mergedData, recordCount, err = mergeParquetFilesSorted(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rewritePlan) + mergedData, recordCount, err = mergeParquetFilesSorted(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rewritePlan, rowGroupRows) } else { - mergedData, recordCount, err = mergeParquetFiles(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema) + mergedData, recordCount, err = mergeParquetFiles(ctx, filerClient, bucketName, dataPath, bin.Entries, positionDeletes, eqDeleteGroups, schema, rowGroupRows) } if err != nil { glog.Warningf("iceberg compact: failed to merge bin %d (%d files): %v", binIdx, len(bin.Entries), err) @@ -314,6 +317,14 @@ func (h *Handler) compactDataFiles( } writtenArtifacts = append(writtenArtifacts, artifact{dir: dataDir, fileName: mergedFileName}) + // Record the column statistics readers prune files by; without + // them every scan has to read every compacted file. + if stats, statsErr := collectColumnStats(mergedData, schema); statsErr != nil { + glog.Warningf("iceberg compact: no column statistics for %s: %v", mergedFileName, statsErr) + } else { + stats.applyTo(dfBuilder) + } + mergedDataFile := dfBuilder.Build() summary.addFile(mergedDataFile) newEntry := iceberg.NewManifestEntry( @@ -527,7 +538,13 @@ func (h *Handler) compactDataFiles( // buildCompactionBins groups small data files by partition for bin-packing. // A file is "small" if it's below targetSize. A bin must have at least // minFiles entries to be worth compacting. -func buildCompactionBins(entries []iceberg.ManifestEntry, targetSize int64, minFiles int) []compactionBin { +// +// With an order, every bin lists its files by their bounds on the ordering +// column and an oversized partition is split into runs of consecutive files, +// so each merged file covers one contiguous range of that column. Without one +// the files keep manifest order and an oversized partition is packed +// largest-first. +func buildCompactionBins(entries []iceberg.ManifestEntry, targetSize int64, minFiles int, meta table.Metadata) []compactionBin { if minFiles < 2 { minFiles = 2 } @@ -560,14 +577,37 @@ func buildCompactionBins(entries []iceberg.ManifestEntry, targetSize int64, minF bin.TotalSize += df.FileSizeBytes() } - // Filter to bins with enough files, splitting oversized bins + // Filter to bins with enough files, splitting oversized bins. The merge + // order resolves per group over the files that survived the eligibility + // filter, so a file that can never participate (oversized, another + // format) cannot veto ordering for the ones that can. var result []compactionBin for _, bin := range groups { if len(bin.Entries) < minFiles { continue } + order := resolveCompactionOrder(meta, bin.Entries) + order.sortEntries(bin.Entries) if bin.TotalSize <= targetSize { result = append(result, *bin) + } else if order != nil { + runs, leftover := splitOrderedBin(*bin, targetSize, minFiles) + result = append(result, runs...) + if order.bestEffort && len(leftover) > 0 { + runtBin := compactionBin{PartitionKey: bin.PartitionKey, Partition: bin.Partition, SpecID: bin.SpecID, Entries: leftover} + packed := splitOversizedBin(runtBin, targetSize, minFiles) + if len(packed) == 0 { + // Stragglers cannot merge even with each other; repacking + // the whole bin lets them pair across the ordered runs. + // Only replace the runs when the repack yields bins — + // otherwise the runs stay valid compaction work. + if full := splitOversizedBin(*bin, targetSize, minFiles); len(full) > 0 { + result = result[:len(result)-len(runs)] + packed = full + } + } + result = append(result, packed...) + } } else { result = append(result, splitOversizedBin(*bin, targetSize, minFiles)...) } @@ -918,6 +958,46 @@ func resolveEqualityColIndices(pqSchema *parquet.Schema, fieldIDs []int, iceberg return indices, nil } +const ( + // Iceberg's write.parquet.row-group-size-bytes default and PyIceberg's + // write.parquet.row-group-limit default. + defaultRowGroupSizeBytes = 128 * 1024 * 1024 + defaultRowGroupRowLimit = 1048576 + minRowGroupRows = 1024 +) + +// rowsPerRowGroup is the most rows one row group of a compacted file holds: +// the row limit, lowered so a row group stays near the byte size, estimated +// from the bin's inputs' bytes per row. parquet-go's writers have no limit of +// their own, so without one every compacted file is a single row group. The +// floor only protects the estimate — a configured row limit always wins. +func rowsPerRowGroup(bin compactionBin, config Config) int64 { + rows, sizeBytes := config.RowGroupRowLimit, config.RowGroupSizeBytes + explicit := rows > 0 + if !explicit { + rows = defaultRowGroupRowLimit + } + if sizeBytes <= 0 { + sizeBytes = defaultRowGroupSizeBytes + } + var totalRows int64 + for _, entry := range bin.Entries { + totalRows += entry.DataFile().Count() + } + if totalRows > 0 && bin.TotalSize > 0 { + bytesPerRow := max(bin.TotalSize/totalRows, 1) + estimated := sizeBytes / bytesPerRow + if !explicit { + estimated = max(estimated, minRowGroupRows) + } + rows = min(rows, estimated) + } + if !explicit { + rows = max(rows, minRowGroupRows) + } + return rows +} + // mergeParquetFiles reads multiple small Parquet files and merges them into // a single Parquet file, optionally filtering out rows matching position or // equality deletes. Files are processed one at a time to keep memory usage @@ -930,6 +1010,7 @@ func mergeParquetFiles( positionDeletes map[string][]int64, eqDeleteGroups []equalityDeleteGroup, icebergSchema *iceberg.Schema, + rowGroupRows int64, ) ([]byte, int64, error) { if len(entries) == 0 { return nil, 0, fmt.Errorf("no entries to merge") @@ -954,7 +1035,7 @@ func mergeParquetFiles( } var outputBuf bytes.Buffer - writer := parquet.NewWriter(&outputBuf, parquetSchema) + writer := parquet.NewWriter(&outputBuf, parquetSchema, parquet.MaxRowsPerRowGroup(rowGroupRows)) drainReader := func(reader *parquet.Reader, source string) (int64, error) { return visitFilteredParquetRows(ctx, reader, source, bucketName, dataPath, positionDeletes, resolvedEqGroups, func(filtered []parquet.Row) error { @@ -1119,6 +1200,7 @@ func mergeParquetFilesSorted( eqDeleteGroups []equalityDeleteGroup, icebergSchema *iceberg.Schema, rewritePlan *compactionRewritePlan, + rowGroupRows int64, ) ([]byte, int64, error) { if len(entries) == 0 { return nil, 0, fmt.Errorf("no entries to merge") @@ -1168,6 +1250,7 @@ func mergeParquetFilesSorted( var outputBuf bytes.Buffer writer := parquet.NewSortingWriter[any](&outputBuf, sortBufferRows, parquetSchema, + parquet.MaxRowsPerRowGroup(rowGroupRows), parquet.SortingWriterConfig( parquet.SortingColumns(sortingColumns...), parquet.SortingBuffers(parquet.NewFileBufferPool(spillDir, "seaweedfs-iceberg-sort-*")), diff --git a/weed/worker/tasks/iceberg/compact_order.go b/weed/worker/tasks/iceberg/compact_order.go new file mode 100644 index 000000000..56b6176fb --- /dev/null +++ b/weed/worker/tasks/iceberg/compact_order.go @@ -0,0 +1,201 @@ +package iceberg + +import ( + "sort" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" +) + +// compactionOrder names the column by whose bounds a bin's input files are +// ordered before they are merged: the first identity field of the table's +// sort order when it declares one, otherwise the first column every candidate +// file carries bounds for. Rows are copied file by file, so a merged file is +// as ordered as the sequence of its inputs. +type compactionOrder struct { + fieldID int + typ iceberg.PrimitiveType + descending bool + // bestEffort marks an order inferred from column bounds rather than the + // table's declared sort order; runs it cannot form may still merge by + // size since contiguous ranges are a preference, not a requirement. + bestEffort bool +} + +func resolveCompactionOrder(meta table.Metadata, entries []iceberg.ManifestEntry) *compactionOrder { + if meta == nil || meta.CurrentSchema() == nil || len(entries) == 0 { + return nil + } + schema := meta.CurrentSchema() + if sortOrder := meta.SortOrder(); !sortOrder.IsUnsorted() { + for _, sortField := range sortOrder.Fields() { + if _, ok := sortField.Transform.(iceberg.IdentityTransform); !ok { + continue + } + field, ok := schema.FindFieldByID(sortField.SourceID()) + if !ok { + continue + } + typ, ok := field.Type.(iceberg.PrimitiveType) + if !ok { + continue + } + candidate := &compactionOrder{fieldID: field.ID, typ: typ, descending: sortField.Direction == table.SortDESC} + // The declared order only holds when every entry carries a bound + // for it; otherwise merging would claim sortedness it cannot + // verify. Fall back to inference instead of ordering by a later + // sort field alone. + if candidate.coversEntries(entries) { + return candidate + } + break + } + } + // A column only orders the merge when every entry in the group carries a + // bound for it. The check is scoped to the entries handed in — a bin's + // eligible files — so an unrelated file without bounds (oversized, a + // different format, or another partition) cannot disable ordering here. + for _, field := range schema.Fields() { + typ, ok := field.Type.(iceberg.PrimitiveType) + if !ok { + continue + } + candidate := &compactionOrder{fieldID: field.ID, typ: typ, bestEffort: true} + if candidate.coversEntries(entries) { + return candidate + } + } + return nil +} + +func (o *compactionOrder) coversEntries(entries []iceberg.ManifestEntry) bool { + for _, entry := range entries { + if _, ok := o.key(entry); !ok { + return false + } + } + return true +} + +// key is the bound a file is ordered by: its lower bound on the column for an +// ascending order, its upper bound for a descending one. +func (o *compactionOrder) key(entry iceberg.ManifestEntry) (iceberg.Literal, bool) { + df := entry.DataFile() + raw := df.LowerBoundValues()[o.fieldID] + if o.descending { + raw = df.UpperBoundValues()[o.fieldID] + } + if raw == nil { + return nil, false + } + lit, err := iceberg.LiteralFromBytes(o.typ, raw) + return lit, err == nil +} + +// sortEntries orders a bin's files by their key. Files without a key sort +// after the rest, in their original order. +func (o *compactionOrder) sortEntries(entries []iceberg.ManifestEntry) { + if o == nil || len(entries) < 2 { + return + } + type keyed struct { + key iceberg.Literal + ok bool + } + keys := make([]keyed, len(entries)) + idx := make([]int, len(entries)) + for i, entry := range entries { + k, ok := o.key(entry) + keys[i], idx[i] = keyed{k, ok}, i + } + sort.SliceStable(idx, func(i, j int) bool { + a, b := keys[idx[i]], keys[idx[j]] + if a.ok != b.ok { + return a.ok + } + if !a.ok { + return false + } + c, ok := compareLiterals(a.key, b.key) + if o.descending { + c = -c + } + return ok && c < 0 + }) + sorted := make([]iceberg.ManifestEntry, len(entries)) + for i, j := range idx { + sorted[i] = entries[j] + } + copy(entries, sorted) +} + +// splitOrderedBin splits a bin whose files are in bound order into runs of +// consecutive files that stay under targetSize, so every output covers one +// contiguous range of the ordering column. The returned entries are the runs +// too short to reach minFiles: the caller can still merge them by size when +// contiguous ranges are only a preference, or leave them for a later pass +// when order must hold. +func splitOrderedBin(bin compactionBin, targetSize int64, minFiles int) (valid []compactionBin, leftover []iceberg.ManifestEntry) { + newBin := func() compactionBin { + return compactionBin{PartitionKey: bin.PartitionKey, Partition: bin.Partition, SpecID: bin.SpecID} + } + current := newBin() + flush := func() { + if len(current.Entries) >= minFiles { + valid = append(valid, current) + } else { + leftover = append(leftover, current.Entries...) + } + current = newBin() + } + for _, entry := range bin.Entries { + size := entry.DataFile().FileSizeBytes() + if current.TotalSize > 0 && current.TotalSize+size > targetSize { + flush() + } + current.Entries = append(current.Entries, entry) + current.TotalSize += size + } + flush() + return valid, leftover +} + +// compareLiterals orders two literals of the same type; false when they are +// not comparable. +func compareLiterals(a, b iceberg.Literal) (int, bool) { + switch x := a.(type) { + case iceberg.TypedLiteral[bool]: + return compareTyped(x, b) + case iceberg.TypedLiteral[int32]: + return compareTyped(x, b) + case iceberg.TypedLiteral[int64]: + return compareTyped(x, b) + case iceberg.TypedLiteral[float32]: + return compareTyped(x, b) + case iceberg.TypedLiteral[float64]: + return compareTyped(x, b) + case iceberg.TypedLiteral[iceberg.Date]: + return compareTyped(x, b) + case iceberg.TypedLiteral[iceberg.Time]: + return compareTyped(x, b) + case iceberg.TypedLiteral[iceberg.Timestamp]: + return compareTyped(x, b) + case iceberg.TypedLiteral[iceberg.TimestampNano]: + return compareTyped(x, b) + case iceberg.TypedLiteral[string]: + return compareTyped(x, b) + case iceberg.TypedLiteral[[]byte]: + return compareTyped(x, b) + case iceberg.TypedLiteral[iceberg.Decimal]: + return compareTyped(x, b) + } + return 0, false +} + +func compareTyped[T iceberg.LiteralType](a iceberg.TypedLiteral[T], b iceberg.Literal) (int, bool) { + bt, ok := b.(iceberg.TypedLiteral[T]) + if !ok { + return 0, false + } + return a.Comparator()(a.Value(), bt.Value()), true +} diff --git a/weed/worker/tasks/iceberg/compact_order_test.go b/weed/worker/tasks/iceberg/compact_order_test.go new file mode 100644 index 000000000..8a4b1fdf7 --- /dev/null +++ b/weed/worker/tasks/iceberg/compact_order_test.go @@ -0,0 +1,388 @@ +package iceberg + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "path" + "strings" + "testing" + "time" + + "github.com/apache/iceberg-go" + "github.com/apache/iceberg-go/table" + + filer_pb "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +type boundedFile struct { + Name string + IDs []int64 + Lower, Upper int64 + SizeBytes int64 + NoBounds bool +} + +// populateBoundedTable is populateTableWithDeleteFiles narrowed to data +// files whose manifest entries carry bounds on the id column. +func populateBoundedTable(t *testing.T, fs *fakeFilerServer, setup tableSetup, files []boundedFile) { + t.Helper() + populateBoundedTableSorted(t, fs, setup, files, table.UnsortedSortOrder) +} + +func populateBoundedTableSorted(t *testing.T, fs *fakeFilerServer, setup tableSetup, files []boundedFile, sortOrder table.SortOrder) { + t.Helper() + schema := newTestSchema() + spec := *iceberg.UnpartitionedSpec + + meta, err := table.NewMetadata(schema, &spec, sortOrder, "s3://"+setup.BucketName+"/"+setup.dataPath(), nil) + if err != nil { + t.Fatalf("create metadata: %v", err) + } + + bucketPath := path.Join(s3tables.TablesPath, setup.BucketName) + nsPath := path.Join(bucketPath, setup.Namespace) + tableFilerPath := path.Join(bucketPath, setup.dataPath()) + metaDir := path.Join(tableFilerPath, "metadata") + dataDir := path.Join(tableFilerPath, "data") + version := meta.Version() + + var entries []iceberg.ManifestEntry + for _, f := range files { + rows := make([]struct { + ID int64 + Name string + }, len(f.IDs)) + for i, id := range f.IDs { + rows[i] = struct { + ID int64 + Name string + }{id, fmt.Sprintf("r%d", id)} + } + data := writeTestParquetFile(t, fs, dataDir, f.Name, rows) + size := int64(len(data)) + if f.SizeBytes > 0 { + size = f.SizeBytes + } + dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, setup.fileRef("data", f.Name), + iceberg.ParquetFile, map[int]any{}, nil, nil, int64(len(f.IDs)), size) + if err != nil { + t.Fatalf("build data file %s: %v", f.Name, err) + } + if !f.NoBounds { + lo, _ := iceberg.Int64Literal(f.Lower).MarshalBinary() + hi, _ := iceberg.Int64Literal(f.Upper).MarshalBinary() + dfb.LowerBoundValues(map[int][]byte{1: lo}).UpperBoundValues(map[int][]byte{1: hi}) + } + snapID := int64(1) + entries = append(entries, iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build())) + } + + var manifestBuf bytes.Buffer + mf, err := iceberg.WriteManifest(setup.fileRef("metadata", "data-manifest-1.avro"), &manifestBuf, + version, spec, schema, 1, entries) + if err != nil { + t.Fatalf("write manifest: %v", err) + } + fs.putEntry(metaDir, "data-manifest-1.avro", &filer_pb.Entry{ + Name: "data-manifest-1.avro", Content: manifestBuf.Bytes(), + Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(manifestBuf.Len())}, + }) + + var mlBuf bytes.Buffer + seqNum := int64(1) + if err := iceberg.WriteManifestList(version, &mlBuf, 1, nil, &seqNum, 0, []iceberg.ManifestFile{mf}); err != nil { + t.Fatalf("write manifest list: %v", err) + } + fs.putEntry(metaDir, "snap-1.avro", &filer_pb.Entry{ + Name: "snap-1.avro", Content: mlBuf.Bytes(), + Attributes: &filer_pb.FuseAttributes{Mtime: time.Now().Unix(), FileSize: uint64(mlBuf.Len())}, + }) + + snap := table.Snapshot{SnapshotID: 1, TimestampMs: time.Now().UnixMilli(), + ManifestList: setup.fileRef("metadata", "snap-1.avro")} + builder, err := table.MetadataBuilderFromBase(meta, "s3://"+setup.BucketName+"/"+setup.dataPath()) + if err != nil { + t.Fatalf("metadata builder: %v", err) + } + if err := builder.AddSnapshot(&snap); err != nil { + t.Fatalf("add snapshot: %v", err) + } + if err := builder.SetSnapshotRef(table.MainBranch, snap.SnapshotID, table.BranchRef); err != nil { + t.Fatalf("set snapshot ref: %v", err) + } + meta, err = builder.Build() + if err != nil { + t.Fatalf("build metadata: %v", err) + } + + fullMetadataJSON, _ := json.Marshal(meta) + xattr, _ := json.Marshal(map[string]interface{}{ + "metadataVersion": 1, + "metadataLocation": setup.fileRef("metadata", "v1.metadata.json"), + "metadata": map[string]interface{}{"fullMetadata": json.RawMessage(fullMetadataJSON)}, + }) + fs.putEntry(path.Join(s3tables.TablesPath), setup.BucketName, &filer_pb.Entry{ + Name: setup.BucketName, IsDirectory: true, + Extended: map[string][]byte{s3tables.ExtendedKeyTableBucket: []byte("true")}, + }) + fs.putEntry(bucketPath, setup.Namespace, &filer_pb.Entry{Name: setup.Namespace, IsDirectory: true}) + fs.putEntry(nsPath, setup.TableName, &filer_pb.Entry{ + Name: setup.TableName, IsDirectory: true, + Extended: map[string][]byte{s3tables.ExtendedKeyMetadata: xattr}, + }) +} + +// A bin's files are merged in the order of their bounds on the +// ordering column, not manifest order, so a merged file is as ordered +// as the sequence of its inputs. +func TestCompactDataFilesKeepsInputOrderByBounds(t *testing.T) { + fs, client := startFakeFiler(t) + setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} + // manifest order is the reverse of bound order + populateBoundedTable(t, fs, setup, []boundedFile{ + {Name: "hi.parquet", IDs: []int64{3, 4}, Lower: 3, Upper: 4}, + {Name: "lo.parquet", IDs: []int64{1, 2}, Lower: 1, Upper: 2}, + }) + + handler := NewHandler(nil) + config := Config{TargetFileSizeBytes: 256 * 1024 * 1024, MinInputFiles: 2, MaxCommitRetries: 3} + if _, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil); err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + + df := compactedDataFile(t, client, setup) + ids := compactedIDs(t, client, setup, df.FilePath()) + want := []int64{1, 2, 3, 4} + if fmt.Sprint(ids) != fmt.Sprint(want) { + t.Errorf("merged rows = %v, want %v", ids, want) + } +} + +// An oversized partition is split into runs of consecutive files in +// bound order, so each output covers one contiguous range. +func TestCompactDataFilesSplitsOrderedRuns(t *testing.T) { + fs, client := startFakeFiler(t) + setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} + // four disjoint ranges, each "4 MB": one bin, split into two runs of two + populateBoundedTable(t, fs, setup, []boundedFile{ + {Name: "q4.parquet", IDs: []int64{7, 8}, Lower: 7, Upper: 8, SizeBytes: 4 << 20}, + {Name: "q1.parquet", IDs: []int64{1, 2}, Lower: 1, Upper: 2, SizeBytes: 4 << 20}, + {Name: "q3.parquet", IDs: []int64{5, 6}, Lower: 5, Upper: 6, SizeBytes: 4 << 20}, + {Name: "q2.parquet", IDs: []int64{3, 4}, Lower: 3, Upper: 4, SizeBytes: 4 << 20}, + }) + + handler := NewHandler(nil) + config := Config{TargetFileSizeBytes: 8 << 20, MinInputFiles: 2, MaxCommitRetries: 3} + result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil) + if err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + if !strings.Contains(result, "compacted 4 files into 2") { + t.Fatalf("expected two ordered runs, got %q", result) + } + + // each output must cover one contiguous range + state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath()) + if err != nil { + t.Fatalf("loadCurrentMetadata: %v", err) + } + manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, state.DataPath, state.Metadata) + if err != nil { + t.Fatalf("loadCurrentManifests: %v", err) + } + var ranges [][2]int64 + for _, mf := range manifests { + if mf.ManifestContent() != iceberg.ManifestContentData { + continue + } + manifestData, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, state.DataPath, mf.FilePath()) + if err != nil { + t.Fatalf("load manifest: %v", err) + } + entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true) + if err != nil { + t.Fatalf("read manifest: %v", err) + } + for _, entry := range entries { + if entry.Status() == iceberg.EntryStatusDELETED { + continue + } + lo, _ := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Int64, entry.DataFile().LowerBoundValues()[1]) + hi, _ := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Int64, entry.DataFile().UpperBoundValues()[1]) + ranges = append(ranges, [2]int64{int64(lo.(iceberg.Int64Literal)), int64(hi.(iceberg.Int64Literal))}) + } + } + if len(ranges) != 2 { + t.Fatalf("expected 2 output files, got %d", len(ranges)) + } + // ranges must be contiguous runs, not interleaved: {1..4} and {5..8} + for _, r := range ranges { + if r[1]-r[0] != 3 { + t.Errorf("output covers non-contiguous range %v", r) + } + } +} + +// An oversized file without bounds can never join a merge, so it must not +// disable bounds ordering for the files that can. +func TestCompactDataFilesOrdersPastIneligibleFile(t *testing.T) { + fs, client := startFakeFiler(t) + setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} + populateBoundedTable(t, fs, setup, []boundedFile{ + {Name: "hi.parquet", IDs: []int64{3, 4}, Lower: 3, Upper: 4}, + {Name: "lo.parquet", IDs: []int64{1, 2}, Lower: 1, Upper: 2}, + {Name: "huge.parquet", IDs: []int64{9}, SizeBytes: 512 << 20, NoBounds: true}, + }) + + handler := NewHandler(nil) + config := Config{TargetFileSizeBytes: 256 << 20, MinInputFiles: 2, MaxCommitRetries: 3} + result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil) + if err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + if !strings.Contains(result, "compacted 2 files into 1") { + t.Fatalf("expected only the two eligible files to merge, got %q", result) + } + + var merged iceberg.DataFile + for _, df := range liveDataFiles(t, client, setup) { + if !strings.HasSuffix(df.FilePath(), "huge.parquet") { + merged = df + } + } + ids := compactedIDs(t, client, setup, merged.FilePath()) + want := []int64{1, 2, 3, 4} + if fmt.Sprint(ids) != fmt.Sprint(want) { + t.Errorf("merged rows = %v, want %v", ids, want) + } +} + +// Runs too short to merge in bound order still merge by size, so files that +// cannot form a contiguous run are not stranded for every later pass. +func TestCompactDataFilesMergesStrandedRun(t *testing.T) { + fs, client := startFakeFiler(t) + setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} + populateBoundedTable(t, fs, setup, []boundedFile{ + {Name: "r1.parquet", IDs: []int64{1}, Lower: 1, Upper: 2, SizeBytes: 4 << 20}, + {Name: "r2.parquet", IDs: []int64{3}, Lower: 3, Upper: 4, SizeBytes: 4 << 20}, + {Name: "r3.parquet", IDs: []int64{5}, Lower: 5, Upper: 6, SizeBytes: 6 << 20}, + {Name: "r4.parquet", IDs: []int64{7}, Lower: 7, Upper: 8, SizeBytes: 3 << 20}, + {Name: "r5.parquet", IDs: []int64{9}, Lower: 9, Upper: 10, SizeBytes: 3 << 20}, + }) + + handler := NewHandler(nil) + config := Config{TargetFileSizeBytes: 10 << 20, MinInputFiles: 3, MaxCommitRetries: 3} + result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil) + if err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + if !strings.Contains(result, "compacted 3 files into 1") { + t.Fatalf("expected stranded files to merge by size, got %q", result) + } +} + +// A declared sort order is preserved at the cost of compaction progress: +// runs too short to merge are left for later passes instead of being +// repacked out of order. +func TestCompactDataFilesKeepsStrandedRunForSortedTable(t *testing.T) { + fs, client := startFakeFiler(t) + setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} + sortOrder, err := table.NewSortOrder(1, []table.SortField{{ + SourceIDs: []int{1}, + Transform: iceberg.IdentityTransform{}, + Direction: table.SortASC, + NullOrder: table.NullsFirst, + }}) + if err != nil { + t.Fatalf("new sort order: %v", err) + } + populateBoundedTableSorted(t, fs, setup, []boundedFile{ + {Name: "r1.parquet", IDs: []int64{1}, Lower: 1, Upper: 2, SizeBytes: 4 << 20}, + {Name: "r2.parquet", IDs: []int64{3}, Lower: 3, Upper: 4, SizeBytes: 4 << 20}, + {Name: "r3.parquet", IDs: []int64{5}, Lower: 5, Upper: 6, SizeBytes: 6 << 20}, + {Name: "r4.parquet", IDs: []int64{7}, Lower: 7, Upper: 8, SizeBytes: 3 << 20}, + {Name: "r5.parquet", IDs: []int64{9}, Lower: 9, Upper: 10, SizeBytes: 3 << 20}, + }, sortOrder) + + handler := NewHandler(nil) + config := Config{TargetFileSizeBytes: 10 << 20, MinInputFiles: 3, MaxCommitRetries: 3} + result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil) + if err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + if !strings.Contains(result, "no files eligible for compaction") { + t.Fatalf("expected underfilled ordered runs to be kept, got %q", result) + } +} + +// A declared sort order only applies when every file carries a bound for its +// column; a file without one falls back to inference or unordered packing. +func TestResolveCompactionOrderSkipsUncoveredSortField(t *testing.T) { + schema := newTestSchema() + spec := *iceberg.UnpartitionedSpec + sortOrder, err := table.NewSortOrder(1, []table.SortField{{ + SourceIDs: []int{1}, + Transform: iceberg.IdentityTransform{}, + Direction: table.SortASC, + NullOrder: table.NullsFirst, + }}) + if err != nil { + t.Fatalf("new sort order: %v", err) + } + meta, err := table.NewMetadata(schema, &spec, sortOrder, "s3://b/p", nil) + if err != nil { + t.Fatalf("create metadata: %v", err) + } + + entry := func(name string, bounded bool) iceberg.ManifestEntry { + dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, + "s3://b/data/"+name, iceberg.ParquetFile, map[int]any{}, nil, nil, 2, 16) + if err != nil { + t.Fatalf("build data file: %v", err) + } + if bounded { + lo, _ := iceberg.Int64Literal(1).MarshalBinary() + hi, _ := iceberg.Int64Literal(2).MarshalBinary() + dfb.LowerBoundValues(map[int][]byte{1: lo}).UpperBoundValues(map[int][]byte{1: hi}) + } + snapID := int64(1) + return iceberg.NewManifestEntry(iceberg.EntryStatusADDED, &snapID, nil, nil, dfb.Build()) + } + + if order := resolveCompactionOrder(meta, []iceberg.ManifestEntry{ + entry("a.parquet", true), entry("b.parquet", true), + }); order == nil || order.bestEffort { + t.Fatalf("covered sort field: got %+v, want strict order", order) + } + if order := resolveCompactionOrder(meta, []iceberg.ManifestEntry{ + entry("a.parquet", true), entry("b.parquet", false), + }); order != nil { + t.Fatalf("uncovered sort field: got %+v, want nil", order) + } +} +// A full-bin repack that produces nothing must not discard the ordered runs +// it was meant to replace — they are still valid compaction work. +func TestCompactDataFilesKeepsRunsWhenRepackFails(t *testing.T) { + fs, client := startFakeFiler(t) + setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} + populateBoundedTable(t, fs, setup, []boundedFile{ + {Name: "r1.parquet", IDs: []int64{1}, Lower: 1, Upper: 2, SizeBytes: 4 << 20}, + {Name: "r2.parquet", IDs: []int64{3}, Lower: 3, Upper: 4, SizeBytes: 3 << 20}, + {Name: "r3.parquet", IDs: []int64{5}, Lower: 5, Upper: 6, SizeBytes: 3 << 20}, + {Name: "r4.parquet", IDs: []int64{7}, Lower: 7, Upper: 8, SizeBytes: 6 << 20}, + {Name: "r5.parquet", IDs: []int64{9}, Lower: 9, Upper: 10, SizeBytes: 6 << 20}, + }) + + handler := NewHandler(nil) + config := Config{TargetFileSizeBytes: 10 << 20, MinInputFiles: 3, MaxCommitRetries: 3} + result, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil) + if err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + if !strings.Contains(result, "compacted 3 files into 1") { + t.Fatalf("expected the ordered run to survive the failed repack, got %q", result) + } +} diff --git a/weed/worker/tasks/iceberg/compact_rowgroup_test.go b/weed/worker/tasks/iceberg/compact_rowgroup_test.go new file mode 100644 index 000000000..1160bda1f --- /dev/null +++ b/weed/worker/tasks/iceberg/compact_rowgroup_test.go @@ -0,0 +1,97 @@ +package iceberg + +import ( + "bytes" + "context" + "fmt" + "testing" + + "github.com/apache/iceberg-go" + "github.com/parquet-go/parquet-go" +) + +// Compacted files get row groups from the table's +// write.parquet.row-group-limit instead of one row group per file. +func TestCompactDataFilesWritesSeveralRowGroups(t *testing.T) { + fs, client := startFakeFiler(t) + setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} + + type rowsT = struct { + ID int64 + Name string + } + makeRows := func(lo, hi int64) []rowsT { + rows := make([]rowsT, 0, hi-lo) + for i := lo; i < hi; i++ { + rows = append(rows, rowsT{i, fmt.Sprintf("r%d", i)}) + } + return rows + } + populateTableWithDeleteFiles(t, fs, setup, + []struct { + Name string + Rows []rowsT + }{ + {"d1.parquet", makeRows(0, 1500)}, + {"d2.parquet", makeRows(1500, 3000)}, + }, + nil, nil, + ) + + handler := NewHandler(nil) + config := Config{ + TargetFileSizeBytes: 256 * 1024 * 1024, + MinInputFiles: 2, + MaxCommitRetries: 3, + RowGroupRowLimit: 2000, + } + if _, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil); err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + + state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath()) + if err != nil { + t.Fatalf("loadCurrentMetadata: %v", err) + } + df := compactedDataFile(t, client, setup) + data, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, state.DataPath, df.FilePath()) + if err != nil { + t.Fatalf("load compacted file: %v", err) + } + f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data))) + if err != nil { + t.Fatalf("open compacted file: %v", err) + } + if got := len(f.Metadata().RowGroups); got != 2 { + t.Errorf("row groups = %d, want 2 (3000 rows at row-group-limit 2000)", got) + } +} + +// A configured row-group limit is a cap, not a floor: small explicit limits +// are honored instead of being raised to the estimate floor. +func TestRowsPerRowGroupHonorsConfiguredLimit(t *testing.T) { + entries := make([]iceberg.ManifestEntry, 0, 2) + spec := *iceberg.UnpartitionedSpec + for _, name := range []string{"a.parquet", "b.parquet"} { + dfb, err := iceberg.NewDataFileBuilder(spec, iceberg.EntryContentData, + "s3://b/data/"+name, iceberg.ParquetFile, map[int]any{}, nil, nil, 300, 9000) + 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())) + } + bin := compactionBin{Entries: entries, TotalSize: 18000} + + if got := rowsPerRowGroup(bin, Config{RowGroupRowLimit: 128}); got != 128 { + t.Errorf("explicit limit: rowsPerRowGroup = %d, want 128", got) + } + // 30 bytes/row, byte cap 128MB: the estimate of ~4.4M rows exceeds the 2M + // default, so the default holds; the floor lifts a degenerate estimate. + if got := rowsPerRowGroup(bin, Config{RowGroupSizeBytes: 1024}); got != minRowGroupRows { + t.Errorf("floored estimate: rowsPerRowGroup = %d, want %d", got, minRowGroupRows) + } + if got := rowsPerRowGroup(bin, Config{}); got != defaultRowGroupRowLimit { + t.Errorf("default: rowsPerRowGroup = %d, want %d", got, defaultRowGroupRowLimit) + } +} diff --git a/weed/worker/tasks/iceberg/compact_stats.go b/weed/worker/tasks/iceberg/compact_stats.go new file mode 100644 index 000000000..5ee092eec --- /dev/null +++ b/weed/worker/tasks/iceberg/compact_stats.go @@ -0,0 +1,231 @@ +package iceberg + +import ( + "bytes" + "math" + "sort" + "strings" + "time" + "unicode/utf8" + + "github.com/apache/iceberg-go" + "github.com/parquet-go/parquet-go" + "github.com/parquet-go/parquet-go/format" +) + +// boundTruncateLength is the default metrics mode of the Iceberg reference +// implementation, truncate(16), applied to string and binary bounds. +const boundTruncateLength = 16 + +// columnStats is what a data file's manifest entry records about its columns, +// keyed by Iceberg field id. Readers plan scans from it. +type columnStats struct { + sizes, values, nulls map[int]int64 + lower, upper map[int][]byte + splitOffsets []int64 +} + +func (s *columnStats) applyTo(b *iceberg.DataFileBuilder) { + b.ColumnSizes(s.sizes).ValueCounts(s.values).NullValueCounts(s.nulls). + LowerBoundValues(s.lower).UpperBoundValues(s.upper).SplitOffsets(s.splitOffsets) +} + +// collectColumnStats reads back the footer of a Parquet file the compactor +// wrote. Counts and sizes are summed over the row groups and the bounds are +// the minimum and maximum of the row group statistics, encoded with the +// spec's single-value serialization. A column whose bounds cannot be read +// for every row group, or converted exactly, gets no bounds. +func collectColumnStats(data []byte, schema *iceberg.Schema) (*columnStats, error) { + file, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data)), + parquet.SkipPageIndex(true), parquet.SkipBloomFilters(true)) + if err != nil { + return nil, err + } + var leaves []*parquet.Column + var walk func(*parquet.Column) + walk = func(c *parquet.Column) { + if c.Leaf() { + leaves = append(leaves, c) + return + } + for _, child := range c.Columns() { + walk(child) + } + } + walk(file.Root()) + + type bounds struct { + field iceberg.NestedField + typ parquet.Type + min, max parquet.Value + seen bool + invalid bool + } + stats := &columnStats{sizes: map[int]int64{}, values: map[int]int64{}, nulls: map[int]int64{}, + lower: map[int][]byte{}, upper: map[int][]byte{}} + agg := map[int]*bounds{} + meta := file.Metadata() + for rgIdx, rg := range file.RowGroups() { + stats.splitOffsets = append(stats.splitOffsets, rowGroupOffset(&meta.RowGroups[rgIdx])) + for col, cc := range rg.ColumnChunks() { + field, ok := icebergFieldOf(leaves[col], schema) + if !ok { + continue + } + stats.sizes[field.ID] += meta.RowGroups[rgIdx].Columns[col].MetaData.TotalCompressedSize + stats.values[field.ID] += cc.NumValues() + b := agg[field.ID] + if b == nil { + b = &bounds{field: field, typ: leaves[col].Type()} + agg[field.ID] = b + } + chunk, ok := cc.(*parquet.FileColumnChunk) + if !ok { + b.invalid = true + continue + } + stats.nulls[field.ID] += chunk.NullCount() + if chunk.NullCount() == chunk.NumValues() { + continue // an all-null chunk constrains nothing + } + lo, hi, ok := chunk.Bounds() + switch { + case !ok: + b.invalid = true + case !b.seen: + b.min, b.max, b.seen = lo.Clone(), hi.Clone(), true + default: + if b.typ.Compare(lo, b.min) < 0 { + b.min = lo.Clone() + } + if b.typ.Compare(hi, b.max) > 0 { + b.max = hi.Clone() + } + } + } + } + + for id, b := range agg { + if b.invalid || !b.seen { + continue + } + lo, okLo := boundLiteral(b.min, b.typ, b.field.Type) + hi, okHi := boundLiteral(b.max, b.typ, b.field.Type) + if !okLo || !okHi { + continue + } + lo, hi, okHi = truncateBounds(lo, hi) + if raw, err := lo.MarshalBinary(); err == nil { + stats.lower[id] = raw + } + if raw, err := hi.MarshalBinary(); err == nil && okHi { + stats.upper[id] = raw + } + } + sort.Slice(stats.splitOffsets, func(i, j int) bool { return stats.splitOffsets[i] < stats.splitOffsets[j] }) + return stats, nil +} + +// icebergFieldOf finds the Iceberg field of a Parquet leaf column by the field +// id the writer stored, or by its dotted path in a file written without ids. +// Only primitive fields outside lists and maps carry statistics. +func icebergFieldOf(leaf *parquet.Column, schema *iceberg.Schema) (iceberg.NestedField, bool) { + var field iceberg.NestedField + var ok bool + if leaf.MaxRepetitionLevel() > 0 { + return field, false + } + if id := leaf.ID(); id > 0 { + field, ok = schema.FindFieldByID(id) + } else { + field, ok = schema.FindFieldByName(strings.Join(leaf.Path(), ".")) + } + if !ok { + return field, false + } + _, primitive := field.Type.(iceberg.PrimitiveType) + return field, primitive +} + +// rowGroupOffset is where a row group's bytes start: the split offset readers +// plan one task per row group from. +func rowGroupOffset(rg *format.RowGroup) int64 { + if rg.FileOffset > 0 || len(rg.Columns) == 0 { + return rg.FileOffset + } + first := rg.Columns[0].MetaData + if first.DictionaryPageOffset > 0 && first.DictionaryPageOffset < first.DataPageOffset { + return first.DictionaryPageOffset + } + return first.DataPageOffset +} + +// boundLiteral converts a Parquet statistics value into a literal of the +// column's Iceberg type, when the conversion is exact. Other types (decimal, +// time, uuid, fixed, timestamps not in the type's unit) get no bound. +func boundLiteral(v parquet.Value, pt parquet.Type, t iceberg.Type) (iceberg.Literal, bool) { + if v.IsNull() { + return nil, false + } + switch t.(type) { + case iceberg.BooleanType: + return iceberg.BoolLiteral(v.Boolean()), v.Kind() == parquet.Boolean + case iceberg.Int32Type: + return iceberg.Int32Literal(v.Int32()), v.Kind() == parquet.Int32 + case iceberg.Int64Type: + switch v.Kind() { + case parquet.Int64: + return iceberg.Int64Literal(v.Int64()), true + case parquet.Int32: + return iceberg.Int64Literal(int64(v.Int32())), true + } + case iceberg.Float32Type: + f := v.Float() + return iceberg.Float32Literal(f), v.Kind() == parquet.Float && !math.IsNaN(float64(f)) + case iceberg.Float64Type: + f := v.Double() + return iceberg.Float64Literal(f), v.Kind() == parquet.Double && !math.IsNaN(f) + case iceberg.DateType: + return iceberg.DateLiteral(iceberg.Date(v.Int32())), v.Kind() == parquet.Int32 + case iceberg.TimestampType, iceberg.TimestampTzType: + return iceberg.TimestampLiteral(iceberg.Timestamp(v.Int64())), v.Kind() == parquet.Int64 && timeUnit(pt) == time.Microsecond + case iceberg.TimestampNsType: + return iceberg.TimestampNsLiteral(iceberg.TimestampNano(v.Int64())), v.Kind() == parquet.Int64 && timeUnit(pt) == time.Nanosecond + case iceberg.StringType: + b := v.ByteArray() + return iceberg.StringLiteral(string(b)), v.Kind() == parquet.ByteArray && utf8.Valid(b) + case iceberg.BinaryType: + return iceberg.BinaryLiteral(bytes.Clone(v.ByteArray())), v.Kind() == parquet.ByteArray + } + return nil, false +} + +func timeUnit(pt parquet.Type) time.Duration { + if pt == nil || pt.LogicalType() == nil { + return 0 + } + if ts, ok := pt.LogicalType().Value.(*format.TimestampType); ok && ts.Unit.Value != nil { + return ts.Unit.Value.Duration() + } + return 0 +} + +// truncateBounds applies truncate(16) to string and binary bounds. A prefix is +// still a lower bound; a shortened upper bound is not, so a value longer than +// the limit keeps no upper bound (the reference implementation increments the +// last character instead). +func truncateBounds(lo, hi iceberg.Literal) (iceberg.Literal, iceberg.Literal, bool) { + switch l := lo.(type) { + case iceberg.StringLiteral: + if r := []rune(string(l)); len(r) > boundTruncateLength { + lo = iceberg.StringLiteral(string(r[:boundTruncateLength])) + } + return lo, hi, utf8.RuneCountInString(string(hi.(iceberg.StringLiteral))) <= boundTruncateLength + case iceberg.BinaryLiteral: + if len(l) > boundTruncateLength { + lo = l[:boundTruncateLength] + } + return lo, hi, len(hi.(iceberg.BinaryLiteral)) <= boundTruncateLength + } + return lo, hi, true +} diff --git a/weed/worker/tasks/iceberg/compact_stats_test.go b/weed/worker/tasks/iceberg/compact_stats_test.go new file mode 100644 index 000000000..b96756461 --- /dev/null +++ b/weed/worker/tasks/iceberg/compact_stats_test.go @@ -0,0 +1,148 @@ +package iceberg + +import ( + "bytes" + "context" + "testing" + + "github.com/apache/iceberg-go" + "github.com/parquet-go/parquet-go" + + filer_pb "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// compactedDataFile returns the data file the current snapshot's data +// manifest references after a compaction run. +func compactedDataFile(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup) iceberg.DataFile { + t.Helper() + found := liveDataFiles(t, client, setup) + if len(found) != 1 { + t.Fatalf("expected 1 live data file after compaction, got %d", len(found)) + } + return found[0] +} + +func liveDataFiles(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup) []iceberg.DataFile { + t.Helper() + state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath()) + if err != nil { + t.Fatalf("loadCurrentMetadata: %v", err) + } + manifests, err := loadCurrentManifests(context.Background(), client, setup.BucketName, state.DataPath, state.Metadata) + if err != nil { + t.Fatalf("loadCurrentManifests: %v", err) + } + var found []iceberg.DataFile + for _, mf := range manifests { + if mf.ManifestContent() != iceberg.ManifestContentData { + continue + } + manifestData, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, state.DataPath, mf.FilePath()) + if err != nil { + t.Fatalf("load manifest: %v", err) + } + entries, err := iceberg.ReadManifest(mf, bytes.NewReader(manifestData), true) + if err != nil { + t.Fatalf("read manifest: %v", err) + } + for _, entry := range entries { + if entry.Status() != iceberg.EntryStatusDELETED { + found = append(found, entry.DataFile()) + } + } + } + return found +} + +func compactedFileBytes(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup, filePath string) []byte { + t.Helper() + state, err := loadCurrentMetadata(context.Background(), client, setup.BucketName, setup.tablePath()) + if err != nil { + t.Fatalf("loadCurrentMetadata: %v", err) + } + data, err := loadFileByIcebergPath(context.Background(), client, setup.BucketName, state.DataPath, filePath) + if err != nil { + t.Fatalf("load compacted file: %v", err) + } + return data +} + +func compactedIDs(t *testing.T, client filer_pb.SeaweedFilerClient, setup tableSetup, filePath string) []int64 { + t.Helper() + data := compactedFileBytes(t, client, setup, filePath) + f, err := parquet.OpenFile(bytes.NewReader(data), int64(len(data))) + if err != nil { + t.Fatalf("open compacted file: %v", err) + } + var ids []int64 + for _, rg := range f.RowGroups() { + rows := rg.Rows() + buf := make([]parquet.Row, 8) + for { + n, err := rows.ReadRows(buf) + for _, row := range buf[:n] { + ids = append(ids, row[0].Int64()) + } + if err != nil { + break + } + } + rows.Close() + } + return ids +} + +// A compacted file's manifest entry carries the column metrics readers +// prune files by: counts, sizes, bounds and split offsets read back +// from the file's own footer. +func TestCompactDataFilesRecordsColumnStatistics(t *testing.T) { + fs, client := startFakeFiler(t) + setup := tableSetup{BucketName: "tb", Namespace: "ns", TableName: "tbl"} + populateTableWithDeleteFiles(t, fs, setup, + []struct { + Name string + Rows []struct { + ID int64 + Name string + } + }{ + {"d1.parquet", []struct { + ID int64 + Name string + }{{1, "a"}, {2, "b"}}}, + {"d2.parquet", []struct { + ID int64 + Name string + }{{3, "c"}}}, + }, + nil, nil, + ) + + handler := NewHandler(nil) + config := Config{TargetFileSizeBytes: 256 * 1024 * 1024, MinInputFiles: 2, MaxCommitRetries: 3} + if _, _, err := handler.compactDataFiles(context.Background(), client, setup.BucketName, setup.tablePath(), config, nil); err != nil { + t.Fatalf("compactDataFiles: %v", err) + } + + df := compactedDataFile(t, client, setup) + if df.ValueCounts()[1] != 3 { + t.Errorf("value_counts[1] = %d, want 3", df.ValueCounts()[1]) + } + if len(df.LowerBoundValues()) == 0 || len(df.UpperBoundValues()) == 0 { + t.Fatal("compacted file has no bounds") + } + lo, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Int64, df.LowerBoundValues()[1]) + if err != nil { + t.Fatalf("decode lower bound: %v", err) + } + hi, err := iceberg.LiteralFromBytes(iceberg.PrimitiveTypes.Int64, df.UpperBoundValues()[1]) + if err != nil { + t.Fatalf("decode upper bound: %v", err) + } + if lo.(iceberg.Int64Literal) != 1 || hi.(iceberg.Int64Literal) != 3 { + t.Errorf("id bounds = %v..%v, want 1..3", lo, hi) + } + if len(df.SplitOffsets()) == 0 { + t.Error("compacted file has no split_offsets") + } +} diff --git a/weed/worker/tasks/iceberg/config.go b/weed/worker/tasks/iceberg/config.go index 930501bc3..a2a5619ea 100644 --- a/weed/worker/tasks/iceberg/config.go +++ b/weed/worker/tasks/iceberg/config.go @@ -130,6 +130,10 @@ type Config struct { SortMaxInputBytes int64 SortBufferRows int64 SortSpillDir string + // Row groups of a compacted file, from the table's write.parquet.* + // properties; zero means the default. + RowGroupSizeBytes int64 + RowGroupRowLimit int64 } // ParseConfig extracts an iceberg maintenance Config from plugin config values. diff --git a/weed/worker/tasks/iceberg/detection.go b/weed/worker/tasks/iceberg/detection.go index f32d3f642..4bfc693bf 100644 --- a/weed/worker/tasks/iceberg/detection.go +++ b/weed/worker/tasks/iceberg/detection.go @@ -477,7 +477,7 @@ func hasEligibleCompaction( } targetSize := compactionTargetSizeForPlan(config, rewritePlan) - bins := buildCompactionBins(candidateEntries, targetSize, minInputFiles) + bins := buildCompactionBins(candidateEntries, targetSize, minInputFiles, meta) bins = filterCompactionBinsByPlan(bins, config, rewritePlan) return len(bins) > 0, nil } diff --git a/weed/worker/tasks/iceberg/handler_test.go b/weed/worker/tasks/iceberg/handler_test.go index a2e68ddc3..be5de8b2e 100644 --- a/weed/worker/tasks/iceberg/handler_test.go +++ b/weed/worker/tasks/iceberg/handler_test.go @@ -701,7 +701,7 @@ func TestBuildCompactionBins(t *testing.T) { {path: "data/f3.parquet", size: 4096, partition: map[int]any{}}, }) - bins := buildCompactionBins(entries, targetSize, minFiles) + bins := buildCompactionBins(entries, targetSize, minFiles, nil) if len(bins) != 1 { t.Fatalf("expected 1 bin, got %d", len(bins)) } @@ -720,7 +720,7 @@ func TestBuildCompactionBinsFiltersLargeFiles(t *testing.T) { {path: "data/large.parquet", size: 5000, partition: map[int]any{}}, }) - bins := buildCompactionBins(entries, targetSize, minFiles) + bins := buildCompactionBins(entries, targetSize, minFiles, nil) if len(bins) != 1 { t.Fatalf("expected 1 bin, got %d", len(bins)) } @@ -756,7 +756,7 @@ func TestBuildCompactionBinsLowercaseParquetFormat(t *testing.T) { entries[0] = lowercaseFormatEntry{entries[0]} entries[1] = lowercaseFormatEntry{entries[1]} - bins := buildCompactionBins(entries, targetSize, minFiles) + bins := buildCompactionBins(entries, targetSize, minFiles, nil) if len(bins) != 1 { t.Fatalf("expected 1 bin, got %d", len(bins)) } @@ -774,7 +774,7 @@ func TestBuildCompactionBinsMinFilesThreshold(t *testing.T) { {path: "data/f2.parquet", size: 2048, partition: map[int]any{}}, }) - bins := buildCompactionBins(entries, targetSize, minFiles) + bins := buildCompactionBins(entries, targetSize, minFiles, nil) if len(bins) != 0 { t.Errorf("expected 0 bins (below min threshold), got %d", len(bins)) } @@ -801,7 +801,7 @@ func TestBuildCompactionBinsMultiplePartitions(t *testing.T) { {path: "data/b3.parquet", size: 4096, partition: partB, partitionSpec: &partitionSpec}, }) - bins := buildCompactionBins(entries, targetSize, minFiles) + bins := buildCompactionBins(entries, targetSize, minFiles, nil) if len(bins) != 2 { t.Fatalf("expected 2 bins (one per partition), got %d", len(bins)) } @@ -977,7 +977,7 @@ func TestBuildCompactionBinsMultipleSpecs(t *testing.T) { {path: "data/s1-f2.parquet", size: 2048, partition: map[int]any{}, specID: 1}, }, partSpecs) - bins := buildCompactionBins(entries, targetSize, minFiles) + bins := buildCompactionBins(entries, targetSize, minFiles, nil) if len(bins) != 2 { t.Fatalf("expected 2 bins (one per spec), got %d", len(bins)) } @@ -1011,7 +1011,7 @@ func TestBuildCompactionBinsSingleSpec(t *testing.T) { {path: "data/f3.parquet", size: 4096, partition: map[int]any{}, specID: 0}, }, partSpecs) - bins := buildCompactionBins(entries, targetSize, minFiles) + bins := buildCompactionBins(entries, targetSize, minFiles, nil) if len(bins) != 1 { t.Fatalf("expected 1 bin, got %d", len(bins)) } @@ -1303,7 +1303,7 @@ func TestMergeParquetFilesWithPositionDeletes(t *testing.T) { merged, count, err := mergeParquetFiles( context.Background(), client, "test-bucket", "ns/tbl", - entries, posDeletes, nil, nil, + entries, posDeletes, nil, nil, 0, ) if err != nil { t.Fatalf("mergeParquetFiles: %v", err) @@ -1419,7 +1419,7 @@ func TestMergeParquetFilesWithEqualityDeletes(t *testing.T) { merged, count, err := mergeParquetFiles( context.Background(), client, "test-bucket", "ns/tbl", - entries, nil, eqGroups, schema, + entries, nil, eqGroups, schema, 0, ) if err != nil { t.Fatalf("mergeParquetFiles: %v", err) @@ -1479,7 +1479,7 @@ func TestMergeParquetFilesDictionaryEncodedInput(t *testing.T) { merged, count, err := mergeParquetFiles( context.Background(), client, "test-bucket", "ns/tbl", - entries, nil, nil, nil, + entries, nil, nil, nil, 0, ) if err != nil { t.Fatalf("mergeParquetFiles: %v", err) diff --git a/weed/worker/tasks/iceberg/table_config.go b/weed/worker/tasks/iceberg/table_config.go index b220d6c21..b4e2f5475 100644 --- a/weed/worker/tasks/iceberg/table_config.go +++ b/weed/worker/tasks/iceberg/table_config.go @@ -19,6 +19,8 @@ import ( const ( propTargetFileSize = "write.target-file-size-bytes" propDeleteTargetFileSize = "write.delete.target-file-size-bytes" + propRowGroupSize = "write.parquet.row-group-size-bytes" + propRowGroupRowLimit = "write.parquet.row-group-limit" propMaxSnapshotAgeMs = "history.expire.max-snapshot-age-ms" propMinSnapshotsToKeep = "history.expire.min-snapshots-to-keep" ) @@ -50,6 +52,12 @@ func applyTableProperties(cfg Config, props iceberg.Properties) Config { if v, ok := propInt64(props, propDeleteTargetFileSize); ok { cfg.DeleteTargetFileSizeBytes = v } + if v, ok := propInt64(props, propRowGroupSize); ok { + cfg.RowGroupSizeBytes = v + } + if v, ok := propInt64(props, propRowGroupRowLimit); ok { + cfg.RowGroupRowLimit = v + } if v, ok := propInt64(props, propMaxSnapshotAgeMs); ok { cfg.SnapshotRetentionMs = v }