VStream: Prevent buffering entire transactions (OOM risk), instead se…

go/test/endtoend/vreplication/initial_data_test.go

@@ -20,8 +20,10 @@ import (
 	"fmt"
 	"math/rand"
 	"os"
+	"strings"
 	"testing"
 
+	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/vt/log"
 )
 
@@ -43,6 +45,12 @@ func insertInitialData(t *testing.T) {
 			`[[VARCHAR("Monoprice") VARCHAR("eléctronics")] [VARCHAR("newegg") VARCHAR("elec†ronics")]]`)
 
 		insertJSONValues(t)
+
+		insertLargeTransactionForChunkTesting(t, vtgateConn, defaultSourceKs+":0", 50000)
+		log.Infof("Inserted large transaction for chunking tests")
+
+		execVtgateQuery(t, vtgateConn, defaultSourceKs, "delete from customer where cid >= 50000 and cid < 50100")
+		log.Infof("Cleaned up chunk testing rows from source keyspace")
 	})
 }
 
@@ -140,3 +148,15 @@ func insertIntoBlobTable(t *testing.T) {
 		execVtgateQuery(t, vtgateConn, "product:0", query)
 	}
 }
+
+// insertLargeTransactionForChunkTesting inserts a transaction large enough to exceed the 1KB chunking threshold.
+func insertLargeTransactionForChunkTesting(t *testing.T, vtgateConn *mysql.Conn, keyspace string, startID int) {
+	execVtgateQuery(t, vtgateConn, keyspace, "BEGIN")
+	for i := 0; i < 15; i++ {
+		largeData := strings.Repeat("x", 94) + fmt.Sprintf("_%05d", i)
+		query := fmt.Sprintf("INSERT INTO customer (cid, name) VALUES (%d, '%s')",
+			startID+i, largeData)
+		execVtgateQuery(t, vtgateConn, keyspace, query)
+	}
+	execVtgateQuery(t, vtgateConn, keyspace, "COMMIT")
+}

go/test/endtoend/vreplication/vreplication_test.go

@@ -55,11 +55,17 @@ var (
 	defaultReplicas        int
 	sourceKsOpts           = make(map[string]string)
 	targetKsOpts           = make(map[string]string)
+	defaultTargetKsOpts    = make(map[string]string)
 	httpClient             = throttlebase.SetupHTTPClient(time.Second)
 	sourceThrottlerAppName = throttlerapp.VStreamerName
 	targetThrottlerAppName = throttlerapp.VReplicationName
 )
 
+const (
+	defaultSourceKs = "product"
+	defaultTargetKs = "customer"
+)
+
 const (
 	// for some tests we keep an open transaction during a Switch writes and commit it afterwards, to reproduce https://github.com/vitessio/vitess/issues/9400
 	// we also then delete the extra row (if) added so that the row counts for the future count comparisons stay the same