Merge #148488 #148607

148488: sql: avoid contention between long mutation and auto stats r=yuzefovich a=yuzefovich

Previously, when processing large mutations in batches (see `BatchedNext` method), we would notify the stats refresher about the number of rows affected after each batch. If the mutation is long, then this could have triggered the auto stats collection that would contend with the mutation that triggered it. There is actually no reason to try collecting auto stats on those just-mutated rows since in most cases they haven't been committed yet. This commit fixes this issue by notifying the stats refresher about the total number of rows affected when the last batch of the mutation has just been processed. The bug has been present since about 19.1 version.

Note that we generally don't perform well with long mutations, but it seems like a worthwhile improvement nonetheless. We can't remove contention with auto stats triggered by _other_ mutations, but at least we can do so for self-inflicted ones. (Also probably a more important retry reason is the closed TS interval.)

Fixes: #148487.

Release note (bug fix): Large mutation statements (INSERTs, UPDATEs, DELETEs, UPSERTs) are now less likely to encounter contention with automatic table stats collection in some cases. The bug has been present since 19.1 version.

148607: *: consolidate `InternalClient` RPC client creation r=cthumuluru-crdb a=cthumuluru-crdb

This commit consolidates `InternalClient` RPC client creation logic and replaces RPC clients with adapters. It is a continuation of the work done in #147606.

Epic: CRDB-48923
Fixes: #148353
Release note: none

Co-authored-by: Yahor Yuzefovich <[email protected]>
Co-authored-by: Chandra Thumuluru <[email protected]>

Author: 3 people

pkg/cli/BUILD.bazel

@@ -141,6 +141,7 @@ go_library(
         "//pkg/raft/raftpb",
         "//pkg/roachpb",
         "//pkg/rpc",
+        "//pkg/rpc/rpcbase",
         "//pkg/security",
         "//pkg/security/certnames",
         "//pkg/security/clientsecopts",
@@ -274,6 +275,7 @@ go_library(
         "@com_google_cloud_go_storage//:storage",
         "@in_gopkg_yaml_v2//:yaml_v2",
         "@in_gopkg_yaml_v3//:yaml_v3",
+        "@io_storj_drpc//:drpc",
         "@org_golang_google_api//option",
         "@org_golang_google_grpc//:grpc",
         "@org_golang_google_grpc//codes",

pkg/cli/rpc_clients.go

@@ -10,11 +10,13 @@ import (
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/rpc"
+	"github.com/cockroachdb/cockroach/pkg/rpc/rpcbase"
 	"github.com/cockroachdb/cockroach/pkg/server"
 	"github.com/cockroachdb/cockroach/pkg/server/serverpb"
 	"github.com/cockroachdb/cockroach/pkg/ts/tspb"
 	"github.com/cockroachdb/errors"
 	"google.golang.org/grpc"
+	"storj.io/drpc"
 )
 
 // rpcConn defines a common interface for creating RPC clients. It hides the
@@ -25,7 +27,7 @@ type rpcConn interface {
 	NewAdminClient() serverpb.RPCAdminClient
 	NewInitClient() serverpb.RPCInitClient
 	NewTimeSeriesClient() tspb.RPCTimeSeriesClient
-	NewInternalClient() kvpb.InternalClient
+	NewInternalClient() kvpb.RPCInternalClient
 }
 
 // grpcConn is an implementation of rpcConn that provides methods to create
@@ -51,8 +53,35 @@ func (c *grpcConn) NewTimeSeriesClient() tspb.RPCTimeSeriesClient {
 	return tspb.NewGRPCTimeSeriesClientAdapter(c.conn)
 }
 
-func (c *grpcConn) NewInternalClient() kvpb.InternalClient {
-	return kvpb.NewInternalClient(c.conn)
+func (c *grpcConn) NewInternalClient() kvpb.RPCInternalClient {
+	return kvpb.NewGRPCInternalClientAdapter(c.conn)
+}
+
+// drpcConn is an implementation of rpcConn that provides methods to create
+// various RPC clients. This allows the CLI to interact with the server using
+// DRPC without exposing the underlying connection details.
+type drpcConn struct {
+	conn drpc.Conn
+}
+
+func (c *drpcConn) NewStatusClient() serverpb.RPCStatusClient {
+	return serverpb.NewDRPCStatusClientAdapter(c.conn)
+}
+
+func (c *drpcConn) NewAdminClient() serverpb.RPCAdminClient {
+	return serverpb.NewDRPCAdminClientAdapter(c.conn)
+}
+
+func (c *drpcConn) NewInitClient() serverpb.RPCInitClient {
+	return serverpb.NewDRPCInitClientAdapter(c.conn)
+}
+
+func (c *drpcConn) NewTimeSeriesClient() tspb.RPCTimeSeriesClient {
+	return tspb.NewDRPCTimeSeriesClientAdapter(c.conn)
+}
+
+func (c *drpcConn) NewInternalClient() kvpb.RPCInternalClient {
+	return kvpb.NewDRPCInternalClientAdapter(c.conn)
 }
 
 func makeRPCClientConfig(cfg server.Config) rpc.ClientConnConfig {
@@ -72,11 +101,19 @@ func makeRPCClientConfig(cfg server.Config) rpc.ClientConnConfig {
 
 func newClientConn(ctx context.Context, cfg server.Config) (rpcConn, func(), error) {
 	ccfg := makeRPCClientConfig(cfg)
-	cc, finish, err := rpc.NewClientConn(ctx, ccfg)
-	if err != nil {
-		return nil, nil, errors.Wrap(err, "failed to connect to the node")
+	if !rpcbase.TODODRPC {
+		cc, finish, err := rpc.NewClientConn(ctx, ccfg)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "failed to connect to the node")
+		}
+		return &grpcConn{conn: cc}, finish, nil
+	} else {
+		dc, finish, err := rpc.NewDRPCClientConn(ctx, ccfg)
+		if err != nil {
+			return nil, nil, errors.Wrap(err, "failed to connect to the node")
+		}
+		return &drpcConn{conn: dc}, finish, nil
 	}
-	return &grpcConn{conn: cc}, finish, nil
 }
 
 // dialAdminClient dials a client connection and returns an AdminClient and a

pkg/kv/kvclient/kvcoord/dist_sender_rangefeed_mock_test.go

@@ -112,9 +112,9 @@ func TestDistSenderRangeFeedRetryOnTransportErrors(t *testing.T) {
 				// cancels the context and closes the range feed stream.
 				if spec.expectRetry {
 					rangeDB.EXPECT().RangeLookup(gomock.Any(), roachpb.RKeyMin, kvpb.INCONSISTENT, false).MinTimes(1).Return([]roachpb.RangeDescriptor{desc}, nil, nil) //.FirstRange().Return(&desc, nil)
-					client := kvpbmock.NewMockInternalClient(ctrl)
+					client := kvpbmock.NewMockRPCInternalClient(ctrl)
 
-					stream := kvpbmock.NewMockInternal_MuxRangeFeedClient(ctrl)
+					stream := kvpbmock.NewMockRPCInternal_MuxRangeFeedClient(ctrl)
 					stream.EXPECT().Send(gomock.Any()).Return(nil)
 					stream.EXPECT().Recv().Do(func() {
 						cancel()

pkg/kv/kvclient/kvcoord/dist_sender_rangefeed_test.go

@@ -49,10 +49,10 @@ type testRangefeedClient struct {
 }
 
 func (c *testRangefeedClient) MuxRangeFeed(
-	ctx context.Context, opts ...grpc.CallOption,
-) (kvpb.Internal_MuxRangeFeedClient, error) {
+	ctx context.Context,
+) (kvpb.RPCInternal_MuxRangeFeedClient, error) {
 	defer c.count()
-	return c.RestrictedInternalClient.MuxRangeFeed(ctx, opts...)
+	return c.RestrictedInternalClient.MuxRangeFeed(ctx)
 }
 
 type internalClientCounts struct {

pkg/kv/kvclient/kvcoord/transport_test.go

@@ -19,7 +19,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing/tracingpb"
 	"github.com/stretchr/testify/require"
-	"google.golang.org/grpc"
 )
 
 func TestTransportMoveToFront(t *testing.T) {
@@ -221,17 +220,17 @@ type mockInternalClient struct {
 	pErr *kvpb.Error
 }
 
-var _ kvpb.InternalClient = &mockInternalClient{}
+var _ kvpb.RPCInternalClient = &mockInternalClient{}
 
 func (*mockInternalClient) ResetQuorum(
-	context.Context, *kvpb.ResetQuorumRequest, ...grpc.CallOption,
+	context.Context, *kvpb.ResetQuorumRequest,
 ) (*kvpb.ResetQuorumResponse, error) {
 	panic("unimplemented")
 }
 
 // Batch is part of the kvpb.InternalClient interface.
 func (m *mockInternalClient) Batch(
-	ctx context.Context, in *kvpb.BatchRequest, opts ...grpc.CallOption,
+	ctx context.Context, in *kvpb.BatchRequest,
 ) (*kvpb.BatchResponse, error) {
 	var sp *tracing.Span
 	if m.tr != nil {
@@ -256,75 +255,75 @@ func (m *mockInternalClient) Batch(
 }
 
 func (m *mockInternalClient) BatchStream(
-	ctx context.Context, opts ...grpc.CallOption,
-) (kvpb.Internal_BatchStreamClient, error) {
+	ctx context.Context,
+) (kvpb.RPCInternal_BatchStreamClient, error) {
 	return nil, fmt.Errorf("unsupported BatchStream call")
 }
 
 // RangeLookup implements the kvpb.InternalClient interface.
 func (m *mockInternalClient) RangeLookup(
-	ctx context.Context, rl *kvpb.RangeLookupRequest, _ ...grpc.CallOption,
+	ctx context.Context, rl *kvpb.RangeLookupRequest,
 ) (*kvpb.RangeLookupResponse, error) {
 	return nil, fmt.Errorf("unsupported RangeLookup call")
 }
 
 func (m *mockInternalClient) MuxRangeFeed(
-	ctx context.Context, opts ...grpc.CallOption,
-) (kvpb.Internal_MuxRangeFeedClient, error) {
+	ctx context.Context,
+) (kvpb.RPCInternal_MuxRangeFeedClient, error) {
 	return nil, fmt.Errorf("unsupported MuxRangeFeed call")
 }
 
 // GossipSubscription is part of the kvpb.InternalClient interface.
 func (m *mockInternalClient) GossipSubscription(
-	ctx context.Context, args *kvpb.GossipSubscriptionRequest, _ ...grpc.CallOption,
-) (kvpb.Internal_GossipSubscriptionClient, error) {
+	ctx context.Context, args *kvpb.GossipSubscriptionRequest,
+) (kvpb.RPCInternal_GossipSubscriptionClient, error) {
 	return nil, fmt.Errorf("unsupported GossipSubscripion call")
 }
 
 func (m *mockInternalClient) Join(
-	context.Context, *kvpb.JoinNodeRequest, ...grpc.CallOption,
+	context.Context, *kvpb.JoinNodeRequest,
 ) (*kvpb.JoinNodeResponse, error) {
 	return nil, fmt.Errorf("unsupported Join call")
 }
 
 func (m *mockInternalClient) TokenBucket(
-	ctx context.Context, in *kvpb.TokenBucketRequest, _ ...grpc.CallOption,
+	ctx context.Context, in *kvpb.TokenBucketRequest,
 ) (*kvpb.TokenBucketResponse, error) {
 	return nil, fmt.Errorf("unsupported TokenBucket call")
 }
 
 func (m *mockInternalClient) GetSpanConfigs(
-	_ context.Context, _ *roachpb.GetSpanConfigsRequest, _ ...grpc.CallOption,
+	_ context.Context, _ *roachpb.GetSpanConfigsRequest,
 ) (*roachpb.GetSpanConfigsResponse, error) {
 	return nil, fmt.Errorf("unsupported GetSpanConfigs call")
 }
 
 func (m *mockInternalClient) SpanConfigConformance(
-	_ context.Context, _ *roachpb.SpanConfigConformanceRequest, _ ...grpc.CallOption,
+	_ context.Context, _ *roachpb.SpanConfigConformanceRequest,
 ) (*roachpb.SpanConfigConformanceResponse, error) {
 	return nil, fmt.Errorf("unsupported SpanConfigConformance call")
 }
 
 func (m *mockInternalClient) GetAllSystemSpanConfigsThatApply(
-	context.Context, *roachpb.GetAllSystemSpanConfigsThatApplyRequest, ...grpc.CallOption,
+	context.Context, *roachpb.GetAllSystemSpanConfigsThatApplyRequest,
 ) (*roachpb.GetAllSystemSpanConfigsThatApplyResponse, error) {
 	return nil, fmt.Errorf("unsupported GetAllSystemSpanConfigsThatApply call")
 }
 
 func (m *mockInternalClient) UpdateSpanConfigs(
-	_ context.Context, _ *roachpb.UpdateSpanConfigsRequest, _ ...grpc.CallOption,
+	_ context.Context, _ *roachpb.UpdateSpanConfigsRequest,
 ) (*roachpb.UpdateSpanConfigsResponse, error) {
 	return nil, fmt.Errorf("unsupported UpdateSpanConfigs call")
 }
 
 func (m *mockInternalClient) TenantSettings(
-	context.Context, *kvpb.TenantSettingsRequest, ...grpc.CallOption,
-) (kvpb.Internal_TenantSettingsClient, error) {
+	context.Context, *kvpb.TenantSettingsRequest,
+) (kvpb.RPCInternal_TenantSettingsClient, error) {
 	return nil, fmt.Errorf("unsupported TenantSettings call")
 }
 
 func (n *mockInternalClient) GetRangeDescriptors(
-	context.Context, *kvpb.GetRangeDescriptorsRequest, ...grpc.CallOption,
-) (kvpb.Internal_GetRangeDescriptorsClient, error) {
+	context.Context, *kvpb.GetRangeDescriptorsRequest,
+) (kvpb.RPCInternal_GetRangeDescriptorsClient, error) {
 	return nil, fmt.Errorf("unsupported GetRangeDescriptors call")
 }

pkg/kv/kvclient/kvtenant/connector.go

@@ -222,7 +222,7 @@ type connector struct {
 
 // client represents an RPC client that proxies to a KV instance.
 type client struct {
-	kvpb.InternalClient
+	kvpb.RPCInternalClient
 	serverpb.RPCStatusClient
 	serverpb.RPCAdminClient
 	tspb.RPCTimeSeriesClient
@@ -983,7 +983,7 @@ func (c *connector) dialAddrs(ctx context.Context) (*client, error) {
 					continue
 				}
 				return &client{
-					InternalClient:      kvpb.NewInternalClient(conn),
+					RPCInternalClient:   kvpb.NewGRPCInternalClientAdapter(conn),
 					RPCStatusClient:     serverpb.NewGRPCStatusClientAdapter(conn),
 					RPCAdminClient:      serverpb.NewGRPCAdminClientAdapter(conn),
 					RPCTimeSeriesClient: tspb.NewGRPCTimeSeriesClientAdapter(conn),
@@ -1005,7 +1005,7 @@ func (c *connector) dialAddr(ctx context.Context, addr string) (conn *grpc.Clien
 	return conn, err
 }
 
-func (c *connector) tryForgetClient(ctx context.Context, client kvpb.InternalClient) {
+func (c *connector) tryForgetClient(ctx context.Context, client *client) {
 	if ctx.Err() != nil {
 		// Error (may be) due to context. Don't forget client.
 		return

pkg/kv/kvpb/kvpbmock/BUILD.bazel

@@ -4,8 +4,8 @@ gomock(
     name = "mock_kvpb",
     out = "mocks_generated.go",
     interfaces = [
-        "InternalClient",
-        "Internal_MuxRangeFeedClient",
+        "RPCInternalClient",
+        "RPCInternal_MuxRangeFeedClient",
     ],
     library = "//pkg/kv/kvpb",
     package = "kvpbmock",