Change DeleteFolderChildren to DeleteRange api

2 files changed, 24 insertions, 7 deletions

diff --git a/weed/filer/configuration.go b/weed/filer/configuration.go
index da75b575b..9ef2f3e0f 100644
--- a/weed/filer/configuration.go
+++ b/weed/filer/configuration.go
@@ -82,7 +82,6 @@ func (f *Filer) LoadConfiguration(config *util.ViperProxy) {
 func validateOneEnabledStore(config *util.ViperProxy) {
 	enabledStore := ""
 	for _, store := range Stores {
-		glog.V(0).Infof("Store Engines: %v", store.GetName())
 		if config.GetBool(store.GetName() + ".enabled") {
 			if enabledStore == "" {
 				enabledStore = store.GetName()
diff --git a/weed/filer/tikv/tikv_store.go b/weed/filer/tikv/tikv_store.go
index 9bee57283..e14cf5190 100644
--- a/weed/filer/tikv/tikv_store.go
+++ b/weed/filer/tikv/tikv_store.go
@@ -21,7 +21,8 @@ func init() {
 }
 
 type TikvStore struct {
-	client *tikv.KVStore
+	client                 *tikv.KVStore
+	deleteRangeConcurrency int
 }
 
 // Basic APIs
@@ -35,6 +36,7 @@ func (store *TikvStore) Initialize(config util.Configuration, prefix string) err
 	for _, item := range strings.Split(pdAddrsStr, ",") {
 		pdAddrs = append(pdAddrs, strings.TrimSpace(item))
 	}
+	store.deleteRangeConcurrency = 1
 	return store.initialize(pdAddrs)
 }
 
@@ -142,7 +144,10 @@ func (store *TikvStore) DeleteFolderChildren(ctx context.Context, path util.Full
 	if err != nil {
 		return err
 	}
-
+	var (
+		startKey []byte = nil
+		endKey   []byte = nil
+	)
 	err = txn.RunInTxn(func(txn *txnkv.KVTxn) error {
 		iter, err := txn.Iter(directoryPrefix, nil)
 		if err != nil {
@@ -151,24 +156,37 @@ func (store *TikvStore) DeleteFolderChildren(ctx context.Context, path util.Full
 		defer iter.Close()
 		for iter.Valid() {
 			key := iter.Key()
+			endKey = key
 			if !bytes.HasPrefix(key, directoryPrefix) {
 				break
 			}
-			err = txn.Delete(key)
-			if err != nil {
-				return err
+			if startKey == nil {
+				startKey = key
 			}
+
 			err = iter.Next()
 			if err != nil {
 				return err
 			}
 		}
+		// Only one Key matched just delete it.
+		if startKey != nil && bytes.Equal(startKey, endKey) {
+			return txn.Delete(startKey)
+		}
 		return nil
 	})
 	if err != nil {
 		return fmt.Errorf("delete %s : %v", path, err)
 	}
-	return nil
+
+	if startKey != nil && endKey != nil && !bytes.Equal(startKey, endKey) {
+		// has startKey and endKey and they are not equals, so use delete range
+		_, err = store.client.DeleteRange(context.Background(), startKey, endKey, store.deleteRangeConcurrency)
+		if err != nil {
+			return fmt.Errorf("delete %s : %v", path, err)
+		}
+	}
+	return err
 }
 
 func (store *TikvStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (string, error) {