fix: Ensure Index.Walk fetches matching foreign keys only

This commit breaks other tests, so it is unclear if the
`Influx.Walk` function is intended to match by prefix.

If so, it would suggest code documentation clarifying
`Index.Walk` is a prefix match. Additionally, a new `Index.WalkExact`
function that performs an exact match by the `foreignKey` argument.

Closes #20096

kv/index.go

@@ -195,16 +195,18 @@ func (i *Index) Walk(ctx context.Context, tx Tx, foreignKey []byte, visitFn Visi
 		return err
 	}
 
-	return indexWalk(ctx, cursor, sourceBucket, visitFn)
+	return indexWalk(ctx, foreignKey, cursor, sourceBucket, visitFn)
 }
 
 // indexWalk consumes the indexKey and primaryKey pairs in the index bucket and looks up their
 // associated primaryKey's value in the provided source bucket.
 // When an item is located in the source, the provided visit function is called with primary key and associated value.
-func indexWalk(ctx context.Context, indexCursor ForwardCursor, sourceBucket Bucket, visit VisitFunc) (err error) {
+func indexWalk(ctx context.Context, foreignKey []byte, indexCursor ForwardCursor, sourceBucket Bucket, visit VisitFunc) (err error) {
 	var keys [][]byte
 	for ik, pk := indexCursor.Next(); ik != nil; ik, pk = indexCursor.Next() {
-		keys = append(keys, pk)
+		if string(foreignKey) == string(ik) {
+			keys = append(keys, pk)
+		}
 	}
 
 	if err := indexCursor.Err(); err != nil {

kv/index_test.go

@@ -7,9 +7,14 @@ import (
 	"os"
 	"testing"
 
+	"github.com/golang/mock/gomock"
 	"github.com/influxdata/influxdb/v2/bolt"
 	"github.com/influxdata/influxdb/v2/inmem"
+	"github.com/influxdata/influxdb/v2/kv"
+	"github.com/influxdata/influxdb/v2/kv/mock"
 	influxdbtesting "github.com/influxdata/influxdb/v2/testing"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.uber.org/zap/zaptest"
 )
 
@@ -33,6 +38,94 @@ func Test_Bolt_Index(t *testing.T) {
 	influxdbtesting.TestIndex(t, s)
 }
 
+func TestIndex_Walk(t *testing.T) {
+	t.Run("only selects exact keys", func(t *testing.T) {
+		ctrl := gomock.NewController(t)
+		t.Cleanup(ctrl.Finish)
+
+		// shorthand to cast string to []byte
+		b := func(s string) []byte { return []byte(s) }
+
+		var (
+			sourceBucket = b("source")
+			indexBucket  = b("index")
+			foreignKey   = b("jenkins")
+			idxkeyvals   = []struct{ key, val []byte }{
+				{b("jenkins-aws"), b("pk1")},
+				{b("jenkins-aws"), b("pk2")},
+				{b("jenkins-aws"), b("pk3")},
+				{b("jenkins"), b("pk4")},
+				{b("jenkins"), b("pk5")},
+			}
+			srckeyvals = []struct{ key, val []byte }{
+				{b("pk4"), b("val4")},
+				{b("pk5"), b("val5")},
+			}
+		)
+
+		mapping := kv.NewIndexMapping(sourceBucket, indexBucket, func(data []byte) ([]byte, error) {
+			return nil, nil
+		})
+
+		tx := mock.NewMockTx(ctrl)
+
+		src := mock.NewMockBucket(ctrl)
+		src.EXPECT().
+			GetBatch(srckeyvals[0].key, srckeyvals[1].key).
+			Return([][]byte{srckeyvals[0].val, srckeyvals[1].val}, nil)
+
+		tx.EXPECT().
+			Bucket(sourceBucket).
+			Return(src, nil)
+
+		idx := mock.NewMockBucket(ctrl)
+		tx.EXPECT().
+			Bucket(indexBucket).
+			Return(idx, nil)
+
+		cur := mock.NewMockForwardCursor(ctrl)
+
+		i := 0
+		cur.EXPECT().
+			Next().
+			DoAndReturn(func() ([]byte, []byte) {
+				var k, v []byte
+				if i < len(idxkeyvals) {
+					elem := idxkeyvals[i]
+					i++
+					k, v = elem.key, elem.val
+				}
+
+				return k, v
+			}).
+			Times(len(idxkeyvals) + 1)
+		cur.EXPECT().
+			Err().
+			Return(nil)
+		cur.EXPECT().
+			Close().
+			Return(nil)
+		idx.EXPECT().
+			ForwardCursor(foreignKey, gomock.Any()).
+			Return(cur, nil)
+
+		ctx := context.Background()
+		index := kv.NewIndex(mapping, kv.WithIndexReadPathEnabled)
+
+		j := 0
+		err := index.Walk(ctx, tx, foreignKey, func(k, v []byte) (bool, error) {
+			require.Less(t, j, len(srckeyvals))
+			assert.Equal(t, srckeyvals[j].key, k)
+			assert.Equal(t, srckeyvals[j].val, v)
+			j++
+			return true, nil
+		})
+
+		assert.NoError(t, err)
+	})
+
+}
+
 func Benchmark_Inmem_Index_Walk(b *testing.B) {
 	influxdbtesting.BenchmarkIndexWalk(b, inmem.NewKVStore(), 1000, 200)
 }