fix(bigtable/bttest): emulator too lenient regarding RowFilter and Ch…

…eckAndMutateRow conditions (#4095)

- Sanitize some test requests to include Mutations (these are rejected by real bigtable)
- Sanitize Chain and Interleave filters to have at least 2  (these are rejected by real bigtable)
- BlockAllFilter and PassAllFilter are required to be true
- bttest emulates some of these error conditions

bigtable/bttest/inmem.go

@@ -483,10 +483,19 @@ func filterRow(f *btpb.RowFilter, r *row) (bool, error) {
 	// Handle filters that apply beyond just including/excluding cells.
 	switch f := f.Filter.(type) {
 	case *btpb.RowFilter_BlockAllFilter:
-		return !f.BlockAllFilter, nil
+		if !f.BlockAllFilter {
+			return false, status.Errorf(codes.InvalidArgument, "block_all_filter must be true if set")
+		}
+		return false, nil
 	case *btpb.RowFilter_PassAllFilter:
-		return f.PassAllFilter, nil
+		if !f.PassAllFilter {
+			return false, status.Errorf(codes.InvalidArgument, "pass_all_filter must be true if set")
+		}
+		return true, nil
 	case *btpb.RowFilter_Chain_:
+		if len(f.Chain.Filters) < 2 {
+			return false, status.Errorf(codes.InvalidArgument, "Chain must contain at least two RowFilters")
+		}
 		for _, sub := range f.Chain.Filters {
 			match, err := filterRow(sub, r)
 			if err != nil {
@@ -498,6 +507,9 @@ func filterRow(f *btpb.RowFilter, r *row) (bool, error) {
 		}
 		return true, nil
 	case *btpb.RowFilter_Interleave_:
+		if len(f.Interleave.Filters) < 2 {
+			return false, status.Errorf(codes.InvalidArgument, "Interleave must contain at least two RowFilters")
+		}
 		srs := make([]*row, 0, len(f.Interleave.Filters))
 		for _, sub := range f.Interleave.Filters {
 			sr := r.copy()

bigtable/bttest/inmem_test.go

@@ -34,6 +34,8 @@ import (
 	btapb "google.golang.org/genproto/googleapis/bigtable/admin/v2"
 	btpb "google.golang.org/genproto/googleapis/bigtable/v2"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 func TestConcurrentMutationsReadModifyAndGC(t *testing.T) {
@@ -903,27 +905,29 @@ func TestCheckAndMutateRowWithoutPredicate(t *testing.T) {
 		t.Fatalf("Creating table: %v", err)
 	}
 
-	// Populate the table
 	val := []byte("value")
+	muts := []*btpb.Mutation{{
+		Mutation: &btpb.Mutation_SetCell_{SetCell: &btpb.Mutation_SetCell{
+			FamilyName:      "cf",
+			ColumnQualifier: []byte("col"),
+			TimestampMicros: 1000,
+			Value:           val,
+		}},
+	}}
+
 	mrreq := &btpb.MutateRowRequest{
 		TableName: tbl.Name,
 		RowKey:    []byte("row-present"),
-		Mutations: []*btpb.Mutation{{
-			Mutation: &btpb.Mutation_SetCell_{SetCell: &btpb.Mutation_SetCell{
-				FamilyName:      "cf",
-				ColumnQualifier: []byte("col"),
-				TimestampMicros: 1000,
-				Value:           val,
-			}},
-		}},
+		Mutations: muts,
 	}
 	if _, err := s.MutateRow(ctx, mrreq); err != nil {
 		t.Fatalf("Populating table: %v", err)
 	}
 
 	req := &btpb.CheckAndMutateRowRequest{
-		TableName: tbl.Name,
-		RowKey:    []byte("row-not-present"),
+		TableName:      tbl.Name,
+		RowKey:         []byte("row-not-present"),
+		FalseMutations: muts,
 	}
 	if res, err := s.CheckAndMutateRow(ctx, req); err != nil {
 		t.Errorf("CheckAndMutateRow error: %v", err)
@@ -932,8 +936,9 @@ func TestCheckAndMutateRowWithoutPredicate(t *testing.T) {
 	}
 
 	req = &btpb.CheckAndMutateRowRequest{
-		TableName: tbl.Name,
-		RowKey:    []byte("row-present"),
+		TableName:      tbl.Name,
+		RowKey:         []byte("row-present"),
+		FalseMutations: muts,
 	}
 	if res, err := s.CheckAndMutateRow(ctx, req); err != nil {
 		t.Errorf("CheckAndMutateRow error: %v", err)
@@ -993,6 +998,14 @@ func TestCheckAndMutateRowWithPredicate(t *testing.T) {
 		}
 	}
 
+	var bogusMutations = []*btpb.Mutation{{
+		Mutation: &btpb.Mutation_DeleteFromFamily_{
+			DeleteFromFamily: &btpb.Mutation_DeleteFromFamily{
+				FamilyName: "bogus_family",
+			},
+		},
+	}}
+
 	tests := []struct {
 		req       *btpb.CheckAndMutateRowRequest
 		wantMatch bool
@@ -1005,11 +1018,13 @@ func TestCheckAndMutateRowWithPredicate(t *testing.T) {
 		{
 			req: &btpb.CheckAndMutateRowRequest{
 				TableName: tbl.Name,
+				RowKey:    []byte("row1"),
 				PredicateFilter: &btpb.RowFilter{
 					Filter: &btpb.RowFilter_RowKeyRegexFilter{
 						RowKeyRegexFilter: []byte("not-one"),
 					},
 				},
+				TrueMutations: bogusMutations,
 			},
 			name: "no match",
 		},
@@ -1022,6 +1037,7 @@ func TestCheckAndMutateRowWithPredicate(t *testing.T) {
 						RowKeyRegexFilter: []byte("ro.+"),
 					},
 				},
+				FalseMutations: bogusMutations,
 			},
 			wantMatch: true,
 			name:      "rowkey regex",
@@ -1035,6 +1051,7 @@ func TestCheckAndMutateRowWithPredicate(t *testing.T) {
 						PassAllFilter: true,
 					},
 				},
+				FalseMutations: bogusMutations,
 			},
 			wantMatch: true,
 			name:      "pass all",
@@ -1344,13 +1361,14 @@ func populateTable(ctx context.Context, s *server) (*btapb.Table, error) {
 
 func TestFilters(t *testing.T) {
 	tests := []struct {
-		in  *btpb.RowFilter
-		out int
+		in   *btpb.RowFilter
+		code codes.Code
+		out  int
 	}{
 		{in: &btpb.RowFilter{Filter: &btpb.RowFilter_BlockAllFilter{true}}, out: 0},
-		{in: &btpb.RowFilter{Filter: &btpb.RowFilter_BlockAllFilter{false}}, out: 1},
+		{in: &btpb.RowFilter{Filter: &btpb.RowFilter_BlockAllFilter{false}}, code: codes.InvalidArgument},
 		{in: &btpb.RowFilter{Filter: &btpb.RowFilter_PassAllFilter{true}}, out: 1},
-		{in: &btpb.RowFilter{Filter: &btpb.RowFilter_PassAllFilter{false}}, out: 0},
+		{in: &btpb.RowFilter{Filter: &btpb.RowFilter_PassAllFilter{false}}, code: codes.InvalidArgument},
 	}
 
 	ctx := context.Background()
@@ -1373,7 +1391,16 @@ func TestFilters(t *testing.T) {
 		req.Filter = tc.in
 
 		mock := &MockReadRowsServer{}
-		if err = s.ReadRows(req, mock); err != nil {
+		err := s.ReadRows(req, mock)
+		if tc.code != codes.OK {
+			s, _ := status.FromError(err)
+			if s.Code() != tc.code {
+				t.Errorf("error code: got %d, want %d", s.Code(), tc.code)
+			}
+			continue
+		}
+
+		if err != nil {
 			t.Errorf("ReadRows error: %v", err)
 			continue
 		}
@@ -1742,14 +1769,19 @@ func TestFilterRowWithSingleColumnQualifier(t *testing.T) {
 												EndValue: &btpb.ValueRange_EndValueClosed{EndValueClosed: []byte("a")},
 											}},
 										},
+										{Filter: &btpb.RowFilter_PassAllFilter{PassAllFilter: true}},
 									}},
 								}},
 								TrueFilter: &btpb.RowFilter{Filter: &btpb.RowFilter_PassAllFilter{PassAllFilter: true}},
 							},
-						}}},
+						}},
+							{Filter: &btpb.RowFilter_BlockAllFilter{BlockAllFilter: true}},
+						},
 					},
 				},
-			}}},
+			},
+				{Filter: &btpb.RowFilter_PassAllFilter{PassAllFilter: true}},
+			}},
 		}},
 	}