extract functions out of functors

Author: zhangchiqing

storage/operation/badgerimpl/writer.go

@@ -106,7 +106,7 @@ func (b *ReaderBatchWriter) Delete(key []byte) error {
 // It returns error if endPrefix < startPrefix
 // no other errors are expected during normal operation
 func (b *ReaderBatchWriter) DeleteByRange(globalReader storage.Reader, startPrefix, endPrefix []byte) error {
-	err := operation.IterateKeysInPrefixRange(startPrefix, endPrefix, func(key []byte) error {
+	err := operation.Iterate(startPrefix, endPrefix, func(key []byte) error {
 		err := b.batch.Delete(key)
 		if err != nil {
 			return fmt.Errorf("could not add key to delete batch (%v): %w", key, err)

storage/operation/pebbleimpl/writer.go

@@ -117,11 +117,11 @@ func (b *ReaderBatchWriter) DeleteByRange(globalReader storage.Reader, startPref
 		return b.batch.DeleteRange(start, end, pebble.Sync)
 	}
 
-	return operation.IterateKeysInPrefixRange(startPrefix, endPrefix, func(key []byte) error {
+	return operation.IterateKeysByPrefixRange(globalReader, startPrefix, endPrefix, func(key []byte) error {
 		err := b.batch.Delete(key, pebble.Sync)
 		if err != nil {
 			return fmt.Errorf("could not add key to delete batch (%v): %w", key, err)
 		}
 		return nil
-	})(globalReader)
+	})
 }

storage/operation/reads.go

@@ -14,7 +14,10 @@ import (
 // CheckFunc is a function that checks if the value should be read and decoded.
 // return (true, nil) to read the value and pass it to the CreateFunc and HandleFunc for decoding
 // return (false, nil) to skip reading the value
-// return (false, err) if running into any exception, the iteration should be stopped.
+// return (false, err) if running into any error, the iteration should be stopped.
+// when making a CheckFunc to be used in the IterationFunc to iterate over the keys, a sentinel error
+// can be defined and checked to stop the iteration early, such as finding the first key that match
+// certain condition.
 type CheckFunc func(key []byte) (bool, error)
 
 // CreateFunc returns a pointer to an initialized entity that we can potentially
@@ -28,12 +31,13 @@ type CreateFunc func() interface{}
 type HandleFunc func() error
 type IterationFunc func() (CheckFunc, CreateFunc, HandleFunc)
 
-// IterateKeysInPrefixRange will iterate over all entries in the database, where the key starts with a prefixes in
+// IterateKey will iterate over all entries in the database, where the key starts with a prefixes in
 // the range [startPrefix, endPrefix] (both inclusive). We require that startPrefix <= endPrefix (otherwise this
 // function errors). On every such key, the `check` function is called. If `check` errors, iteration is aborted.
+// In other words, error returned by the iteration functions will be propagated to the caller.
 // No errors expected during normal operations.
-func IterateKeysInPrefixRange(startPrefix []byte, endPrefix []byte, check func(key []byte) error) func(storage.Reader) error {
-	return Iterate(startPrefix, endPrefix, func() (CheckFunc, CreateFunc, HandleFunc) {
+func IterateKeysByPrefixRange(r storage.Reader, startPrefix []byte, endPrefix []byte, check func(key []byte) error) error {
+	return IterateKeys(r, startPrefix, endPrefix, func() (CheckFunc, CreateFunc, HandleFunc) {
 		return func(key []byte) (bool, error) {
 			err := check(key)
 			if err != nil {
@@ -44,100 +48,88 @@ func IterateKeysInPrefixRange(startPrefix []byte, endPrefix []byte, check func(k
 	}, storage.IteratorOption{IterateKeyOnly: true})
 }
 
-// Iterate will iterate over all keys with prefixes in the given range [startPrefix, endPrefix] (both inclusive)
-func Iterate(startPrefix []byte, endPrefix []byte, iterFunc IterationFunc, opt storage.IteratorOption) func(storage.Reader) error {
-	return func(r storage.Reader) error {
+// IterateKey will iterate over all entries in the database, where the key starts with a prefixes in
+// the range [startPrefix, endPrefix] (both inclusive).
+// No errors expected during normal operations.
+func IterateKeys(r storage.Reader, startPrefix []byte, endPrefix []byte, iterFunc IterationFunc, opt storage.IteratorOption) error {
+	if len(startPrefix) == 0 {
+		return fmt.Errorf("startPrefix prefix is empty")
+	}
 
-		if len(startPrefix) == 0 {
-			return fmt.Errorf("startPrefix prefix is empty")
-		}
+	if len(endPrefix) == 0 {
+		return fmt.Errorf("endPrefix prefix is empty")
+	}
 
-		if len(endPrefix) == 0 {
-			return fmt.Errorf("endPrefix prefix is empty")
-		}
+	// Reverse iteration is not supported by pebble
+	if bytes.Compare(startPrefix, endPrefix) > 0 {
+		return fmt.Errorf("startPrefix key must be less than or equal to endPrefix key")
+	}
 
-		// Reverse iteration is not supported by pebble
-		if bytes.Compare(startPrefix, endPrefix) > 0 {
-			return fmt.Errorf("startPrefix key must be less than or equal to endPrefix key")
-		}
+	it, err := r.NewIter(startPrefix, endPrefix, opt)
+	if err != nil {
+		return fmt.Errorf("can not create iterator: %w", err)
+	}
+	defer it.Close()
 
-		it, err := r.NewIter(startPrefix, endPrefix, opt)
-		if err != nil {
-			return fmt.Errorf("can not create iterator: %w", err)
-		}
-		defer it.Close()
+	for it.First(); it.Valid(); it.Next() {
+		item := it.IterItem()
+		key := item.Key()
+
+		// initialize processing functions for iteration
+		check, create, handle := iterFunc()
 
-		for it.First(); it.Valid(); it.Next() {
-			item := it.IterItem()
-			key := item.Key()
+		keyCopy := make([]byte, len(key))
 
-			// initialize processing functions for iteration
-			check, create, handle := iterFunc()
+		// The underlying database may re-use and modify the backing memory of the returned key.
+		// Tor safety we proactively make a copy before passing the key to the upper layer.
+		copy(keyCopy, key)
 
-			keyCopy := make([]byte, len(key))
+		// check if we should process the item at all
+		shouldReadValue, err := check(keyCopy)
+		if err != nil {
+			return err
+		}
+		if !shouldReadValue { // skip reading value
+			continue
+		}
 
-			// The underlying database may re-use and modify the backing memory of the returned key.
-			// Tor safety we proactively make a copy before passing the key to the upper layer.
-			copy(keyCopy, key)
+		err = item.Value(func(val []byte) error {
 
-			// check if we should process the item at all
-			shouldReadValue, err := check(keyCopy)
+			// decode into the entity
+			entity := create()
+			err = msgpack.Unmarshal(val, entity)
 			if err != nil {
-				return err
+				return irrecoverable.NewExceptionf("could not decode entity: %w", err)
 			}
-			if !shouldReadValue { // skip reading value
-				continue
-			}
-
-			err = item.Value(func(val []byte) error {
-
-				// decode into the entity
-				entity := create()
-				err = msgpack.Unmarshal(val, entity)
-				if err != nil {
-					return irrecoverable.NewExceptionf("could not decode entity: %w", err)
-				}
-
-				// process the entity
-				err = handle()
-				if err != nil {
-					return fmt.Errorf("could not handle entity: %w", err)
-				}
-
-				return nil
-			})
 
+			// process the entity
+			err = handle()
 			if err != nil {
-				return fmt.Errorf("could not process value: %w", err)
+				return fmt.Errorf("could not handle entity: %w", err)
 			}
-		}
 
-		return nil
+			return nil
+		})
+
+		if err != nil {
+			return fmt.Errorf("could not process value: %w", err)
+		}
 	}
+
+	return nil
 }
 
 // Traverse will iterate over all keys with the given prefix
-func Traverse(prefix []byte, iterFunc IterationFunc, opt storage.IteratorOption) func(storage.Reader) error {
-	return Iterate(prefix, prefix, iterFunc, opt)
+// error returned by the iteration functions will be propagated to the caller.
+// No other errors are expected during normal operation.
+func TraverseByPrefix(r storage.Reader, prefix []byte, iterFunc IterationFunc, opt storage.IteratorOption) error {
+	return IterateKeys(r, prefix, prefix, iterFunc, opt)
 }
 
-// Exists takes a key and a pointer to an a boolean variable `keyExists` as inputs and returns an function.
+// KeyExists returns true if a key exists in the database.
 // When this returned function is executed (and only then), it will write into the `keyExists` whether
 // the key exists.
 // No errors are expected during normal operation.
-func Exists(key []byte, keyExists *bool) func(storage.Reader) error {
-	return func(r storage.Reader) error {
-		exists, err := KeyExists(r, key)
-		if err != nil {
-			return err
-		}
-		*keyExists = exists
-		return nil
-	}
-}
-
-// KeyExists returns true if a key exists in the database.
-// No errors are expected during normal operation.
 func KeyExists(r storage.Reader, key []byte) (bool, error) {
 	_, closer, err := r.Get(key)
 	if err != nil {
@@ -154,13 +146,6 @@ func KeyExists(r storage.Reader, key []byte) (bool, error) {
 	return true, nil
 }
 
-// Retrieve returns a functor that retrieves the binary data under the given key from the database
-func Retrieve(key []byte, entity interface{}) func(storage.Reader) error {
-	return func(r storage.Reader) error {
-		return RetrieveByKey(r, key, entity)
-	}
-}
-
 // RetrieveByKey will retrieve the binary data under the given key from the database
 // and decode it into the given entity. The provided entity needs to be a
 // pointer to an initialized entity of the correct type.
@@ -183,49 +168,47 @@ func RetrieveByKey(r storage.Reader, key []byte, entity interface{}) error {
 	return nil
 }
 
-// FindHighestAtOrBelow is for database entries that are indexed by block height. It is suitable to search
+// FindHighestAtOrBelowByPrefix is for database entries that are indexed by block height. It is suitable to search
 // keys with the format prefix` + `height` (where "+" denotes concatenation of binary strings). The height
 // is encoded as Big-Endian (entries with numerically smaller height have lexicographically smaller key).
 // The function finds the *highest* key with the given prefix and height equal to or below the given height.
-func FindHighestAtOrBelow(prefix []byte, height uint64, entity interface{}) func(storage.Reader) error {
-	return func(r storage.Reader) error {
-		if len(prefix) == 0 {
-			return fmt.Errorf("prefix must not be empty")
-		}
-
-		key := append(prefix, EncodeKeyPart(height)...)
-		it, err := r.NewIter(prefix, key, storage.DefaultIteratorOptions())
-		if err != nil {
-			return fmt.Errorf("can not create iterator: %w", err)
-		}
-		defer it.Close()
+func FindHighestAtOrBelowByPrefix(r storage.Reader, prefix []byte, height uint64, entity interface{}) error {
+	if len(prefix) == 0 {
+		return fmt.Errorf("prefix must not be empty")
+	}
 
-		var highestKey []byte
+	key := append(prefix, EncodeKeyPart(height)...)
+	it, err := r.NewIter(prefix, key, storage.DefaultIteratorOptions())
+	if err != nil {
+		return fmt.Errorf("can not create iterator: %w", err)
+	}
+	defer it.Close()
 
-		// find highest value below the given height
-		for it.First(); it.Valid(); it.Next() {
-			// copy the key to avoid the underlying slices of the key
-			// being modified by the Next() call
-			highestKey = it.IterItem().KeyCopy(highestKey)
-		}
+	var highestKey []byte
 
-		if len(highestKey) == 0 {
-			return storage.ErrNotFound
-		}
+	// find highest value below the given height
+	for it.First(); it.Valid(); it.Next() {
+		// copy the key to avoid the underlying slices of the key
+		// being modified by the Next() call
+		highestKey = it.IterItem().KeyCopy(highestKey)
+	}
 
-		// read the value of the highest key
-		val, closer, err := r.Get(highestKey)
-		if err != nil {
-			return err
-		}
+	if len(highestKey) == 0 {
+		return storage.ErrNotFound
+	}
 
-		defer closer.Close()
+	// read the value of the highest key
+	val, closer, err := r.Get(highestKey)
+	if err != nil {
+		return err
+	}
 
-		err = msgpack.Unmarshal(val, entity)
-		if err != nil {
-			return irrecoverable.NewExceptionf("failed to decode value: %w", err)
-		}
+	defer closer.Close()
 
-		return nil
+	err = msgpack.Unmarshal(val, entity)
+	if err != nil {
+		return irrecoverable.NewExceptionf("failed to decode value: %w", err)
 	}
+
+	return nil
 }

storage/operation/reads_functors.go

@@ -0,0 +1,44 @@
+package operation
+
+import "github.com/onflow/flow-go/storage"
+
+// Leo: This package includes deprecated functions that wraps the operation of reading from the database.
+// They are needed because the original badger implementation is also implemented in the same wrapped function manner,
+// since badger requires reads to be done in a transaction, which is stateful.
+// Using these deprecated functions could minimize the changes during refactor and easier to review the changes.
+// The simplified implementation of the functions are in the reads.go file, which are encouraged to be used instead.
+
+func Iterate(startPrefix []byte, endPrefix []byte, check func(key []byte) error) func(storage.Reader) error {
+	return func(r storage.Reader) error {
+		return IterateKeysByPrefixRange(r, startPrefix, endPrefix, check)
+	}
+}
+
+func Traverse(prefix []byte, iterFunc IterationFunc, opt storage.IteratorOption) func(storage.Reader) error {
+	return func(r storage.Reader) error {
+		return TraverseByPrefix(r, prefix, iterFunc, opt)
+	}
+}
+
+func Retrieve(key []byte, entity interface{}) func(storage.Reader) error {
+	return func(r storage.Reader) error {
+		return RetrieveByKey(r, key, entity)
+	}
+}
+
+func Exists(key []byte, keyExists *bool) func(storage.Reader) error {
+	return func(r storage.Reader) error {
+		exists, err := KeyExists(r, key)
+		if err != nil {
+			return err
+		}
+		*keyExists = exists
+		return nil
+	}
+}
+
+func FindHighestAtOrBelow(prefix []byte, height uint64, entity interface{}) func(storage.Reader) error {
+	return func(r storage.Reader) error {
+		return FindHighestAtOrBelowByPrefix(r, prefix, height, entity)
+	}
+}

storage/operation/reads_test.go

@@ -53,7 +53,7 @@ func TestIterateKeysInPrefixRange(t *testing.T) {
 
 		// Forward iteration and check boundaries
 		var found [][]byte
-		require.NoError(t, operation.IterateKeysInPrefixRange(prefixStart, prefixEnd, func(key []byte) error {
+		require.NoError(t, operation.Iterate(prefixStart, prefixEnd, func(key []byte) error {
 			found = append(found, key)
 			return nil
 		})(r), "should iterate forward without error")
@@ -66,7 +66,7 @@ func TestIterateInvalidRange(t *testing.T) {
 	dbtest.RunWithStorages(t, func(t *testing.T, r storage.Reader, withWriter dbtest.WithWriter) {
 
 		var found [][]byte
-		require.Error(t, operation.IterateKeysInPrefixRange([]byte{0x02}, []byte{0x01}, func(key []byte) error {
+		require.Error(t, operation.Iterate([]byte{0x02}, []byte{0x01}, func(key []byte) error {
 			found = append(found, key)
 			return nil
 		})(r))
@@ -118,7 +118,7 @@ func TestIterationBoundary(t *testing.T) {
 
 		// Forward iteration and check boundaries
 		var found [][]byte
-		require.NoError(t, operation.IterateKeysInPrefixRange(prefixStart, prefixEnd, func(key []byte) error {
+		require.NoError(t, operation.Iterate(prefixStart, prefixEnd, func(key []byte) error {
 			found = append(found, key)
 			return nil
 		})(r), "should iterate forward without error")