Merge remote-tracking branch 'upstream/main' into mhamza/pool-reuse

Signed-off-by: Mohamed Hamza <[email protected]>

Author: mhamza15

Makefile

@@ -286,7 +286,7 @@ $(PROTO_GO_OUTS): minimaltools install_protoc-gen-go proto/*.proto
 # This rule builds the bootstrap images for all flavors.
 DOCKER_IMAGES_FOR_TEST = mysql80 mysql84 percona80
 DOCKER_IMAGES = common $(DOCKER_IMAGES_FOR_TEST)
-BOOTSTRAP_VERSION=44
+BOOTSTRAP_VERSION=45
 ensure_bootstrap_version:
 	find docker/ -type f -exec sed -i "s/^\(ARG bootstrap_version\)=.*/\1=${BOOTSTRAP_VERSION}/" {} \;
 	sed -i 's/\(^.*flag.String(\"bootstrap-version\",\) *\"[^\"]\+\"/\1 \"${BOOTSTRAP_VERSION}\"/' test.go

build.env

@@ -17,7 +17,7 @@
 source ./tools/shell_functions.inc
 
 go version >/dev/null 2>&1 || fail "Go is not installed or is not in \$PATH. See https://vitess.io/contributing/build-from-source for install instructions."
-goversion_min 1.24.2 || echo "Go version reported: `go version`. Version 1.24.2+ recommended. See https://vitess.io/contributing/build-from-source for install instructions."
+goversion_min 1.24.3 || echo "Go version reported: `go version`. Version 1.24.3+ recommended. See https://vitess.io/contributing/build-from-source for install instructions."
 
 mkdir -p dist
 mkdir -p bin

docker/bootstrap/CHANGELOG.md

@@ -174,4 +174,8 @@ List of changes between bootstrap image versions.
 
 ## [44] - 2025-04-02
 ### Changes
-- Update build to golang 1.24.2
+- Update build to golang 1.24.2
+
+## [45] - 2025-05-07
+### Changes
+- Update build to golang 1.24.3

docker/bootstrap/Dockerfile.common

@@ -1,4 +1,4 @@
-FROM --platform=linux/amd64 golang:1.24.2-bookworm
+FROM --platform=linux/amd64 golang:1.24.3-bookworm
 
 # Install Vitess build dependencies
 RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \

docker/lite/Dockerfile

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM --platform=linux/amd64 golang:1.24.2-bookworm AS builder
+FROM --platform=linux/amd64 golang:1.24.3-bookworm AS builder
 
 # Allows docker builds to set the BUILD_NUMBER
 ARG BUILD_NUMBER

docker/lite/Dockerfile.mysql84

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM --platform=linux/amd64 golang:1.24.2-bookworm AS builder
+FROM --platform=linux/amd64 golang:1.24.3-bookworm AS builder
 
 # Allows docker builds to set the BUILD_NUMBER
 ARG BUILD_NUMBER

docker/lite/Dockerfile.percona80

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM --platform=linux/amd64 golang:1.24.2-bookworm AS builder
+FROM --platform=linux/amd64 golang:1.24.3-bookworm AS builder
 
 # Allows docker builds to set the BUILD_NUMBER
 ARG BUILD_NUMBER

docker/vttestserver/Dockerfile.mysql80

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM --platform=linux/amd64 golang:1.24.2-bookworm AS builder
+FROM --platform=linux/amd64 golang:1.24.3-bookworm AS builder
 
 # Allows docker builds to set the BUILD_NUMBER
 ARG BUILD_NUMBER

docker/vttestserver/Dockerfile.mysql84

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM --platform=linux/amd64 golang:1.24.2-bookworm AS builder
+FROM --platform=linux/amd64 golang:1.24.3-bookworm AS builder
 
 # Allows docker builds to set the BUILD_NUMBER
 ARG BUILD_NUMBER

go.mod

@@ -1,6 +1,6 @@
 module vitess.io/vitess
 
-go 1.24.2
+go 1.24.3
 
 require (
 	cloud.google.com/go/storage v1.52.0

go/test/endtoend/vreplication/vstream_test.go

@@ -26,8 +26,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"vitess.io/vitess/go/sets"
 	"vitess.io/vitess/go/vt/log"
 	_ "vitess.io/vitess/go/vt/vtgate/grpcvtgateconn"
 	"vitess.io/vitess/go/vt/vtgate/vtgateconn"
@@ -37,6 +39,166 @@ import (
 	vtgatepb "vitess.io/vitess/go/vt/proto/vtgate"
 )
 
+func TestVStreamWithTablesToSkipCopyFlag(t *testing.T) {
+	vc = NewVitessCluster(t, nil)
+	defer vc.TearDown()
+
+	require.NotNil(t, vc)
+	defaultReplicas = 2
+	defaultRdonly = 0
+
+	defaultCell := vc.Cells[vc.CellNames[0]]
+	vc.AddKeyspace(t, []*Cell{defaultCell}, "product", "0", initialProductVSchema, initialProductSchema, defaultReplicas, defaultRdonly, 100, nil)
+	verifyClusterHealth(t, vc)
+
+	ctx := context.Background()
+	vstreamConn, err := vtgateconn.Dial(ctx, fmt.Sprintf("%s:%d", vc.ClusterConfig.hostname, vc.ClusterConfig.vtgateGrpcPort))
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer vstreamConn.Close()
+	vgtid := &binlogdatapb.VGtid{
+		ShardGtids: []*binlogdatapb.ShardGtid{{
+			Keyspace: "product",
+			Shard:    "0",
+			Gtid:     "",
+		}}}
+
+	filter := &binlogdatapb.Filter{
+		Rules: []*binlogdatapb.Rule{
+			{
+				Match:  "customer",
+				Filter: "select * from customer",
+			}, {
+				Match:  "product",
+				Filter: "select * from product",
+			}, {
+				Match:  "merchant",
+				Filter: "select * from merchant",
+			},
+		},
+	}
+	flags := &vtgatepb.VStreamFlags{
+		TablesToCopy: []string{"product", "customer"},
+	}
+	id := 0
+	vtgateConn := vc.GetVTGateConn(t)
+	defer vtgateConn.Close()
+
+	// To test the copy phase, let's insert 10 rows intitally in each table
+	// present in the filter before running the VStream.
+	for range 10 {
+		id++
+		execVtgateQuery(t, vtgateConn, "product", fmt.Sprintf("insert into customer (cid, name) values (%d, 'customer%d')", id+100, id))
+		execVtgateQuery(t, vtgateConn, "product", fmt.Sprintf("insert into product (pid, description) values (%d, 'description%d')", id+100, id))
+		execVtgateQuery(t, vtgateConn, "product", fmt.Sprintf("insert into merchant (mname, category) values ('mname%d', 'category%d')", id+100, id))
+	}
+
+	// Stream events from the VStream API
+	reader, err := vstreamConn.VStream(ctx, topodatapb.TabletType_PRIMARY, vgtid, filter, flags)
+	require.NoError(t, err)
+	var numRowEvents int64
+
+	copyPhaseCompleted := atomic.Bool{}
+	copyPhaseCompleted.Store(false)
+
+	done := atomic.Bool{}
+	done.Store(false)
+
+	copiedTables := make(sets.Set[string])
+	// Start reading events from the VStream.
+	go func() {
+		for {
+			evs, err := reader.Recv()
+			switch err {
+			case nil:
+				for _, ev := range evs {
+					if ev.Type == binlogdatapb.VEventType_ROW {
+						if !copyPhaseCompleted.Load() {
+							escapedTableNameParts := strings.Split(ev.RowEvent.TableName, ".")
+							require.Len(t, escapedTableNameParts, 2)
+							copiedTables.Insert(escapedTableNameParts[1])
+						}
+						numRowEvents++
+					}
+					if ev.Type == binlogdatapb.VEventType_COPY_COMPLETED {
+						copyPhaseCompleted.Store(true)
+					}
+				}
+			case io.EOF:
+				log.Infof("Stream Ended")
+			default:
+				log.Infof("%s:: remote error: %v", time.Now(), err)
+			}
+
+			if done.Load() {
+				return
+			}
+		}
+	}()
+
+	// Wait for copy phase to complete.
+	ticker := time.NewTicker(100 * time.Millisecond)
+	for {
+		<-ticker.C
+		if copyPhaseCompleted.Load() {
+			break
+		}
+	}
+
+	stopInserting := atomic.Bool{}
+	stopInserting.Store(false)
+	var insertMu sync.Mutex
+	go func() {
+		for {
+			if stopInserting.Load() {
+				return
+			}
+			insertMu.Lock()
+			id++
+			execVtgateQuery(t, vtgateConn, "product", fmt.Sprintf("insert into customer (cid, name) values (%d, 'customer%d')", id+100, id))
+			execVtgateQuery(t, vtgateConn, "product", fmt.Sprintf("insert into product (pid, description) values (%d, 'description%d')", id+100, id))
+			execVtgateQuery(t, vtgateConn, "product", fmt.Sprintf("insert into merchant (mname, category) values ('mname%d', 'category%d')", id+100, id))
+			insertMu.Unlock()
+		}
+	}()
+
+	time.Sleep(100 * time.Millisecond)
+	stopInserting.Store(true)
+	time.Sleep(10 * time.Second) // Give the vstream plenty of time to catchup
+	done.Store(true)
+
+	qr1 := execVtgateQuery(t, vtgateConn, "product", "select count(*) from customer")
+	qr2 := execVtgateQuery(t, vtgateConn, "product", "select count(*) from product")
+	qr3 := execVtgateQuery(t, vtgateConn, "product", "select count(*) from merchant")
+	require.NotNil(t, qr1)
+	require.NotNil(t, qr2)
+	require.NotNil(t, qr3)
+
+	// Total number of rows.
+	insertedRows1, err := qr1.Rows[0][0].ToCastInt64()
+	require.NoError(t, err)
+	require.NotZero(t, insertedRows1)
+	insertedRows2, err := qr2.Rows[0][0].ToCastInt64()
+	require.NoError(t, err)
+	require.NotZero(t, insertedRows2)
+	insertedRows3, err := qr3.Rows[0][0].ToCastInt64()
+	require.NoError(t, err)
+	require.NotZero(t, insertedRows3)
+
+	assert.Len(t, copiedTables, 2)
+	for _, expectedCopiedTableName := range flags.TablesToCopy {
+		assert.Truef(t, copiedTables.Has(expectedCopiedTableName), "expected table %s to be copied", expectedCopiedTableName)
+	}
+	// We don't expect merchant table to be part of copy phase.
+	assert.False(t, copiedTables.Has("merchant"), "expected table merchant not to be copied")
+
+	// Since we don't expect merchant table to be part of copy phase, we can
+	// subtract 10 from the total rows found in the 3 tables.
+	wantTotalRows := insertedRows1 + insertedRows2 + insertedRows3 - 10
+	assert.Equal(t, wantTotalRows, numRowEvents)
+}
+
 // Validates that we have a working VStream API
 // If Failover is enabled:
 //   - We ensure that this works through active reparents and doesn't miss any events