feat(bigtable): Add support for reverse scans

bigtable/bigtable.go

@@ -207,7 +207,14 @@ func (t *Table) ReadRows(ctx context.Context, arg RowSet, f func(Row) bool, opts
 		if err != nil {
 			return err
 		}
-		cr := newChunkReader()
+
+		var cr *chunkReader
+		if req.Reversed {
+			cr = newReverseChunkReader()
+		} else {
+			cr = newChunkReader()
+		}
+
 		for {
 			res, err := stream.Recv()
 			if err == io.EOF {
@@ -306,6 +313,10 @@ type RowSet interface {
 	// given row key or any row key lexicographically less than it.
 	retainRowsAfter(lastRowKey string) RowSet
 
+	// retainRowsBefore returns a new RowSet that does not include the
+	// given row key or any row key lexicographically greater than it.
+	retainRowsBefore(lastRowKey string) RowSet
+
 	// Valid reports whether this set can cover at least one row.
 	valid() bool
 }
@@ -331,17 +342,39 @@ func (r RowList) retainRowsAfter(lastRowKey string) RowSet {
 	return retryKeys
 }
 
+func (r RowList) retainRowsBefore(lastRowKey string) RowSet {
+	var retryKeys RowList
+	for _, key := range r {
+		if key < lastRowKey {
+			retryKeys = append(retryKeys, key)
+		}
+	}
+	return retryKeys
+}
+
 func (r RowList) valid() bool {
 	return len(r) > 0
 }
 
+type BoundType int64
+
+const (
+	undefined BoundType = iota
+	unbounded
+	closed
+	open
+)
+
 // A RowRange is a half-open interval [Start, Limit) encompassing
 // all the rows with keys at least as large as Start, and less than Limit.
 // (Bigtable string comparison is the same as Go's.)
 // A RowRange can be unbounded, encompassing all keys at least as large as Start.
 type RowRange struct {
-	start string
-	limit string
+	start      string
+	startBound BoundType
+	// todo rename this field. it's module private so we won't break anybody
+	limit    string
+	endBound BoundType
 }
 
 // NewRange returns the new RowRange [begin, end).
@@ -352,9 +385,44 @@ func NewRange(begin, end string) RowRange {
 	}
 }
 
+func NewClosedOpenRange(begin, limit string) RowRange {
+	return RowRange{
+		start:      begin,
+		limit:      limit,
+		startBound: closed,
+		endBound:   open,
+	}
+}
+func NewOpenClosedRange(start, limit string) RowRange {
+	return RowRange{
+		start:      start,
+		limit:      limit,
+		startBound: open,
+		endBound:   closed,
+	}
+}
+
+func NewOpenRange(start, limit string) RowRange {
+	return RowRange{
+		start:      start,
+		limit:      limit,
+		startBound: open,
+		endBound:   open,
+	}
+}
+
+func NewClosedRange(begin, end string) RowRange {
+	return RowRange{
+		start:      begin,
+		limit:      end,
+		startBound: closed,
+		endBound:   closed,
+	}
+}
+
 // Unbounded tests whether a RowRange is unbounded.
 func (r RowRange) Unbounded() bool {
-	return r.limit == ""
+	return r.limit == "" || r.endBound == unbounded
 }
 
 // Contains says whether the RowRange contains the key.
@@ -372,13 +440,27 @@ func (r RowRange) String() string {
 }
 
 func (r RowRange) proto() *btpb.RowSet {
-	rr := &btpb.RowRange{
-		StartKey: &btpb.RowRange_StartKeyClosed{StartKeyClosed: []byte(r.start)},
+	var rr = btpb.RowRange{}
+
+	// empty strings are NOT considered unbounded for backwards compatibility
+	if r.startBound == unbounded {
+		// leave unbounded
+	} else if r.startBound == closed || r.startBound == undefined {
+		rr.StartKey = &btpb.RowRange_StartKeyClosed{StartKeyClosed: []byte(r.start)}
+	} else {
+		rr.StartKey = &btpb.RowRange_StartKeyOpen{StartKeyOpen: []byte(r.start)}
 	}
-	if !r.Unbounded() {
+
+	// empty strings are considered unbounded for backwards compatibility
+	if r.endBound == unbounded || r.limit == "" {
+		// leave unbounded
+	} else if r.endBound == closed {
+		rr.EndKey = &btpb.RowRange_EndKeyClosed{EndKeyClosed: []byte(r.limit)}
+	} else {
 		rr.EndKey = &btpb.RowRange_EndKeyOpen{EndKeyOpen: []byte(r.limit)}
 	}
-	return &btpb.RowSet{RowRanges: []*btpb.RowRange{rr}}
+
+	return &btpb.RowSet{RowRanges: []*btpb.RowRange{&rr}}
 }
 
 func (r RowRange) retainRowsAfter(lastRowKey string) RowSet {
@@ -393,6 +475,14 @@ func (r RowRange) retainRowsAfter(lastRowKey string) RowSet {
 	return NewRange(start, r.limit)
 }
 
+func (r RowRange) retainRowsBefore(lastRowKey string) RowSet {
+	if lastRowKey == "" || (r.limit != "" && r.limit <= lastRowKey) {
+		return r
+	}
+
+	return NewRange(r.start, lastRowKey)
+}
+
 func (r RowRange) valid() bool {
 	return r.Unbounded() || r.start < r.limit
 }
@@ -424,6 +514,21 @@ func (r RowRangeList) retainRowsAfter(lastRowKey string) RowSet {
 	return ranges
 }
 
+func (r RowRangeList) retainRowsBefore(lastRowKey string) RowSet {
+	if lastRowKey == "" {
+		return r
+	}
+	// Return a list of any range that has not yet been completely processed
+	var ranges RowRangeList
+	for _, rr := range r {
+		retained := rr.retainRowsBefore(lastRowKey)
+		if retained.valid() {
+			ranges = append(ranges, retained.(RowRange))
+		}
+	}
+	return ranges
+}
+
 func (r RowRangeList) valid() bool {
 	for _, rr := range r {
 		if rr.valid() {
@@ -450,8 +555,18 @@ func PrefixRange(prefix string) RowRange {
 // large as start.
 func InfiniteRange(start string) RowRange {
 	return RowRange{
-		start: start,
-		limit: "",
+		start:    start,
+		endBound: unbounded,
+	}
+}
+
+// InfiniteReverseRange returns the RowRange consisting of all keys at least as
+// large as start.
+func InfiniteReverseRange(limit string) RowRange {
+	return RowRange{
+		limit:      limit,
+		startBound: unbounded,
+		endBound:   closed,
 	}
 }
 
@@ -577,6 +692,25 @@ func (wrs withFullReadStats) set(settings *readSettings) {
 	settings.fullReadStatsFunc = wrs.f
 }
 
+// ReverseScan returns a RadOption that will reverse the results of a Scan.
+// The rows will be streamed in reverse lexiographic order of the keys. The row key ranges of the RowSet are
+// still expected to be oriented the same way as forwards. ie [a,c] where a <= c. The row content
+// will remain unchanged from the ordering forward scans. This is particularly useful to get the
+// last N records before a key:
+//
+//	table.ReadRows(ctx, NewOpenClosedRange("", "key"), func(row bigtable.Row) bool {
+//	   return true
+//	}, bigtable.ReverseScan(), bigtable.LimitRows(10))
+func ReverseScan() ReadOption {
+	return reverseScan{}
+}
+
+type reverseScan struct{}
+
+func (rs reverseScan) set(settings *readSettings) {
+	settings.req.Reversed = true
+}
+
 // mutationsAreRetryable returns true if all mutations are idempotent
 // and therefore retryable. A mutation is idempotent iff all cell timestamps
 // have an explicit timestamp set and do not rely on the timestamp being set on the server.