Repository: vitessio/vitess
Pull request #10739 (2022-07-19 09:22:08)

ApplySchema: do not ReloadSchema on ExecuteFetchAsDba

## Description

Followup to #10719 
There is yet another location where `ReloadSchema` is called: from within the VTTablet, and this is controlled as a parameter in  `ExecuteFetchAsDba`. We now strictly do not call `ReloadSchema` anywhere until all SQLs have been applied.

This specific PR applies to `direct` DDLs, not to Online DDLs.

## Related Issue(s)

#10719 

## Checklist

-   [ ] "Backport me!" label has been added if this change should be backported
-   [x] Tests were added or are not required
-   [ ] Documentation was added or is not required

## Deployment Notes

<!-- Notes regarding deployment of the contained body of work. These should note any db migrations, etc. -->


vitess-bot[bot]: ### Review Checklist
            
Hello reviewers! :wave: Please follow this checklist when reviewing this Pull Request.

#### General
- [ ] Ensure that the Pull Request has a descriptive title.
- [ ] If this is a change that users need to know about, please apply the `release notes (needs details)` label so that merging is blocked unless the summary release notes document is included.
- [ ] If a new flag is being introduced, review whether it is really needed. The flag names should be clear and intuitive (as far as possible), and the flag's help should be descriptive. Additionally, flag names should use dashes (`-`) as word separators rather than underscores (`_`).
- [ ] If a workflow is added or modified, each items in `Jobs` should be named in order to mark it as `required`. If the workflow should be required, the GitHub Admin should be notified.

#### Bug fixes
- [ ] There should be at least one unit or end-to-end test.
- [ ] The Pull Request description should either include a link to an issue that describes the bug OR an actual description of the bug and how to reproduce, along with a description of the fix.

#### Non-trivial changes
- [ ] There should be some code comments as to why things are implemented the way they are.

#### New/Existing features
- [ ] Should be documented, either by modifying the existing documentation or creating new documentation.
- [ ] New features should have a link to a feature request issue or an RFC that documents the use cases, corner cases and test cases.

#### Backward compatibility            
- [ ] Protobuf changes should be wire-compatible.
- [ ] Changes to `_vt` tables and RPCs need to be backward compatible.
- [ ] `vtctl` command output order should be stable and `awk`-able.

shlomi-noach: `Cluster (12)` consistent failure.

```
2022-07-19T10:36:39.1386134Z === RUN   TestSharded
2022-07-19T10:36:39.1563691Z     message_test.go:511: 
2022-07-19T10:36:39.1565147Z         	Error Trace:	message_test.go:511
2022-07-19T10:36:39.1566356Z         	            				message_test.go:368
2022-07-19T10:36:39.1567617Z         	Error:      	Expected nil, but got: Code: INVALID_ARGUMENT
2022-07-19T10:36:39.1570579Z         	            	rpc error: code = InvalidArgument desc = target: user.-80.primary: vttablet: rpc error: code = InvalidArgument desc = table sharded_message not found in schema (CallerID: unsecure_grpc_client)
2022-07-19T10:36:39.1573523Z         	            	target: user.80-.primary: vttablet: rpc error: code = InvalidArgument desc = table sharded_message not found in schema (CallerID: unsecure_grpc_client)
2022-07-19T10:36:39.1575588Z         	Test:       	TestSharded
2022-07-19T10:36:39.1576858Z --- FAIL: TestSharded (0.02s)
2022-07-19T10:36:39.1578162Z === RUN   TestUnsharded
2022-07-19T10:36:52.1909586Z --- PASS: TestUnsharded (13.03s)
2022-07-19T10:36:52.1911359Z === RUN   TestReparenting
2022-07-19T10:36:52.1913145Z I0719 10:36:52.190102   31342 message_test.go:600] Message stream ended: Code: CANCELED
2022-07-19T10:36:52.1915318Z rpc error: code = Canceled desc = grpc: the client connection is closing
2022-07-19T10:36:52.1943090Z     message_test.go:388: 
2022-07-19T10:36:52.1944030Z         	Error Trace:	message_test.go:388
2022-07-19T10:36:52.1945143Z         	Error:      	Expected nil, but got: Code: INVALID_ARGUMENT
2022-07-19T10:36:52.1948260Z         	            	rpc error: code = InvalidArgument desc = target: user.-80.primary: vttablet: rpc error: code = InvalidArgument desc = table sharded_message not found in schema (CallerID: unsecure_grpc_client)
2022-07-19T10:36:52.1951154Z         	            	target: user.80-.primary: vttablet: rpc error: code = InvalidArgument desc = table sharded_message not found in schema (CallerID: unsecure_grpc_client)
2022-07-19T10:36:52.1952601Z         	Test:       	TestReparenting
2022-07-19T10:36:52.1953736Z --- FAIL: TestReparenting (0.00s)
2022-07-19T10:36:52.1954919Z === RUN   TestConnection
2022-07-19T10:36:53.2056838Z     message_test.go:455: 
2022-07-19T10:36:53.2058851Z         	Error Trace:	message_test.go:455
2022-07-19T10:36:53.2060450Z         	Error:      	Expected nil, but got: Code: INVALID_ARGUMENT
2022-07-19T10:36:53.2064123Z         	            	rpc error: code = InvalidArgument desc = target: user.-80.primary: vttablet: rpc error: code = InvalidArgument desc = table sharded_message not found in schema (CallerID: unsecure_grpc_client)
2022-07-19T10:36:53.2068323Z         	            	target: user.80-.primary: vttablet: rpc error: code = InvalidArgument desc = table sharded_message not found in schema (CallerID: unsecure_grpc_client)
2022-07-19T10:36:53.2070517Z         	Test:       	TestConnection
2022-07-19T10:36:53.2072065Z --- FAIL: TestConnection (1.01s)
2022-07-19T10:36:53.2073130Z FAIL
2022-07-19T10:37:05.6304128Z FAIL	vitess.io/vitess/go/test/endtoend/messaging	52.892s
2022-07-19T10:37:05.6330069Z FAIL
```
seems related.
shlomi-noach: I'm lost with the test failures.
shlomi-noach: `vtgate_queries`, `vtgate_reservedconn` keep failing though I dont see how they're related. `vstream*` tests seem related but I can't make the connection.
shlomi-noach: OK I get it!
```
				schematools.ReloadShard(
					reloadCtx,
					exec.ts,
					exec.tmc,
					exec.logger,
					exec.keyspace,
					result.Shard,
					result.Position,
					concurrency,
					false, /* includePrimary */
```
says `false, /* includePrimary */`; we really do want to reload the schema on the primary though!
shlomi-noach: OK, tests still failing, and they're consistent, and they're persistent, and it has to do with `ApplySchema` and therefore also have to do with `ReloadSchema()`.
shlomi-noach: e.g. https://github.com/vitessio/vitess/runs/7413675616?check_suite_focus=true
shlomi-noach: Thanks @dbussink for helping to find the root cause, which was reusing an iteration argument's address.

Patch:
diff --git a/.golangci.yml b/.golangci.yml
index 155a64a451..f57fb88dea 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -31,6 +31,7 @@ linters:
     # Extras
     - gofmt
     - goimports
+    - exportloopref
 
     # revive is a replacement for golint, but we do not run it in CI for now.
     # This is only enabled as a post-commit hook
diff --git a/go/timer/rate_limiter.go b/go/timer/rate_limiter.go
index d9369f9aef..25bc2b32f6 100644
--- a/go/timer/rate_limiter.go
+++ b/go/timer/rate_limiter.go
@@ -42,10 +42,10 @@ func NewRateLimiter(d time.Duration) *RateLimiter {
 	r.cancel = cancel
 	go func() {
 		ticker := time.NewTicker(d)
+		defer ticker.Stop()
 		for {
 			select {
 			case <-ctx.Done():
-				ticker.Stop()
 				return
 			case <-ticker.C:
 				atomic.StoreInt64(&r.tickerValue, r.tickerValue+1)
diff --git a/go/vt/schemamanager/tablet_executor.go b/go/vt/schemamanager/tablet_executor.go
index aee9241d21..417412ed99 100644
--- a/go/vt/schemamanager/tablet_executor.go
+++ b/go/vt/schemamanager/tablet_executor.go
@@ -363,35 +363,22 @@ func (exec *TabletExecutor) Execute(ctx context.Context, sqls []string) *Execute
 	defer rl.Stop()
 
 	syncOperationExecuted := false
-	for index, sql := range sqls {
-		// Attempt to renew lease:
-		if err := rl.Do(func() error { return topo.CheckKeyspaceLockedAndRenew(ctx, exec.keyspace) }); err != nil {
-			execResult.ExecutorErr = vterrors.Wrapf(err, "CheckKeyspaceLocked in ApplySchemaKeyspace %v", exec.keyspace).Error()
-			return &execResult
-		}
-		execResult.CurSQLIndex = index
-		if exec.hasProvidedUUIDs() {
-			providedUUID = exec.uuids[index]
-		}
-		executedAsynchronously, err := exec.executeSQL(ctx, sql, providedUUID, &execResult)
-		if err != nil {
-			execResult.ExecutorErr = err.Error()
-			return &execResult
-		}
-		if !executedAsynchronously {
-			syncOperationExecuted = true
-		}
-		if len(execResult.FailedShards) > 0 {
-			break
-		}
-	}
 
-	if syncOperationExecuted {
+	// ReloadSchema once. Do it even if we do an early return on error
+	defer func() {
+		if !syncOperationExecuted {
+			exec.logger.Infof("Skipped ReloadSchema since all SQLs executed asynchronously")
+			return
+		}
 		// same shards will appear multiple times in execResult.SuccessShards when there are
 		// multiple SQLs
 		uniqueShards := map[string]*ShardResult{}
-		for _, result := range execResult.SuccessShards {
-			uniqueShards[result.Shard] = &result
+		for i := range execResult.SuccessShards {
+			// Please do not change the above iteration to "for result := range ...".
+			// This is because we want to end up grabbing a pointer to the result. But golang's "for"
+			// implementation reuses the iteration parameter, and we end up reusing the same pointer.
+			result := &execResult.SuccessShards[i]
+			uniqueShards[result.Shard] = result
 		}
 		var wg sync.WaitGroup
 		// If all shards succeeded, wait (up to waitReplicasTimeout) for replicas to
@@ -419,9 +406,31 @@ func (exec *TabletExecutor) Execute(ctx context.Context, sqls []string) *Execute
 			}(result)
 		}
 		wg.Wait()
-	} else {
-		exec.logger.Infof("Skipped ReloadSchema since all SQLs executed asynchronously")
+	}()
+
+	for index, sql := range sqls {
+		// Attempt to renew lease:
+		if err := rl.Do(func() error { return topo.CheckKeyspaceLockedAndRenew(ctx, exec.keyspace) }); err != nil {
+			execResult.ExecutorErr = vterrors.Wrapf(err, "CheckKeyspaceLocked in ApplySchemaKeyspace %v", exec.keyspace).Error()
+			return &execResult
+		}
+		execResult.CurSQLIndex = index
+		if exec.hasProvidedUUIDs() {
+			providedUUID = exec.uuids[index]
+		}
+		executedAsynchronously, err := exec.executeSQL(ctx, sql, providedUUID, &execResult)
+		if err != nil {
+			execResult.ExecutorErr = err.Error()
+			return &execResult
+		}
+		if !executedAsynchronously {
+			syncOperationExecuted = true
+		}
+		if len(execResult.FailedShards) > 0 {
+			break
+		}
 	}
+
 	return &execResult
 }
 
@@ -482,7 +491,7 @@ func (exec *TabletExecutor) executeOneTablet(
 				sql = sqlparser.String(ddlStmt)
 			}
 		}
-		result, err = exec.tmc.ExecuteFetchAsDba(ctx, tablet, false, []byte(sql), 10, false, true)
+		result, err = exec.tmc.ExecuteFetchAsDba(ctx, tablet, false, []byte(sql), 10, false, false /* do not ReloadSchema */)
 	}
 	if err != nil {
 		errChan <- ShardWithError{Shard: tablet.Shard, Err: err.Error()}
diff --git a/go/vt/vtgate/planbuilder/physical/route.go b/go/vt/vtgate/planbuilder/physical/route.go
index 24b87caf4f..871324b042 100644
--- a/go/vt/vtgate/planbuilder/physical/route.go
+++ b/go/vt/vtgate/planbuilder/physical/route.go
@@ -187,7 +187,7 @@ func (r *Route) searchForNewVindexes(ctx *plancontext.PlanningContext, predicate
 }
 
 func (r *Route) planComparison(ctx *plancontext.PlanningContext, cmp *sqlparser.ComparisonExpr) (found bool, exitEarly bool, err error) {
-	if sqlparser.IsNull(cmp.Left) || sqlparser.IsNull(cmp.Right) {
+	if cmp.Operator != sqlparser.NullSafeEqualOp && (sqlparser.IsNull(cmp.Left) || sqlparser.IsNull(cmp.Right)) {
 		// we are looking at ANDed predicates in the WHERE clause.
 		// since we know that nothing returns true when compared to NULL,
 		// so we can safely bail out here
diff --git a/go/vt/wrangler/traffic_switcher.go b/go/vt/wrangler/traffic_switcher.go
index 32f0bfdd83..03267eb9cc 100644
--- a/go/vt/wrangler/traffic_switcher.go
+++ b/go/vt/wrangler/traffic_switcher.go
@@ -57,9 +57,6 @@ const (
 	renameTableTemplate = "_%.59s_old" // limit table name to 64 characters
 
 	sqlDeleteWorkflow = "delete from _vt.vreplication where db_name = %s and workflow = %s"
-	sqlDeleteVDiffs   = `delete from vd, vdt, vdl using _vt.vdiff as vd inner join _vt.vdiff_table as vdt on (vd.id = vdt.vdiff_id)
-						inner join _vt.vdiff_log as vdl on (vd.id = vdl.vdiff_id)
-						where vd.keyspace = %s and vd.workflow = %s`
 )
 
 // accessType specifies the type of access for a shard (allow/disallow writes).
@@ -1246,10 +1243,7 @@ func (ts *trafficSwitcher) deleteReverseVReplication(ctx context.Context) error
 		if _, err := ts.TabletManagerClient().VReplicationExec(ctx, source.GetPrimary().Tablet, query); err != nil {
 			return err
 		}
-		query = fmt.Sprintf(sqlDeleteVDiffs, encodeString(source.GetPrimary().Keyspace), encodeString(ts.reverseWorkflow))
-		if _, err := ts.wr.tmc.ExecuteFetchAsDba(ctx, source.GetPrimary().Tablet, false, []byte(query), -1, false, false); err != nil {
-			ts.Logger().Infof("Error deleting vdiff data for %s.%s workflow: %v", source.GetPrimary().Keyspace, ts.reverseWorkflow, err)
-		}
+		ts.wr.deleteWorkflowVDiffData(ctx, source.GetPrimary().Tablet, ts.reverseWorkflow)
 		return nil
 	})
 }
@@ -1572,10 +1566,7 @@ func (ts *trafficSwitcher) dropTargetVReplicationStreams(ctx context.Context) er
 		if _, err := ts.TabletManagerClient().VReplicationExec(ctx, target.GetPrimary().Tablet, query); err != nil {
 			return err
 		}
-		query = fmt.Sprintf(sqlDeleteVDiffs, encodeString(target.GetPrimary().Keyspace), encodeString(ts.WorkflowName()))
-		if _, err := ts.wr.tmc.ExecuteFetchAsDba(ctx, target.GetPrimary().Tablet, false, []byte(query), -1, false, false); err != nil {
-			ts.Logger().Infof("Error deleting vdiff data for %s.%s workflow: %v", target.GetPrimary().Keyspace, ts.WorkflowName(), err)
-		}
+		ts.wr.deleteWorkflowVDiffData(ctx, target.GetPrimary().Tablet, ts.WorkflowName())
 		return nil
 	})
 }
@@ -1587,10 +1578,7 @@ func (ts *trafficSwitcher) dropSourceReverseVReplicationStreams(ctx context.Cont
 		if _, err := ts.TabletManagerClient().VReplicationExec(ctx, source.GetPrimary().Tablet, query); err != nil {
 			return err
 		}
-		query = fmt.Sprintf(sqlDeleteVDiffs, encodeString(source.GetPrimary().Keyspace), encodeString(workflow.ReverseWorkflowName(ts.WorkflowName())))
-		if _, err := ts.wr.tmc.ExecuteFetchAsDba(ctx, source.GetPrimary().Tablet, false, []byte(query), -1, false, false); err != nil {
-			ts.Logger().Infof("Error deleting vdiff data for %s.%s workflow: %v", source.GetPrimary().Keyspace, workflow.ReverseWorkflowName(ts.WorkflowName()), err)
-		}
+		ts.wr.deleteWorkflowVDiffData(ctx, source.GetPrimary().Tablet, workflow.ReverseWorkflowName(ts.WorkflowName()))
 		return nil
 	})
 }
diff --git a/go/vt/wrangler/vexec.go b/go/vt/wrangler/vexec.go
index 4b036d2d34..9709f8b5d6 100644
--- a/go/vt/wrangler/vexec.go
+++ b/go/vt/wrangler/vexec.go
@@ -216,10 +216,7 @@ func (vx *vexec) exec() (map[*topo.TabletInfo]*querypb.QueryResult, error) {
 				// If we deleted a workflow then let's make a best effort attempt to clean
 				// up any related data.
 				if vx.query == sqlVReplicationDelete {
-					query := fmt.Sprintf(sqlDeleteVDiffs, encodeString(primary.Keyspace), encodeString(vx.workflow))
-					if _, err := vx.wr.tmc.ExecuteFetchAsDba(ctx, primary.Tablet, false, []byte(query), -1, false, false); err != nil {
-						vx.wr.Logger().Infof("Error deleting vdiff data for %s.%s workflow: %v", primary.Keyspace, vx.workflow, err)
-					}
+					vx.wr.deleteWorkflowVDiffData(ctx, primary.Tablet, vx.workflow)
 				}
 				mu.Lock()
 				results[primary] = qr
diff --git a/go/vt/wrangler/workflow.go b/go/vt/wrangler/workflow.go
index 6c3c8d8152..c1f8f6e022 100644
--- a/go/vt/wrangler/workflow.go
+++ b/go/vt/wrangler/workflow.go
@@ -8,6 +8,7 @@ import (
 	"sync"
 	"time"
 
+	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/sqltypes"
 	"vitess.io/vitess/go/vt/discovery"
 	"vitess.io/vitess/go/vt/log"
@@ -668,3 +669,20 @@ func (vrw *VReplicationWorkflow) GetCopyProgress() (*CopyProgress, error) {
 }
 
 // endregion
+
+// region Workflow related utility functions
+
+// deleteWorkflowVDiffData cleans up any potential VDiff related data associated with the workflow on the given tablet
+func (wr *Wrangler) deleteWorkflowVDiffData(ctx context.Context, tablet *topodatapb.Tablet, workflow string) {
+	sqlDeleteVDiffs := `delete from vd, vdt, vdl using _vt.vdiff as vd inner join _vt.vdiff_table as vdt on (vd.id = vdt.vdiff_id)
+						inner join _vt.vdiff_log as vdl on (vd.id = vdl.vdiff_id)
+						where vd.keyspace = %s and vd.workflow = %s`
+	query := fmt.Sprintf(sqlDeleteVDiffs, encodeString(tablet.Keyspace), encodeString(workflow))
+	if _, err := wr.tmc.ExecuteFetchAsDba(ctx, tablet, false, []byte(query), -1, false, false); err != nil {
+		if sqlErr, ok := err.(*mysql.SQLError); ok && sqlErr.Num != mysql.ERNoSuchTable { // the tables may not exist if no vdiffs have been run
+			wr.Logger().Errorf("Error deleting vdiff data for %s.%s workflow: %v", tablet.Keyspace, workflow, err)
+		}
+	}
+}
+
+// endregion


Test patch:
diff --git a/go/test/endtoend/vtgate/misc_test.go b/go/test/endtoend/vtgate/misc_test.go
index 5cf14b761d..1527735725 100644
--- a/go/test/endtoend/vtgate/misc_test.go
+++ b/go/test/endtoend/vtgate/misc_test.go
@@ -43,6 +43,7 @@ func TestSelectNull(t *testing.T) {
 	utils.AssertMatches(t, conn, "select id, idx from t5_null_vindex order by id", "[[INT64(1) VARCHAR(\"a\")] [INT64(2) VARCHAR(\"b\")] [INT64(3) NULL]]")
 	utils.AssertIsEmpty(t, conn, "select id, idx from t5_null_vindex where idx = null")
 	utils.AssertMatches(t, conn, "select id, idx from t5_null_vindex where idx is null", "[[INT64(3) NULL]]")
+	utils.AssertMatches(t, conn, "select id, idx from t5_null_vindex where idx <=> null", "[[INT64(3) NULL]]")
 	utils.AssertMatches(t, conn, "select id, idx from t5_null_vindex where idx is not null order by id", "[[INT64(1) VARCHAR(\"a\")] [INT64(2) VARCHAR(\"b\")]]")
 	utils.AssertIsEmpty(t, conn, "select id, idx from t5_null_vindex where id IN (null)")
 	utils.AssertMatches(t, conn, "select id, idx from t5_null_vindex where id IN (1,2,null) order by id", "[[INT64(1) VARCHAR(\"a\")] [INT64(2) VARCHAR(\"b\")]]")
diff --git a/go/vt/vtgate/planbuilder/testdata/filter_cases.txt b/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
index 35b9d599af..35456ef0e2 100644
--- a/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
@@ -36,6 +36,25 @@ Gen4 plan same as above
 }
 Gen4 plan same as above
 
+# Null Safe Equality Operator is handled correctly
+"SELECT id from user where someColumn <=> null"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT id from user where someColumn \u003c=\u003e null",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "Scatter",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "FieldQuery": "select id from `user` where 1 != 1",
+    "Query": "select id from `user` where someColumn \u003c=\u003e null",
+    "Table": "`user`"
+  }
+}
+Gen4 plan same as above
+
 # Single table unique vindex route
 "select id from user where user.id = 5"
 {