OnlineDDL: Improve Safety of CREATE/DROP TABLE phases (#19242)

Signed-off-by: Matt Lord <mattalord@gmail.com>

Author: mattlord

go/vt/vttablet/onlineddl/executor.go

@@ -48,10 +48,6 @@ import (
 	"vitess.io/vitess/go/vt/binlog/binlogplayer"
 	"vitess.io/vitess/go/vt/dbconnpool"
 	"vitess.io/vitess/go/vt/log"
-	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
-	querypb "vitess.io/vitess/go/vt/proto/query"
-	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
-	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/schema"
 	"vitess.io/vitess/go/vt/schemadiff"
 	"vitess.io/vitess/go/vt/servenv"
@@ -66,6 +62,11 @@ import (
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/throttle"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/throttle/throttlerapp"
 	"vitess.io/vitess/go/vt/vttablet/tmclient"
+
+	binlogdatapb "vitess.io/vitess/go/vt/proto/binlogdata"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 )
 
 var (
@@ -79,12 +80,6 @@ var (
 
 var staleMigrationMinutesStats = stats.NewGauge("OnlineDDLStaleMigrationMinutes", "longest stale migration in minutes")
 
-// fixCompletedTimestampDone fixes a nil `completed_timestamp` columns, see
-// https://github.com/vitessio/vitess/issues/13927
-// The fix is in release-18.0
-// TODO: remove in release-19.0
-var fixCompletedTimestampDone bool
-
 var (
 	emptyResult                           = &sqltypes.Result{}
 	acceptableDropTableIfExistsErrorCodes = []sqlerror.ErrorCode{sqlerror.ERCantFindFile, sqlerror.ERNoSuchTable}
@@ -118,7 +113,6 @@ func registerOnlineDDLFlags(fs *pflag.FlagSet) {
 }
 
 const (
-	maxPasswordLength                        = 32 // MySQL's *replication* password may not exceed 32 characters
 	staleMigrationFailMinutes                = 180
 	staleMigrationWarningMinutes             = 5
 	progressPctStarted               float64 = 0
@@ -468,7 +462,8 @@ func (e *Executor) getCreateTableStatement(ctx context.Context, tableName string
 	return createTable, nil
 }
 
-// executeDirectly runs a DDL query directly on the backend MySQL server
+// executeDirectly runs a DDL query directly on the backend MySQL server.
+// This is primarily used for CREATE/DROP/RENAME TABLE statements.
 func (e *Executor) executeDirectly(ctx context.Context, onlineDDL *schema.OnlineDDL, acceptableMySQLErrorCodes ...sqlerror.ErrorCode) (acceptableErrorCodeFound bool, err error) {
 	conn, err := dbconnpool.NewDBConnection(ctx, e.env.Config().DB.DbaWithDB())
 	if err != nil {
@@ -498,6 +493,11 @@ func (e *Executor) executeDirectly(ctx context.Context, onlineDDL *schema.Online
 		}
 		defer conn.ExecuteFetch("SET foreign_key_checks=@vt_onlineddl_foreign_key_checks", 0, false)
 	}
+	restoreLockWaitTimeout, err := e.initDBConnectionLockWaitTimeout(conn, onlineDDL.CutOverThreshold)
+	if err != nil {
+		return false, vterrors.Wrap(err, "failed to set lock_wait_timeout on direct connection")
+	}
+	defer restoreLockWaitTimeout()
 	_, err = conn.ExecuteFetch(onlineDDL.SQL, 0, false)
 	if err != nil {
 		// let's see if this error is actually acceptable
@@ -1228,6 +1228,24 @@ func (e *Executor) initConnectionLockWaitTimeout(ctx context.Context, conn *conn
 	return deferFunc, nil
 }
 
+// initDBConnectionLockWaitTimeout sets the given lock_wait_timeout for the given direct connection, with a deferred value restoration function.
+func (e *Executor) initDBConnectionLockWaitTimeout(conn *dbconnpool.DBConnection, lockWaitTimeout time.Duration) (deferFunc func(), err error) {
+	deferFunc = func() {}
+
+	if _, err := conn.ExecuteFetch("set @lock_wait_timeout=@@session.lock_wait_timeout", 0, false); err != nil {
+		return deferFunc, vterrors.Wrap(err, "could not read lock_wait_timeout")
+	}
+	timeoutSeconds := int64(lockWaitTimeout.Seconds())
+	setQuery := fmt.Sprintf("set @@session.lock_wait_timeout=%d", timeoutSeconds)
+	if _, err := conn.ExecuteFetch(setQuery, 0, false); err != nil {
+		return deferFunc, err
+	}
+	deferFunc = func() {
+		conn.ExecuteFetch("set @@session.lock_wait_timeout=@lock_wait_timeout", 0, false)
+	}
+	return deferFunc, nil
+}
+
 // createDuplicateTableLike creates the table named by `newTableName` in the likeness of onlineDDL.Table
 // This function emulates MySQL's `CREATE TABLE LIKE ...` statement. The difference is that this function takes control over the generated CONSTRAINT names,
 // if any, such that they are deterministic across shards, as well as preserve original names where possible.
@@ -3532,17 +3550,6 @@ func (e *Executor) gcArtifacts(ctx context.Context) error {
 	e.migrationMutex.Lock()
 	defer e.migrationMutex.Unlock()
 
-	// v18 fix. Remove in v19
-	if !fixCompletedTimestampDone {
-		if _, err := e.execQuery(ctx, sqlFixCompletedTimestamp); err != nil {
-			// This query fixes a bug where stale migrations were marked as 'cancelled' or 'failed' without updating 'completed_timestamp'
-			// Running this query retroactively sets completed_timestamp
-			// This fix is created in v18 and can be removed in v19
-			return err
-		}
-		fixCompletedTimestampDone = true
-	}
-
 	query, err := sqlparser.ParseAndBind(sqlSelectUncollectedArtifacts,
 		sqltypes.Int64BindVariable(int64((retainOnlineDDLTables).Seconds())),
 	)

go/vt/vttablet/onlineddl/executor_test.go

@@ -21,13 +21,26 @@ Functionality of this Executor is tested in go/test/endtoend/onlineddl/...
 package onlineddl
 
 import (
+	"context"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"vitess.io/vitess/go/mysql/fakesqldb"
+	"vitess.io/vitess/go/sqltypes"
+	"vitess.io/vitess/go/timer"
+	"vitess.io/vitess/go/vt/dbconfigs"
+	"vitess.io/vitess/go/vt/dbconnpool"
 	"vitess.io/vitess/go/vt/schema"
+	"vitess.io/vitess/go/vt/topo/memorytopo"
+	"vitess.io/vitess/go/vt/vtenv"
+	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
+	"vitess.io/vitess/go/vt/vttablet/tmclient"
+	"vitess.io/vitess/go/vt/vttablet/tmclienttest"
+
+	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 )
 
 func TestShouldCutOverAccordingToBackoff(t *testing.T) {
@@ -249,3 +262,90 @@ func TestGetInOrderCompletionPendingCount(t *testing.T) {
 		require.Equal(t, uint64(3), getInOrderCompletionPendingCount(onlineDDL, pendingMigrationsUUIDs))
 	}
 }
+
+func TestInitDBConnectionLockWaitTimeout(t *testing.T) {
+	db := fakesqldb.New(t)
+	defer db.Close()
+	params := db.ConnParams()
+	connector := dbconfigs.NewTestDBConfigs(*params, *params, params.DbName).DbaWithDB()
+	conn, err := dbconnpool.NewDBConnection(context.Background(), connector)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	db.AddQuery("set @lock_wait_timeout=@@session.lock_wait_timeout", &sqltypes.Result{})
+	db.AddQuery("set @@session.lock_wait_timeout=5", &sqltypes.Result{})
+	db.AddQuery("set @@session.lock_wait_timeout=@lock_wait_timeout", &sqltypes.Result{})
+
+	executor := &Executor{}
+	deferFunc, err := executor.initDBConnectionLockWaitTimeout(conn, 5*time.Second)
+	require.NoError(t, err)
+	queryLog := db.QueryLog()
+	assert.Contains(t, queryLog, "set @lock_wait_timeout=@@session.lock_wait_timeout")
+	assert.Contains(t, queryLog, "set @@session.lock_wait_timeout=5")
+
+	deferFunc()
+	assert.Contains(t, db.QueryLog(), "set @@session.lock_wait_timeout=@lock_wait_timeout")
+}
+
+func TestExecuteDirectlySetsLockWaitTimeout(t *testing.T) {
+	ctx := t.Context()
+	db := fakesqldb.New(t)
+	defer db.Close()
+	params := db.ConnParams()
+	connector := dbconfigs.NewTestDBConfigs(*params, *params, params.DbName).DbaWithDB()
+	conn, err := dbconnpool.NewDBConnection(ctx, connector)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	db.AddQuery("set @lock_wait_timeout=@@session.lock_wait_timeout", &sqltypes.Result{})
+	db.AddQuery("set @@session.lock_wait_timeout=5", &sqltypes.Result{})
+	db.AddQuery("set @@session.lock_wait_timeout=@lock_wait_timeout", &sqltypes.Result{})
+	db.AddQuery("create table test_lock_wait(id int)", &sqltypes.Result{})
+
+	venv := vtenv.NewTestEnv()
+	cfg := tabletenv.NewDefaultConfig()
+	cfg.DB = dbconfigs.NewTestDBConfigs(*params, *params, params.DbName)
+	protocolName := t.Name()
+	resetProtocol := tmclienttest.SetProtocol(t.Name(), protocolName)
+	defer resetProtocol()
+	tmclient.RegisterTabletManagerClientFactory(protocolName, func() tmclient.TabletManagerClient {
+		return &fakeTabletManagerClient{}
+	})
+	alias := &topodatapb.TabletAlias{Cell: "cell", Uid: 1}
+	ts := memorytopo.NewServer(ctx, "cell")
+	err = ts.CreateTablet(ctx, &topodatapb.Tablet{
+		Alias:    alias,
+		Keyspace: "ks",
+		Shard:    "0",
+		Type:     topodatapb.TabletType_PRIMARY,
+	})
+	require.NoError(t, err)
+	executor := &Executor{
+		env:         tabletenv.NewEnv(venv, cfg, "ExecutorTest"),
+		ts:          ts,
+		tabletAlias: alias,
+		execQuery: func(ctx context.Context, query string) (*sqltypes.Result, error) {
+			return &sqltypes.Result{}, nil
+		},
+		ticks: timer.NewTimer(migrationCheckInterval),
+	}
+
+	onlineDDL := &schema.OnlineDDL{SQL: "create table test_lock_wait(id int)", CutOverThreshold: 5 * time.Second, UUID: "uuid"}
+	_, err = executor.executeDirectly(ctx, onlineDDL)
+	require.NoError(t, err)
+
+	queryLog := db.QueryLog()
+	assert.Contains(t, queryLog, "set @lock_wait_timeout=@@session.lock_wait_timeout")
+	assert.Contains(t, queryLog, "set @@session.lock_wait_timeout=5")
+	assert.Contains(t, queryLog, "set @@session.lock_wait_timeout=@lock_wait_timeout")
+}
+
+type fakeTabletManagerClient struct {
+	tmclient.TabletManagerClient
+}
+
+func (fakeTabletManagerClient) Close() {}
+
+func (fakeTabletManagerClient) ReloadSchema(ctx context.Context, tablet *topodatapb.Tablet, waitPosition string) error {
+	return nil
+}

go/vt/vttablet/tabletserver/gc/tablegc.go

@@ -18,8 +18,11 @@ package gc
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"maps"
 	"sort"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -29,6 +32,7 @@ import (
 	"vitess.io/vitess/go/mysql/capabilities"
 	"vitess.io/vitess/go/mysql/sqlerror"
 
+	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/timer"
 	"vitess.io/vitess/go/vt/dbconnpool"
 	"vitess.io/vitess/go/vt/log"
@@ -37,6 +41,7 @@ import (
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/topo"
 	"vitess.io/vitess/go/vt/utils"
+	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/connpool"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/tabletenv"
 	"vitess.io/vitess/go/vt/vttablet/tabletserver/throttle"
@@ -127,6 +132,11 @@ type TableGC struct {
 	lifecycleStates map[schema.TableGCState]bool
 }
 
+type capabilityConn interface {
+	ExecuteFetch(query string, maxrows int, wantfields bool) (*sqltypes.Result, error)
+	SupportsCapability(capabilities.FlavorCapability) (bool, error)
+}
+
 // Status published some status values from the collector
 type Status struct {
 	Keyspace string
@@ -191,19 +201,11 @@ func (collector *TableGC) Open() (err error) {
 		return err
 	}
 	defer conn.Close()
-	serverSupportsFastDrops, err := conn.SupportsCapability(capabilities.FastDropTableFlavorCapability)
+	collector.lifecycleStates, err = adjustLifecycleForFastDrops(conn, collector.lifecycleStates)
 	if err != nil {
 		return err
 	}
-	if serverSupportsFastDrops {
-		// MySQL 8.0.23 and onwards supports fast DROP TABLE operations. This means we don't have to
-		// go through the purging & evac cycle: once the table has been held for long enough, we can just
-		// move on to dropping it. Dropping a large table in 8.0.23 is expected to take several seconds, and
-		// should not block other queries or place any locks on the buffer pool.
-		delete(collector.lifecycleStates, schema.PurgeTableGCState)
-		delete(collector.lifecycleStates, schema.EvacTableGCState)
-	}
-	log.Infof("TableGC: MySQL version=%v, serverSupportsFastDrops=%v, lifecycleStates=%v", conn.ServerVersion, serverSupportsFastDrops, collector.lifecycleStates)
+	log.Infof("TableGC: MySQL version=%v, lifecycleStates=%v", conn.ServerVersion, collector.lifecycleStates)
 
 	ctx := context.Background()
 	ctx, collector.cancelOperation = context.WithCancel(ctx)
@@ -212,6 +214,44 @@ func (collector *TableGC) Open() (err error) {
 	return nil
 }
 
+// adjustLifecycleForFastDrops adjusts the default lifecycle phases because MySQL 8.0.23 and onwards
+// supports fast DROP TABLE operations. This means we don't have to go through the purge & evac
+// cycles: once the table has been held for long enough, we can just move on to dropping it. Dropping
+// a large table in 8.0.23 is expected to take several seconds, and should not block other queries
+// or place any locks on the buffer pool.
+func adjustLifecycleForFastDrops(conn capabilityConn, lifecycleStates map[schema.TableGCState]bool) (map[schema.TableGCState]bool, error) {
+	if len(lifecycleStates) == 0 {
+		return lifecycleStates, nil
+	}
+
+	serverSupportsFastDrops, err := conn.SupportsCapability(capabilities.FastDropTableFlavorCapability)
+	if err != nil {
+		return lifecycleStates, err
+	}
+	if !serverSupportsFastDrops {
+		return lifecycleStates, nil
+	}
+
+	// Unfortunately you can still encounter problems if the Adaptive Hash Indexes are enabled: https://bugs.mysql.com/bug.php?id=113312
+	// So if AHI is enabled, we cannot safely skip PURGE and EVAC even on MySQL versions that support fast DROP TABLE.
+	res, err := conn.ExecuteFetch("SELECT variable_value FROM performance_schema.global_variables WHERE variable_name = 'innodb_adaptive_hash_index'", 1, false)
+	if err != nil {
+		return nil, vterrors.Wrap(err, "failed to check innodb_adaptive_hash_index setting")
+	}
+	if res == nil || len(res.Rows) == 0 || len(res.Rows[0]) == 0 {
+		return nil, errors.New("failed to check innodb_adaptive_hash_index setting: unexpected result")
+	}
+	if strings.ToLower(res.Rows[0][0].ToString()) == "on" {
+		return lifecycleStates, nil
+	}
+
+	updated := make(map[schema.TableGCState]bool, len(lifecycleStates))
+	maps.Copy(updated, lifecycleStates)
+	delete(updated, schema.PurgeTableGCState)
+	delete(updated, schema.EvacTableGCState)
+	return updated, nil
+}
+
 // Close frees resources
 func (collector *TableGC) Close() {
 	log.Infof("TableGC - started execution of Close. Acquiring initMutex lock")