Schema Tracking Refactor: Merge schema-tracking in health-streamer into schema.Engine

changelog/17.0/17.0.0/summary.md

@@ -321,6 +321,9 @@ Affected flags and YAML config keys:
 - `shutdown_grace_period`
 - `unhealthy_threshold`
 
+The flag `queryserver-config-schema-change-signal-interval` is deprecated and will be removed in a later release.
+Schema-tracking has been refactored in this release to not use polling anymore, therefore the signal interval isn't required anymore.
+
 ### Online DDL
 
 #### <a id="online-ddl-cut-over-threshold-flag" /> --cut-over-threshold DDL strategy flag

go/flags/endtoend/vttablet.txt

@@ -226,7 +226,6 @@ Usage of vttablet:
       --queryserver-config-query-pool-waiter-cap int                     query server query pool waiter limit, this is the maximum number of queries that can be queued waiting to get a connection (default 5000)
       --queryserver-config-query-timeout duration                        query server query timeout (in seconds), this is the query timeout in vttablet side. If a query takes more than this timeout, it will be killed. (default 30s)
       --queryserver-config-schema-change-signal                          query server schema signal, will signal connected vtgates that schema has changed whenever this is detected. VTGates will need to have -schema_change_signal enabled for this to work (default true)
-      --queryserver-config-schema-change-signal-interval duration        query server schema change signal interval defines at which interval the query server shall send schema updates to vtgate. (default 5s)
       --queryserver-config-schema-reload-time duration                   query server schema reload time, how often vttablet reloads schemas from underlying MySQL instance in seconds. vttablet keeps table schemas in its own memory and periodically refreshes it from MySQL. This config controls the reload time. (default 30m0s)
       --queryserver-config-stream-buffer-size int                        query server stream buffer size, the maximum number of bytes sent from vttablet for each stream call. It's recommended to keep this value in sync with vtgate's stream_buffer_size. (default 32768)
       --queryserver-config-stream-pool-size int                          query server stream connection pool size, stream pool is used by stream queries: queries that return results to client in a streaming fashion (default 200)

go/test/endtoend/cluster/cluster_process.go

@@ -929,6 +929,46 @@ func (cluster *LocalProcessCluster) StreamTabletHealth(ctx context.Context, vtta
 	return responses, nil
 }
 
+// StreamTabletHealthUntil invokes a HealthStream on a local cluster Vttablet and
+// returns the responses. It waits until a certain condition is met. The amount of time to wait is an input that it takes.
+func (cluster *LocalProcessCluster) StreamTabletHealthUntil(ctx context.Context, vttablet *Vttablet, timeout time.Duration, condition func(shr *querypb.StreamHealthResponse) bool) error {
+	tablet, err := cluster.VtctlclientGetTablet(vttablet)
+	if err != nil {
+		return err
+	}
+
+	conn, err := tabletconn.GetDialer()(tablet, grpcclient.FailFast(false))
+	if err != nil {
+		return err
+	}
+
+	conditionSuccess := false
+	timeoutExceeded := false
+	go func() {
+		time.Sleep(timeout)
+		timeoutExceeded = true
+	}()
+
+	err = conn.StreamHealth(ctx, func(shr *querypb.StreamHealthResponse) error {
+		if condition(shr) {
+			conditionSuccess = true
+		}
+		if timeoutExceeded || conditionSuccess {
+			return io.EOF
+		}
+		return nil
+	})
+
+	if conditionSuccess {
+		return nil
+	}
+
+	if timeoutExceeded {
+		return errors.New("timeout exceed while waiting for the condition in StreamHealth")
+	}
+	return err
+}
+
 func (cluster *LocalProcessCluster) VtctlclientGetTablet(tablet *Vttablet) (*topodatapb.Tablet, error) {
 	result, err := cluster.VtctlclientProcess.ExecuteCommandWithOutput("GetTablet", "--", tablet.Alias)
 	if err != nil {

go/test/endtoend/cluster/vtctldclient_process.go

@@ -93,6 +93,18 @@ func VtctldClientProcessInstance(hostname string, grpcPort int, tmpDirectory str
 	return vtctldclient
 }
 
+// PlannedReparentShard executes vtctlclient command to make specified tablet the primary for the shard.
+func (vtctldclient *VtctldClientProcess) PlannedReparentShard(Keyspace string, Shard string, alias string) (err error) {
+	output, err := vtctldclient.ExecuteCommandWithOutput(
+		"PlannedReparentShard",
+		fmt.Sprintf("%s/%s", Keyspace, Shard),
+		"--new-primary", alias)
+	if err != nil {
+		log.Errorf("error in PlannedReparentShard output %s, err %s", output, err.Error())
+	}
+	return err
+}
+
 // CreateKeyspace executes the vtctl command to create a keyspace
 func (vtctldclient *VtctldClientProcess) CreateKeyspace(keyspaceName string, sidecarDBName string) (err error) {
 	var output string

go/test/endtoend/tabletmanager/main_test.go

@@ -95,6 +95,7 @@ func TestMain(m *testing.M) {
 		// List of users authorized to execute vschema ddl operations
 		clusterInstance.VtGateExtraArgs = []string{
 			"--vschema_ddl_authorized_users=%",
+			"--enable-views",
 			"--discovery_low_replication_lag", tabletUnhealthyThreshold.String(),
 		}
 		// Set extra tablet args for lock timeout

go/test/endtoend/tabletmanager/tablet_health_test.go

@@ -27,12 +27,12 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"k8s.io/utils/strings/slices"
 
 	"vitess.io/vitess/go/json2"
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/test/endtoend/cluster"
 	"vitess.io/vitess/go/test/endtoend/utils"
-
 	querypb "vitess.io/vitess/go/vt/proto/query"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 )
@@ -195,6 +195,136 @@ func TestHealthCheck(t *testing.T) {
 	killTablets(t, rTablet)
 }
 
+// TestHealthCheckSchemaChangeSignal tests the tables and views, which report their schemas have changed in the output of a StreamHealth.
+func TestHealthCheckSchemaChangeSignal(t *testing.T) {
+	// Add one replica that starts not initialized
+	defer cluster.PanicHandler(t)
+	ctx := context.Background()
+
+	vtParams := clusterInstance.GetVTParams(keyspaceName)
+	conn, err := mysql.Connect(ctx, &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	// Make sure the primary is the primary when the test starts.
+	// This state should be ensured before we actually test anything.
+	checkTabletType(t, primaryTablet.Alias, "PRIMARY")
+
+	// Run a bunch of DDL queries and verify that the tables/views changed show up in the health stream.
+	// These tests are for the part where `--queryserver-enable-views` flag is not set.
+	verifyHealthStreamSchemaChangeSignals(t, conn, &primaryTablet, false)
+
+	// We start a new vttablet, this time with `--queryserver-enable-views` flag specified.
+	tempTablet := clusterInstance.NewVttabletInstance("replica", 0, "")
+	// Start Mysql Processes and return connection
+	_, err = cluster.StartMySQLAndGetConnection(ctx, tempTablet, username, clusterInstance.TmpDirectory)
+	require.NoError(t, err)
+	oldArgs := clusterInstance.VtTabletExtraArgs
+	clusterInstance.VtTabletExtraArgs = append(clusterInstance.VtTabletExtraArgs, "--queryserver-enable-views")
+	defer func() {
+		clusterInstance.VtTabletExtraArgs = oldArgs
+	}()
+	// start vttablet process, should be in SERVING state as we already have a primary.
+	err = clusterInstance.StartVttablet(tempTablet, "SERVING", false, cell, keyspaceName, hostname, shardName)
+	require.NoError(t, err)
+
+	defer func() {
+		// Restore the primary tablet back to the original.
+		err = clusterInstance.VtctldClientProcess.PlannedReparentShard(keyspaceName, shardName, primaryTablet.Alias)
+		require.NoError(t, err)
+		// Manual cleanup of processes
+		killTablets(t, tempTablet)
+	}()
+
+	// Now we reparent the cluster to the new tablet we have.
+	err = clusterInstance.VtctldClientProcess.PlannedReparentShard(keyspaceName, shardName, tempTablet.Alias)
+	require.NoError(t, err)
+
+	checkTabletType(t, tempTablet.Alias, "PRIMARY")
+	// Run a bunch of DDL queries and verify that the tables/views changed show up in the health stream.
+	// These tests are for the part where `--queryserver-enable-views` flag is set.
+	verifyHealthStreamSchemaChangeSignals(t, conn, tempTablet, true)
+}
+
+func verifyHealthStreamSchemaChangeSignals(t *testing.T, vtgateConn *mysql.Conn, primaryTablet *cluster.Vttablet, viewsEnabled bool) {
+	verifyTableDDLSchemaChangeSignal(t, vtgateConn, primaryTablet, "CREATE TABLE `area` (`id` int NOT NULL, `country` varchar(30), PRIMARY KEY (`id`))", "area")
+	verifyTableDDLSchemaChangeSignal(t, vtgateConn, primaryTablet, "CREATE TABLE `area2` (`id` int NOT NULL, PRIMARY KEY (`id`))", "area2")
+	verifyViewDDLSchemaChangeSignal(t, vtgateConn, primaryTablet, "CREATE VIEW v2 as select * from area", viewsEnabled)
+	verifyTableDDLSchemaChangeSignal(t, vtgateConn, primaryTablet, "ALTER TABLE `area` ADD COLUMN name varchar(30) NOT NULL", "area")
+	verifyTableDDLSchemaChangeSignal(t, vtgateConn, primaryTablet, "DROP TABLE `area2`", "area2")
+	verifyViewDDLSchemaChangeSignal(t, vtgateConn, primaryTablet, "ALTER VIEW v2 as select id from area", viewsEnabled)
+	verifyViewDDLSchemaChangeSignal(t, vtgateConn, primaryTablet, "DROP VIEW v2", viewsEnabled)
+	verifyTableDDLSchemaChangeSignal(t, vtgateConn, primaryTablet, "DROP TABLE `area`", "area")
+}
+
+func verifyTableDDLSchemaChangeSignal(t *testing.T, vtgateConn *mysql.Conn, primaryTablet *cluster.Vttablet, query string, table string) {
+	ctx := context.Background()
+	var streamErr error
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	ranOnce := false
+	go func() {
+		defer wg.Done()
+		streamErr = clusterInstance.StreamTabletHealthUntil(ctx, primaryTablet, 30*time.Second, func(shr *querypb.StreamHealthResponse) bool {
+			ranOnce = true
+			if shr != nil && shr.RealtimeStats != nil && slices.Contains(shr.RealtimeStats.TableSchemaChanged, table) {
+				return true
+			}
+			return false
+		})
+	}()
+	// The test becomes flaky if we run the DDL immediately after starting the above go routine because the client for the Stream
+	// sometimes isn't registered by the time DDL runs, and it misses the update we get. To prevent this situation, we wait for one Stream packet
+	// to have returned. Once we know we received a Stream packet, then we know that we are registered for the health stream and can execute the DDL.
+	for i := 0; i < 30; i++ {
+		if ranOnce {
+			break
+		}
+		time.Sleep(1 * time.Second)
+	}
+	require.True(t, ranOnce, "We should have received atleast 1 health stream")
+	_, err := vtgateConn.ExecuteFetch(query, 10000, false)
+	require.NoError(t, err)
+	wg.Wait()
+	require.NoError(t, streamErr)
+}
+
+func verifyViewDDLSchemaChangeSignal(t *testing.T, vtgateConn *mysql.Conn, primaryTablet *cluster.Vttablet, query string, viewsEnabled bool) {
+	ctx := context.Background()
+	var streamErr error
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	ranOnce := false
+	go func() {
+		defer wg.Done()
+		streamErr = clusterInstance.StreamTabletHealthUntil(ctx, primaryTablet, 30*time.Second, func(shr *querypb.StreamHealthResponse) bool {
+			ranOnce = true
+			listToUse := shr.RealtimeStats.TableSchemaChanged
+			if viewsEnabled {
+				listToUse = shr.RealtimeStats.ViewSchemaChanged
+			}
+			if shr != nil && shr.RealtimeStats != nil && slices.Contains(listToUse, "v2") {
+				return true
+			}
+			return false
+		})
+	}()
+	// The test becomes flaky if we run the DDL immediately after starting the above go routine because the client for the Stream
+	// sometimes isn't registered by the time DDL runs, and it misses the update we get. To prevent this situation, we wait for one Stream packet
+	// to have returned. Once we know we received a Stream packet, then we know that we are registered for the health stream and can execute the DDL.
+	for i := 0; i < 30; i++ {
+		if ranOnce {
+			break
+		}
+		time.Sleep(1 * time.Second)
+	}
+	require.True(t, ranOnce, "We should have received atleast 1 health stream")
+	_, err := vtgateConn.ExecuteFetch(query, 10000, false)
+	require.NoError(t, err)
+	wg.Wait()
+	require.NoError(t, streamErr)
+}
+
 func checkHealth(t *testing.T, port int, shouldError bool) {
 	url := fmt.Sprintf("http://localhost:%d/healthz", port)
 	resp, err := http.Get(url)

go/test/fuzzing/tabletserver_schema_fuzzer.go

@@ -72,5 +72,5 @@ func newTestLoadTable(tableName, comment string, db *fakesqldb.DB) (*schema.Tabl
 	}
 	defer conn.Recycle()
 
-	return schema.LoadTable(conn, "fakesqldb", tableName, comment)
+	return schema.LoadTable(conn, "fakesqldb", tableName, "BASE_TABLE", comment)
 }

go/vt/sidecardb/schema/schemaengine/schema_engine_tables.sql

@@ -0,0 +1,24 @@
+/*
+Copyright 2023 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.
+*/
+
+CREATE TABLE IF NOT EXISTS schema_engine_tables
+(
+    TABLE_SCHEMA varchar(64) NOT NULL,
+    TABLE_NAME varchar(64) NOT NULL,
+    CREATE_STATEMENT longtext,
+    CREATE_TIME BIGINT,
+    PRIMARY KEY (TABLE_SCHEMA, TABLE_NAME)
+) engine = InnoDB

go/vt/sidecardb/schema/schemaengine/schema_engine_views.sql

@@ -0,0 +1,24 @@
+/*
+Copyright 2023 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.
+*/
+
+CREATE TABLE IF NOT EXISTS schema_engine_views
+(
+    VIEW_SCHEMA varchar(64) NOT NULL,
+    VIEW_NAME varchar(64) NOT NULL,
+    CREATE_STATEMENT longtext,
+    VIEW_DEFINITION longtext NOT NULL,
+    PRIMARY KEY (VIEW_SCHEMA, VIEW_NAME)
+) engine = InnoDB

go/vt/vttablet/endtoend/framework/server.go

@@ -119,7 +119,6 @@ func StartServer(connParams, connAppDebugParams mysql.ConnParams, dbName string)
 	config.HotRowProtection.Mode = tabletenv.Enable
 	config.TrackSchemaVersions = true
 	_ = config.GracePeriods.ShutdownSeconds.Set("2s")
-	_ = config.SignalSchemaChangeReloadIntervalSeconds.Set("2100ms")
 	config.SignalWhenSchemaChange = true
 	_ = config.Healthcheck.IntervalSeconds.Set("100ms")
 	_ = config.Oltp.TxTimeoutSeconds.Set("5s")

go/vt/vttablet/tabletserver/binlog_watcher.go

@@ -17,11 +17,10 @@ limitations under the License.
 package tabletserver
 
 import (
+	"context"
 	"sync"
 	"time"
 
-	"context"
-
 	"vitess.io/vitess/go/vt/log"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"