iceberg maintenance: keep compacted files prunable (stats, bound order, row groups) (#11654)

* iceberg maintenance: record column statistics on compacted files

A compacted file's manifest entry was built with no column_sizes,
value_counts, null_value_counts, lower_bounds, upper_bounds or
split_offsets, so no reader could skip a compacted file on any
predicate. parquet-go already writes exact per-chunk min/max and null
counts into the footer; read that footer back after the merge and
record it on the data file, bounds as the spec's single-value
serialization with string and binary truncated as truncate(16). A
column whose bounds cannot be converted exactly gets none, and a
statistics failure is logged while the compaction commits anyway:
metrics are an optimization, not a correctness requirement.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* iceberg maintenance: merge bins in bound order so compacted files stay prunable

A bin's files were concatenated in manifest order, or largest-first
when a partition was split under the target size, so inputs disjoint
on a column came out as outputs that overlapped on it. Order each
bin's files by their bounds on one column before merging and split an
oversized partition into runs of consecutive files, so every output
covers one contiguous range. The column is the first identity field
of the table's sort order when it declares one (a descending order
sorts by upper bound), otherwise the first schema column every
candidate file has bounds for; detection resolves the same order so
it plans the bins execution builds. Files without bounds keep the old
behavior.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* iceberg maintenance: cap compacted files' row groups from table config

Neither merge writer set a row-group limit (parquet-go's default is
unlimited rows), so every compacted file was a single row group and
readers could not skip inside it either. Rows per row group now come
from the table's write.parquet.row-group-limit and
write.parquet.row-group-size-bytes, defaulting to PyIceberg's
1 048 576 rows and Iceberg's 128 MiB, with the byte size turned into
rows from the bin's inputs' compressed bytes per row and a floor of
1 024 rows. Both writers take the cap, and the statistics the entry
records list one split offset per row group.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* iceberg: keep compaction order eligibility per group, rescue stranded runs

The merge order resolved over all candidates, so one oversized or
non-Parquet file without bounds disabled ordering for files that could
participate. Resolve it per partition group over the eligible entries.

Ordered runs too short to merge were dropped entirely. Runs from an
inferred bounds order now fall back to size-based packing — ordering is
a preference there — while runs under a declared sort order are still
left for later passes so the sort contract holds.

An explicit write.parquet.row-group-limit is a cap, not a floor: values
below the estimate floor are now honored instead of being raised.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* iceberg: only use the declared sort order when its bounds are complete

An entry without bounds on the sort column sorted to the tail and merged
into an output claiming an order it cannot verify. Fall back to
bound inference instead of ordering by a later sort field alone.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* iceberg maintenance: keep ordered runs when the full repack yields nothing

The bestEffort leftover fallback removed the ordered runs before checking
whether repacking the whole bin produced any bins, discarding valid
compaction work.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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-*")),

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
+}

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)
+	}
+}

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)
+	}
+}

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
+}

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")
+	}
+}

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.

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
 }

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)

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
 	}