Merge branch 'vitessio:main' into vtadmin_schemas_sortable

changelog/22.0/22.0.0/summary.md

@@ -15,6 +15,7 @@
   - **[Stalled Disk Recovery in VTOrc](#stall-disk-recovery)**
   - **[Update default MySQL version to 8.0.40](#mysql-8-0-40)**
   - **[Update lite images to Debian Bookworm](#debian-bookworm)**
+  - **[KeyRanges in `--clusters_to_watch` in VTOrc](#key-range-vtorc)**
   - **[Support for Filtering Query logs on Error](#query-logs)**
 - **[Minor Changes](#minor-changes)**
   - **[VTTablet Flags](#flags-vttablet)**
@@ -135,6 +136,11 @@ This is the last time this will be needed in the `8.0.x` series, as starting wit
 
 The base system now uses Debian Bookworm instead of Debian Bullseye for the `vitess/lite` images. This change was brought by [Pull Request #17552].
 
+### <a id="key-range-vtorc"/>KeyRanges in `--clusters_to_watch` in VTOrc</a>
+VTOrc now supports specifying keyranges in the `--clusters_to_watch` flag. This means that there is no need to restart a VTOrc instance with a different flag value when you reshard a keyspace.
+For example, if a VTOrc is configured to watch `ks/-80`, then it would watch all the shards that fall under the keyrange `-80`. If a reshard is performed and `-80` is split into new shards `-40` and `40-80`, the VTOrc instance will automatically start watching the new shards without needing a restart. In the previous logic, specifying `ks/-80` for the flag would mean that VTOrc would watch only 1 (or no) shard. In the new system, since we interpret `-80` as a key range, it can watch multiple shards as described in the example.
+Users can continue to specify exact keyranges. The new feature is backward compatible.
+
 ### <a id="query-logs"/>Support for Filtering Query logs on Error</a>
 
 The `querylog-mode` setting can be configured to `error` to log only queries that result in errors. This option is supported in both VTGate and VTTablet.
@@ -147,6 +153,8 @@ The `querylog-mode` setting can be configured to `error` to log only queries tha
 While the flag will continue to accept float values (interpreted as seconds) for backward compatibility, 
 **float inputs are deprecated** and will be removed in a future release.
 
+- `--consolidator-query-waiter-cap` flag to set the maximum number of clients allowed to wait on the consolidator. The default value is set to 0 for unlimited wait. Users can adjust  this value based on the performance of VTTablet to avoid excessive memory usage and the risk of being OOMKilled, particularly in Kubernetes deployments.
+
 ### <a id="topo-read-concurrency-changes"/>`--topo_read_concurrency` behaviour changes
 
 The `--topo_read_concurrency` flag was added to all components that access the topology and the provided limit is now applied separately for each global or local cell _(default `32`)_.

go/flags/endtoend/vtcombo.txt

@@ -44,6 +44,7 @@ Flags:
       --config-path strings                                              Paths to search for config files in. (default [{{ .Workdir }}])
       --config-persistence-min-interval duration                         minimum interval between persisting dynamic config changes back to disk (if no change has occurred, nothing is done). (default 1s)
       --config-type string                                               Config file type (omit to infer config type from file extension).
+      --consolidator-query-waiter-cap int                                Configure the maximum number of clients allowed to wait on the consolidator.
       --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/vtorc.txt

@@ -24,7 +24,7 @@ Flags:
       --bind-address string                                         Bind address for the server. If empty, the server will listen on all available unicast and anycast IP addresses of the local system.
       --catch-sigpipe                                               catch and ignore SIGPIPE on stdout and stderr if specified
       --change-tablets-with-errant-gtid-to-drained                  Whether VTOrc should be changing the type of tablets with errant GTIDs to DRAINED
-      --clusters_to_watch strings                                   Comma-separated list of keyspaces or keyspace/shards that this instance will monitor and repair. Defaults to all clusters in the topology. Example: "ks1,ks2/-80"
+      --clusters_to_watch strings                                   Comma-separated list of keyspaces or keyspace/keyranges that this instance will monitor and repair. Defaults to all clusters in the topology. Example: "ks1,ks2/-80"
       --config-file string                                          Full path of the config file (with extension) to use. If set, --config-path, --config-type, and --config-name are ignored.
       --config-file-not-found-handling ConfigFileNotFoundHandling   Behavior when a config file is not found. (Options: error, exit, ignore, warn) (default warn)
       --config-name string                                          Name of the config file (without extension) to search for. (default "vtconfig")

go/flags/endtoend/vttablet.txt

@@ -78,6 +78,7 @@ Flags:
       --config-path strings                                              Paths to search for config files in. (default [{{ .Workdir }}])
       --config-persistence-min-interval duration                         minimum interval between persisting dynamic config changes back to disk (if no change has occurred, nothing is done). (default 1s)
       --config-type string                                               Config file type (omit to infer config type from file extension).
+      --consolidator-query-waiter-cap int                                Configure the maximum number of clients allowed to wait on the consolidator.
       --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/sync2/consolidator.go

@@ -40,6 +40,7 @@ type PendingResult interface {
 	SetResult(*sqltypes.Result)
 	Result() *sqltypes.Result
 	Wait()
+	AddWaiterCounter(int64) *int64
 }
 
 type consolidator struct {
@@ -77,6 +78,7 @@ func (co *consolidator) Create(query string) (PendingResult, bool) {
 	defer co.mu.Unlock()
 	var r *pendingResult
 	if r, ok := co.queries[query]; ok {
+		r.AddWaiterCounter(1)
 		return r, false
 	}
 	r = &pendingResult{consolidator: co, query: query}
@@ -122,17 +124,23 @@ func (rs *pendingResult) Wait() {
 	rs.executing.RLock()
 }
 
+func (rs *pendingResult) AddWaiterCounter(c int64) *int64 {
+	atomic.AddInt64(rs.consolidator.totalWaiterCount, c)
+	return rs.consolidator.totalWaiterCount
+}
+
 // ConsolidatorCache is a thread-safe object used for counting how often recent
 // queries have been consolidated.
 // It is also used by the txserializer package to count how often transactions
 // have been queued and had to wait because they targeted the same row (range).
 type ConsolidatorCache struct {
 	*cache.LRUCache[*ccount]
+	totalWaiterCount *int64
 }
 
 // NewConsolidatorCache creates a new cache with the given capacity.
 func NewConsolidatorCache(capacity int64) *ConsolidatorCache {
-	return &ConsolidatorCache{cache.NewLRUCache[*ccount](capacity)}
+	return &ConsolidatorCache{cache.NewLRUCache[*ccount](capacity), new(int64)}
 }
 
 // Record increments the count for "query" by 1.

go/sync2/consolidator_test.go

@@ -18,11 +18,42 @@ package sync2
 
 import (
 	"reflect"
+	"sync"
 	"testing"
 
 	"vitess.io/vitess/go/sqltypes"
 )
 
+func TestAddWaiterCount(t *testing.T) {
+	con := NewConsolidator()
+	sql := "select * from SomeTable"
+	pr, _ := con.Create(sql)
+	var wgAdd sync.WaitGroup
+	var wgSub sync.WaitGroup
+
+	var concurrent = 1000
+
+	for i := 0; i < concurrent; i++ {
+		wgAdd.Add(1)
+		wgSub.Add(1)
+		go func() {
+			defer wgAdd.Done()
+			pr.AddWaiterCounter(1)
+		}()
+		go func() {
+			defer wgSub.Done()
+			pr.AddWaiterCounter(-1)
+		}()
+	}
+
+	wgAdd.Wait()
+	wgSub.Wait()
+
+	if *pr.AddWaiterCounter(0) != 0 {
+		t.Fatalf("Expect 0 totalWaiterCount but got: %v", *pr.AddWaiterCounter(0))
+	}
+}
+
 func TestConsolidator(t *testing.T) {
 	con := NewConsolidator()
 	sql := "select * from SomeTable"

go/sync2/fake_consolidator.go

@@ -112,3 +112,8 @@ func (fr *FakePendingResult) SetResult(result *sqltypes.Result) {
 func (fr *FakePendingResult) Wait() {
 	fr.WaitCalls++
 }
+
+// AddWaiterCounter is currently a no-op.
+func (fr *FakePendingResult) AddWaiterCounter(int64) *int64 {
+	return new(int64)
+}

go/vt/key/key.go

@@ -95,6 +95,14 @@ func NewKeyRange(start []byte, end []byte) *topodatapb.KeyRange {
 	return &topodatapb.KeyRange{Start: start, End: end}
 }
 
+// NewCompleteKeyRange returns a complete key range.
+func NewCompleteKeyRange() *topodatapb.KeyRange {
+	return &topodatapb.KeyRange{
+		Start: nil,
+		End:   nil,
+	}
+}
+
 // KeyRangeAdd adds two adjacent KeyRange values (in any order) into a single value. If the values are not adjacent,
 // it returns false.
 func KeyRangeAdd(a, b *topodatapb.KeyRange) (*topodatapb.KeyRange, bool) {