messages: review comments

Author: sougou

go/vt/proto/query/query.pb.go

@@ -31,8 +31,8 @@ var (
 
 func init() {
 	flag.IntVar(&qsConfig.PoolSize, "queryserver-config-pool-size", DefaultQsConfig.PoolSize, "query server connection pool size, connection pool is used by regular queries (non streaming, not in a transaction)")
-	flag.IntVar(&qsConfig.StreamPoolSize, "queryserver-config-stream-pool-size", DefaultQsConfig.StreamPoolSize, "query server stream pool size, stream pool is used by stream queries: queries that return results to client in a streaming fashion")
-	flag.IntVar(&qsConfig.MessagePoolSize, "queryserver-config-message-pool-size", DefaultQsConfig.MessagePoolSize, "query server message pool size, message pool is used by message managers: recommended value is one per message table")
+	flag.IntVar(&qsConfig.StreamPoolSize, "queryserver-config-stream-pool-size", DefaultQsConfig.StreamPoolSize, "query server stream connection pool size, stream pool is used by stream queries: queries that return results to client in a streaming fashion")
+	flag.IntVar(&qsConfig.MessagePoolSize, "queryserver-config-message-conn-pool-size", DefaultQsConfig.MessagePoolSize, "query server message connection pool size, message pool is used by message managers: recommended value is one per message table")
 	flag.IntVar(&qsConfig.TransactionCap, "queryserver-config-transaction-cap", DefaultQsConfig.TransactionCap, "query server transaction cap is the maximum number of transactions allowed to happen at any given point of a time for a single vttablet. E.g. by setting transaction cap to 100, there are at most 100 transactions will be processed by a vttablet and the 101th transaction will be blocked (and fail if it cannot get connection within specified timeout)")
 	flag.Float64Var(&qsConfig.TransactionTimeout, "queryserver-config-transaction-timeout", DefaultQsConfig.TransactionTimeout, "query server transaction timeout (in seconds), a transaction will be killed if it takes longer than this value")
 	flag.Float64Var(&qsConfig.TxShutDownGracePeriod, "transaction_shutdown_grace_period", DefaultQsConfig.TxShutDownGracePeriod, "how long to wait (in seconds) for transactions to complete during graceful shutdown.")

go/vt/tabletserver/config.go

@@ -330,7 +330,9 @@ func (q *query) MessageAck(ctx context.Context, request *querypb.MessageAckReque
 		return nil, vterrors.ToGRPCError(err)
 	}
 	return &querypb.MessageAckResponse{
-		Count: count,
+		Result: &querypb.QueryResult{
+			RowsAffected: uint64(count),
+		},
 	}, nil
 }
 

go/vt/tabletserver/grpcqueryservice/server.go

@@ -547,7 +547,7 @@ func (conn *gRPCQueryClient) MessageAck(ctx context.Context, target *querypb.Tar
 	if err != nil {
 		return 0, tabletconn.TabletErrorFromGRPC(err)
 	}
-	return reply.Count, nil
+	return int64(reply.Result.RowsAffected), nil
 }
 
 // SplitQuery is the stub for TabletServer.SplitQuery RPC

go/vt/tabletserver/grpctabletconn/conn.go

@@ -490,7 +490,8 @@ func (si *SchemaInfo) GetTable(tableName sqlparser.TableIdent) *TableInfo {
 	return si.tables[tableName.String()]
 }
 
-// GetSchema returns a copy of the schema.
+// GetSchema returns the current schema. The Tables are a shared
+// data strucutre and must be treated as read-only.
 func (si *SchemaInfo) GetSchema() map[string]*schema.Table {
 	si.mu.Lock()
 	defer si.mu.Unlock()

go/vt/tabletserver/schema_info.go

@@ -484,7 +484,7 @@ func (conn *vtgateConn) MessageAck(ctx context.Context, keyspace string, name st
 	if err != nil {
 		return 0, vterrors.FromGRPCError(err)
 	}
-	return r.Count, nil
+	return int64(r.Result.RowsAffected), nil
 }
 
 func (conn *vtgateConn) SplitQuery(

go/vt/vtgate/grpcvtgateconn/conn.go

@@ -416,7 +416,11 @@ func (vtg *VTGate) MessageAck(ctx context.Context, request *vtgatepb.MessageAckR
 	if vtgErr != nil {
 		return nil, vterrors.ToGRPCError(vtgErr)
 	}
-	return &querypb.MessageAckResponse{Count: count}, nil
+	return &querypb.MessageAckResponse{
+		Result: &querypb.QueryResult{
+			RowsAffected: uint64(count),
+		},
+	}, nil
 }
 
 // SplitQuery is the RPC version of vtgateservice.VTGateService method

go/vt/vtgate/grpcvtgateservice/server.go

@@ -806,7 +806,8 @@ func (vtg *VTGate) GetSrvKeyspace(ctx context.Context, keyspace string) (*topoda
 	return vtg.resolver.toposerv.GetSrvKeyspace(ctx, vtg.resolver.cell, keyspace)
 }
 
-// MessageStream is part of the vtgate service API.
+// MessageStream is part of the vtgate service API. This is a V2 level API that's sent
+// to the Resolver.
 func (vtg *VTGate) MessageStream(ctx context.Context, keyspace string, shard string, keyRange *topodatapb.KeyRange, name string, sendReply func(*sqltypes.Result) error) error {
 	startTime := time.Now()
 	ltt := topoproto.TabletTypeLString(topodatapb.TabletType_MASTER)
@@ -835,7 +836,9 @@ func (vtg *VTGate) MessageStream(ctx context.Context, keyspace string, shard str
 	return formatError(err)
 }
 
-// MessageAck is part of the vtgate service API.
+// MessageAck is part of the vtgate service API. This is a V3 level API that's sent
+// to the Router. The table name will be resolved using V3 rules, and the routing
+// will make use of vindexes for sharded keyspaces.
 func (vtg *VTGate) MessageAck(ctx context.Context, keyspace string, name string, ids []*querypb.Value) (int64, error) {
 	startTime := time.Now()
 	ltt := topoproto.TabletTypeLString(topodatapb.TabletType_MASTER)

go/vt/vtgate/vtgate.go

@@ -724,7 +724,7 @@ func (f *fakeVTGateService) MessageAck(ctx context.Context, keyspace string, nam
 	if !reflect.DeepEqual(ids, messageids) {
 		return 0, errors.New("MessageAck ids mismatch")
 	}
-	return 1, nil
+	return messageAckRowsAffected, nil
 }
 
 // querySplitQuery contains all the fields we use to test SplitQuery
@@ -1882,7 +1882,10 @@ func testMessageStreamPanic(t *testing.T, conn *vtgateconn.VTGateConn) {
 
 func testMessageAck(t *testing.T, conn *vtgateconn.VTGateConn) {
 	ctx := newContext()
-	_, err := conn.MessageAck(ctx, "", messageName, messageids)
+	got, err := conn.MessageAck(ctx, "", messageName, messageids)
+	if got != messageAckRowsAffected {
+		t.Errorf("MessageAck: %d, want %d", got, messageAckRowsAffected)
+	}
 	if err != nil {
 		t.Error(err)
 	}
@@ -2635,3 +2638,4 @@ var messageids = []*querypb.Value{
 	sqltypes.MakeString([]byte("1")).ToProtoValue(),
 	sqltypes.MakeString([]byte("3")).ToProtoValue(),
 }
+var messageAckRowsAffected = int64(1)

go/vt/vtgate/vtgateconntest/client.go

@@ -588,8 +588,10 @@ message MessageAckRequest {
 
 // MessageAckResponse is the response for MessageAck.
 message MessageAckResponse {
-  // count is the number of messages acked.
-  int64 count = 1;
+  // result contains the result of the ack operation.
+  // Since this acts like a DML, only
+  // RowsAffected is returned in the result.
+  QueryResult result = 1;
 }
 
 // SplitQueryRequest is the payload for SplitQuery sent by VTGate to a VTTablet.

proto/query.proto

@@ -298,7 +298,7 @@ def message_ack(
           e, 'MessageAck', name=name, ids=ids,
           keyspace=keyspace)
 
-    return response.count
+    return response.result.rows_affected
 
 
 def _convert_exception(exc, *args, **kwargs):