From cdea7e79e0d100dcd2540e1656897f8ec8ddd736 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 6 Nov 2026 13:41:02 -0800 Subject: [PATCH] filer: fold small chunks into manifests 256 at a time (#12462) * filer: fold small chunks into manifests 256 at a time Every metadata event carries the file's whole chunk list, so a file grown by small appends re-sent each chunk record on every append until the list reached ManifestBatch (10000). Chunks under 1 MiB now fold once 256 of them build up; full-size chunks still fold at ManifestBatch, so large files keep their flat lists and read without a manifest fetch. Co-Authored-By: Claude Opus 5.5 * filer: merge appended small chunks at the small-chunk fold count The append path merges small chunks into full-size ones, but only at 1000 of them; the fold now hides small chunks in a manifest at 256, so the merge would never see enough of them. Merging at the same count keeps rewriting them into full chunks first. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/filer/filechunk_manifest.go | 49 ++++++++-- .../filer/filechunk_manifest_rollback_test.go | 26 ++++- weed/filer/filechunk_manifest_test.go | 95 ++++++++++++++++++- .../filer_server_handlers_write_merge.go | 4 +- 4 files changed, 161 insertions(+), 13 deletions(-) diff --git a/weed/filer/filechunk_manifest.go b/weed/filer/filechunk_manifest.go index 3eacae5cdf..654761b395 100644 --- a/weed/filer/filechunk_manifest.go +++ b/weed/filer/filechunk_manifest.go @@ -499,13 +499,24 @@ func retriedStreamFetchChunkData(ctx context.Context, writer io.Writer, urlStrin } -// MaybeManifestize folds a flat chunk list into manifest chunks once it passes -// ManifestBatch, so an entry's chunk list stays within what the metadata store -// will hold. A fold that fails partway returns inputChunks unchanged -- never -// the half-folded list, which drops the manifests the caller came in with -- -// and hands the blobs it had already saved to deleteChunks, since the flat list -// it returns references none of them. deleteChunks may be nil where the caller -// has no deleter to offer; then the blobs are only named in the log. +// SmallChunkManifestBatch is how many chunks under smallChunkSize fold into one +// manifest. Appends and partial flushes leave such chunks, and every metadata +// event for the file carries its whole chunk list, so without an earlier fold a +// file grown by small appends re-sends each of them on every append. +const ( + SmallChunkManifestBatch = 256 + smallChunkSize = 1 << 20 +) + +// MaybeManifestize folds a flat chunk list into manifest chunks: chunks under +// smallChunkSize once SmallChunkManifestBatch of them build up, and the rest +// once they pass ManifestBatch, so an entry's chunk list stays within what the +// metadata store will hold. A fold that fails partway returns inputChunks +// unchanged -- never the half-folded list, which drops the manifests the caller +// came in with -- and hands the blobs it had already saved to deleteChunks, +// since the flat list it returns references none of them. deleteChunks may be +// nil where the caller has no deleter to offer; then the blobs are only named +// in the log. func MaybeManifestize(saveFunc SaveDataAsChunkFunctionType, deleteChunks func([]*filer_pb.FileChunk), inputChunks []*filer_pb.FileChunk) (chunks []*filer_pb.FileChunk, err error) { var saved []*filer_pb.FileChunk record := func(reader io.Reader, name string, offset int64, tsNs int64, expectedDataSize uint64) (*filer_pb.FileChunk, error) { @@ -516,7 +527,10 @@ func MaybeManifestize(saveFunc SaveDataAsChunkFunctionType, deleteChunks func([] return chunk, saveErr } - chunks, err = doMaybeManifestize(record, inputChunks, ManifestBatch, mergeIntoManifest) + chunks, err = foldSmallChunks(record, inputChunks) + if err == nil { + chunks, err = doMaybeManifestize(record, chunks, ManifestBatch, mergeIntoManifest) + } if err == nil { return chunks, nil } @@ -530,6 +544,25 @@ func MaybeManifestize(saveFunc SaveDataAsChunkFunctionType, deleteChunks func([] return inputChunks, err } +func foldSmallChunks(saveFunc SaveDataAsChunkFunctionType, inputChunks []*filer_pb.FileChunk) ([]*filer_pb.FileChunk, error) { + var smallChunks, otherChunks []*filer_pb.FileChunk + for _, chunk := range inputChunks { + if !chunk.IsChunkManifest && chunk.Size < smallChunkSize { + smallChunks = append(smallChunks, chunk) + } else { + otherChunks = append(otherChunks, chunk) + } + } + if len(smallChunks) < SmallChunkManifestBatch { + return inputChunks, nil + } + folded, err := doMaybeManifestize(saveFunc, smallChunks, SmallChunkManifestBatch, mergeIntoManifest) + if err != nil { + return inputChunks, err + } + return append(otherChunks, folded...), nil +} + func doMaybeManifestize(saveFunc SaveDataAsChunkFunctionType, inputChunks []*filer_pb.FileChunk, mergeFactor int, mergefn func(saveFunc SaveDataAsChunkFunctionType, dataChunks []*filer_pb.FileChunk) (manifestChunk *filer_pb.FileChunk, err error)) (chunks []*filer_pb.FileChunk, err error) { var dataChunks []*filer_pb.FileChunk diff --git a/weed/filer/filechunk_manifest_rollback_test.go b/weed/filer/filechunk_manifest_rollback_test.go index 7dcf072170..c53eb77ff6 100644 --- a/weed/filer/filechunk_manifest_rollback_test.go +++ b/weed/filer/filechunk_manifest_rollback_test.go @@ -13,8 +13,8 @@ func flatTestChunks(n int) []*filer_pb.FileChunk { for i := range chunks { chunks[i] = &filer_pb.FileChunk{ FileId: fmt.Sprintf("1,%x", i+1), - Offset: int64(i) * 8, - Size: 8, + Offset: int64(i) * smallChunkSize, + Size: smallChunkSize, } } return chunks @@ -57,7 +57,7 @@ func TestMaybeManifestizeFolds(t *testing.T) { if store.saves != 1 || len(store.deleted) != 0 { t.Errorf("expected 1 save and no deletes, got %d saves, %d deleted", store.saves, len(store.deleted)) } - if manifests[0].Offset != 0 || manifests[0].Size != uint64(ManifestBatch*8) { + if manifests[0].Offset != 0 || manifests[0].Size != uint64(ManifestBatch*smallChunkSize) { t.Errorf("manifest span [%d,%d) wrong", manifests[0].Offset, manifests[0].Size) } } @@ -131,3 +131,23 @@ func TestMaybeManifestizeRollbackWithoutDeleter(t *testing.T) { t.Fatalf("expected fallback to %d flat chunks, got %d", len(chunks), len(result)) } } + +// Small-chunk manifests saved before a later batch fails are deleted too. +func TestMaybeManifestizeRollsBackSmallFold(t *testing.T) { + store := &fakeManifestStore{failOn: 3} + chunks := flatTestChunks(2*ManifestBatch + 50) + for i := 0; i < SmallChunkManifestBatch; i++ { + chunks = append(chunks, &filer_pb.FileChunk{FileId: fmt.Sprintf("3,%x", i+1), Offset: int64(i) * 8, Size: 8}) + } + + result, err := MaybeManifestize(store.save, store.delete, chunks) + if err == nil { + t.Fatal("expected the failed save to be reported") + } + if len(result) != len(chunks) || HasChunkManifest(result) { + t.Fatalf("expected fallback to %d flat chunks, got %d", len(chunks), len(result)) + } + if len(store.deleted) != 2 || store.deleted[0].FileId != "2,1" || store.deleted[1].FileId != "2,2" { + t.Fatalf("expected the small-chunk and first regular manifests deleted, got %+v", store.deleted) + } +} diff --git a/weed/filer/filechunk_manifest_test.go b/weed/filer/filechunk_manifest_test.go index 039399f31e..7cf8e42ab0 100644 --- a/weed/filer/filechunk_manifest_test.go +++ b/weed/filer/filechunk_manifest_test.go @@ -256,7 +256,7 @@ func TestMaybeManifestizePreservesSSEMetadata(t *testing.T) { store := newTestManifestStore() chunks := make([]*filer_pb.FileChunk, ManifestBatch+1) for i := range chunks { - chunks[i] = testChunk(1, uint64(i+1), uint32(i+1), int64(i*8), 8, int64(i+1)) + chunks[i] = testChunk(1, uint64(i+1), uint32(i+1), int64(i)*smallChunkSize, smallChunkSize, int64(i+1)) chunks[i].SseType = sseType chunks[i].SseMetadata = []byte(fmt.Sprintf("metadata-%d", i)) } @@ -807,3 +807,96 @@ type failOnWrite struct{} func (w *failOnWrite) Write(p []byte) (int, error) { return 0, fmt.Errorf("no space left") } + +func TestMaybeManifestizeBoundsSmallAppends(t *testing.T) { + const appends = 2726 + store := newTestManifestStore() + var chunks, before []*filer_pb.FileChunk + var appended []*filer_pb.FileChunk + for i := 0; i < appends; i++ { + chunk := testChunk(1, uint64(i+1), uint32(i+1), int64(i)*308, 308, int64(i+1)) + chunk.ETag = fmt.Sprintf("etag-%022d", i) + appended = append(appended, proto.Clone(chunk).(*filer_pb.FileChunk)) + before = chunks + var err error + chunks, err = MaybeManifestize(store.saveFunc(), nil, append(append([]*filer_pb.FileChunk(nil), chunks...), chunk)) + require.NoError(t, err) + require.Less(t, len(chunks), SmallChunkManifestBatch+i/SmallChunkManifestBatch+1, "append %d", i) + } + + manifests, flat := SeparateManifestChunks(chunks) + assert.Len(t, manifests, appends/SmallChunkManifestBatch) + assert.Len(t, flat, appends%SmallChunkManifestBatch) + assert.Equal(t, appends/SmallChunkManifestBatch, store.nextID, "each manifest is saved once") + + resolved, _, err := store.resolveAll(chunks) + require.NoError(t, err) + require.Len(t, resolved, appends) + seen := make(map[string]bool) + for _, chunk := range resolved { + seen[chunk.GetFileIdString()] = true + } + for _, chunk := range appended { + assert.True(t, seen[chunk.GetFileIdString()], "chunk %s lost", chunk.GetFileIdString()) + } + + eventSize := func(oldChunks, newChunks []*filer_pb.FileChunk) int { + return proto.Size(&filer_pb.SubscribeMetadataResponse{EventNotification: &filer_pb.EventNotification{ + OldEntry: &filer_pb.Entry{Name: "journal.jsonl", Chunks: oldChunks}, + NewEntry: &filer_pb.Entry{Name: "journal.jsonl", Chunks: newChunks}, + }}) + } + folded := eventSize(before, chunks) + unfolded := eventSize(appended[:appends-1], appended) + assert.Less(t, folded*10, unfolded, "folded event %d bytes, unfolded %d", folded, unfolded) +} + +func TestMaybeManifestizeLeavesLargeChunksToManifestBatch(t *testing.T) { + store := newTestManifestStore() + var chunks []*filer_pb.FileChunk + var offset int64 + for i := 0; i < 3*SmallChunkManifestBatch; i++ { + size := uint64(smallChunkSize) + if i%2 == 0 { + size = 100 + } + chunks = append(chunks, testChunk(1, uint64(i+1), uint32(i+1), offset, size, int64(i+1))) + offset += int64(size) + } + + packed, err := MaybeManifestize(store.saveFunc(), nil, chunks) + require.NoError(t, err) + + manifests, flat := SeparateManifestChunks(packed) + require.Len(t, manifests, 1) + assert.Len(t, flat, len(chunks)-SmallChunkManifestBatch) + for _, chunk := range flat { + if chunk.Size < smallChunkSize { + continue + } + assert.Equal(t, uint64(smallChunkSize), chunk.Size) + } + folded, err := store.resolve(manifests[0]) + require.NoError(t, err) + require.Len(t, folded, SmallChunkManifestBatch) + for _, chunk := range folded { + assert.Equal(t, uint64(100), chunk.Size) + } +} + +func TestMaybeManifestizeKeepsOrderBelowSmallBatch(t *testing.T) { + store := newTestManifestStore() + var chunks []*filer_pb.FileChunk + for i := 0; i < 2*(SmallChunkManifestBatch-1); i++ { + size := uint64(smallChunkSize) + if i%2 == 0 { + size = 100 + } + chunks = append(chunks, testChunk(1, uint64(i+1), uint32(i+1), int64(i)*smallChunkSize, size, int64(i+1))) + } + + packed, err := MaybeManifestize(store.saveFunc(), nil, chunks) + require.NoError(t, err) + assert.Zero(t, store.nextID) + assertEqualChunks(t, chunks, packed) +} diff --git a/weed/server/filer_server_handlers_write_merge.go b/weed/server/filer_server_handlers_write_merge.go index 157d96c797..a167c1a856 100644 --- a/weed/server/filer_server_handlers_write_merge.go +++ b/weed/server/filer_server_handlers_write_merge.go @@ -12,7 +12,9 @@ import ( "github.com/seaweedfs/seaweedfs/weed/stats" ) -const MergeChunkMinCount int = 1000 +// MergeChunkMinCount matches the small-chunk fold so a merge rewrites them +// before MaybeManifestize hides them in a manifest. +const MergeChunkMinCount int = filer.SmallChunkManifestBatch func (fs *FilerServer) maybeMergeChunks(ctx context.Context, so *operation.StorageOption, inputChunks []*filer_pb.FileChunk) (mergedChunks []*filer_pb.FileChunk, err error) { // Don't merge SSE-encrypted chunks to preserve per-chunk metadata