feat: add --consolidator-reject-on-cap flag to reject queries when waiter cap exceeded (#19836)

Signed-off-by: Eduardo Ortega <5791035+ejortegau@users.noreply.github.com>
Co-authored-by: Claude <svc-devxp-claude@slack-corp.com>

Author: ejortegau

changelog/25.0/25.0.0/summary.md

@@ -12,6 +12,8 @@
         - [Default data protection for `_reverse` workflow cancel/complete](#vreplication-reverse-workflow-data-protection)
     - **[VTGate](#minor-changes-vtgate)**
         - [New controls for cross-keyspace reads](#vtgate-cross-keyspace-reads)
+    - **[VTTablet](#minor-changes-vttablet)**
+        - [Consolidator Reject on Waiter Cap](#vttablet-consolidator-reject-on-cap)
     - **[VTTablet](#minor-changes-vttablet)**
         - [Schema engine table-count limit is now configurable](#vttablet-schema-max-table-count)
 
@@ -67,6 +69,16 @@ The VTGate flag prevents cross-keyspace reads globally, regardless of per-keyspa
 
 ### <a id="minor-changes-vttablet"/>VTTablet</a>
 
+#### <a id="vttablet-consolidator-reject-on-cap"/>Consolidator Reject on Waiter Cap</a>
+
+A new `--consolidator-reject-on-cap` flag (default `false`) has been added to VTTablet. When enabled alongside a non-zero `--consolidator-query-waiter-cap`, queries that would join a consolidated result but exceed the **global** consolidator waiter cap are rejected with a `RESOURCE_EXHAUSTED` error instead of silently falling back to independent MySQL execution.
+
+**Important:** The cap is enforced against the consolidator's global `totalWaiterCount` across all queries, not a per-query waiter count. This means a duplicate for query B can be rejected because query A has already consumed most of the global waiter budget. This provides backpressure when the consolidator as a whole is saturated, rather than when any single query has too many waiters.
+
+See [#19836](https://github.com/vitessio/vitess/pull/19836) for details.
+
+### <a id="minor-changes-vttablet"/>VTTablet</a>
+
 #### <a id="vttablet-schema-max-table-count"/>Schema engine table-count limit is now configurable</a>
 
 Previously the schema engine had a hardcoded cap of 10,000 tables: a vttablet whose underlying MySQL had more than 10,000 tables would fail to load its schema and could not serve queries. This made recovery from `EmergencyReparentShard` impossible without dropping tables directly on MySQL.

go/flags/endtoend/vtcombo.txt

@@ -49,6 +49,7 @@ Flags:
       --config-type string                                               Config file type (omit to infer config type from file extension).
       --consolidator-cache-proto3-rows                                   If true, the consolidation leader pre-caches proto3-encoded rows so that waiters avoid redundant encoding work.
       --consolidator-query-waiter-cap int                                Configure the maximum number of clients allowed to wait on the consolidator.
+      --consolidator-reject-on-cap                                       If true, reject queries with a RESOURCE_EXHAUSTED error when the global consolidator waiter cap is exceeded, instead of falling back to independent execution. The cap is enforced globally across all consolidated queries, not per-query.
       --consolidator-stream-query-size int                               Configure the stream consolidator query size in bytes. Setting to 0 disables the stream consolidator. (default 2097152)
       --consolidator-stream-total-size int                               Configure the stream consolidator total size in bytes. Setting to 0 disables the stream consolidator. (default 134217728)
       --consul-auth-static-file string                                   JSON File to read the topos/tokens from.

go/flags/endtoend/vttablet.txt

@@ -82,6 +82,7 @@ Flags:
       --config-type string                                               Config file type (omit to infer config type from file extension).
       --consolidator-cache-proto3-rows                                   If true, the consolidation leader pre-caches proto3-encoded rows so that waiters avoid redundant encoding work.
       --consolidator-query-waiter-cap int                                Configure the maximum number of clients allowed to wait on the consolidator.
+      --consolidator-reject-on-cap                                       If true, reject queries with a RESOURCE_EXHAUSTED error when the global consolidator waiter cap is exceeded, instead of falling back to independent execution. The cap is enforced globally across all consolidated queries, not per-query.
       --consolidator-stream-query-size int                               Configure the stream consolidator query size in bytes. Setting to 0 disables the stream consolidator. (default 2097152)
       --consolidator-stream-total-size int                               Configure the stream consolidator total size in bytes. Setting to 0 disables the stream consolidator. (default 134217728)
       --consul-auth-static-file string                                   JSON File to read the topos/tokens from.

go/vt/vttablet/tabletserver/query_executor.go

@@ -775,6 +775,9 @@ func (qre *QueryExecutor) execSelect() (*sqltypes.Result, error) {
 				startTime := time.Now()
 				q.Wait()
 				qre.tsv.stats.WaitTimings.Record("Consolidations", startTime)
+			} else if qre.tsv.config.ConsolidatorRejectOnCap {
+				q.AddWaiterCounter(-1)
+				return nil, vterrors.Errorf(vtrpcpb.Code_RESOURCE_EXHAUSTED, "consolidator waiter cap exceeded")
 			} else {
 				// Waiter cap exceeded, fall back to independent query execution
 				waiterCapExceeded = true

go/vt/vttablet/tabletserver/query_executor_test.go

@@ -1697,6 +1697,101 @@ func TestQueryExecutorConsolidatorWaiterCapFallback(t *testing.T) {
 	db.VerifyAllExecutedOrFail()
 }
 
+func TestQueryExecutorConsolidatorRejectOnCap(t *testing.T) {
+	tests := []struct {
+		name            string
+		rejectOnCap     bool
+		wantErr         bool
+		wantErrCode     vtrpcpb.Code
+		wantErrMsg      string
+		wantDBQueries   int
+		wantWaitCalls   int
+		wantCounterArgs []int64
+	}{
+		{
+			name:            "reject enabled",
+			rejectOnCap:     true,
+			wantErr:         true,
+			wantErrCode:     vtrpcpb.Code_RESOURCE_EXHAUSTED,
+			wantErrMsg:      "consolidator waiter cap exceeded",
+			wantDBQueries:   0,
+			wantWaitCalls:   0,
+			wantCounterArgs: []int64{-1},
+		},
+		{
+			name:            "reject disabled falls back to independent execution",
+			rejectOnCap:     false,
+			wantErr:         false,
+			wantDBQueries:   1,
+			wantWaitCalls:   0,
+			wantCounterArgs: []int64{-1},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			db := setUpQueryExecutorTest(t)
+			defer db.Close()
+
+			ctx := context.Background()
+			tsv := newTestTabletServer(ctx, enableConsolidator, db)
+			defer tsv.StopService()
+
+			tsv.config.ConsolidatorQueryWaiterCap = 1
+			tsv.config.ConsolidatorRejectOnCap = tc.rejectOnCap
+
+			fakeConsolidator := sync2.NewFakeConsolidator()
+			tsv.qe.consolidator = fakeConsolidator
+
+			input := "select * from t limit 10001"
+			result := &sqltypes.Result{
+				Fields: getTestTableFields(),
+				Rows: [][]sqltypes.Value{{
+					sqltypes.NewInt32(1),
+					sqltypes.NewInt32(100),
+					sqltypes.NewInt32(200),
+				}},
+			}
+
+			fakePendingResult := &sync2.FakePendingResult{Consolidator: fakeConsolidator}
+			fakePendingResult.SetResult(result)
+			fakePendingResult.WaiterCount = 2
+			fakeConsolidator.SetTotalWaiterCount(2)
+
+			fakeConsolidator.CreateReturn = &sync2.FakeConsolidatorCreateReturn{
+				Created:       false,
+				PendingResult: fakePendingResult,
+			}
+
+			db.AddQuery(input, result)
+
+			qre := newTestQueryExecutor(context.Background(), tsv, input, 0)
+			qre.options = &querypb.ExecuteOptions{Consolidator: querypb.ExecuteOptions_CONSOLIDATOR_ENABLED}
+
+			got, err := qre.Execute()
+			if tc.wantErr {
+				require.Error(t, err)
+				assert.Equal(t, tc.wantErrCode, vterrors.Code(err))
+				assert.ErrorContains(t, err, tc.wantErrMsg)
+			} else {
+				require.NoError(t, err)
+				require.NotNil(t, got)
+			}
+
+			require.Len(t, fakeConsolidator.CreateCalls, 1)
+			require.Equal(t, tc.wantWaitCalls, fakePendingResult.WaitCalls)
+			require.Equal(t, 0, fakePendingResult.BroadcastCalls)
+
+			require.Len(t, fakePendingResult.AddWaiterCounterCalls, len(tc.wantCounterArgs))
+			for i, wantArg := range tc.wantCounterArgs {
+				require.Equal(t, wantArg, fakePendingResult.AddWaiterCounterCalls[i])
+			}
+
+			require.Equal(t, tc.wantDBQueries, db.GetQueryCalledNum(input))
+		})
+	}
+}
+
 func TestGetConnectionLogStats(t *testing.T) {
 	db := setUpQueryExecutorTest(t)
 	defer db.Close()

go/vt/vttablet/tabletserver/tabletenv/config.go

@@ -207,6 +207,7 @@ func registerTabletEnvFlags(fs *pflag.FlagSet) {
 
 	fs.Int64Var(&currentConfig.ConsolidatorQueryWaiterCap, "consolidator-query-waiter-cap", 0, "Configure the maximum number of clients allowed to wait on the consolidator.")
 	fs.BoolVar(&currentConfig.ConsolidatorCacheProto3Rows, "consolidator-cache-proto3-rows", defaultConfig.ConsolidatorCacheProto3Rows, "If true, the consolidation leader pre-caches proto3-encoded rows so that waiters avoid redundant encoding work.")
+	fs.BoolVar(&currentConfig.ConsolidatorRejectOnCap, "consolidator-reject-on-cap", defaultConfig.ConsolidatorRejectOnCap, "If true, reject queries with a RESOURCE_EXHAUSTED error when the global consolidator waiter cap is exceeded, instead of falling back to independent execution. The cap is enforced globally across all consolidated queries, not per-query.")
 	utils.SetFlagDurationVar(fs, &healthCheckInterval, "health-check-interval", defaultConfig.Healthcheck.Interval, "Interval between health checks")
 	utils.SetFlagDurationVar(fs, &degradedThreshold, "degraded-threshold", defaultConfig.Healthcheck.DegradedThreshold, "replication lag after which a replica is considered degraded")
 	utils.SetFlagDurationVar(fs, &unhealthyThreshold, "unhealthy-threshold", defaultConfig.Healthcheck.UnhealthyThreshold, "replication lag after which a replica is considered unhealthy")
@@ -342,6 +343,7 @@ type TabletConfig struct {
 	ConsolidatorStreamTotalSize int64         `json:"consolidatorStreamTotalSize,omitempty"`
 	ConsolidatorStreamQuerySize int64         `json:"consolidatorStreamQuerySize,omitempty"`
 	ConsolidatorQueryWaiterCap  int64         `json:"consolidatorMaxQueryWait,omitempty"`
+	ConsolidatorRejectOnCap     bool          `json:"consolidatorRejectOnCap,omitempty"`
 	ConsolidatorCacheProto3Rows bool          `json:"consolidatorCacheProto3Rows,omitempty"`
 	QueryCacheMemory            int64         `json:"queryCacheMemory,omitempty"`
 	QueryCacheDoorkeeper        bool          `json:"queryCacheDoorkeeper,omitempty"`