ledger: add kvIter to storage package (11 of N)

ledger/catchpointtracker.go

@@ -1431,15 +1431,13 @@ func (ct *catchpointTracker) initializeHashes(ctx context.Context, tx *sql.Tx, r
 
 		// Now add the kvstore hashes
 		pendingTrieHashes = 0
-		kvs, err := tx.QueryContext(ctx, "SELECT key, value FROM kvstore")
+		kvs, err := store.MakeKVsIter(ctx, tx)
 		if err != nil {
 			return err
 		}
 		defer kvs.Close()
 		for kvs.Next() {
-			var k []byte
-			var v []byte
-			err := kvs.Scan(&k, &v)
+			k, v, err := kvs.KeyValue()
 			if err != nil {
 				return err
 			}

ledger/catchpointwriter.go

@@ -61,7 +61,13 @@ type catchpointWriter struct {
 	accountsIterator     accountsBatchIter
 	maxResourcesPerChunk int
 	accountsDone         bool
-	kvRows               *sql.Rows
+	kvRows               kvIter
+}
+
+type kvIter interface {
+	Next() bool
+	KeyValue() ([]byte, []byte, error)
+	Close()
 }
 
 type accountsBatchIter interface {
@@ -280,7 +286,7 @@ func (cw *catchpointWriter) readDatabaseStep(ctx context.Context, tx *sql.Tx) er
 
 	// Create the *Rows iterator JIT
 	if cw.kvRows == nil {
-		rows, err := tx.QueryContext(ctx, "SELECT key, value FROM kvstore")
+		rows, err := store.MakeKVsIter(ctx, tx)
 		if err != nil {
 			return err
 		}
@@ -289,9 +295,7 @@ func (cw *catchpointWriter) readDatabaseStep(ctx context.Context, tx *sql.Tx) er
 
 	kvrs := make([]encoded.KVRecordV6, 0, BalancesPerCatchpointFileChunk)
 	for cw.kvRows.Next() {
-		var k []byte
-		var v []byte
-		err := cw.kvRows.Scan(&k, &v)
+		k, v, err := cw.kvRows.KeyValue()
 		if err != nil {
 			return err
 		}

ledger/store/kvsIter.go

@@ -0,0 +1,53 @@
+// Copyright (C) 2019-2022 Algorand, Inc.
+// This file is part of go-algorand
+//
+// go-algorand is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// go-algorand is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with go-algorand.  If not, see <https://www.gnu.org/licenses/>.
+
+package store
+
+import (
+	"context"
+	"database/sql"
+)
+
+type kvsIter struct {
+	tx   *sql.Tx
+	rows *sql.Rows
+}
+
+// MakeKVsIter creates a KV iterator.
+func MakeKVsIter(ctx context.Context, tx *sql.Tx) (*kvsIter, error) {
+	rows, err := tx.QueryContext(ctx, "SELECT key, value FROM kvstore")
+	if err != nil {
+		return nil, err
+	}
+
+	return &kvsIter{
+		tx:   tx,
+		rows: rows,
+	}, nil
+}
+
+func (iter *kvsIter) Next() bool {
+	return iter.rows.Next()
+}
+
+func (iter *kvsIter) KeyValue() (k []byte, v []byte, err error) {
+	err = iter.rows.Scan(&k, &v)
+	return k, v, err
+}
+
+func (iter *kvsIter) Close() {
+	iter.rows.Close()
+}