fix: return error if the first fillBatch fails

Signed-off-by: ekexium <[email protected]>

Author: ekexium

internal/unionstore/arena/arena.go

@@ -36,13 +36,11 @@ package arena
 
 import (
 	"encoding/binary"
-	"math"
-
 	"github.com/tikv/client-go/v2/internal/logutil"
-	"go.uber.org/zap"
-
 	"github.com/tikv/client-go/v2/kv"
 	"go.uber.org/atomic"
+	"go.uber.org/zap"
+	"math"
 )
 
 const (

internal/unionstore/memdb_art.go

@@ -17,7 +17,6 @@ package unionstore
 import (
 	"context"
 	"fmt"
-	"go.uber.org/zap"
 	"sync"
 
 	"github.com/tikv/client-go/v2/internal/logutil"
@@ -204,6 +203,7 @@ type snapshotBatchedIter struct {
 	lower                 []byte
 	upper                 []byte
 	reverse               bool
+	err                   error
 
 	// current batch
 	keys      [][]byte
@@ -230,10 +230,7 @@ func (db *artDBWithContext) BatchedSnapshotIter(lower, upper []byte, reverse boo
 	}
 
 	iter.snapshot = db.GetSnapshot()
-	err := iter.fillBatch()
-	if err != nil {
-		logutil.BgLogger().Error("failed to fill batch for snapshotBatchedIter", zap.Error(err))
-	}
+	iter.err = iter.fillBatch()
 	return iter
 }
 
@@ -325,10 +322,14 @@ func (it *snapshotBatchedIter) fillBatch() error {
 
 func (it *snapshotBatchedIter) Valid() bool {
 	return it.snapshotTruncateSeqNo == it.db.SnapshotSeqNo &&
-		it.pos < len(it.keys)
+		it.pos < len(it.keys) &&
+		it.err == nil
 }
 
 func (it *snapshotBatchedIter) Next() error {
+	if it.err != nil {
+		return it.err
+	}
 	if it.snapshotTruncateSeqNo != it.db.SnapshotSeqNo {
 		return errors.New(
 			fmt.Sprintf(

internal/unionstore/union_store.go

@@ -162,11 +162,6 @@ func (us *KVUnionStore) SetEntrySizeLimit(entryLimit, bufferLimit uint64) {
 	us.memBuffer.SetEntrySizeLimit(entryLimit, bufferLimit)
 }
 
-type KvPair struct {
-	Key   []byte
-	Value []byte
-}
-
 // MemBuffer is an interface that stores mutations that written during transaction execution.
 // It now unifies MemDB and PipelinedMemDB.
 // The implementations should follow the transaction guarantees:

tikv/unionstore_export.go

@@ -60,5 +60,3 @@ type MemDBCheckpoint = unionstore.MemDBCheckpoint
 
 // Metrics is the metrics of unionstore.
 type Metrics = unionstore.Metrics
-
-type KvPair = unionstore.KvPair