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,6 +342,16 @@ 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
 }
@@ -352,6 +373,38 @@ func NewRange(begin, end string) RowRange {
 	}
 }
 
+func NewClosedOpenRange(begin, limit string) RowRange {
+	return NewRange(begin, limit)
+}
+func NewOpenClosedRange(start, limit string) RowRange {
+	// contract the start to exclude the first key
+	if start != "" {
+		start = start + "\x00"
+	}
+	// expand the end to include the last key
+	if limit != "" {
+		limit = limit + "\x00"
+	}
+	return NewRange(start, limit)
+}
+
+func NewOpenRange(start, limit string) RowRange {
+	// contract the start to exclude the first key
+	if start != "" {
+		start = start + "\x00"
+	}
+	return NewRange(start, limit)
+}
+
+func NewClosedRange(begin, end string) RowRange {
+	// expand the end to include the last key
+	if end != "" {
+		end = end + "\x00"
+	}
+	return NewRange(begin, end)
+}
+
+
 // Unbounded tests whether a RowRange is unbounded.
 func (r RowRange) Unbounded() bool {
 	return r.limit == ""
@@ -393,6 +446,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 +485,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() {
@@ -577,6 +653,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.

bigtable/bigtable_test.go

@@ -51,6 +51,142 @@ func TestPrefix(t *testing.T) {
 	}
 }
 
+func TestNewClosedOpenRange(t *testing.T) {
+	start := "b"
+	limit := "b\x01"
+	r := NewClosedOpenRange(start, limit)
+	for _, test := range []struct {
+		k string
+		contains bool
+	}{
+		{"a", false},
+		{"b", true},
+		{"b\x00", true},
+		{"b\x01", false},
+	} {
+		if want, got := test.contains, r.Contains(test.k); want != got {
+			t.Errorf("NewClosedOpenRange(%q, %q).Contains(%q) = %t, want %t", start, limit, test.k, got, want)
+		}
+	}
+
+	for _, test := range []struct {
+		start, limit string
+		valid bool
+	} {
+		{"a", "a", false},
+		{"b", "a", false},
+		{"a", "a\x00", true},
+		{"a", "b", true},
+	} {
+		r := NewClosedOpenRange(test.start, test.limit)
+		if want, got := test.valid, r.valid(); want != got {
+			t.Errorf("NewClosedOpenRange(%q, %q).valid() = %t, want %t", test.start, test.limit, got, want)
+		}
+	}
+}
+func TestNewOpenClosedRange(t *testing.T) {
+	start := "b"
+	limit := "b\x01"
+	r := NewOpenClosedRange(start, limit)
+	for _, test := range []struct {
+		k string
+		contains bool
+	}{
+		{"a", false},
+		{"b", false},
+		{"b\x00", true},
+		{"b\x01", true},
+		{"b\x01\x00", false},
+	} {
+		if want, got := test.contains, r.Contains(test.k); want != got {
+			t.Errorf("NewOpenClosedRange(%q, %q).Contains(%q) = %t, want %t", start, limit, test.k,  got, want)
+		}
+	}
+
+	for _, test := range []struct {
+		start, limit string
+		valid bool
+	} {
+		{"a", "a", false},
+		{"b", "a", false},
+		{"a", "a\x00", true},
+		{"a", "b", true},
+	} {
+		r := NewOpenClosedRange(test.start, test.limit)
+		if want, got := test.valid, r.valid(); want != got {
+			t.Errorf("NewOpenClosedRange(%q, %q).valid() = %t, want %t", test.start, test.limit, got, want)
+		}
+	}
+}
+func TestNewClosedRange(t *testing.T) {
+	start := "b"
+	limit := "b"
+
+	r := NewClosedRange(start, limit)
+	for _, test := range []struct {
+		k string
+		contains bool
+	}{
+		{"a", false},
+		{"b", true},
+		{"b\x01", false},
+	} {
+		if want, got := test.contains, r.Contains(test.k); want != got {
+			t.Errorf("NewClosedRange(%q, %q).Contains(%q) = %t, want %t", "a", "a\x01", test.k,  got, test.contains)
+		}
+	}
+
+	for _, test := range []struct {
+		start, limit string
+		valid bool
+	} {
+		{"a", "b", true},
+		{"b", "b", true},
+		{"b", "b\x00", true},
+		{"b\x00", "b", false},
+	} {
+		r := NewClosedRange(test.start, test.limit)
+		if want, got := test.valid, r.valid(); want != got {
+			t.Errorf("NewClosedRange(%q, %q).valid() = %t, want %t", test.start, test.limit, got, want)
+		}
+	}
+}
+
+func TestNewOpenRange(t *testing.T) {
+	start := "b"
+	limit := "b\x01"
+
+	r := NewOpenRange(start, limit)
+	for _, test := range []struct {
+		k string
+		contains bool
+	}{
+		{"a", false},
+		{"b", false},
+		{"b\x00", true},
+		{"b\x01", false},
+	} {
+		if want, got := test.contains, r.Contains(test.k); want != got {
+			t.Errorf("NewOpenRange(%q, %q).Contains(%q) = %t, want %t", "a", "a\x01", test.k,  got, test.contains)
+		}
+	}
+
+	for _, test := range []struct {
+		start, limit string
+		valid bool
+	} {
+		{"a", "a", false},
+		{"a", "b", true},
+		{"a", "a\x00", false},
+		{"a", "a\x01", true},
+	} {
+		r := NewOpenRange(test.start, test.limit)
+		if want, got := test.valid, r.valid(); want != got {
+			t.Errorf("NewOpenRange(%q, %q).valid() = %t, want %t", test.start, test.limit, got, want)
+		}
+	}
+}
+
 func TestApplyErrors(t *testing.T) {
 	ctx := context.Background()
 	table := &Table{

bigtable/reader.go

@@ -19,6 +19,7 @@ package bigtable
 import (
 	"bytes"
 	"fmt"
+	"strings"
 
 	btpb "google.golang.org/genproto/googleapis/bigtable/v2"
 )
@@ -58,6 +59,7 @@ const (
 // chunkReader handles cell chunks from the read rows response and combines
 // them into full Rows.
 type chunkReader struct {
+	reversed  bool
 	state     rrState
 	curKey    []byte
 	curLabels []string
@@ -71,9 +73,14 @@ type chunkReader struct {
 
 // newChunkReader returns a new chunkReader for handling read rows responses.
 func newChunkReader() *chunkReader {
-	return &chunkReader{state: newRow}
+	return &chunkReader{reversed: false, state: newRow}
 }
 
+func newReverseChunkReader() *chunkReader {
+	return &chunkReader{reversed: true, state: newRow}
+}
+
+
 // Process takes a cell chunk and returns a new Row if the given chunk
 // completes a Row, or nil otherwise.
 func (cr *chunkReader) Process(cc *btpb.ReadRowsResponse_CellChunk) (Row, error) {
@@ -200,9 +207,19 @@ func (cr *chunkReader) validateNewRow(cc *btpb.ReadRowsResponse_CellChunk) error
 	if cc.RowKey == nil || cc.FamilyName == nil || cc.Qualifier == nil {
 		return fmt.Errorf("missing key field for new row %v", cc)
 	}
-	if cr.lastKey != "" && cr.lastKey >= string(cc.RowKey) {
-		return fmt.Errorf("out of order row key: %q, %q", cr.lastKey, string(cc.RowKey))
+
+	if cr.lastKey != "" {
+		r := strings.Compare(string(cc.RowKey), cr.lastKey)
+		direction := ">"
+		if cr.reversed {
+			r *= -1
+			direction = "<"
+		}
+		if r <= 0 {
+			return fmt.Errorf("out of order row key, new key %q must be %s prev row: %q", cc.RowKey, direction, cr.lastKey)
+		}
 	}
+
 	return nil
 }