allow innodb_lock_wait_timeout as system variable (vitessio#16574)

Signed-off-by: Harshit Gangal <[email protected]>

Author: harshit-gangal

go/test/endtoend/vtgate/reservedconn/sysvar_test.go

@@ -421,3 +421,74 @@ func checkOltpAndOlapInterchangingTx(t *testing.T, conn *mysql.Conn) {
 	utils.Exec(t, conn, "set workload = oltp")
 	utils.AssertMatches(t, conn, "select id, val1 from test where id = 80", "[[INT64(80) NULL]]")
 }
+
+func TestSysVarTxIsolation(t *testing.T) {
+	conn, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	// will run every check twice to see that the isolation level is set for all the queries in the session and
+
+	// default from mysql
+	utils.AssertMatches(t, conn, "select @@transaction_isolation", `[[VARCHAR("REPEATABLE-READ")]]`)
+	// ensuring it goes to mysql
+	utils.AssertContains(t, conn, "select @@transaction_isolation, connection_id()", `REPEATABLE-READ`)
+	// second run, ensuring it has the same value.
+	utils.AssertContains(t, conn, "select @@transaction_isolation, connection_id()", `REPEATABLE-READ`)
+
+	// setting to different value.
+	utils.Exec(t, conn, "set @@transaction_isolation = 'read-committed'")
+	utils.AssertMatches(t, conn, "select @@transaction_isolation", `[[VARCHAR("READ-COMMITTED")]]`)
+	// ensuring it goes to mysql
+	utils.AssertContains(t, conn, "select @@transaction_isolation, connection_id()", `READ-COMMITTED`)
+	// second run, to ensuring the setting is applied on the session and not just on next query after settings.
+	utils.AssertContains(t, conn, "select @@transaction_isolation, connection_id()", `READ-COMMITTED`)
+
+	// changing setting to different value.
+	utils.Exec(t, conn, "set session transaction isolation level read uncommitted")
+	utils.AssertMatches(t, conn, "select @@transaction_isolation", `[[VARCHAR("READ-UNCOMMITTED")]]`)
+	// ensuring it goes to mysql
+	utils.AssertContains(t, conn, "select @@transaction_isolation, connection_id()", `READ-UNCOMMITTED`)
+	// second run, to ensuring the setting is applied on the session and not just on next query after settings.
+	utils.AssertContains(t, conn, "select @@transaction_isolation, connection_id()", `READ-UNCOMMITTED`)
+
+	// changing setting to different value.
+	utils.Exec(t, conn, "set transaction isolation level serializable")
+	utils.AssertMatches(t, conn, "select @@transaction_isolation", `[[VARCHAR("SERIALIZABLE")]]`)
+	// ensuring it goes to mysql
+	utils.AssertContains(t, conn, "select @@transaction_isolation, connection_id()", `SERIALIZABLE`)
+	// second run, to ensuring the setting is applied on the session and not just on next query after settings.
+	utils.AssertContains(t, conn, "select @@transaction_isolation, connection_id()", `SERIALIZABLE`)
+}
+
+// TestSysVarInnodbWaitTimeout tests the innodb_lock_wait_timeout system variable
+func TestSysVarInnodbWaitTimeout(t *testing.T) {
+	conn, err := mysql.Connect(context.Background(), &vtParams)
+	require.NoError(t, err)
+	defer conn.Close()
+
+	// default from mysql
+	utils.AssertMatches(t, conn, "select @@innodb_lock_wait_timeout", `[[UINT64(20)]]`)
+	utils.AssertMatches(t, conn, "select @@global.innodb_lock_wait_timeout", `[[UINT64(20)]]`)
+	// ensuring it goes to mysql
+	utils.AssertContains(t, conn, "select @@innodb_lock_wait_timeout", `UINT64(20)`)
+	utils.AssertContains(t, conn, "select @@global.innodb_lock_wait_timeout", `UINT64(20)`)
+
+	// setting to different value.
+	utils.Exec(t, conn, "set @@innodb_lock_wait_timeout = 120")
+	utils.AssertMatches(t, conn, "select @@innodb_lock_wait_timeout", `[[INT64(120)]]`)
+	// ensuring it goes to mysql
+	utils.AssertContains(t, conn, "select @@global.innodb_lock_wait_timeout, connection_id()", `UINT64(20)`)
+	utils.AssertContains(t, conn, "select @@innodb_lock_wait_timeout, connection_id()", `INT64(120)`)
+	// second run, to ensuring the setting is applied on the session and not just on next query after settings.
+	utils.AssertContains(t, conn, "select @@innodb_lock_wait_timeout, connection_id()", `INT64(120)`)
+
+	// changing setting to different value.
+	utils.Exec(t, conn, "set @@innodb_lock_wait_timeout = 240")
+	utils.AssertMatches(t, conn, "select @@innodb_lock_wait_timeout", `[[INT64(240)]]`)
+	// ensuring it goes to mysql
+	utils.AssertContains(t, conn, "select @@global.innodb_lock_wait_timeout, connection_id()", `UINT64(20)`)
+	utils.AssertContains(t, conn, "select @@innodb_lock_wait_timeout, connection_id()", `INT64(240)`)
+	// second run, to ensuring the setting is applied on the session and not just on next query after settings.
+	utils.AssertContains(t, conn, "select @@innodb_lock_wait_timeout, connection_id()", `INT64(240)`)
+}

go/vt/sqlparser/ast_rewriting.go

@@ -415,6 +415,76 @@ func (er *astRewriter) rewrite(cursor *Cursor) bool {
 	return true
 }
 
+func (er *astRewriter) rewriteNotExpr(cursor *Cursor, node *NotExpr) {
+	switch inner := node.Expr.(type) {
+	case *ComparisonExpr:
+		// not col = 42 => col != 42
+		// not col > 42 => col <= 42
+		// etc
+		canChange, inverse := inverseOp(inner.Operator)
+		if canChange {
+			inner.Operator = inverse
+			cursor.Replace(inner)
+		}
+	case *NotExpr:
+		// not not true => true
+		cursor.Replace(inner.Expr)
+	case BoolVal:
+		// not true => false
+		inner = !inner
+		cursor.Replace(inner)
+	}
+}
+
+func (er *astRewriter) rewriteVariable(cursor *Cursor, node *Variable) {
+	// Iff we are in SET, we want to change the scope of variables if a modifier has been set
+	// and only on the lhs of the assignment:
+	// set session sql_mode = @someElse
+	// here we need to change the scope of `sql_mode` and not of `@someElse`
+	if v, isSet := cursor.Parent().(*SetExpr); isSet && v.Var == node {
+		return
+	}
+	// no rewriting for global scope variable.
+	// this should be returned from the underlying database.
+	switch node.Scope {
+	case VariableScope:
+		er.udvRewrite(cursor, node)
+	case SessionScope, NextTxScope:
+		er.sysVarRewrite(cursor, node)
+	}
+}
+
+func (er *astRewriter) visitSelect(node *Select) {
+	for _, col := range node.SelectExprs {
+		if _, hasStar := col.(*StarExpr); hasStar {
+			er.hasStarInSelect = true
+			continue
+		}
+
+		aliasedExpr, ok := col.(*AliasedExpr)
+		if !ok || aliasedExpr.As.NotEmpty() {
+			continue
+		}
+		buf := NewTrackedBuffer(nil)
+		aliasedExpr.Expr.Format(buf)
+		// select last_insert_id() -> select :__lastInsertId as `last_insert_id()`
+		innerBindVarNeeds, err := er.rewriteAliasedExpr(aliasedExpr)
+		if err != nil {
+			er.err = err
+			return
+		}
+		if innerBindVarNeeds.HasRewrites() {
+			aliasedExpr.As = NewIdentifierCI(buf.String())
+		}
+		er.bindVars.MergeWith(innerBindVarNeeds)
+
+	}
+	// set select limit if explicitly not set when sql_select_limit is set on the connection.
+	if er.selectLimit > 0 && node.Limit == nil {
+		node.Limit = &Limit{Rowcount: NewIntLiteral(strconv.Itoa(er.selectLimit))}
+	}
+}
+
 func inverseOp(i ComparisonExprOperator) (bool, ComparisonExprOperator) {
 	switch i {
 	case EqualOp:

go/vt/sysvars/sysvars.go

@@ -171,6 +171,7 @@ var (
 		{Name: "foreign_key_checks", IsBoolean: true, SupportSetVar: true},
 		{Name: "group_concat_max_len", SupportSetVar: true},
 		{Name: "information_schema_stats_expiry"},
+		{Name: "innodb_lock_wait_timeout"},
 		{Name: "max_heap_table_size", SupportSetVar: true},
 		{Name: "max_seeks_for_key", SupportSetVar: true},
 		{Name: "max_tmp_tables"},
@@ -226,7 +227,6 @@ var (
 		{Name: "collation_server"},
 		{Name: "completion_type"},
 		{Name: "div_precision_increment", SupportSetVar: true},
-		{Name: "innodb_lock_wait_timeout"},
 		{Name: "interactive_timeout"},
 		{Name: "lc_time_names"},
 		{Name: "lock_wait_timeout", SupportSetVar: true},

go/vt/vtgate/executor_select_test.go

@@ -3782,3 +3782,85 @@ func TestMain(m *testing.M) {
 	_flag.ParseFlagsForTest()
 	os.Exit(m.Run())
 }
+
+func TestStreamJoinQuery(t *testing.T) {
+	ctx := utils.LeakCheckContext(t)
+
+	// Special setup: Don't use createExecutorEnv.
+	cell := "aa"
+	hc := discovery.NewFakeHealthCheck(nil)
+	u := createSandbox(KsTestUnsharded)
+	s := createSandbox(KsTestSharded)
+	s.VSchema = executorVSchema
+	u.VSchema = unshardedVSchema
+	serv := newSandboxForCells(ctx, []string{cell})
+	resolver := newTestResolver(ctx, hc, serv, cell)
+	shards := []string{"-20", "20-40", "40-60", "60-80", "80-a0", "a0-c0", "c0-e0", "e0-"}
+	for _, shard := range shards {
+		_ = hc.AddTestTablet(cell, shard, 1, "TestExecutor", shard, topodatapb.TabletType_PRIMARY, true, 1, nil)
+	}
+	executor := createExecutor(ctx, serv, cell, resolver)
+	defer executor.Close()
+
+	sql := "select u.foo, u.apa, ue.bar, ue.apa from user u join user_extra ue on u.foo = ue.bar"
+	result, err := executorStream(ctx, executor, sql)
+	require.NoError(t, err)
+	wantResult := &sqltypes.Result{
+		Fields: append(sandboxconn.SingleRowResult.Fields, sandboxconn.SingleRowResult.Fields...),
+	}
+	wantRow := append(sandboxconn.StreamRowResult.Rows[0], sandboxconn.StreamRowResult.Rows[0]...)
+	for i := 0; i < 64; i++ {
+		wantResult.Rows = append(wantResult.Rows, wantRow)
+	}
+	require.Equal(t, len(wantResult.Rows), len(result.Rows))
+	for idx := 0; idx < 64; idx++ {
+		utils.MustMatch(t, wantResult.Rows[idx], result.Rows[idx], "mismatched on: ", strconv.Itoa(idx))
+	}
+}
+
+// TestSysVarGlobalAndSession tests that global and session variables are set correctly.
+// It also tests that setting a global variable does not affect the session variable and vice versa.
+// Also, test what happens on running select @@global and select @@session for a system variable.
+func TestSysVarGlobalAndSession(t *testing.T) {
+	executor, sbc1, _, _, _ := createExecutorEnv(t)
+	executor.normalize = true
+	session := NewAutocommitSession(&vtgatepb.Session{EnableSystemSettings: true, SystemVariables: map[string]string{}})
+
+	sbc1.SetResults([]*sqltypes.Result{
+		sqltypes.MakeTestResult(sqltypes.MakeTestFields("innodb_lock_wait_timeout", "uint64"), "20"),
+		sqltypes.MakeTestResult(sqltypes.MakeTestFields("innodb_lock_wait_timeout", "uint64"), "20"),
+		sqltypes.MakeTestResult(sqltypes.MakeTestFields("1", "int64")),
+		sqltypes.MakeTestResult(sqltypes.MakeTestFields("new", "uint64"), "40"),
+		sqltypes.MakeTestResult(sqltypes.MakeTestFields("reserve_execute", "uint64")),
+		sqltypes.MakeTestResult(sqltypes.MakeTestFields("@@global.innodb_lock_wait_timeout", "uint64"), "20"),
+	})
+	qr, err := executor.Execute(context.Background(), nil, "TestSetStmt", session,
+		"select @@innodb_lock_wait_timeout", nil)
+	require.NoError(t, err)
+	require.Equal(t, `[[UINT64(20)]]`, fmt.Sprintf("%v", qr.Rows))
+
+	qr, err = executor.Execute(context.Background(), nil, "TestSetStmt", session,
+		"select @@global.innodb_lock_wait_timeout", nil)
+	require.NoError(t, err)
+	require.Equal(t, `[[UINT64(20)]]`, fmt.Sprintf("%v", qr.Rows))
+
+	_, err = executor.Execute(context.Background(), nil, "TestSetStmt", session,
+		"set @@global.innodb_lock_wait_timeout = 120", nil)
+	require.NoError(t, err)
+	require.Empty(t, session.SystemVariables["innodb_lock_wait_timeout"])
+
+	_, err = executor.Execute(context.Background(), nil, "TestSetStmt", session,
+		"set @@innodb_lock_wait_timeout = 40", nil)
+	require.NoError(t, err)
+	require.EqualValues(t, "40", session.SystemVariables["innodb_lock_wait_timeout"])
+
+	qr, err = executor.Execute(context.Background(), nil, "TestSetStmt", session,
+		"select @@innodb_lock_wait_timeout", nil)
+	require.NoError(t, err)
+	require.Equal(t, `[[INT64(40)]]`, fmt.Sprintf("%v", qr.Rows))
+
+	qr, err = executor.Execute(context.Background(), nil, "TestSetStmt", session,
+		"select @@global.innodb_lock_wait_timeout", nil)
+	require.NoError(t, err)
+	require.Equal(t, `[[UINT64(20)]]`, fmt.Sprintf("%v", qr.Rows))
+}

go/vt/vtgate/executor_set_test.go

@@ -347,6 +347,17 @@ func TestExecutorSetOp(t *testing.T) {
 	}, {
 		in:     "set global client_found_rows = 1",
 		result: returnNoResult("client_found_rows", "int64"),
+	}, {
+		in:      "set tx_isolation = 'read-committed'",
+		sysVars: map[string]string{"tx_isolation": "'read-committed'"},
+		result:  returnResult("tx_isolation", "varchar", "read-committed"),
+	}, {
+		in:      "set @@innodb_lock_wait_timeout=120",
+		sysVars: map[string]string{"innodb_lock_wait_timeout": "120"},
+		result:  returnResult("innodb_lock_wait_timeout", "int64", "120"),
+	}, {
+		in:     "set @@global.innodb_lock_wait_timeout=120",
+		result: returnResult("innodb_lock_wait_timeout", "int64", "120"),
 	}}
 	for _, tcase := range testcases {
 		t.Run(tcase.in, func(t *testing.T) {