add stream list directory entries

This commit is contained in:
Chris Lu 2021-01-15 23:56:24 -08:00
parent 01dc8a43ba
commit a4063a5437
20 changed files with 267 additions and 242 deletions

View file

@ -172,7 +172,7 @@ func (store *AbstractSqlStore) DeleteFolderChildren(ctx context.Context, fullpat
return nil return nil
} }
func (store *AbstractSqlStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*filer.Entry, hasMore bool, err error) { func (store *AbstractSqlStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
sqlText := store.SqlListExclusive sqlText := store.SqlListExclusive
if includeStartFile { if includeStartFile {
sqlText = store.SqlListInclusive sqlText = store.SqlListInclusive
@ -180,7 +180,7 @@ func (store *AbstractSqlStore) ListDirectoryPrefixedEntries(ctx context.Context,
rows, err := store.getTxOrDB(ctx).QueryContext(ctx, sqlText, util.HashStringToLong(string(dirPath)), startFileName, string(dirPath), prefix+"%", limit+1) rows, err := store.getTxOrDB(ctx).QueryContext(ctx, sqlText, util.HashStringToLong(string(dirPath)), startFileName, string(dirPath), prefix+"%", limit+1)
if err != nil { if err != nil {
return nil, false, fmt.Errorf("list %s : %v", dirPath, err) return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
} }
defer rows.Close() defer rows.Close()
@ -189,30 +189,29 @@ func (store *AbstractSqlStore) ListDirectoryPrefixedEntries(ctx context.Context,
var data []byte var data []byte
if err = rows.Scan(&name, &data); err != nil { if err = rows.Scan(&name, &data); err != nil {
glog.V(0).Infof("scan %s : %v", dirPath, err) glog.V(0).Infof("scan %s : %v", dirPath, err)
return nil, false, fmt.Errorf("scan %s: %v", dirPath, err) return lastFileName, fmt.Errorf("scan %s: %v", dirPath, err)
} }
lastFileName = name
entry := &filer.Entry{ entry := &filer.Entry{
FullPath: util.NewFullPath(string(dirPath), name), FullPath: util.NewFullPath(string(dirPath), name),
} }
if err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); err != nil { if err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); err != nil {
glog.V(0).Infof("scan decode %s : %v", entry.FullPath, err) glog.V(0).Infof("scan decode %s : %v", entry.FullPath, err)
return nil, false, fmt.Errorf("scan decode %s : %v", entry.FullPath, err) return lastFileName, fmt.Errorf("scan decode %s : %v", entry.FullPath, err)
}
if !eachEntryFunc(entry) {
break
} }
entries = append(entries, entry)
} }
hasMore = int64(len(entries)) == limit+1 return lastFileName, nil
if hasMore {
entries = entries[:limit]
}
return entries, hasMore, nil
} }
func (store *AbstractSqlStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *AbstractSqlStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "") return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", nil)
} }
func (store *AbstractSqlStore) Shutdown() { func (store *AbstractSqlStore) Shutdown() {

View file

@ -168,11 +168,11 @@ func (store *CassandraStore) DeleteFolderChildren(ctx context.Context, fullpath
return nil return nil
} }
func (store *CassandraStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*filer.Entry, hasMore bool, err error) { func (store *CassandraStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return nil, false, filer.ErrUnsupportedListDirectoryPrefixed return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
} }
func (store *CassandraStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *CassandraStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
if _, ok := store.isSuperLargeDirectory(string(dirPath)); ok { if _, ok := store.isSuperLargeDirectory(string(dirPath)); ok {
return // nil, filer.ErrUnsupportedSuperLargeDirectoryListing return // nil, filer.ErrUnsupportedSuperLargeDirectoryListing
@ -190,23 +190,21 @@ func (store *CassandraStore) ListDirectoryEntries(ctx context.Context, dirPath u
entry := &filer.Entry{ entry := &filer.Entry{
FullPath: util.NewFullPath(string(dirPath), name), FullPath: util.NewFullPath(string(dirPath), name),
} }
lastFileName = name
if decodeErr := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); decodeErr != nil { if decodeErr := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); decodeErr != nil {
err = decodeErr err = decodeErr
glog.V(0).Infof("list %s : %v", entry.FullPath, err) glog.V(0).Infof("list %s : %v", entry.FullPath, err)
break break
} }
entries = append(entries, entry) if !eachEntryFunc(entry) {
break
}
} }
if err := iter.Close(); err != nil { if err := iter.Close(); err != nil {
glog.V(0).Infof("list iterator close: %v", err) glog.V(0).Infof("list iterator close: %v", err)
} }
hasMore = int64(len(entries)) == limit+1 return lastFileName, err
if hasMore {
entries = entries[:limit]
}
return entries, hasMore, err
} }
func (store *CassandraStore) Shutdown() { func (store *CassandraStore) Shutdown() {

View file

@ -96,8 +96,8 @@ func (store *ElasticStore) RollbackTransaction(ctx context.Context) error {
return nil return nil
} }
func (store *ElasticStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*filer.Entry, hasMore bool, err error) { func (store *ElasticStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return nil, false, filer.ErrUnsupportedListDirectoryPrefixed return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
} }
func (store *ElasticStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) { func (store *ElasticStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
@ -187,26 +187,28 @@ func (store *ElasticStore) deleteEntry(ctx context.Context, index, id string) (e
} }
func (store *ElasticStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) { func (store *ElasticStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
if entries, _, err := store.ListDirectoryEntries(ctx, fullpath, "", false, math.MaxInt32); err == nil { _, err = store.ListDirectoryEntries(ctx, fullpath, "", false, math.MaxInt32, func(entry *filer.Entry) bool {
for _, entry := range entries { if err := store.DeleteEntry(ctx, entry.FullPath); err != nil {
store.DeleteEntry(ctx, entry.FullPath) glog.Errorf("elastic delete %s: %v.", entry.FullPath, err)
return false
} }
} return true
return nil })
return
} }
func (store *ElasticStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *ElasticStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
if string(dirPath) == "/" { if string(dirPath) == "/" {
return store.listRootDirectoryEntries(ctx, startFileName, includeStartFile, limit) return store.listRootDirectoryEntries(ctx, startFileName, includeStartFile, limit, eachEntryFunc)
} }
return store.listDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit) return store.listDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, eachEntryFunc)
} }
func (store *ElasticStore) listRootDirectoryEntries(ctx context.Context, startFileName string, inclusive bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *ElasticStore) listRootDirectoryEntries(ctx context.Context, startFileName string, inclusive bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
indexResult, err := store.client.CatIndices().Do(ctx) indexResult, err := store.client.CatIndices().Do(ctx)
if err != nil { if err != nil {
glog.Errorf("list indices %v.", err) glog.Errorf("list indices %v.", err)
return entries, false, err return
} }
for _, index := range indexResult { for _, index := range indexResult {
if index.Index == indexKV { if index.Index == indexKV {
@ -216,32 +218,33 @@ func (store *ElasticStore) listRootDirectoryEntries(ctx context.Context, startFi
if entry, err := store.FindEntry(ctx, if entry, err := store.FindEntry(ctx,
weed_util.FullPath("/"+strings.Replace(index.Index, indexPrefix, "", 1))); err == nil { weed_util.FullPath("/"+strings.Replace(index.Index, indexPrefix, "", 1))); err == nil {
fileName := getFileName(entry.FullPath) fileName := getFileName(entry.FullPath)
lastFileName = fileName
if fileName == startFileName && !inclusive { if fileName == startFileName && !inclusive {
continue continue
} }
limit-- limit--
if limit < 0 { if limit < 0 {
hasMore = true
break break
} }
entries = append(entries, entry) if !eachEntryFunc(entry) {
break
}
} }
} }
} }
return entries, hasMore, nil return
} }
func (store *ElasticStore) listDirectoryEntries( func (store *ElasticStore) listDirectoryEntries(
ctx context.Context, fullpath weed_util.FullPath, startFileName string, inclusive bool, limit int64, ctx context.Context, fullpath weed_util.FullPath, startFileName string, inclusive bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
) (entries []*filer.Entry, hasMore bool, err error) {
first := true first := true
index := getIndex(fullpath) index := getIndex(fullpath)
nextStart := "" nextStart := ""
parentId := weed_util.Md5String([]byte(fullpath)) parentId := weed_util.Md5String([]byte(fullpath))
if _, err := store.client.Refresh(index).Do(ctx); err != nil { if _, err = store.client.Refresh(index).Do(ctx); err != nil {
if elastic.IsNotFound(err) { if elastic.IsNotFound(err) {
store.client.CreateIndex(index).Do(ctx) store.client.CreateIndex(index).Do(ctx)
return entries, hasMore, nil return
} }
} }
for { for {
@ -249,7 +252,7 @@ func (store *ElasticStore) listDirectoryEntries(
if (startFileName == "" && first) || inclusive { if (startFileName == "" && first) || inclusive {
if result, err = store.search(ctx, index, parentId); err != nil { if result, err = store.search(ctx, index, parentId); err != nil {
glog.Errorf("search (%s,%s,%t,%d) %v.", string(fullpath), startFileName, inclusive, limit, err) glog.Errorf("search (%s,%s,%t,%d) %v.", string(fullpath), startFileName, inclusive, limit, err)
return entries, hasMore, err return
} }
} else { } else {
fullPath := string(fullpath) + "/" + startFileName fullPath := string(fullpath) + "/" + startFileName
@ -259,7 +262,7 @@ func (store *ElasticStore) listDirectoryEntries(
after := weed_util.Md5String([]byte(fullPath)) after := weed_util.Md5String([]byte(fullPath))
if result, err = store.searchAfter(ctx, index, parentId, after); err != nil { if result, err = store.searchAfter(ctx, index, parentId, after); err != nil {
glog.Errorf("searchAfter (%s,%s,%t,%d) %v.", string(fullpath), startFileName, inclusive, limit, err) glog.Errorf("searchAfter (%s,%s,%t,%d) %v.", string(fullpath), startFileName, inclusive, limit, err)
return entries, hasMore, err return
} }
} }
first = false first = false
@ -271,22 +274,21 @@ func (store *ElasticStore) listDirectoryEntries(
if err := jsoniter.Unmarshal(hit.Source, esEntry); err == nil { if err := jsoniter.Unmarshal(hit.Source, esEntry); err == nil {
limit-- limit--
if limit < 0 { if limit < 0 {
hasMore = true return lastFileName, nil
return entries, hasMore, nil
} }
nextStart = string(esEntry.Entry.FullPath) nextStart = string(esEntry.Entry.FullPath)
fileName := getFileName(esEntry.Entry.FullPath) fileName := getFileName(esEntry.Entry.FullPath)
lastFileName = fileName
if fileName == startFileName && !inclusive { if fileName == startFileName && !inclusive {
continue continue
} }
entries = append(entries, esEntry.Entry) if !eachEntryFunc(esEntry.Entry) {
break
}
} }
} }
if len(result.Hits.Hits) < store.maxPageSize {
break
}
} }
return entries, hasMore, nil return
} }
func (store *ElasticStore) search(ctx context.Context, index, parentId string) (result *elastic.SearchResult, err error) { func (store *ElasticStore) search(ctx context.Context, index, parentId string) (result *elastic.SearchResult, err error) {

View file

@ -139,17 +139,17 @@ func (store *EtcdStore) DeleteFolderChildren(ctx context.Context, fullpath weed_
return nil return nil
} }
func (store *EtcdStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*filer.Entry, hasMore bool, err error) { 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) {
return nil, false, filer.ErrUnsupportedListDirectoryPrefixed return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
} }
func (store *EtcdStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *EtcdStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
directoryPrefix := genDirectoryKeyPrefix(dirPath, "") directoryPrefix := genDirectoryKeyPrefix(dirPath, "")
resp, err := store.client.Get(ctx, string(directoryPrefix), resp, err := store.client.Get(ctx, string(directoryPrefix),
clientv3.WithPrefix(), clientv3.WithSort(clientv3.SortByKey, clientv3.SortDescend)) clientv3.WithPrefix(), clientv3.WithSort(clientv3.SortByKey, clientv3.SortDescend))
if err != nil { if err != nil {
return nil, false, fmt.Errorf("list %s : %v", dirPath, err) return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
} }
for _, kv := range resp.Kvs { for _, kv := range resp.Kvs {
@ -160,9 +160,9 @@ func (store *EtcdStore) ListDirectoryEntries(ctx context.Context, dirPath weed_u
if fileName == startFileName && !includeStartFile { if fileName == startFileName && !includeStartFile {
continue continue
} }
lastFileName = fileName
limit-- limit--
if limit < 0 { if limit < 0 {
hasMore = true
break break
} }
entry := &filer.Entry{ entry := &filer.Entry{
@ -173,10 +173,12 @@ func (store *EtcdStore) ListDirectoryEntries(ctx context.Context, dirPath weed_u
glog.V(0).Infof("list %s : %v", entry.FullPath, err) glog.V(0).Infof("list %s : %v", entry.FullPath, err)
break break
} }
entries = append(entries, entry) if !eachEntryFunc(entry) {
break
}
} }
return entries, hasMore, err return lastFileName, err
} }
func genKey(dirPath, fileName string) (key []byte) { func genKey(dirPath, fileName string) (key []byte) {

View file

@ -281,22 +281,19 @@ func (f *Filer) FindEntry(ctx context.Context, p util.FullPath) (entry *Entry, e
} }
func (f *Filer) doListDirectoryEntries(ctx context.Context, p util.FullPath, startFileName string, inclusive bool, limit int64, prefix string) (entries []*Entry, hasMore bool, expiredCount int64, lastFileName string, err error) { func (f *Filer) doListDirectoryEntries(ctx context.Context, p util.FullPath, startFileName string, inclusive bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (expiredCount int64, lastFileName string, err error) {
listedEntries, listHasMore, listErr := f.Store.ListDirectoryPrefixedEntries(ctx, p, startFileName, inclusive, limit, prefix) lastFileName, err = f.Store.ListDirectoryPrefixedEntries(ctx, p, startFileName, inclusive, limit, prefix, func(entry *Entry) bool {
hasMore = listHasMore
if listErr != nil {
return listedEntries, hasMore, expiredCount, "", listErr
}
for _, entry := range listedEntries {
lastFileName = entry.Name()
if entry.TtlSec > 0 { if entry.TtlSec > 0 {
if entry.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) { if entry.Crtime.Add(time.Duration(entry.TtlSec) * time.Second).Before(time.Now()) {
f.Store.DeleteOneEntry(ctx, entry) f.Store.DeleteOneEntry(ctx, entry)
expiredCount++ expiredCount++
continue return true
} }
} }
entries = append(entries, entry) return eachEntryFunc(entry)
})
if err != nil {
return expiredCount, lastFileName, err
} }
return return
} }

View file

@ -32,49 +32,76 @@ func (f *Filer) ListDirectoryEntries(ctx context.Context, p util.FullPath, start
var missedCount int64 var missedCount int64
var lastFileName string var lastFileName string
entries, hasMore, missedCount, lastFileName, err = f.doListPatternMatchedEntries(ctx, p, startFileName, inclusive, limit, prefix, restNamePattern) missedCount, lastFileName, err = f.doListPatternMatchedEntries(ctx, p, startFileName, inclusive, limit+1, prefix, restNamePattern, func(entry *Entry) bool {
entries = append(entries, entry)
return true
})
for missedCount > 0 && err == nil { for missedCount > 0 && err == nil {
var makeupEntries []*Entry missedCount, lastFileName, err = f.doListPatternMatchedEntries(ctx, p, lastFileName, false, missedCount+1, prefix, restNamePattern, func(entry *Entry) bool {
makeupEntries, hasMore, missedCount, lastFileName, err = f.doListPatternMatchedEntries(ctx, p, lastFileName, false, missedCount, prefix, restNamePattern)
for _, entry := range makeupEntries {
entries = append(entries, entry) entries = append(entries, entry)
} return true
})
}
hasMore = int64(len(entries)) >= limit+1
if hasMore {
entries = entries[:limit]
} }
return entries, hasMore, err return entries, hasMore, err
} }
func (f *Filer) doListPatternMatchedEntries(ctx context.Context, p util.FullPath, startFileName string, inclusive bool, limit int64, prefix, restNamePattern string) (matchedEntries []*Entry, hasMore bool, missedCount int64, lastFileName string, err error) { // For now, prefix and namePattern are mutually exclusive
var foundEntries []*Entry func (f *Filer) StreamListDirectoryEntries(ctx context.Context, p util.FullPath, startFileName string, inclusive bool, limit int64, prefix string, namePattern string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
if strings.HasSuffix(string(p), "/") && len(p) > 1 {
p = p[0 : len(p)-1]
}
foundEntries, hasMore, lastFileName, err = f.doListValidEntries(ctx, p, startFileName, inclusive, limit, prefix) prefixInNamePattern, restNamePattern := splitPattern(namePattern)
if err != nil { if prefixInNamePattern != "" {
return prefix = prefixInNamePattern
} }
var missedCount int64
missedCount, lastFileName, err = f.doListPatternMatchedEntries(ctx, p, startFileName, inclusive, limit, prefix, restNamePattern, eachEntryFunc)
for missedCount > 0 && err == nil {
missedCount, lastFileName, err = f.doListPatternMatchedEntries(ctx, p, lastFileName, false, missedCount, prefix, restNamePattern, eachEntryFunc)
}
return
}
func (f *Filer) doListPatternMatchedEntries(ctx context.Context, p util.FullPath, startFileName string, inclusive bool, limit int64, prefix, restNamePattern string, eachEntryFunc ListEachEntryFunc) (missedCount int64, lastFileName string, err error) {
if len(restNamePattern) == 0 { if len(restNamePattern) == 0 {
return foundEntries, false, 0, lastFileName, nil lastFileName, err = f.doListValidEntries(ctx, p, startFileName, inclusive, limit, prefix, eachEntryFunc)
return 0, lastFileName, err
} }
for _, entry := range foundEntries {
lastFileName, err = f.doListValidEntries(ctx, p, startFileName, inclusive, limit, prefix, func(entry *Entry) bool {
nameToTest := strings.ToLower(entry.Name()) nameToTest := strings.ToLower(entry.Name())
if matched, matchErr := filepath.Match(restNamePattern, nameToTest[len(prefix):]); matchErr == nil && matched { if matched, matchErr := filepath.Match(restNamePattern, nameToTest[len(prefix):]); matchErr == nil && matched {
matchedEntries = append(matchedEntries, entry) if !eachEntryFunc(entry) {
return false
}
} else { } else {
missedCount++ missedCount++
} }
return true
})
if err != nil {
return
} }
return return
} }
func (f *Filer) doListValidEntries(ctx context.Context, p util.FullPath, startFileName string, inclusive bool, limit int64, prefix string) (entries []*Entry, hasMore bool, lastFileName string, err error) { func (f *Filer) doListValidEntries(ctx context.Context, p util.FullPath, startFileName string, inclusive bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
var makeupEntries []*Entry
var expiredCount int64 var expiredCount int64
entries, hasMore, expiredCount, lastFileName, err = f.doListDirectoryEntries(ctx, p, startFileName, inclusive, limit, prefix) expiredCount, lastFileName, err = f.doListDirectoryEntries(ctx, p, startFileName, inclusive, limit, prefix, eachEntryFunc)
for expiredCount > 0 && err == nil { for expiredCount > 0 && err == nil {
makeupEntries, hasMore, expiredCount, lastFileName, err = f.doListDirectoryEntries(ctx, p, lastFileName, false, expiredCount, prefix) expiredCount, lastFileName, err = f.doListDirectoryEntries(ctx, p, lastFileName, false, expiredCount, prefix, eachEntryFunc)
if err == nil {
entries = append(entries, makeupEntries...)
}
} }
return return
} }

View file

@ -13,6 +13,8 @@ var (
ErrKvNotFound = errors.New("kv: not found") ErrKvNotFound = errors.New("kv: not found")
) )
type ListEachEntryFunc func(entry *Entry) bool
type FilerStore interface { type FilerStore interface {
// GetName gets the name to locate the configuration in filer.toml file // GetName gets the name to locate the configuration in filer.toml file
GetName() string GetName() string
@ -24,8 +26,8 @@ type FilerStore interface {
FindEntry(context.Context, util.FullPath) (entry *Entry, err error) FindEntry(context.Context, util.FullPath) (entry *Entry, err error)
DeleteEntry(context.Context, util.FullPath) (err error) DeleteEntry(context.Context, util.FullPath) (err error)
DeleteFolderChildren(context.Context, util.FullPath) (err error) DeleteFolderChildren(context.Context, util.FullPath) (err error)
ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64) ([]*Entry, bool, error) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error)
ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) ([]*Entry, bool, error) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error)
BeginTransaction(ctx context.Context) (context.Context, error) BeginTransaction(ctx context.Context) (context.Context, error)
CommitTransaction(ctx context.Context) error CommitTransaction(ctx context.Context) error

View file

@ -106,32 +106,24 @@ func (t *FilerStorePathTranlator) DeleteFolderChildren(ctx context.Context, fp u
return t.actualStore.DeleteFolderChildren(ctx, newFullPath) return t.actualStore.DeleteFolderChildren(ctx, newFullPath)
} }
func (t *FilerStorePathTranlator) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64) ([]*Entry, bool, error) { func (t *FilerStorePathTranlator) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (string, error) {
newFullPath := t.translatePath(dirPath) newFullPath := t.translatePath(dirPath)
entries, hasMore, err := t.actualStore.ListDirectoryEntries(ctx, newFullPath, startFileName, includeStartFile, limit) return t.actualStore.ListDirectoryEntries(ctx, newFullPath, startFileName, includeStartFile, limit, func(entry *Entry) bool {
if err != nil {
return nil, hasMore, err
}
for _, entry := range entries {
entry.FullPath = dirPath[:len(t.storeRoot)-1] + entry.FullPath entry.FullPath = dirPath[:len(t.storeRoot)-1] + entry.FullPath
} return eachEntryFunc(entry)
return entries, hasMore, err })
} }
func (t *FilerStorePathTranlator) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) ([]*Entry, bool, error) { func (t *FilerStorePathTranlator) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (string, error) {
newFullPath := t.translatePath(dirPath) newFullPath := t.translatePath(dirPath)
entries, hasMore, err := t.actualStore.ListDirectoryPrefixedEntries(ctx, newFullPath, startFileName, includeStartFile, limit, prefix) return t.actualStore.ListDirectoryPrefixedEntries(ctx, newFullPath, startFileName, includeStartFile, limit, prefix, func(entry *Entry) bool {
if err != nil {
return nil, hasMore, err
}
for _, entry := range entries {
entry.FullPath = dirPath[:len(t.storeRoot)-1] + entry.FullPath entry.FullPath = dirPath[:len(t.storeRoot)-1] + entry.FullPath
} return eachEntryFunc(entry)
return entries, hasMore, nil })
} }
func (t *FilerStorePathTranlator) BeginTransaction(ctx context.Context) (context.Context, error) { func (t *FilerStorePathTranlator) BeginTransaction(ctx context.Context) (context.Context, error) {

View file

@ -194,7 +194,7 @@ func (fsw *FilerStoreWrapper) DeleteFolderChildren(ctx context.Context, fp util.
return actualStore.DeleteFolderChildren(ctx, fp) return actualStore.DeleteFolderChildren(ctx, fp)
} }
func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64) ([]*Entry, bool, error) { func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc ListEachEntryFunc) (string, error) {
actualStore := fsw.getActualStore(dirPath + "/") actualStore := fsw.getActualStore(dirPath + "/")
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "list").Inc() stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "list").Inc()
start := time.Now() start := time.Now()
@ -203,18 +203,14 @@ func (fsw *FilerStoreWrapper) ListDirectoryEntries(ctx context.Context, dirPath
}() }()
glog.V(4).Infof("ListDirectoryEntries %s from %s limit %d", dirPath, startFileName, limit) glog.V(4).Infof("ListDirectoryEntries %s from %s limit %d", dirPath, startFileName, limit)
entries, hasMore, err := actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit) return actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *Entry) bool {
if err != nil {
return nil, hasMore, err
}
for _, entry := range entries {
fsw.maybeReadHardLink(ctx, entry) fsw.maybeReadHardLink(ctx, entry)
filer_pb.AfterEntryDeserialization(entry.Chunks) filer_pb.AfterEntryDeserialization(entry.Chunks)
} return eachEntryFunc(entry)
return entries, hasMore, err })
} }
func (fsw *FilerStoreWrapper) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) ([]*Entry, bool, error) { func (fsw *FilerStoreWrapper) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
actualStore := fsw.getActualStore(dirPath + "/") actualStore := fsw.getActualStore(dirPath + "/")
stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "prefixList").Inc() stats.FilerStoreCounter.WithLabelValues(actualStore.GetName(), "prefixList").Inc()
start := time.Now() start := time.Now()
@ -222,48 +218,52 @@ func (fsw *FilerStoreWrapper) ListDirectoryPrefixedEntries(ctx context.Context,
stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "prefixList").Observe(time.Since(start).Seconds()) stats.FilerStoreHistogram.WithLabelValues(actualStore.GetName(), "prefixList").Observe(time.Since(start).Seconds())
}() }()
glog.V(4).Infof("ListDirectoryPrefixedEntries %s from %s prefix %s limit %d", dirPath, startFileName, prefix, limit) glog.V(4).Infof("ListDirectoryPrefixedEntries %s from %s prefix %s limit %d", dirPath, startFileName, prefix, limit)
entries, hasMore, err := actualStore.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix) lastFileName, err = actualStore.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix, eachEntryFunc)
if err == ErrUnsupportedListDirectoryPrefixed { if err == ErrUnsupportedListDirectoryPrefixed {
entries, hasMore, err = fsw.prefixFilterEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix) lastFileName, err = fsw.prefixFilterEntries(ctx, dirPath, startFileName, includeStartFile, limit, prefix, func(entry *Entry) bool {
fsw.maybeReadHardLink(ctx, entry)
filer_pb.AfterEntryDeserialization(entry.Chunks)
return eachEntryFunc(entry)
})
} }
if err != nil { return lastFileName, err
return nil, hasMore, err
}
for _, entry := range entries {
fsw.maybeReadHardLink(ctx, entry)
filer_pb.AfterEntryDeserialization(entry.Chunks)
}
return entries, hasMore, nil
} }
func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*Entry, hasMore bool, err error) { func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) {
actualStore := fsw.getActualStore(dirPath + "/") actualStore := fsw.getActualStore(dirPath + "/")
entries, hasMore, err = actualStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit)
if err != nil {
return nil, hasMore, err
}
if prefix == "" { if prefix == "" {
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 {
notPrefixed = append(notPrefixed, entry)
return true
})
if err != nil {
return return
} }
count := int64(0) count := int64(0)
var lastFileName string
notPrefixed := entries
entries = nil
for count < limit && len(notPrefixed) > 0 { for count < limit && len(notPrefixed) > 0 {
for _, entry := range notPrefixed { for _, entry := range notPrefixed {
lastFileName = entry.Name()
if strings.HasPrefix(entry.Name(), prefix) { if strings.HasPrefix(entry.Name(), prefix) {
count++ count++
entries = append(entries, entry) if !eachEntryFunc(entry) {
return
}
if count >= limit { if count >= limit {
break break
} }
} }
} }
if count < limit { if count < limit {
notPrefixed, hasMore, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit) notPrefixed = notPrefixed[:0]
_, err = actualStore.ListDirectoryEntries(ctx, dirPath, lastFileName, false, limit, func(entry *Entry) bool {
notPrefixed = append(notPrefixed, entry)
return true
})
if err != nil { if err != nil {
return return
} }

View file

@ -148,20 +148,18 @@ func (store *HbaseStore) DeleteFolderChildren(ctx context.Context, path util.Ful
return return
} }
func (store *HbaseStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64) ([]*filer.Entry, bool, error) { func (store *HbaseStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (string, error) {
return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "") return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", eachEntryFunc)
} }
func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) ([]*filer.Entry, bool, error) { 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}} family := map[string][]string{store.cfMetaDir: {COLUMN_NAME}}
expectedPrefix := []byte(dirPath.Child(prefix)) expectedPrefix := []byte(dirPath.Child(prefix))
scan, err := hrpc.NewScanRange(ctx, store.table, expectedPrefix, nil, hrpc.Families(family)) scan, err := hrpc.NewScanRange(ctx, store.table, expectedPrefix, nil, hrpc.Families(family))
if err != nil { if err != nil {
return nil, false, err return lastFileName, err
} }
var hasMore bool
var entries []*filer.Entry
scanner := store.Client.Scan(scan) scanner := store.Client.Scan(scan)
defer scanner.Close() defer scanner.Close()
for { for {
@ -170,7 +168,7 @@ func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPa
break break
} }
if err != nil { if err != nil {
return entries, hasMore, err return lastFileName, err
} }
if len(res.Cells) == 0 { if len(res.Cells) == 0 {
continue continue
@ -187,6 +185,8 @@ func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPa
continue continue
} }
lastFileName = fileName
value := cell.Value value := cell.Value
if fileName == startFileName && !includeStartFile { if fileName == startFileName && !includeStartFile {
@ -195,7 +195,6 @@ func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPa
limit-- limit--
if limit < 0 { if limit < 0 {
hasMore = true
break break
} }
entry := &filer.Entry{ entry := &filer.Entry{
@ -206,10 +205,12 @@ func (store *HbaseStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPa
glog.V(0).Infof("list %s : %v", entry.FullPath, err) glog.V(0).Infof("list %s : %v", entry.FullPath, err)
break break
} }
entries = append(entries, entry) if !eachEntryFunc(entry) {
break
}
} }
return entries, hasMore, nil return lastFileName, nil
} }
func (store *HbaseStore) BeginTransaction(ctx context.Context) (context.Context, error) { func (store *HbaseStore) BeginTransaction(ctx context.Context) (context.Context, error) {

View file

@ -162,11 +162,11 @@ func (store *LevelDBStore) DeleteFolderChildren(ctx context.Context, fullpath we
return nil return nil
} }
func (store *LevelDBStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *LevelDBStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "") return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", eachEntryFunc)
} }
func (store *LevelDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*filer.Entry, hasMore bool, err error) { func (store *LevelDBStore) 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) directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
lastFileStart := directoryPrefix lastFileStart := directoryPrefix
@ -187,9 +187,9 @@ func (store *LevelDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir
if fileName == startFileName && !includeStartFile { if fileName == startFileName && !includeStartFile {
continue continue
} }
lastFileName = fileName
limit-- limit--
if limit < 0 { if limit < 0 {
hasMore = true
break break
} }
entry := &filer.Entry{ entry := &filer.Entry{
@ -200,11 +200,13 @@ func (store *LevelDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir
glog.V(0).Infof("list %s : %v", entry.FullPath, err) glog.V(0).Infof("list %s : %v", entry.FullPath, err)
break break
} }
entries = append(entries, entry) if !eachEntryFunc(entry) {
break
}
} }
iter.Release() iter.Release()
return entries, hasMore, err return lastFileName, err
} }
func genKey(dirPath, fileName string) (key []byte) { func genKey(dirPath, fileName string) (key []byte) {

View file

@ -171,11 +171,11 @@ func (store *LevelDB2Store) DeleteFolderChildren(ctx context.Context, fullpath w
return nil return nil
} }
func (store *LevelDB2Store) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *LevelDB2Store) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "") return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", eachEntryFunc)
} }
func (store *LevelDB2Store) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*filer.Entry, hasMore bool, err error) { func (store *LevelDB2Store) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
directoryPrefix, partitionId := genDirectoryKeyPrefix(dirPath, prefix, store.dbCount) directoryPrefix, partitionId := genDirectoryKeyPrefix(dirPath, prefix, store.dbCount)
lastFileStart := directoryPrefix lastFileStart := directoryPrefix
@ -196,9 +196,9 @@ func (store *LevelDB2Store) ListDirectoryPrefixedEntries(ctx context.Context, di
if fileName == startFileName && !includeStartFile { if fileName == startFileName && !includeStartFile {
continue continue
} }
lastFileName = fileName
limit-- limit--
if limit < 0 { if limit < 0 {
hasMore = true
break break
} }
entry := &filer.Entry{ entry := &filer.Entry{
@ -211,11 +211,13 @@ func (store *LevelDB2Store) ListDirectoryPrefixedEntries(ctx context.Context, di
glog.V(0).Infof("list %s : %v", entry.FullPath, err) glog.V(0).Infof("list %s : %v", entry.FullPath, err)
break break
} }
entries = append(entries, entry) if !eachEntryFunc(entry) {
break
}
} }
iter.Release() iter.Release()
return entries, hasMore, err return lastFileName, err
} }
func genKey(dirPath, fileName string, dbCount int) (key []byte, partitionId int) { func genKey(dirPath, fileName string, dbCount int) (key []byte, partitionId int) {

View file

@ -286,15 +286,15 @@ func (store *LevelDB3Store) DeleteFolderChildren(ctx context.Context, fullpath w
return nil return nil
} }
func (store *LevelDB3Store) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *LevelDB3Store) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "") return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", eachEntryFunc)
} }
func (store *LevelDB3Store) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*filer.Entry, hasMore bool, err error) { func (store *LevelDB3Store) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
db, _, shortPath, err := store.findDB(dirPath, true) db, _, shortPath, err := store.findDB(dirPath, true)
if err != nil { if err != nil {
return nil, false, fmt.Errorf("findDB %s : %v", dirPath, err) return lastFileName, fmt.Errorf("findDB %s : %v", dirPath, err)
} }
directoryPrefix := genDirectoryKeyPrefix(shortPath, prefix) directoryPrefix := genDirectoryKeyPrefix(shortPath, prefix)
@ -316,9 +316,9 @@ func (store *LevelDB3Store) ListDirectoryPrefixedEntries(ctx context.Context, di
if fileName == startFileName && !includeStartFile { if fileName == startFileName && !includeStartFile {
continue continue
} }
lastFileName = fileName
limit-- limit--
if limit < 0 { if limit < 0 {
hasMore = true
break break
} }
entry := &filer.Entry{ entry := &filer.Entry{
@ -331,11 +331,13 @@ func (store *LevelDB3Store) ListDirectoryPrefixedEntries(ctx context.Context, di
glog.V(0).Infof("list %s : %v", entry.FullPath, err) glog.V(0).Infof("list %s : %v", entry.FullPath, err)
break break
} }
entries = append(entries, entry) if !eachEntryFunc(entry) {
break
}
} }
iter.Release() iter.Release()
return entries, hasMore, err return lastFileName, err
} }
func genKey(dirPath, fileName string) (key []byte) { func genKey(dirPath, fileName string) (key []byte) {

View file

@ -178,11 +178,11 @@ func (store *MongodbStore) DeleteFolderChildren(ctx context.Context, fullpath ut
return nil return nil
} }
func (store *MongodbStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*filer.Entry, hasMore bool, err error) { func (store *MongodbStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return nil, false, filer.ErrUnsupportedListDirectoryPrefixed return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
} }
func (store *MongodbStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *MongodbStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
var where = bson.M{"directory": string(dirPath), "name": bson.M{"$gt": startFileName}} var where = bson.M{"directory": string(dirPath), "name": bson.M{"$gt": startFileName}}
if includeStartFile { if includeStartFile {
@ -190,38 +190,37 @@ func (store *MongodbStore) ListDirectoryEntries(ctx context.Context, dirPath uti
"$gte": startFileName, "$gte": startFileName,
} }
} }
optLimit := int64(limit + 1) optLimit := int64(limit)
opts := &options.FindOptions{Limit: &optLimit, Sort: bson.M{"name": 1}} opts := &options.FindOptions{Limit: &optLimit, Sort: bson.M{"name": 1}}
cur, err := store.connect.Database(store.database).Collection(store.collectionName).Find(ctx, where, opts) cur, err := store.connect.Database(store.database).Collection(store.collectionName).Find(ctx, where, opts)
for cur.Next(ctx) { for cur.Next(ctx) {
var data Model var data Model
err := cur.Decode(&data) err := cur.Decode(&data)
if err != nil && err != mongo.ErrNoDocuments { if err != nil && err != mongo.ErrNoDocuments {
return nil, false, err return lastFileName, err
} }
entry := &filer.Entry{ entry := &filer.Entry{
FullPath: util.NewFullPath(string(dirPath), data.Name), FullPath: util.NewFullPath(string(dirPath), data.Name),
} }
lastFileName = data.Name
if decodeErr := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data.Meta)); decodeErr != nil { if decodeErr := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data.Meta)); decodeErr != nil {
err = decodeErr err = decodeErr
glog.V(0).Infof("list %s : %v", entry.FullPath, err) glog.V(0).Infof("list %s : %v", entry.FullPath, err)
break break
} }
entries = append(entries, entry) if !eachEntryFunc(entry) {
} break
}
hasMore = int64(len(entries)) == limit+1
if hasMore {
entries = entries[:limit]
} }
if err := cur.Close(ctx); err != nil { if err := cur.Close(ctx); err != nil {
glog.V(0).Infof("list iterator close: %v", err) glog.V(0).Infof("list iterator close: %v", err)
} }
return entries, hasMore, err return lastFileName, err
} }
func (store *MongodbStore) Shutdown() { func (store *MongodbStore) Shutdown() {

View file

@ -125,16 +125,16 @@ func (store *UniversalRedisStore) DeleteFolderChildren(ctx context.Context, full
return nil return nil
} }
func (store *UniversalRedisStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*filer.Entry, hasMore bool, err error) { func (store *UniversalRedisStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return nil, false, filer.ErrUnsupportedListDirectoryPrefixed return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
} }
func (store *UniversalRedisStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *UniversalRedisStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
dirListKey := genDirectoryListKey(string(dirPath)) dirListKey := genDirectoryListKey(string(dirPath))
members, err := store.Client.SMembers(ctx, dirListKey).Result() members, err := store.Client.SMembers(ctx, dirListKey).Result()
if err != nil { if err != nil {
return nil, false, fmt.Errorf("list %s : %v", dirPath, err) return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
} }
// skip // skip
@ -160,15 +160,15 @@ func (store *UniversalRedisStore) ListDirectoryEntries(ctx context.Context, dirP
}) })
// limit // limit
if limit < int64(len(entries)) { if limit < int64(len(members)) {
members = members[:limit] members = members[:limit]
hasMore = true
} }
// fetch entry meta // fetch entry meta
for _, fileName := range members { for _, fileName := range members {
path := util.NewFullPath(string(dirPath), fileName) path := util.NewFullPath(string(dirPath), fileName)
entry, err := store.FindEntry(ctx, path) entry, err := store.FindEntry(ctx, path)
lastFileName = fileName
if err != nil { if err != nil {
glog.V(0).Infof("list %s : %v", path, err) glog.V(0).Infof("list %s : %v", path, err)
if err == filer_pb.ErrNotFound { if err == filer_pb.ErrNotFound {
@ -182,11 +182,13 @@ func (store *UniversalRedisStore) ListDirectoryEntries(ctx context.Context, dirP
continue continue
} }
} }
entries = append(entries, entry) if !eachEntryFunc(entry) {
break
}
} }
} }
return entries, hasMore, err return lastFileName, err
} }
func genDirectoryListKey(dir string) (dirList string) { func genDirectoryListKey(dir string) (dirList string) {

View file

@ -149,11 +149,11 @@ func (store *UniversalRedis2Store) DeleteFolderChildren(ctx context.Context, ful
return nil return nil
} }
func (store *UniversalRedis2Store) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*filer.Entry, hasMore bool, err error) { func (store *UniversalRedis2Store) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return nil, false, filer.ErrUnsupportedListDirectoryPrefixed return lastFileName, filer.ErrUnsupportedListDirectoryPrefixed
} }
func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
dirListKey := genDirectoryListKey(string(dirPath)) dirListKey := genDirectoryListKey(string(dirPath))
start := int64(0) start := int64(0)
@ -163,20 +163,16 @@ func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dir
start++ start++
} }
} }
members, err := store.Client.ZRange(ctx, dirListKey, start, start+int64(limit)-1+1).Result() members, err := store.Client.ZRange(ctx, dirListKey, start, start+int64(limit)-1).Result()
if err != nil { if err != nil {
return nil, false, fmt.Errorf("list %s : %v", dirPath, err) return lastFileName, fmt.Errorf("list %s : %v", dirPath, err)
}
hasMore = int64(len(entries)) == limit+1
if hasMore {
members = members[:len(members)-1]
} }
// fetch entry meta // fetch entry meta
for _, fileName := range members { for _, fileName := range members {
path := util.NewFullPath(string(dirPath), fileName) path := util.NewFullPath(string(dirPath), fileName)
entry, err := store.FindEntry(ctx, path) entry, err := store.FindEntry(ctx, path)
lastFileName = fileName
if err != nil { if err != nil {
glog.V(0).Infof("list %s : %v", path, err) glog.V(0).Infof("list %s : %v", path, err)
if err == filer_pb.ErrNotFound { if err == filer_pb.ErrNotFound {
@ -190,11 +186,13 @@ func (store *UniversalRedis2Store) ListDirectoryEntries(ctx context.Context, dir
continue continue
} }
} }
entries = append(entries, entry) if !eachEntryFunc(entry) {
break
}
} }
} }
return entries, hasMore, err return lastFileName, err
} }
func genDirectoryListKey(dir string) (dirList string) { func genDirectoryListKey(dir string) (dirList string) {

View file

@ -158,7 +158,7 @@ func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath we
iter := store.db.NewIterator(ro) iter := store.db.NewIterator(ro)
defer iter.Close() defer iter.Close()
_, err = enumerate(iter, directoryPrefix, nil, false, -1, func(key, value []byte) bool { err = enumerate(iter, directoryPrefix, nil, false, -1, func(key, value []byte) bool {
batch.Delete(key) batch.Delete(key)
return true return true
}) })
@ -175,7 +175,7 @@ func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath we
return nil return nil
} }
func enumerate(iter *gorocksdb.Iterator, prefix, lastKey []byte, includeLastKey bool, limit int64, fn func(key, value []byte) bool) (hasMore bool, err error) { func enumerate(iter *gorocksdb.Iterator, prefix, lastKey []byte, includeLastKey bool, limit int64, fn func(key, value []byte) bool) (err error) {
if len(lastKey) == 0 { if len(lastKey) == 0 {
iter.Seek(prefix) iter.Seek(prefix)
@ -196,7 +196,6 @@ func enumerate(iter *gorocksdb.Iterator, prefix, lastKey []byte, includeLastKey
if limit > 0 { if limit > 0 {
i++ i++
if i > limit { if i > limit {
hasMore = true
break break
} }
} }
@ -216,16 +215,16 @@ func enumerate(iter *gorocksdb.Iterator, prefix, lastKey []byte, includeLastKey
} }
if err := iter.Err(); err != nil { if err := iter.Err(); err != nil {
return hasMore, fmt.Errorf("prefix scan iterator: %v", err) return fmt.Errorf("prefix scan iterator: %v", err)
} }
return hasMore, nil return nil
} }
func (store *RocksDBStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64) (entries []*filer.Entry, hasMore bool, err error) { func (store *RocksDBStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "") return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", eachEntryFunc)
} }
func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string) (entries []*filer.Entry, hasMore bool, err error) { func (store *RocksDBStore) 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) directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
lastFileStart := directoryPrefix lastFileStart := directoryPrefix
@ -239,7 +238,7 @@ func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir
iter := store.db.NewIterator(ro) iter := store.db.NewIterator(ro)
defer iter.Close() defer iter.Close()
hasMore, err = enumerate(iter, directoryPrefix, lastFileStart, includeStartFile, limit, func(key, value []byte) bool { err = enumerate(iter, directoryPrefix, lastFileStart, includeStartFile, limit, func(key, value []byte) bool {
fileName := getNameFromKey(key) fileName := getNameFromKey(key)
if fileName == "" { if fileName == "" {
return true return true
@ -247,6 +246,7 @@ func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir
entry := &filer.Entry{ entry := &filer.Entry{
FullPath: weed_util.NewFullPath(string(dirPath), fileName), FullPath: weed_util.NewFullPath(string(dirPath), fileName),
} }
lastFileName = fileName
// println("list", entry.FullPath, "chunks", len(entry.Chunks)) // println("list", entry.FullPath, "chunks", len(entry.Chunks))
if decodeErr := entry.DecodeAttributesAndChunks(value); decodeErr != nil { if decodeErr := entry.DecodeAttributesAndChunks(value); decodeErr != nil {
@ -254,14 +254,16 @@ func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dir
glog.V(0).Infof("list %s : %v", entry.FullPath, err) glog.V(0).Infof("list %s : %v", entry.FullPath, err)
return false return false
} }
entries = append(entries, entry) if !eachEntryFunc(entry) {
return false
}
return true return true
}) })
if err != nil { if err != nil {
return entries, false, fmt.Errorf("prefix list %s : %v", dirPath, err) return lastFileName, fmt.Errorf("prefix list %s : %v", dirPath, err)
} }
return entries, false, err return lastFileName, err
} }
func genKey(dirPath, fileName string) (key []byte) { func genKey(dirPath, fileName string) (key []byte) {

View file

@ -319,14 +319,14 @@ func (dir *Dir) ReadDirAll(ctx context.Context) (ret []fuse.Dirent, err error) {
glog.Errorf("dir ReadDirAll %s: %v", dirPath, err) glog.Errorf("dir ReadDirAll %s: %v", dirPath, err)
return nil, fuse.EIO return nil, fuse.EIO
} }
listedEntries, listErr := dir.wfs.metaCache.ListDirectoryEntries(context.Background(), util.FullPath(dir.FullPath()), "", false, int64(math.MaxInt32)) listErr := dir.wfs.metaCache.ListDirectoryEntries(context.Background(), util.FullPath(dir.FullPath()), "", false, int64(math.MaxInt32), func(entry *filer.Entry) bool {
processEachEntryFn(entry.ToProtoEntry(), false)
return true
})
if listErr != nil { if listErr != nil {
glog.Errorf("list meta cache: %v", listErr) glog.Errorf("list meta cache: %v", listErr)
return nil, fuse.EIO return nil, fuse.EIO
} }
for _, cachedEntry := range listedEntries {
processEachEntryFn(cachedEntry.ToProtoEntry(), false)
}
return return
} }

View file

@ -117,22 +117,22 @@ func (mc *MetaCache) DeleteEntry(ctx context.Context, fp util.FullPath) (err err
return mc.localStore.DeleteEntry(ctx, fp) return mc.localStore.DeleteEntry(ctx, fp)
} }
func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64) ([]*filer.Entry, error) { func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) error {
mc.RLock() mc.RLock()
defer mc.RUnlock() defer mc.RUnlock()
if !mc.visitedBoundary.HasVisited(dirPath) { if !mc.visitedBoundary.HasVisited(dirPath) {
return nil, fmt.Errorf("unsynchronized dir: %v", dirPath) return fmt.Errorf("unsynchronized dir: %v", dirPath)
} }
entries, _, err := mc.localStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit) _, err := mc.localStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *filer.Entry) bool {
if err != nil {
return nil, err
}
for _, entry := range entries {
mc.mapIdFromFilerToLocal(entry) mc.mapIdFromFilerToLocal(entry)
return eachEntryFunc(entry)
})
if err != nil {
return err
} }
return entries, err return err
} }
func (mc *MetaCache) Shutdown() { func (mc *MetaCache) Shutdown() {

View file

@ -44,7 +44,7 @@ func (fs *FilerServer) LookupDirectoryEntry(ctx context.Context, req *filer_pb.L
}, nil }, nil
} }
func (fs *FilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error { func (fs *FilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) (err error) {
glog.V(4).Infof("ListEntries %v", req) glog.V(4).Infof("ListEntries %v", req)
@ -60,23 +60,12 @@ func (fs *FilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream file
lastFileName := req.StartFromFileName lastFileName := req.StartFromFileName
includeLastFile := req.InclusiveStartFrom includeLastFile := req.InclusiveStartFrom
var listErr error
for limit > 0 { for limit > 0 {
entries, hasMore, err := fs.filer.ListDirectoryEntries(stream.Context(), util.FullPath(req.Directory), lastFileName, includeLastFile, int64(paginationLimit), req.Prefix, "") var hasEntries bool
lastFileName, listErr = fs.filer.StreamListDirectoryEntries(stream.Context(), util.FullPath(req.Directory), lastFileName, includeLastFile, int64(paginationLimit), req.Prefix, "", func(entry *filer.Entry) bool {
if err != nil { hasEntries = true
return err if err = stream.Send(&filer_pb.ListEntriesResponse{
}
if len(entries) == 0 {
return nil
}
includeLastFile = false
for _, entry := range entries {
lastFileName = entry.Name()
if err := stream.Send(&filer_pb.ListEntriesResponse{
Entry: &filer_pb.Entry{ Entry: &filer_pb.Entry{
Name: entry.Name(), Name: entry.Name(),
IsDirectory: entry.IsDirectory(), IsDirectory: entry.IsDirectory(),
@ -88,18 +77,27 @@ func (fs *FilerServer) ListEntries(req *filer_pb.ListEntriesRequest, stream file
Content: entry.Content, Content: entry.Content,
}, },
}); err != nil { }); err != nil {
return err return false
} }
limit-- limit--
if limit == 0 { if limit == 0 {
return nil return false
} }
return true
})
if listErr != nil {
return listErr
}
if err != nil {
return err
}
if !hasEntries {
return nil
} }
if !hasMore { includeLastFile = false
break
}
} }