CBG-5225: sg-bucket KV range scan implementations

base/bucket.go

@@ -473,6 +473,12 @@ func AsSubdocStore(ds DataStore) (sgbucket.SubdocStore, bool) {
 	return subdocStore, ok && ds.IsSupported(sgbucket.BucketStoreFeatureSubdocOperations)
 }
 
+// AsRangeScanStore returns a RangeScanStore if the dataStore implements and supports range scan operations.
+func AsRangeScanStore(ds DataStore) (sgbucket.RangeScanStore, bool) {
+	rss, ok := ds.(sgbucket.RangeScanStore)
+	return rss, ok && ds.IsSupported(sgbucket.BucketStoreFeatureRangeScan)
+}
+
 // WaitUntilDataStoreReady will try to perform a basic operation in the given DataStore until it can succeed.
 // It's not necessarily the case that a datastore that exists is ready to be used.
 //

base/collection.go

@@ -257,6 +257,8 @@ func (b *GocbV2Bucket) IsSupported(feature sgbucket.BucketStoreFeature) bool {
 		return b.IsMinimumVersion(7, 6)
 	case sgbucket.BucketStoreFeatureMobileXDCR:
 		return b.supportsHLV
+	case sgbucket.BucketStoreFeatureRangeScan:
+		return b.IsMinimumVersion(7, 6)
 	default:
 		return false
 	}
@@ -717,6 +719,19 @@ func (b *GocbV2Bucket) DefaultDataStore(_ context.Context) sgbucket.DataStore {
 	}
 }
 
+// GetMatchingDataStore returns a DataStore on this bucket that matches the scope and collection
+// of the given DataStore. Useful when opening a second connection to the same bucket and needing
+// to operate on the same collection as the original.
+func (b *GocbV2Bucket) GetMatchingDataStore(ctx context.Context, other sgbucket.DataStoreName) (sgbucket.DataStore, error) {
+	if other.ScopeName() == DefaultScope && other.CollectionName() == DefaultCollection {
+		return b.DefaultDataStore(ctx), nil
+	}
+	return b.NamedDataStore(ctx, ScopeAndCollectionName{
+		Scope:      other.ScopeName(),
+		Collection: other.CollectionName(),
+	})
+}
+
 // NamedDataStore returns a collection on a bucket within the given scope and collection.
 func (b *GocbV2Bucket) NamedDataStore(_ context.Context, name sgbucket.DataStoreName) (sgbucket.DataStore, error) {
 	c, err := NewCollection(

base/collection_rangescan.go

@@ -0,0 +1,93 @@
+/*
+Copyright 2025-Present Couchbase, Inc.
+
+Use of this software is governed by the Business Source License included in
+the file licenses/BSL-Couchbase.txt.  As of the Change Date specified in that
+file, in accordance with the Business Source License, use of this software will
+be governed by the Apache License, Version 2.0, included in the file
+licenses/APL2.txt.
+*/
+
+package base
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/couchbase/gocb/v2"
+	sgbucket "github.com/couchbase/sg-bucket"
+)
+
+var _ sgbucket.RangeScanStore = &Collection{}
+
+func (c *Collection) Scan(_ context.Context, scanType sgbucket.ScanType, opts sgbucket.ScanOptions) (sgbucket.ScanResultIterator, error) {
+	c.Bucket.waitForAvailKvOp()
+	defer c.Bucket.releaseKvOp()
+
+	gocbScanType, err := toGocbScanType(scanType)
+	if err != nil {
+		return nil, err
+	}
+
+	scanOpts := &gocb.ScanOptions{
+		IDsOnly:    opts.IDsOnly,
+		Transcoder: gocb.NewRawBinaryTranscoder(),
+	}
+
+	result, err := c.Collection.Scan(gocbScanType, scanOpts)
+	if err != nil {
+		return nil, err
+	}
+
+	return &gocbScanResultIterator{result: result}, nil
+}
+
+func toGocbScanType(scanType sgbucket.ScanType) (gocb.ScanType, error) {
+	switch st := scanType.(type) {
+	case sgbucket.RangeScan:
+		rs := gocb.RangeScan{}
+		if st.From != nil {
+			rs.From = &gocb.ScanTerm{Term: st.From.Term, Exclusive: st.From.Exclusive}
+		}
+		if st.To != nil {
+			rs.To = &gocb.ScanTerm{Term: st.To.Term, Exclusive: st.To.Exclusive}
+		}
+		return rs, nil
+	default:
+		return nil, fmt.Errorf("unsupported scan type: %T", scanType)
+	}
+}
+
+type gocbScanResultIterator struct {
+	result *gocb.ScanResult
+	err    error
+}
+
+func (it *gocbScanResultIterator) Next(_ context.Context) *sgbucket.ScanResultItem {
+	if it.err != nil {
+		return nil
+	}
+	item := it.result.Next()
+	if item == nil {
+		return nil
+	}
+	result := &sgbucket.ScanResultItem{
+		ID:  item.ID(),
+		Cas: uint64(item.Cas()),
+	}
+	if !item.IDOnly() {
+		if err := item.Content(&result.Body); err != nil {
+			it.err = fmt.Errorf("failed to decode scan result body for %s: %w", item.ID(), err)
+			return nil
+		}
+	}
+	return result
+}
+
+func (it *gocbScanResultIterator) Close(_ context.Context) error {
+	closeErr := it.result.Close()
+	if it.err == nil {
+		it.err = closeErr
+	}
+	return it.err
+}