mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-21 22:12:36 +00:00
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>
This commit is contained in:
co-authored by
Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
parent
b4e5a0f0f2
commit
669c3a6f6a
@@ -598,10 +598,13 @@ func buildCompactionBins(entries []iceberg.ManifestEntry, targetSize int64, minF
|
||||
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
|
||||
// instead of stranding them for every later pass.
|
||||
result = result[:len(result)-len(runs)]
|
||||
packed = splitOversizedBin(*bin, targetSize, minFiles)
|
||||
// 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...)
|
||||
}
|
||||
|
||||
@@ -363,3 +363,26 @@ func TestResolveCompactionOrderSkipsUncoveredSortField(t *testing.T) {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user