Merge remote-tracking branch 'origin/release-15.0' into release-15.0-…

…create_release-1

Signed-off-by: Florent Poinsard <florent.poinsard@outlook.fr>

.github/workflows/upgrade_downgrade_test_backups_e2e_next_release.yml

@@ -79,7 +79,7 @@ jobs:
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
  uses: actions/setup-go@v3
  with:
- go-version: 1.20.5
+ go-version: 1.20.8
 
  - name: Set up python
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'

.github/workflows/upgrade_downgrade_test_backups_manual_next_release.yml

@@ -81,7 +81,7 @@ jobs:
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
  uses: actions/setup-go@v3
  with:
- go-version: 1.20.5
+ go-version: 1.20.8
 
  - name: Set up python
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'

.github/workflows/upgrade_downgrade_test_query_serving_queries_next_release.yml

@@ -81,7 +81,7 @@ jobs:
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
  uses: actions/setup-go@v3
  with:
- go-version: 1.20.5
+ go-version: 1.20.8
 
  - name: Set up python
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'

.github/workflows/upgrade_downgrade_test_query_serving_schema_next_release.yml

@@ -81,7 +81,7 @@ jobs:
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
  uses: actions/setup-go@v3
  with:
- go-version: 1.20.5
+ go-version: 1.20.8
 
  - name: Set up python
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'

.github/workflows/upgrade_downgrade_test_reparent_new_vtctl.yml

@@ -81,7 +81,7 @@ jobs:
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
  uses: actions/setup-go@v3
  with:
- go-version: 1.20.5
+ go-version: 1.20.8
 
  - name: Set up python
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'

.github/workflows/upgrade_downgrade_test_reparent_new_vttablet.yml

@@ -81,7 +81,7 @@ jobs:
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'
  uses: actions/setup-go@v3
  with:
- go-version: 1.20.5
+ go-version: 1.20.8
 
  - name: Set up python
  if: steps.skip-workflow.outputs.skip-workflow == 'false' && steps.changes.outputs.end_to_end == 'true'

docker/k8s/Dockerfile

@@ -51,6 +51,7 @@ COPY --from=base /vt/bin/vtgate /vt/bin/
 COPY --from=base /vt/bin/vttablet /vt/bin/
 COPY --from=base /vt/bin/vtbackup /vt/bin/
 COPY --from=base /vt/bin/vtadmin /vt/bin/
+COPY --from=base /vt/bin/vtorc /vt/bin/
 
 # copy web admin files
 COPY --from=base $VTROOT/web /vt/web/

docker/k8s/vtorc/Dockerfile

@@ -0,0 +1,38 @@
+# Copyright 2019 The Vitess Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+ARG VT_BASE_VER=latest
+ARG DEBIAN_VER=stable-slim
+
+FROM vitess/k8s:${VT_BASE_VER} AS k8s
+
+FROM debian:${DEBIAN_VER}
+
+# Set up Vitess environment (just enough to run pre-built Go binaries)
+ENV VTROOT /vt
+
+# Prepare directory structure.
+RUN mkdir -p /vt/bin && mkdir -p /vtdataroot
+
+# Copy certs to allow https calls
+COPY --from=k8s /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
+
+# Copy binaries
+COPY --from=k8s /vt/bin/vtorc /vt/bin/
+
+# add vitess user/group and add permissions
+RUN groupadd -r --gid 2000 vitess && \
+ useradd -r -g vitess --uid 1000 vitess && \
+ chown -R vitess:vitess /vt && \
+ chown -R vitess:vitess /vtdataroot

docker/release.sh

@@ -21,6 +21,11 @@ do
  docker push vitess/vtadmin:$vt_base_version-$debian_version
  if [[ $debian_version == $default_debian_version ]]; then docker push vitess/vtadmin:$vt_base_version; fi
 
+ docker build --platform linux/amd64 --build-arg VT_BASE_VER=$vt_base_version --build-arg DEBIAN_VER=$debian_version-slim -t vitess/vtorc:$vt_base_version-$debian_version k8s/vtorc
+ docker tag vitess/vtorc:$vt_base_version-$debian_version vitess/vtorc:$vt_base_version
+ docker push vitess/vtorc:$vt_base_version-$debian_version
+ if [[ $debian_version == $default_debian_version ]]; then docker push vitess/vtorc:$vt_base_version; fi
+
  docker build --platform linux/amd64 --build-arg VT_BASE_VER=$vt_base_version --build-arg DEBIAN_VER=$debian_version-slim -t vitess/vtgate:$vt_base_version-$debian_version k8s/vtgate
  docker tag vitess/vtgate:$vt_base_version-$debian_version vitess/vtgate:$vt_base_version
  docker push vitess/vtgate:$vt_base_version-$debian_version

go/vt/vtgate/endtoend/vstream_test.go

@@ -20,6 +20,7 @@ import (
  "context"
  "fmt"
  "io"
+ "regexp"
  "sync"
  "testing"
 
@@ -330,9 +331,9 @@ func TestVStreamSharded(t *testing.T) {
  received bool
  }
  expectedEvents := []*expectedEvent{
- {`type:FIELD field_event:{table_name:"ks.t1_sharded" fields:{name:"id1" type:INT64 table:"t1_sharded" org_table:"t1_sharded" database:"vt_ks_-80" org_name:"id1" column_length:20 charset:63 flags:53251} fields:{name:"id2" type:INT64 table:"t1_sharded" org_table:"t1_sharded" database:"vt_ks_-80" org_name:"id2" column_length:20 charset:63 flags:32768} keyspace:"ks" shard:"-80"}`, false},
+ {`type:FIELD field_event:{table_name:"ks.t1_sharded" fields:{name:"id1" type:INT64 table:"t1_sharded" org_table:"t1_sharded" database:"vt_ks_-80" org_name:"id1" column_length:20 charset:63 flags:53251 column_type:"bigint(20)"} fields:{name:"id2" type:INT64 table:"t1_sharded" org_table:"t1_sharded" database:"vt_ks_-80" org_name:"id2" column_length:20 charset:63 flags:32768 column_type:"bigint(20)"} keyspace:"ks" shard:"-80"}`, false},
  {`type:ROW row_event:{table_name:"ks.t1_sharded" row_changes:{after:{lengths:1 lengths:1 values:"11"}} keyspace:"ks" shard:"-80"}`, false},
- {`type:FIELD field_event:{table_name:"ks.t1_sharded" fields:{name:"id1" type:INT64 table:"t1_sharded" org_table:"t1_sharded" database:"vt_ks_80-" org_name:"id1" column_length:20 charset:63 flags:53251} fields:{name:"id2" type:INT64 table:"t1_sharded" org_table:"t1_sharded" database:"vt_ks_80-" org_name:"id2" column_length:20 charset:63 flags:32768} keyspace:"ks" shard:"80-"}`, false},
+ {`type:FIELD field_event:{table_name:"ks.t1_sharded" fields:{name:"id1" type:INT64 table:"t1_sharded" org_table:"t1_sharded" database:"vt_ks_80-" org_name:"id1" column_length:20 charset:63 flags:53251 column_type:"bigint(20)"} fields:{name:"id2" type:INT64 table:"t1_sharded" org_table:"t1_sharded" database:"vt_ks_80-" org_name:"id2" column_length:20 charset:63 flags:32768 column_type:"bigint(20)"} keyspace:"ks" shard:"80-"}`, false},
  {`type:ROW row_event:{table_name:"ks.t1_sharded" row_changes:{after:{lengths:1 lengths:1 values:"44"}} keyspace:"ks" shard:"80-"}`, false},
  }
  for {
@@ -357,7 +358,7 @@ func TestVStreamSharded(t *testing.T) {
  for _, ev := range evs {
  s := fmt.Sprintf("%v", ev)
  for _, expectedEv := range expectedEvents {
- if expectedEv.ev == s {
+ if removeAnyDeprecatedDisplayWidths(expectedEv.ev) == removeAnyDeprecatedDisplayWidths(s) {
  expectedEv.received = true
  break
  }
@@ -381,6 +382,17 @@ func TestVStreamSharded(t *testing.T) {
 
 }
 
+func removeAnyDeprecatedDisplayWidths(orig string) string {
+ var adjusted string
+ baseIntType := "int"
+ intRE := regexp.MustCompile(`(?i)int\(([0-9]*)?\)`)
+ adjusted = intRE.ReplaceAllString(orig, baseIntType)
+ baseYearType := "year"
+ yearRE := regexp.MustCompile(`(?i)year\(([0-9]*)?\)`)
+ adjusted = yearRE.ReplaceAllString(adjusted, baseYearType)
+ return adjusted
+}
+
 var printMu sync.Mutex
 
 func printEvents(evs []*binlogdatapb.VEvent) {

go/vt/vttablet/endtoend/framework/client.go

@@ -19,6 +19,7 @@ package framework
 import (
  "context"
  "errors"
+ "sync"
  "time"
 
  "google.golang.org/protobuf/proto"
@@ -40,6 +41,7 @@ type QueryClient struct {
  target *querypb.Target
  server *tabletserver.TabletServer
  transactionID int64
+ reservedIDMu sync.Mutex
  reservedID int64
  sessionStateChanges string
 }
@@ -114,6 +116,8 @@ func (client *QueryClient) Commit() error {
 func (client *QueryClient) Rollback() error {
  defer func() { client.transactionID = 0 }()
  rID, err := client.server.Rollback(client.ctx, client.target, client.transactionID)
+ client.reservedIDMu.Lock()
+ defer client.reservedIDMu.Unlock()
  client.reservedID = rID
  if err != nil {
  return err
@@ -293,6 +297,8 @@ func (client *QueryClient) MessageAck(name string, ids []string) (int64, error)
 
 // ReserveExecute performs a ReserveExecute.
 func (client *QueryClient) ReserveExecute(query string, preQueries []string, bindvars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
+ client.reservedIDMu.Lock()
+ defer client.reservedIDMu.Unlock()
  if client.reservedID != 0 {
  return nil, errors.New("already reserved a connection")
  }

go/vt/vttablet/endtoend/transaction_test.go

@@ -321,7 +321,7 @@ func TestShutdownGracePeriod(t *testing.T) {
  err := client.Begin(false)
  require.NoError(t, err)
  go func() {
- _, err = client.Execute("select sleep(10) from dual", nil)
+ _, err := client.Execute("select sleep(10) from dual", nil)
  assert.Error(t, err)
  }()
 
@@ -346,7 +346,7 @@ func TestShutdownGracePeriod(t *testing.T) {
  err = client.Begin(false)
  require.NoError(t, err)
  go func() {
- _, err = client.Execute("select sleep(11) from dual", nil)
+ _, err := client.Execute("select sleep(11) from dual", nil)
  assert.Error(t, err)
  }()
 
@@ -373,7 +373,7 @@ func TestShutdownGracePeriodWithStreamExecute(t *testing.T) {
  err := client.Begin(false)
  require.NoError(t, err)
  go func() {
- _, err = client.StreamExecute("select sleep(10) from dual", nil)
+ _, err := client.StreamExecute("select sleep(10) from dual", nil)
  assert.Error(t, err)
  }()
 
@@ -398,7 +398,7 @@ func TestShutdownGracePeriodWithStreamExecute(t *testing.T) {
  err = client.Begin(false)
  require.NoError(t, err)
  go func() {
- _, err = client.StreamExecute("select sleep(11) from dual", nil)
+ _, err := client.StreamExecute("select sleep(11) from dual", nil)
  assert.Error(t, err)
  }()
 
@@ -425,7 +425,7 @@ func TestShutdownGracePeriodWithReserveExecute(t *testing.T) {
  err := client.Begin(false)
  require.NoError(t, err)
  go func() {
- _, err = client.ReserveExecute("select sleep(10) from dual", nil, nil)
+ _, err := client.ReserveExecute("select sleep(10) from dual", nil, nil)
  assert.Error(t, err)
  }()
 
@@ -450,7 +450,7 @@ func TestShutdownGracePeriodWithReserveExecute(t *testing.T) {
  err = client.Begin(false)
  require.NoError(t, err)
  go func() {
- _, err = client.ReserveExecute("select sleep(11) from dual", nil, nil)
+ _, err := client.ReserveExecute("select sleep(11) from dual", nil, nil)
  assert.Error(t, err)
  }()
 

go/vt/vttablet/tabletserver/schema/engine.go

@@ -593,6 +593,32 @@ func (se *Engine) GetSchema() map[string]*Table {
  return tables
 }
 
+// MarshalMinimalSchema returns a protobuf encoded binlogdata.MinimalSchema
+func (se *Engine) MarshalMinimalSchema() ([]byte, error) {
+ se.mu.Lock()
+ defer se.mu.Unlock()
+ dbSchema := &binlogdatapb.MinimalSchema{
+ Tables: make([]*binlogdatapb.MinimalTable, 0, len(se.tables)),
+ }
+ for _, table := range se.tables {
+ dbSchema.Tables = append(dbSchema.Tables, newMinimalTable(table))
+ }
+ return dbSchema.MarshalVT()
+}
+
+func newMinimalTable(st *Table) *binlogdatapb.MinimalTable {
+ table := &binlogdatapb.MinimalTable{
+ Name: st.Name.String(),
+ Fields: st.Fields,
+ }
+ pkc := make([]int64, len(st.PKColumns))
+ for i, pk := range st.PKColumns {
+ pkc[i] = int64(pk)
+ }
+ table.PKColumns = pkc
+ return table
+}
+
 // GetConnection returns a connection from the pool
 func (se *Engine) GetConnection(ctx context.Context) (*connpool.DBConn, error) {
  return se.conns.Get(ctx, nil)

go/vt/vttablet/tabletserver/schema/tracker.go

@@ -23,8 +23,6 @@ import (
  "sync"
  "time"
 
- "google.golang.org/protobuf/proto"
-
  "vitess.io/vitess/go/vt/schema"
 
  "vitess.io/vitess/go/mysql"
@@ -240,14 +238,10 @@ func (tr *Tracker) schemaUpdated(gtid string, ddl string, timestamp int64) error
 }
 
 func (tr *Tracker) saveCurrentSchemaToDb(ctx context.Context, gtid, ddl string, timestamp int64) error {
- tables := tr.engine.GetSchema()
- dbSchema := &binlogdatapb.MinimalSchema{
- Tables: []*binlogdatapb.MinimalTable{},
- }
- for _, table := range tables {
- dbSchema.Tables = append(dbSchema.Tables, newMinimalTable(table))
+ blob, err := tr.engine.MarshalMinimalSchema()
+ if err != nil {
+ return err
  }
- blob, _ := proto.Marshal(dbSchema)
 
  conn, err := tr.engine.GetConnection(ctx)
  if err != nil {
@@ -265,19 +259,6 @@ func (tr *Tracker) saveCurrentSchemaToDb(ctx context.Context, gtid, ddl string,
  return nil
 }
 
-func newMinimalTable(st *Table) *binlogdatapb.MinimalTable {
- table := &binlogdatapb.MinimalTable{
- Name: st.Name.String(),
- Fields: st.Fields,
- }
- var pkc []int64
- for _, pk := range st.PKColumns {
- pkc = append(pkc, int64(pk))
- }
- table.PKColumns = pkc
- return table
-}
-
 func encodeString(in string) string {
  buf := bytes.NewBuffer(nil)
  sqltypes.NewVarChar(in).EncodeSQL(buf)

go/vt/vttablet/tabletserver/throttle/client.go

@@ -50,7 +50,8 @@ type Client struct {
  checkType ThrottleCheckType
  flags CheckFlags
 
- lastSuccessfulThrottle int64
+ lastSuccessfulThrottleMu sync.Mutex
+ lastSuccessfulThrottle int64
 }
 
 // NewProductionClient creates a client suitable for foreground/production jobs, which have normal priority.
@@ -94,6 +95,8 @@ func (c *Client) ThrottleCheckOK(ctx context.Context, overrideAppName string) (t
  // no throttler
  return true
  }
+ c.lastSuccessfulThrottleMu.Lock()
+ defer c.lastSuccessfulThrottleMu.Unlock()
  if c.lastSuccessfulThrottle >= atomic.LoadInt64(&throttleTicks) {
  // if last check was OK just very recently there is no need to check again
  return true

go/vt/vttablet/tabletserver/vstreamer/rowstreamer.go

@@ -153,9 +153,14 @@ func (rs *rowStreamer) buildPlan() error {
  return err
  }
  ti := &Table{
- Name: st.Name,
- Fields: st.Fields,
+ Name: st.Name,
  }
+
+ ti.Fields, err = getFields(rs.ctx, rs.cp, st.Name, rs.cp.DBName(), st.Fields)
+ if err != nil {
+ return err
+ }
+
  // The plan we build is identical to the one for vstreamer.
  // This is because the row format of a read is identical
  // to the row format of a binlog event. So, the same