From 8c1a0a0cc2979a5541a9fe41e4134dfbb41e270e Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 24 Oct 2026 19:59:19 -0700 Subject: [PATCH] filer: fix prefixed listing when the resume point sorts below the prefix (#12049) * test: prefixed directory listing with start below the prefix A listing constrained to prefix "item-b" and resumed from "item" must return item-b.versions. The seek-based stores return an empty page because the iterator lands before the prefix range and breaks on the first non-matching name. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: clamp prefixed-listing seek to the prefix range Seek-based stores (leveldb, leveldb2, leveldb3, rocksdb, tikv, etcd) started the scan at startFileName and stopped at the first key outside the prefix. A startFileName below the prefix landed the iterator on a non-matching sibling first, so the listing returned an empty page even though matching entries followed. etcd had the mirror problem: a range beginning below the prefix end leaked non-matching entries. Begin the seek at max(startFileName, prefix), which is what foundationdb and the sql stores already do. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer(hbase): honor startFileName lower bound in prefixed listings The scan starts at the prefix but only skipped the exact startFileName, so entries sorting before it were returned again on resume. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: keep paging a prefixed listing while matches can still follow The fallback for stores without native prefix support stopped paging as soon as the last scanned name reached the prefix, dropping prefix matches that continue on the next page. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: return the last delivered name from the prefixed-listing fallback The fallback for stores without native prefix support pages raw entries and filters client-side. Keep the scan cursor and the returned cursor separate: resuming must continue after the last delivered match, not the last scanned name, or matches still inside the current raw page are skipped on the next page. Also keep paging while the scan cursor is inside or below the prefix range instead of stopping at the first name that reaches the prefix. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer(hbase): start prefixed-listing scans at the resume key 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/filer/etcd/etcd_store.go | 2 +- weed/filer/filerstore_wrapper.go | 65 ++++++++++------------ weed/filer/filerstore_wrapper_test.go | 38 +++++++++++++ weed/filer/hbase/hbase_store.go | 8 ++- weed/filer/leveldb/leveldb_store.go | 2 +- weed/filer/leveldb/leveldb_store_test.go | 25 +++++++++ weed/filer/leveldb2/leveldb2_store.go | 2 +- weed/filer/leveldb2/leveldb2_store_test.go | 25 +++++++++ weed/filer/leveldb3/leveldb3_store.go | 2 +- weed/filer/leveldb3/leveldb3_store_test.go | 25 +++++++++ weed/filer/rocksdb/rocksdb_store.go | 2 +- weed/filer/tikv/tikv_store.go | 2 +- 12 files changed, 154 insertions(+), 44 deletions(-) diff --git a/weed/filer/etcd/etcd_store.go b/weed/filer/etcd/etcd_store.go index 0b83bacc89..acbeb755a0 100644 --- a/weed/filer/etcd/etcd_store.go +++ b/weed/filer/etcd/etcd_store.go @@ -181,7 +181,7 @@ func (store *EtcdStore) DeleteFolderChildren(ctx context.Context, fullpath weed_ func (store *EtcdStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) { directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName) } diff --git a/weed/filer/filerstore_wrapper.go b/weed/filer/filerstore_wrapper.go index 1a17e22cf6..646bd5650f 100644 --- a/weed/filer/filerstore_wrapper.go +++ b/weed/filer/filerstore_wrapper.go @@ -400,46 +400,39 @@ func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath u return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc) } - var notPrefixed []*Entry - lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) (bool, error) { - notPrefixed = append(notPrefixed, entry) - return true, nil - }) - if err != nil { - return - } - + // lastFileName must stay at the last name delivered to eachEntryFunc; the + // raw scan cursor runs ahead over non-matching names and cannot be returned. count := int64(0) - for count < limit && len(notPrefixed) > 0 { - for _, entry := range notPrefixed { - if strings.HasPrefix(entry.Name(), prefix) { - count++ - res, resErr := eachEntryFunc(entry) - - if resErr != nil { - err = fmt.Errorf("failed to process eachEntryFunc for entry %q: %w", entry.Name(), resErr) - return - } - - if !res { - return - } - if count >= limit { - break - } - } + scanName, scanInclusive := startFileName, includeStartFile + for count < limit { + var page []*Entry + if scanName, err = actualStore.ListDirectoryEntries(ctx, dirPath, scanName, scanInclusive, limit, func(entry *Entry) (bool, error) { + page = append(page, entry) + return true, nil + }); err != nil { + return } - if count < limit && lastFileName < prefix { - notPrefixed = notPrefixed[:0] - lastFileName, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit, func(entry *Entry) (bool, error) { - notPrefixed = append(notPrefixed, entry) - return true, nil - }) - if err != nil { + scanInclusive = false + if len(page) == 0 { + return + } + for _, entry := range page { + if !strings.HasPrefix(entry.Name(), prefix) { + continue + } + count++ + res, resErr := eachEntryFunc(entry) + if resErr != nil { + err = fmt.Errorf("failed to process eachEntryFunc for entry %q: %w", entry.Name(), resErr) return } - } else { - break + lastFileName = entry.Name() + if !res || count >= limit { + return + } + } + if scanName >= prefix && !strings.HasPrefix(scanName, prefix) { + return } } return diff --git a/weed/filer/filerstore_wrapper_test.go b/weed/filer/filerstore_wrapper_test.go index 0b180f63e5..6351d6978f 100644 --- a/weed/filer/filerstore_wrapper_test.go +++ b/weed/filer/filerstore_wrapper_test.go @@ -205,3 +205,41 @@ func TestFilerStoreWrapperRollbackSucceedsWithCancelledContext(t *testing.T) { assert.NoError(t, wrapper.RollbackTransaction(cancelledCtx())) assert.NoError(t, wrapper.RollbackTransaction(expiredCtx())) } + +type noPrefixListStore struct { + *stubFilerStore +} + +func (s *noPrefixListStore) ListDirectoryPrefixedEntries(context.Context, util.FullPath, string, bool, int64, string, ListEachEntryFunc) (string, error) { + return "", ErrUnsupportedListDirectoryPrefixed +} + +// The wrapper fallback must return the last delivered name as the cursor, not +// the end of the raw page, and must keep paging while prefix matches can still +// follow the raw cursor. +func TestPrefixFilterEntriesCursorStaysAtLastDelivered(t *testing.T) { + store := &noPrefixListStore{newStubFilerStore()} + wrapper := NewFilerStoreWrapper(store) + ctx := context.Background() + for _, name := range []string{"a", "b1", "b2", "b3", "b4"} { + require.NoError(t, store.InsertEntry(ctx, &Entry{FullPath: util.FullPath("/d/" + name)})) + } + + var delivered []string + lastFileName, err := wrapper.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "", false, 2, "b", func(entry *Entry) (bool, error) { + delivered = append(delivered, entry.Name()) + return true, nil + }) + require.NoError(t, err) + assert.Equal(t, []string{"b1", "b2"}, delivered) + assert.Equal(t, "b2", lastFileName) + + delivered = delivered[:0] + lastFileName, err = wrapper.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), lastFileName, false, 10, "b", func(entry *Entry) (bool, error) { + delivered = append(delivered, entry.Name()) + return true, nil + }) + require.NoError(t, err) + assert.Equal(t, []string{"b3", "b4"}, delivered) + assert.Equal(t, "b4", lastFileName) +} diff --git a/weed/filer/hbase/hbase_store.go b/weed/filer/hbase/hbase_store.go index 7663fef9d3..cfa027f3c1 100644 --- a/weed/filer/hbase/hbase_store.go +++ b/weed/filer/hbase/hbase_store.go @@ -156,7 +156,11 @@ func (store *HbaseStore) ListDirectoryEntries(ctx context.Context, dirPath util. func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) { family := map[string][]string{store.cfMetaDir: {COLUMN_NAME}} expectedPrefix := []byte(dirPath.Child(prefix)) - scan, err := hrpc.NewScanRange(ctx, store.table, expectedPrefix, nil, hrpc.Families(family)) + scanStart := expectedPrefix + if resumeKey := []byte(dirPath.Child(startFileName)); bytes.Compare(resumeKey, scanStart) > 0 { + scanStart = resumeKey + } + scan, err := hrpc.NewScanRange(ctx, store.table, scanStart, nil, hrpc.Families(family)) if err != nil { return lastFileName, err } @@ -188,7 +192,7 @@ func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPa value := cell.Value - if fileName == startFileName && !includeStartFile { + if fileName < startFileName || (fileName == startFileName && !includeStartFile) { continue } diff --git a/weed/filer/leveldb/leveldb_store.go b/weed/filer/leveldb/leveldb_store.go index 606166da90..341327e433 100644 --- a/weed/filer/leveldb/leveldb_store.go +++ b/weed/filer/leveldb/leveldb_store.go @@ -221,7 +221,7 @@ func (store *LevelDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName) } diff --git a/weed/filer/leveldb/leveldb_store_test.go b/weed/filer/leveldb/leveldb_store_test.go index b5384fc778..b55586ec98 100644 --- a/weed/filer/leveldb/leveldb_store_test.go +++ b/weed/filer/leveldb/leveldb_store_test.go @@ -110,3 +110,28 @@ func BenchmarkInsertEntry(b *testing.B) { store.InsertEntry(ctx, entry) } } + +func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) { + store := &LevelDBStore{} + store.initialize(t.TempDir()) + defer store.Shutdown() + + ctx := context.Background() + for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} { + if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil { + t.Fatalf("insert %s: %v", name, err) + } + } + + var names []string + _, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) { + names = append(names, entry.Name()) + return true, nil + }) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(names) != 1 || names[0] != "item-b.versions" { + t.Fatalf("want [item-b.versions], got %v", names) + } +} diff --git a/weed/filer/leveldb2/leveldb2_store.go b/weed/filer/leveldb2/leveldb2_store.go index 5c05033b99..e861db47e1 100644 --- a/weed/filer/leveldb2/leveldb2_store.go +++ b/weed/filer/leveldb2/leveldb2_store.go @@ -239,7 +239,7 @@ func (store *LevelDB2Store) ListDirectoryPrefixedEntries(ctx context.Context, di directoryPrefix, partitionId := genDirectoryKeyPrefix(dirPath, prefix, store.dbCount) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart, _ = genDirectoryKeyPrefix(dirPath, startFileName, store.dbCount) } diff --git a/weed/filer/leveldb2/leveldb2_store_test.go b/weed/filer/leveldb2/leveldb2_store_test.go index 9651711d92..8d1d46ccc3 100644 --- a/weed/filer/leveldb2/leveldb2_store_test.go +++ b/weed/filer/leveldb2/leveldb2_store_test.go @@ -151,6 +151,31 @@ func TestInitializeRejectsOversizedOption(t *testing.T) { } } +func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) { + store := &LevelDB2Store{} + store.initialize(t.TempDir(), 2) + defer store.Shutdown() + + ctx := context.Background() + for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} { + if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil { + t.Fatalf("insert %s: %v", name, err) + } + } + + var names []string + _, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) { + names = append(names, entry.Name()) + return true, nil + }) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(names) != 1 || names[0] != "item-b.versions" { + t.Fatalf("want [item-b.versions], got %v", names) + } +} + func TestEmptyRoot(t *testing.T) { testFiler := filer.NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil) dir := t.TempDir() diff --git a/weed/filer/leveldb3/leveldb3_store.go b/weed/filer/leveldb3/leveldb3_store.go index 029d902715..0b8748347a 100644 --- a/weed/filer/leveldb3/leveldb3_store.go +++ b/weed/filer/leveldb3/leveldb3_store.go @@ -370,7 +370,7 @@ func (store *LevelDB3Store) ListDirectoryPrefixedEntries(ctx context.Context, di directoryPrefix := genDirectoryKeyPrefix(shortPath, prefix) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart = genDirectoryKeyPrefix(shortPath, startFileName) } diff --git a/weed/filer/leveldb3/leveldb3_store_test.go b/weed/filer/leveldb3/leveldb3_store_test.go index 62896a299f..a9f3be10da 100644 --- a/weed/filer/leveldb3/leveldb3_store_test.go +++ b/weed/filer/leveldb3/leveldb3_store_test.go @@ -182,3 +182,28 @@ func TestEmptyRoot(t *testing.T) { } } + +func TestListDirectoryPrefixedEntries_StartBelowPrefix(t *testing.T) { + store := &LevelDB3Store{} + store.initialize(t.TempDir()) + defer store.Shutdown() + + ctx := context.Background() + for _, name := range []string{"item-a", "item-a.versions", "item-b.versions"} { + if err := store.InsertEntry(ctx, &filer.Entry{FullPath: util.FullPath("/d/" + name)}); err != nil { + t.Fatalf("insert %s: %v", name, err) + } + } + + var names []string + _, err := store.ListDirectoryPrefixedEntries(ctx, util.FullPath("/d"), "item", true, 100, "item-b", func(entry *filer.Entry) (bool, error) { + names = append(names, entry.Name()) + return true, nil + }) + if err != nil { + t.Fatalf("list: %v", err) + } + if len(names) != 1 || names[0] != "item-b.versions" { + t.Fatalf("want [item-b.versions], got %v", names) + } +} diff --git a/weed/filer/rocksdb/rocksdb_store.go b/weed/filer/rocksdb/rocksdb_store.go index 2283efa6fb..dd88d73aa3 100644 --- a/weed/filer/rocksdb/rocksdb_store.go +++ b/weed/filer/rocksdb/rocksdb_store.go @@ -243,7 +243,7 @@ func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName) } diff --git a/weed/filer/tikv/tikv_store.go b/weed/filer/tikv/tikv_store.go index d6ba8f2844..92c34d70d6 100644 --- a/weed/filer/tikv/tikv_store.go +++ b/weed/filer/tikv/tikv_store.go @@ -246,7 +246,7 @@ func (store *TikvStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPat lastFileName := "" directoryPrefix := store.genDirectoryKeyPrefix(dirPath, prefix) lastFileStart := directoryPrefix - if startFileName != "" { + if startFileName > prefix { lastFileStart = store.genDirectoryKeyPrefix(dirPath, startFileName) }