hrpc/scan: add support for scan attribute (#238)

hrpc/hrpc_test.go

@@ -395,6 +395,31 @@ func TestScanToProto(t *testing.T) {
 				},
 			},
 		},
+		{ // set scan attribute
+			s: func() *Scan {
+				s, _ := NewScanStr(ctx, "",
+					Attribute("key1", []byte("value1")),
+					Attribute("key2", []byte("value2")),
+				)
+				return s
+			}(),
+			expProto: &pb.ScanRequest{
+				Region:                  rs,
+				NumberOfRows:            proto.Uint32(DefaultNumberOfRows),
+				CloseScanner:            proto.Bool(false),
+				ClientHandlesPartials:   proto.Bool(true),
+				ClientHandlesHeartbeats: proto.Bool(true),
+				Scan: &pb.Scan{
+					MaxResultSize: proto.Uint64(DefaultMaxResultSize),
+					Column:        []*pb.Column{},
+					TimeRange:     &pb.TimeRange{},
+					Attribute: []*pb.NameBytesPair{
+						{Name: proto.String("key1"), Value: []byte("value1")},
+						{Name: proto.String("key2"), Value: []byte("value2")},
+					},
+				},
+			},
+		},
 		{ // scan key range
 			s: func() *Scan {
 				s, _ := NewScanRange(ctx, nil, startRow, stopRow)

hrpc/scan.go

@@ -67,6 +67,7 @@ type Scan struct {
 	maxResultSize uint64
 	numberOfRows  uint32
 	reversed      bool
+	attribute     []*pb.NameBytesPair
 
 	closeScanner        bool
 	allowPartialResults bool
@@ -227,6 +228,7 @@ func (s *Scan) ToProto() proto.Message {
 	if s.consistency != DefaultConsistency {
 		scan.Scan.Consistency = s.consistency.toProto()
 	}
+	scan.Scan.Attribute = s.attribute
 	scan.Scan.Filter = s.filter
 	return scan
 }
@@ -344,3 +346,19 @@ func Reversed() func(Call) error {
 		return nil
 	}
 }
+
+// Attribute is a Scan-only option which set metadata-like attribute on the request. Attribute
+// option can be used multiple times and will be appended to a list. Attribute are useful to
+// communicate special information about the Scan request to HBase, such as:
+// - retrieve MOB metadata
+// - change behaviour of coprocessors
+func Attribute(key string, val []byte) func(Call) error {
+	return func(g Call) error {
+		scan, ok := g.(*Scan)
+		if !ok {
+			return errors.New("'Attributes' option can only be used with Scan queries")
+		}
+		scan.attribute = append(scan.attribute, &pb.NameBytesPair{Name: &key, Value: val})
+		return nil
+	}
+}