Repository: vitessio/vitess
Pull request #11050 (2022-08-20 09:23:00)

Add possibility of viewing plans with graphviz
## Description
To make it easier to see and understand our query plans, this PR makes it possible for our plan tests to output not just JSON but also some nice graphviz visualisations.

As an example, the TPCH query #10 is hard to read as JSON:
```json
"select c_custkey, c_name, sum(l_extendedprice * (1 - l_discount)) as revenue, c_acctbal, n_name, c_address, c_phone, c_comment from customer, orders, lineitem, nation where c_custkey = o_custkey and l_orderkey = o_orderkey and o_orderdate \u003e= date('1993-10-01') and o_orderdate \u003c date('1993-10-01') + interval '3' month and l_returnflag = 'R' and c_nationkey = n_nationkey group by c_custkey, c_name, c_acctbal, c_phone, n_name, c_address, c_comment order by revenue desc limit 20"
"unsupported: cross-shard query with aggregates"
{
  "QueryType": "SELECT",
  "Original": "select c_custkey, c_name, sum(l_extendedprice * (1 - l_discount)) as revenue, c_acctbal, n_name, c_address, c_phone, c_comment from customer, orders, lineitem, nation where c_custkey = o_custkey and l_orderkey = o_orderkey and o_orderdate \u003e= date('1993-10-01') and o_orderdate \u003c date('1993-10-01') + interval '3' month and l_returnflag = 'R' and c_nationkey = n_nationkey group by c_custkey, c_name, c_acctbal, c_phone, n_name, c_address, c_comment order by revenue desc limit 20",
  "Instructions": {
    "OperatorType": "Limit",
    "Count": "INT64(20)",
    "Inputs": [
      {
        "OperatorType": "Sort",
        "Variant": "Memory",
        "OrderBy": "2 DESC",
        "ResultColumns": 8,
        "Inputs": [
          {
            "OperatorType": "Aggregate",
            "Variant": "Ordered",
            "Aggregates": "sum(2) AS revenue",
            "GroupBy": "(0|14), (1|13), (3|12), (6|11), (4|10), (5|9), (7|8)",
            "Inputs": [
              {
                "OperatorType": "Projection",
                "Expressions": [
                  "[COLUMN 0] as c_custkey",
                  "[COLUMN 1] as c_name",
                  "(([COLUMN 14] * [COLUMN 15]) * [COLUMN 16]) * [COLUMN 17] as revenue",
                  "[COLUMN 2] as c_acctbal",
                  "[COLUMN 4] as n_name",
                  "[COLUMN 5] as c_address",
                  "[COLUMN 3] as c_phone",
                  "[COLUMN 6] as c_comment",
                  "[COLUMN 13]",
                  "[COLUMN 12]",
                  "[COLUMN 11]",
                  "[COLUMN 10]",
                  "[COLUMN 9]",
                  "[COLUMN 8]",
                  "[COLUMN 7]"
                ],
                "Inputs": [
                  {
                    "OperatorType": "Sort",
                    "Variant": "Memory",
                    "OrderBy": "(0|7) ASC, (1|8) ASC, (2|9) ASC, (3|10) ASC, (4|11) ASC, (5|12) ASC, (6|13) ASC",
                    "Inputs": [
                      {
                        "OperatorType": "Join",
                        "Variant": "Join",
                        "JoinColumnIndexes": "R:0,R:1,R:2,R:3,R:4,R:5,R:6,R:7,R:8,R:9,R:10,R:11,R:12,R:13,L:3,L:4,R:14,R:15",
                        "JoinVars": {
                          "o_custkey": 0
                        },
                        "TableName": "orders_lineitem_customer_nation",
                        "Inputs": [
                          {
                            "OperatorType": "Join",
                            "Variant": "Join",
                            "JoinColumnIndexes": "L:1,L:1,L:4,L:2,R:1",
                            "JoinVars": {
                              "o_orderkey": 0
                            },
                            "TableName": "orders_lineitem",
                            "Inputs": [
                              {
                                "OperatorType": "Route",
                                "Variant": "Scatter",
                                "Keyspace": {
                                  "Name": "main",
                                  "Sharded": true
                                },
                                "FieldQuery": "select o_orderkey, o_custkey, count(*), weight_string(o_orderkey), weight_string(o_custkey) from orders where 1 != 1 group by o_orderkey, weight_string(o_orderkey), o_custkey, weight_string(o_custkey)",
                                "Query": "select o_orderkey, o_custkey, count(*), weight_string(o_orderkey), weight_string(o_custkey) from orders where o_orderdate \u003e= date('1993-10-01') and o_orderdate \u003c date('1993-10-01') + interval '3' month group by o_orderkey, weight_string(o_orderkey), o_custkey, weight_string(o_custkey)",
                                "Table": "orders"
                              },
                              {
                                "OperatorType": "Route",
                                "Variant": "EqualUnique",
                                "Keyspace": {
                                  "Name": "main",
                                  "Sharded": true
                                },
                                "FieldQuery": "select 1, sum(l_extendedprice * (1 - l_discount)) as revenue from lineitem where 1 != 1 group by 1",
                                "Query": "select 1, sum(l_extendedprice * (1 - l_discount)) as revenue from lineitem where l_returnflag = 'R' and l_orderkey = :o_orderkey group by 1",
                                "Table": "lineitem",
                                "Values": [
                                  ":o_orderkey"
                                ],
                                "Vindex": "lineitem_map"
                              }
                            ]
                          },
                          {
                            "OperatorType": "Join",
                            "Variant": "Join",
                            "JoinColumnIndexes": "L:3,L:5,L:7,L:9,R:1,L:11,L:13,L:4,L:6,L:8,L:10,R:2,L:12,L:14,L:1,R:0",
                            "JoinVars": {
                              "c_nationkey": 0
                            },
                            "TableName": "customer_nation",
                            "Inputs": [
                              {
                                "OperatorType": "Route",
                                "Variant": "EqualUnique",
                                "Keyspace": {
                                  "Name": "main",
                                  "Sharded": true
                                },
                                "FieldQuery": "select c_nationkey, count(*), weight_string(c_nationkey), c_custkey, weight_string(c_custkey), c_name, weight_string(c_name), c_acctbal, weight_string(c_acctbal), c_phone, weight_string(c_phone), c_address, weight_string(c_address), c_comment, weight_string(c_comment) from customer where 1 != 1 group by c_nationkey, weight_string(c_nationkey), c_custkey, weight_string(c_custkey), c_name, weight_string(c_name), c_acctbal, weight_string(c_acctbal), c_phone, weight_string(c_phone), c_address, weight_string(c_address), c_comment, weight_string(c_comment)",
                                "Query": "select c_nationkey, count(*), weight_string(c_nationkey), c_custkey, weight_string(c_custkey), c_name, weight_string(c_name), c_acctbal, weight_string(c_acctbal), c_phone, weight_string(c_phone), c_address, weight_string(c_address), c_comment, weight_string(c_comment) from customer where c_custkey = :o_custkey group by c_nationkey, weight_string(c_nationkey), c_custkey, weight_string(c_custkey), c_name, weight_string(c_name), c_acctbal, weight_string(c_acctbal), c_phone, weight_string(c_phone), c_address, weight_string(c_address), c_comment, weight_string(c_comment)",
                                "Table": "customer",
                                "Values": [
                                  ":o_custkey"
                                ],
                                "Vindex": "hash"
                              },
                              {
                                "OperatorType": "Route",
                                "Variant": "EqualUnique",
                                "Keyspace": {
                                  "Name": "main",
                                  "Sharded": true
                                },
                                "FieldQuery": "select count(*), n_name, weight_string(n_name) from nation where 1 != 1 group by n_name, weight_string(n_name)",
                                "Query": "select count(*), n_name, weight_string(n_name) from nation where n_nationkey = :c_nationkey group by n_name, weight_string(n_name)",
                                "Table": "nation",
                                "Values": [
                                  ":c_nationkey"
                                ],
                                "Vindex": "hash"
                              }
                            ]
                          }
                        ]
                      }
                    ]
                  }
                ]
              }
            ]
          }
        ]
      }
    ]
  },
  "TablesUsed": [
    "main.customer",
    "main.lineitem",
    "main.nation",
    "main.orders"
  ]
}
```

As graphviz, it's much easier to see the shape of the query:
![image](https://user-images.githubusercontent.com/402248/185742584-69b254ec-0857-4f0e-85c1-fb8ea77657cf.png)

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.

frouioui: I think we should backport this to the supported releases (`12`, `13`, `14`). Might be helpful the next time we have a bug to solve on those branches :)

Patch:
diff --git a/go/flags/endtoend/vtctld.txt b/go/flags/endtoend/vtctld.txt
index 6e51bf570c..e3c18f1fac 100644
--- a/go/flags/endtoend/vtctld.txt
+++ b/go/flags/endtoend/vtctld.txt
@@ -278,7 +278,7 @@ Usage of vtctld:
       --vreplication_healthcheck_timeout duration                        healthcheck retry delay (default 1m0s)
       --vreplication_healthcheck_topology_refresh duration               refresh interval for re-reading the topology (default 30s)
       --vreplication_heartbeat_update_interval int                       Frequency (in seconds, default 1, max 60) at which the time_updated column of a vreplication stream when idling (default 1)
-      --vreplication_max_time_to_retry_on_error duration                 stop automatically retrying when we've had consecutive failures with the same error for this long after the first occurrence (default 15m0s)
+      --vreplication_max_time_to_retry_on_error duration                 stop automatically retrying when we've had consecutive failures with the same error for this long after the first occurrence (default 0s)
       --vreplication_replica_lag_tolerance duration                      Replica lag threshold duration: once lag is below this we switch from copy phase to the replication (streaming) phase (default 1m0s)
       --vreplication_retry_delay duration                                delay before retrying a failed workflow event in the replication phase (default 5s)
       --vreplication_store_compressed_gtid                               Store compressed gtids in the pos column of _vt.vreplication
diff --git a/go/flags/endtoend/vtexplain.txt b/go/flags/endtoend/vtexplain.txt
index eceb3cb92e..f292904df3 100644
--- a/go/flags/endtoend/vtexplain.txt
+++ b/go/flags/endtoend/vtexplain.txt
@@ -267,7 +267,7 @@ Usage of vtexplain:
       --vreplication_healthcheck_timeout duration                        healthcheck retry delay (default 1m0s)
       --vreplication_healthcheck_topology_refresh duration               refresh interval for re-reading the topology (default 30s)
       --vreplication_heartbeat_update_interval int                       Frequency (in seconds, default 1, max 60) at which the time_updated column of a vreplication stream when idling (default 1)
-      --vreplication_max_time_to_retry_on_error duration                 stop automatically retrying when we've had consecutive failures with the same error for this long after the first occurrence (default 15m0s)
+      --vreplication_max_time_to_retry_on_error duration                 stop automatically retrying when we've had consecutive failures with the same error for this long after the first occurrence (default 0s)
       --vreplication_replica_lag_tolerance duration                      Replica lag threshold duration: once lag is below this we switch from copy phase to the replication (streaming) phase (default 1m0s)
       --vreplication_retry_delay duration                                delay before retrying a failed workflow event in the replication phase (default 5s)
       --vreplication_store_compressed_gtid                               Store compressed gtids in the pos column of _vt.vreplication
diff --git a/go/flags/endtoend/vttablet.txt b/go/flags/endtoend/vttablet.txt
index 287cce1d98..e0dca3c6b5 100644
--- a/go/flags/endtoend/vttablet.txt
+++ b/go/flags/endtoend/vttablet.txt
@@ -460,7 +460,7 @@ Usage of vttablet:
       --vreplication_healthcheck_timeout duration                        healthcheck retry delay (default 1m0s)
       --vreplication_healthcheck_topology_refresh duration               refresh interval for re-reading the topology (default 30s)
       --vreplication_heartbeat_update_interval int                       Frequency (in seconds, default 1, max 60) at which the time_updated column of a vreplication stream when idling (default 1)
-      --vreplication_max_time_to_retry_on_error duration                 stop automatically retrying when we've had consecutive failures with the same error for this long after the first occurrence (default 15m0s)
+      --vreplication_max_time_to_retry_on_error duration                 stop automatically retrying when we've had consecutive failures with the same error for this long after the first occurrence (default 0s)
       --vreplication_replica_lag_tolerance duration                      Replica lag threshold duration: once lag is below this we switch from copy phase to the replication (streaming) phase (default 1m0s)
       --vreplication_retry_delay duration                                delay before retrying a failed workflow event in the replication phase (default 5s)
       --vreplication_store_compressed_gtid                               Store compressed gtids in the pos column of _vt.vreplication
diff --git a/go/tools/graphviz/graph.go b/go/tools/graphviz/graph.go
new file mode 100644
index 0000000000..b3b12a33ea
--- /dev/null
+++ b/go/tools/graphviz/graph.go
@@ -0,0 +1,205 @@
+/*
+Copyright 2022 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.
+*/
+
+package graphviz
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"runtime"
+	"strconv"
+	"strings"
+)
+
+type (
+	Graph struct {
+		lastID int
+		nodes  []*Node
+		edges  []*Edge
+	}
+	Node struct {
+		id      int
+		Name    string
+		attrs   []string
+		tooltip string
+	}
+	Edge struct {
+		From, To *Node
+	}
+)
+
+func escape(s string) string {
+	s = strings.ReplaceAll(s, "<", "\\<")
+	s = strings.ReplaceAll(s, ">", "\\>")
+	s = strings.ReplaceAll(s, "\"", "\\\"")
+	s = strings.ReplaceAll(s, "|", "\\|")
+	s = strings.ReplaceAll(s, "{", "\\{")
+	s = strings.ReplaceAll(s, "}", "\\}")
+	return s
+}
+
+func (n *Node) AddAttribute(s string) {
+	n.attrs = append(n.attrs, escape(s))
+}
+func (n *Node) AddTooltip(s string) {
+	n.tooltip = escape(s)
+}
+
+func (g *Graph) produceDot() string {
+	var dot strings.Builder
+	dot.WriteString(`digraph {
+node [shape=record, fontsize=10]
+`)
+	for _, node := range g.nodes {
+		labels := node.Name
+		for i, attr := range node.attrs {
+			if i == 0 {
+				labels += "|{" + attr
+			} else {
+				labels += "|" + attr
+			}
+
+		}
+		labels += "}"
+		if node.tooltip != "" {
+			dot.WriteString(fmt.Sprintf(`n%d [label="%s", tooltip="%s"]`, node.id, labels, node.tooltip))
+		} else {
+			dot.WriteString(fmt.Sprintf(`n%d [label="%s"]`, node.id, labels))
+		}
+		dot.WriteString(";\n")
+	}
+	for _, edge := range g.edges {
+		dot.WriteString(fmt.Sprintf(`n%d -> n%d`, edge.From.id, edge.To.id))
+		dot.WriteString(";\n")
+	}
+	dot.WriteString("}")
+	return dot.String()
+}
+
+func (g *Graph) AddNode(name string) *Node {
+	n := &Node{
+		id:   g.lastID,
+		Name: name,
+	}
+	g.lastID++
+	g.nodes = append(g.nodes, n)
+	return n
+}
+
+func (g *Graph) AddEdge(from, to *Node) *Edge {
+	e := &Edge{
+		From: from,
+		To:   to,
+	}
+	g.edges = append(g.edges, e)
+	return e
+}
+
+const htmlTemplate = `
+<!DOCTYPE html>
+<html>
+<head>
+    <meta charset="UTF-8">
+    <title>GraphViz Viewer</title>
+    <script src="https://cdn.jsdelivr.net/npm/@hpcc-js/wasm/dist/index.min.js"></script>
+	<script src="https://cdn.jsdelivr.net/npm/svg-pan-zoom/dist/svg-pan-zoom.min.js"></script>
+    <script>
+        var hpccWasm = window["@hpcc-js/wasm"];
+    </script>
+	<style>
+		#graph svg {
+			width: 90vw;
+			height: 90vh;
+			border-width: 1px;
+			border-style: dotted;
+		}
+	</style>
+</head>
+<body>
+    <div id="graph"></div>
+    <script>
+		const dot = %s;
+        hpccWasm.graphviz.layout(dot, "svg", "dot").then(svg => {
+            const div = document.getElementById("graph");
+            div.innerHTML = svg;
+			svgPanZoom(div.querySelector('svg'), { controlIconsEnabled: true });
+        });
+    </script>
+</body>
+</html>
+`
+
+func (g *Graph) Render() error {
+
+	dot := g.produceDot()
+
+	browsers := func() []string {
+		var cmds []string
+		if userBrowser := os.Getenv("BROWSER"); userBrowser != "" {
+			cmds = append(cmds, userBrowser)
+		}
+		switch runtime.GOOS {
+		case "darwin":
+			cmds = append(cmds, "/usr/bin/open")
+		case "windows":
+			cmds = append(cmds, "cmd /c start")
+		default:
+			// Commands opening browsers are prioritized over xdg-open, so browser()
+			// command can be used on linux to open the .svg file generated by the -web
+			// command (the .svg file includes embedded javascript so is best viewed in
+			// a browser).
+			cmds = append(cmds, []string{"chrome", "google-chrome", "chromium", "firefox", "sensible-browser"}...)
+			if os.Getenv("DISPLAY") != "" {
+				// xdg-open is only for use in a desktop environment.
+				cmds = append(cmds, "xdg-open")
+			}
+		}
+		return cmds
+	}
+
+	tmpfile, err := os.CreateTemp("", "graphviz_*.html")
+	if err != nil {
+		return err
+	}
+
+	_, err = fmt.Fprintf(tmpfile, htmlTemplate, strconv.Quote(dot))
+	if err != nil {
+		return err
+	}
+	err = tmpfile.Close()
+	if err != nil {
+		return err
+	}
+
+	for _, b := range browsers() {
+		args := strings.Split(b, " ")
+		if len(args) == 0 {
+			continue
+		}
+		viewer := exec.Command(args[0], append(args[1:], tmpfile.Name())...)
+		viewer.Stderr = os.Stderr
+		if err := viewer.Start(); err == nil {
+			return nil
+		}
+	}
+
+	return fmt.Errorf("failed to open browser for SVG debugging")
+}
+
+func New() *Graph {
+	return &Graph{}
+}
diff --git a/go/vt/schemadiff/column.go b/go/vt/schemadiff/column.go
index 4d816ea8ad..27169a698f 100644
--- a/go/vt/schemadiff/column.go
+++ b/go/vt/schemadiff/column.go
@@ -22,13 +22,6 @@ import (
 	"vitess.io/vitess/go/vt/sqlparser"
 )
 
-func colWithMaskedName(col *sqlparser.ColumnDefinition) string {
-	col = sqlparser.CloneRefOfColumnDefinition(col)
-	col.Name = sqlparser.NewIdentifierCI("mask")
-	return sqlparser.CanonicalString(col)
-
-}
-
 // columnDetails decorates a column with more details, used by diffing logic
 type columnDetails struct {
 	col     *sqlparser.ColumnDefinition
@@ -40,7 +33,7 @@ func (c *columnDetails) identicalOtherThanName(other *sqlparser.ColumnDefinition
 	if other == nil {
 		return false
 	}
-	return colWithMaskedName(c.col) == colWithMaskedName(other)
+	return sqlparser.EqualsColumnType(c.col.Type, other.Type)
 }
 
 func (c *columnDetails) prevColName() string {
@@ -88,10 +81,8 @@ func NewColumnDefinitionEntity(c *sqlparser.ColumnDefinition) *ColumnDefinitionE
 // change this table to look like the other table.
 // It returns an AlterTable statement if changes are found, or nil if not.
 // the other table may be of different name; its name is ignored.
-func (c *ColumnDefinitionEntity) ColumnDiff(other *ColumnDefinitionEntity, hints *DiffHints) *ModifyColumnDiff {
-	format := sqlparser.CanonicalString(c.columnDefinition)
-	otherFormat := sqlparser.CanonicalString(other.columnDefinition)
-	if format == otherFormat {
+func (c *ColumnDefinitionEntity) ColumnDiff(other *ColumnDefinitionEntity, _ *DiffHints) *ModifyColumnDiff {
+	if sqlparser.EqualsRefOfColumnDefinition(c.columnDefinition, other.columnDefinition) {
 		return nil
 	}
 
diff --git a/go/vt/schemadiff/schema.go b/go/vt/schemadiff/schema.go
index abf2357993..7c8dd674b3 100644
--- a/go/vt/schemadiff/schema.go
+++ b/go/vt/schemadiff/schema.go
@@ -69,7 +69,7 @@ func NewSchemaFromEntities(entities []Entity) (*Schema, error) {
 
 // NewSchemaFromStatements creates a valid and normalized schema based on list of valid statements
 func NewSchemaFromStatements(statements []sqlparser.Statement) (*Schema, error) {
-	entities := []Entity{}
+	entities := make([]Entity, 0, len(statements))
 	for _, s := range statements {
 		switch stmt := s.(type) {
 		case *sqlparser.CreateTable:
@@ -93,7 +93,7 @@ func NewSchemaFromStatements(statements []sqlparser.Statement) (*Schema, error)
 
 // NewSchemaFromQueries creates a valid and normalized schema based on list of queries
 func NewSchemaFromQueries(queries []string) (*Schema, error) {
-	statements := []sqlparser.Statement{}
+	statements := make([]sqlparser.Statement, 0, len(queries))
 	for _, q := range queries {
 		stmt, err := sqlparser.ParseStrictDDL(q)
 		if err != nil {
@@ -107,7 +107,7 @@ func NewSchemaFromQueries(queries []string) (*Schema, error) {
 // NewSchemaFromSQL creates a valid and normalized schema based on a SQL blob that contains
 // CREATE statements for various objects (tables, views)
 func NewSchemaFromSQL(sql string) (*Schema, error) {
-	statements := []sqlparser.Statement{}
+	var statements []sqlparser.Statement
 	tokenizer := sqlparser.NewStringTokenizer(sql)
 	for {
 		stmt, err := sqlparser.ParseNextStrictDDL(tokenizer)
@@ -143,8 +143,8 @@ func getViewDependentTableNames(createView *sqlparser.CreateView) (names []strin
 // normalize is called as part of Schema creation process. The user may only get a hold of normalized schema.
 // It validates some cross-entity constraints, and orders entity based on dependencies (e.g. tables, views that read from tables, 2nd level views, etc.)
 func (s *Schema) normalize() error {
-	s.named = map[string]Entity{}
-	s.sorted = []Entity{}
+	s.named = make(map[string]Entity, len(s.tables)+len(s.views))
+	s.sorted = make([]Entity, 0, len(s.tables)+len(s.views))
 	// Verify no two entities share same name
 	for _, t := range s.tables {
 		name := t.Name()
@@ -174,7 +174,7 @@ func (s *Schema) normalize() error {
 	// We actually prioritise all tables first, then views.
 	// If a view v1 depends on v2, then v2 must come before v1, even though v1
 	// precedes v2 alphabetically
-	dependencyLevels := map[string]int{}
+	dependencyLevels := make(map[string]int, len(s.tables)+len(s.views))
 	for _, t := range s.tables {
 		s.sorted = append(s.sorted, t)
 		dependencyLevels[t.Name()] = 0
@@ -204,14 +204,14 @@ func (s *Schema) normalize() error {
 	// - etc.
 	// we stop when we have been unable to find a view in an iteration.
 	for iterationLevel := 1; ; iterationLevel++ {
-		handledAnyViewsInItration := false
+		handledAnyViewsInIteration := false
 		for _, v := range s.views {
 			name := v.Name()
 			if _, ok := dependencyLevels[name]; ok {
 				// already handled; skip
 				continue
 			}
-			// Not handled. Is this view dependant on already handled objects?
+			// Not handled. Is this view dependent on already handled objects?
 			dependentNames, err := getViewDependentTableNames(&v.CreateView)
 			if err != nil {
 				return err
@@ -219,10 +219,10 @@ func (s *Schema) normalize() error {
 			if allNamesFoundInLowerLevel(dependentNames, iterationLevel) {
 				s.sorted = append(s.sorted, v)
 				dependencyLevels[v.Name()] = iterationLevel
-				handledAnyViewsInItration = true
+				handledAnyViewsInIteration = true
 			}
 		}
-		if !handledAnyViewsInItration {
+		if !handledAnyViewsInIteration {
 			break
 		}
 	}
@@ -275,7 +275,7 @@ func (s *Schema) TableNames() []string {
 	return names
 }
 
-// Tables returns this schema's views in good order (may be applied without error)
+// Views returns this schema's views in good order (may be applied without error)
 func (s *Schema) Views() []*CreateViewEntity {
 	var views []*CreateViewEntity
 	for _, entity := range s.sorted {
@@ -362,7 +362,7 @@ func (s *Schema) heuristicallyDetectTableRenames(
 
 	findRenamedTable := func() bool {
 		// What we're doing next is to try and identify a table RENAME.
-		// We do so by cross referencing dropped and created tables.
+		// We do so by cross-referencing dropped and created tables.
 		// The check is heuristic, and looks like this:
 		// We consider a table renamed iff:
 		// - the DROP and CREATE table definitions are identical other than the table name
@@ -371,7 +371,7 @@ func (s *Schema) heuristicallyDetectTableRenames(
 		// Once we heuristically decide that we found a RENAME, we cancel the DROP,
 		// cancel the CREATE, and inject a RENAME in place of both.
 
-		// findRenamedTable cross references dropped and created tables to find a single renamed table. If such is found:
+		// findRenamedTable cross-references dropped and created tables to find a single renamed table. If such is found:
 		// we remove the entry from DROPped tables, remove the entry from CREATEd tables, add an entry for RENAMEd tables,
 		// and return 'true'.
 		// Successive calls to this function will then find the next heuristic RENAMEs.
@@ -443,7 +443,7 @@ func (s *Schema) View(name string) *CreateViewEntity {
 
 // ToStatements returns an ordered list of statements which can be applied to create the schema
 func (s *Schema) ToStatements() []sqlparser.Statement {
-	stmts := []sqlparser.Statement{}
+	stmts := make([]sqlparser.Statement, 0, len(s.Entities()))
 	for _, e := range s.Entities() {
 		stmts = append(stmts, e.Create().Statement())
 	}
@@ -452,7 +452,7 @@ func (s *Schema) ToStatements() []sqlparser.Statement {
 
 // ToQueries returns an ordered list of queries which can be applied to create the schema
 func (s *Schema) ToQueries() []string {
-	queries := []string{}
+	queries := make([]string, 0, len(s.Entities()))
 	for _, e := range s.Entities() {
 		queries = append(queries, e.Create().CanonicalStatementString())
 	}
@@ -469,17 +469,21 @@ func (s *Schema) ToSQL() string {
 	return buf.String()
 }
 
-// Clone returns a deep copy of the schema. This is used when applying changes for example. It
-// will first create copies of all entities which depend on the sqlparser deep copy logic and
-// then creates a new schema from those copies.
-func (s *Schema) Clone() *Schema {
-	entities := make([]Entity, 0, len(s.Entities()))
-	for _, e := range s.Entities() {
-		entities = append(entities, e.Clone())
+// copy returns a shallow copy of the schema. This is used when applying changes for example.
+// applying changes will ensure we copy new entities themselves separately.
+func (s *Schema) copy() *Schema {
+	dup := newEmptySchema()
+	dup.tables = make([]*CreateTableEntity, len(s.tables))
+	copy(dup.tables, s.tables)
+	dup.views = make([]*CreateViewEntity, len(s.views))
+	copy(dup.views, s.views)
+	dup.named = make(map[string]Entity, len(s.named))
+	for k, v := range s.named {
+		dup.named[k] = v
 	}
-	// Can't error since we're valid ourselves.
-	schema, _ := NewSchemaFromEntities(entities)
-	return schema
+	dup.sorted = make([]Entity, len(s.sorted))
+	copy(dup.sorted, s.sorted)
+	return dup
 }
 
 // apply attempts to apply given list of diffs to this object.
@@ -604,7 +608,7 @@ func (s *Schema) apply(diffs []EntityDiff) error {
 // These diffs are CREATE/DROP/ALTER TABLE/VIEW.
 // The operation does not modify this object. Instead, if successful, a new (modified) Schema is returned.
 func (s *Schema) Apply(diffs []EntityDiff) (*Schema, error) {
-	dup := s.Clone()
+	dup := s.copy()
 	for k, v := range s.named {
 		dup.named[k] = v
 	}
diff --git a/go/vt/schemadiff/table.go b/go/vt/schemadiff/table.go
index e613abca69..c2227bdf2a 100644
--- a/go/vt/schemadiff/table.go
+++ b/go/vt/schemadiff/table.go
@@ -174,7 +174,7 @@ func (d *DropTableEntityDiff) IsEmpty() bool {
 	return d.Statement() == nil
 }
 
-// IsEmpty implements EntityDiff
+// Entities implements EntityDiff
 func (d *DropTableEntityDiff) Entities() (from Entity, to Entity) {
 	return d.from, nil
 }
@@ -244,7 +244,7 @@ func (d *RenameTableEntityDiff) Statement() sqlparser.Statement {
 	return d.renameTable
 }
 
-// CreateTable returns the underlying sqlparser.CreateTable that was generated for the diff.
+// RenameTable returns the underlying sqlparser.RenameTable that was generated for the diff.
 func (d *RenameTableEntityDiff) RenameTable() *sqlparser.RenameTable {
 	if d == nil {
 		return nil
@@ -646,7 +646,7 @@ func (c *CreateTableEntity) TableDiff(other *CreateTableEntity, hints *DiffHints
 		// ordered columns for both tables:
 		t1Columns := c.CreateTable.TableSpec.Columns
 		t2Columns := other.CreateTable.TableSpec.Columns
-		c.diffColumns(alterTable, t1Columns, t2Columns, hints, (diffedTableCharset != ""))
+		c.diffColumns(alterTable, t1Columns, t2Columns, hints, diffedTableCharset != "")
 	}
 	{
 		// diff keys
@@ -943,7 +943,7 @@ func (c *CreateTableEntity) isRangePartitionsRotation(
 	if len(definitions2) == 0 {
 		return false, nil, nil
 	}
-	droppedPartitions1 := []*sqlparser.PartitionDefinition{}
+	var droppedPartitions1 []*sqlparser.PartitionDefinition
 	// It's OK for prefix of t1 partitions to be nonexistent in t2 (as they may have been rotated away in t2)
 	for len(definitions1) > 0 && !sqlparser.EqualsRefOfPartitionDefinition(definitions1[0], definitions2[0]) {
 		droppedPartitions1 = append(droppedPartitions1, definitions1[0])
@@ -970,8 +970,8 @@ func (c *CreateTableEntity) isRangePartitionsRotation(
 		definitions1 = definitions1[1:]
 		definitions2 = definitions2[1:]
 	}
-	partitionSpecs := []*sqlparser.PartitionSpec{}
 	addedPartitions2 := definitions2
+	partitionSpecs := make([]*sqlparser.PartitionSpec, 0, len(droppedPartitions1)+len(addedPartitions2))
 	for _, p := range droppedPartitions1 {
 		partitionSpec := &sqlparser.PartitionSpec{
 			Action: sqlparser.DropAction,
@@ -1179,7 +1179,7 @@ func (c *CreateTableEntity) diffKeys(alterTable *sqlparser.AlterTable,
 					continue
 				}
 
-				// For other changes, we're gonna drop and create.
+				// For other changes, we're going to drop and create.
 				dropKey := dropKeyStatement(t1Key.Info.Name)
 				addKey := &sqlparser.AddIndexDefinition{
 					IndexDefinition: t2Key,
@@ -1249,11 +1249,11 @@ func indexOnlyVisibilityChange(t1Key, t2Key *sqlparser.IndexDefinition) (bool, b
 func evaluateColumnReordering(t1SharedColumns, t2SharedColumns []*sqlparser.ColumnDefinition) map[string]int {
 	minimalColumnReordering := map[string]int{}
 
-	t1SharedColNames := []interface{}{}
+	t1SharedColNames := make([]interface{}, 0, len(t1SharedColumns))
 	for _, col := range t1SharedColumns {
 		t1SharedColNames = append(t1SharedColNames, col.Name.Lowered())
 	}
-	t2SharedColNames := []interface{}{}
+	t2SharedColNames := make([]interface{}, 0, len(t2SharedColumns))
 	for _, col := range t2SharedColumns {
 		t2SharedColNames = append(t2SharedColNames, col.Name.Lowered())
 	}
@@ -1306,9 +1306,9 @@ func (c *CreateTableEntity) diffColumns(alterTable *sqlparser.AlterTable,
 
 	// For purpose of column reordering detection, we maintain a list of
 	// shared columns, by order of appearance in t1
-	t1SharedColumns := []*sqlparser.ColumnDefinition{}
+	var t1SharedColumns []*sqlparser.ColumnDefinition
 
-	dropColumns := []*sqlparser.DropColumn{}
+	var dropColumns []*sqlparser.DropColumn
 	// evaluate dropped columns
 	//
 	for _, t1Col := range t1Columns {
@@ -1325,7 +1325,7 @@ func (c *CreateTableEntity) diffColumns(alterTable *sqlparser.AlterTable,
 
 	// For purpose of column reordering detection, we maintain a list of
 	// shared columns, by order of appearance in t2
-	t2SharedColumns := []*sqlparser.ColumnDefinition{}
+	var t2SharedColumns []*sqlparser.ColumnDefinition
 	for _, t2Col := range t2Columns {
 		if _, ok := t1ColumnsMap[t2Col.Name.Lowered()]; ok {
 			// column exists in both tables
@@ -1335,7 +1335,7 @@ func (c *CreateTableEntity) diffColumns(alterTable *sqlparser.AlterTable,
 
 	// evaluate modified columns
 	//
-	modifyColumns := []*sqlparser.ModifyColumn{}
+	var modifyColumns []*sqlparser.ModifyColumn
 	columnReordering := evaluateColumnReordering(t1SharedColumns, t2SharedColumns)
 	for _, t2Col := range t2SharedColumns {
 		t2ColName := t2Col.Name.Lowered()
@@ -1379,7 +1379,7 @@ func (c *CreateTableEntity) diffColumns(alterTable *sqlparser.AlterTable,
 	// Every added column is obviously a diff. But on top of that, we are also interested to know
 	// if the column is added somewhere in between existing columns rather than appended to the
 	// end of existing columns list.
-	addColumns := []*sqlparser.AddColumns{}
+	var addColumns []*sqlparser.AddColumns
 	expectAppendIndex := len(t2SharedColumns)
 	for t2ColIndex, t2Col := range t2Columns {
 		if _, ok := t1ColumnsMap[t2Col.Name.Lowered()]; !ok {
@@ -1421,10 +1421,10 @@ func heuristicallyDetectColumnRenames(
 	t2ColumnsMap map[string]*columnDetails,
 	hints *DiffHints,
 ) ([]*sqlparser.DropColumn, []*sqlparser.AddColumns, []*sqlparser.RenameColumn) {
-	renameColumns := []*sqlparser.RenameColumn{}
+	var renameColumns []*sqlparser.RenameColumn
 	findRenamedColumn := func() bool {
 		// What we're doing next is to try and identify a column RENAME.
-		// We do so by cross referencing dropped and added columns.
+		// We do so by cross-referencing dropped and added columns.
 		// The check is heuristic, and looks like this:
 		// We consider a column renamed iff:
 		// - the DROP and ADD column definitions are identical other than the column name, and
@@ -1436,7 +1436,7 @@ func heuristicallyDetectColumnRenames(
 		// At any case, once we heuristically decide that we found a RENAME, we cancel the DROP,
 		// cancel the ADD, and inject a RENAME in place of both.
 
-		// findRenamedColumn cross references dropped and added columns to find a single renamed column. If such is found:
+		// findRenamedColumn cross-references dropped and added columns to find a single renamed column. If such is found:
 		// we remove the entry from DROPped columns, remove the entry from ADDed columns, add an entry for RENAMEd columns,
 		// and return 'true'.
 		// Successive calls to this function will then find the next heuristic RENAMEs.
@@ -1868,7 +1868,7 @@ func (c *CreateTableEntity) postApplyNormalize() error {
 	for _, col := range c.CreateTable.TableSpec.Columns {
 		columnExists[col.Name.Lowered()] = true
 	}
-	nonEmptyIndexes := []*sqlparser.IndexDefinition{}
+	var nonEmptyIndexes []*sqlparser.IndexDefinition
 
 	keyHasNonExistentColumns := func(keyCol *sqlparser.IndexColumn) bool {
 		if keyCol.Column.Lowered() != "" {
@@ -1879,7 +1879,7 @@ func (c *CreateTableEntity) postApplyNormalize() error {
 		return false
 	}
 	for _, key := range c.CreateTable.TableSpec.Indexes {
-		existingKeyColumns := []*sqlparser.IndexColumn{}
+		var existingKeyColumns []*sqlparser.IndexColumn
 		for _, keyCol := range key.Columns {
 			if !keyHasNonExistentColumns(keyCol) {
 				existingKeyColumns = append(existingKeyColumns, keyCol)
@@ -1892,14 +1892,14 @@ func (c *CreateTableEntity) postApplyNormalize() error {
 	}
 	c.CreateTable.TableSpec.Indexes = nonEmptyIndexes
 
-	keptConstraints := []*sqlparser.ConstraintDefinition{}
+	var keptConstraints []*sqlparser.ConstraintDefinition
 	for _, constraint := range c.CreateTable.TableSpec.Constraints {
 		check, ok := constraint.Details.(*sqlparser.CheckConstraintDefinition)
 		if !ok {
 			keptConstraints = append(keptConstraints, constraint)
 			continue
 		}
-		referencedColumns := []string{}
+		var referencedColumns []string
 		err := sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
 			switch node := node.(type) {
 			case *sqlparser.ColName:
@@ -1974,7 +1974,7 @@ func (c *CreateTableEntity) validate() error {
 	// validate all columns referenced by generated columns do in fact exist
 	for _, col := range c.CreateTable.TableSpec.Columns {
 		if col.Type.Options != nil && col.Type.Options.As != nil {
-			referencedColumns := []string{}
+			var referencedColumns []string
 			err := sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
 				switch node := node.(type) {
 				case *sqlparser.ColName:
@@ -1995,7 +1995,7 @@ func (c *CreateTableEntity) validate() error {
 	// validate all columns referenced by functional indexes do in fact exist
 	for _, idx := range c.CreateTable.TableSpec.Indexes {
 		for _, idxCol := range idx.Columns {
-			referencedColumns := []string{}
+			var referencedColumns []string
 			err := sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
 				switch node := node.(type) {
 				case *sqlparser.ColName:
@@ -2031,7 +2031,7 @@ func (c *CreateTableEntity) validate() error {
 		if !ok {
 			continue
 		}
-		referencedColumns := []string{}
+		var referencedColumns []string
 		err := sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
 			switch node := node.(type) {
 			case *sqlparser.ColName:
@@ -2060,7 +2060,7 @@ func (c *CreateTableEntity) validate() error {
 		}
 		// validate columns referenced by partitions do in fact exist
 		// also, validate that all unique keys include partitioned columns
-		partitionColNames := []string{}
+		var partitionColNames []string
 		err := sqlparser.Walk(func(node sqlparser.SQLNode) (kontinue bool, err error) {
 			switch node := node.(type) {
 			case *sqlparser.ColName:
@@ -2105,12 +2105,6 @@ func (c *CreateTableEntity) identicalOtherThanName(other *CreateTableEntity) boo
 	if other == nil {
 		return false
 	}
-	return sqlparser.EqualsRefOfCreateTable(tableWithMaskedName(&c.CreateTable), tableWithMaskedName(&other.CreateTable))
-}
-
-// tableWithMaskedName returns the CREATE TABLE statement but with table's name replaced with a constant arbitrary name
-func tableWithMaskedName(createTable *sqlparser.CreateTable) *sqlparser.CreateTable {
-	createTable = sqlparser.CloneRefOfCreateTable(createTable)
-	createTable.Table.Name = sqlparser.NewIdentifierCS("mask")
-	return createTable
+	return sqlparser.EqualsRefOfTableSpec(c.TableSpec, other.TableSpec) &&
+		sqlparser.EqualsRefOfParsedComments(c.Comments, other.Comments)
 }
diff --git a/go/vt/schemadiff/types.go b/go/vt/schemadiff/types.go
index d11b6e5347..dce0d10c84 100644
--- a/go/vt/schemadiff/types.go
+++ b/go/vt/schemadiff/types.go
@@ -45,9 +45,9 @@ type EntityDiff interface {
 	// Statement returns a valid SQL statement that applies the diff, e.g. an ALTER TABLE ...
 	// It returns nil if the diff is empty
 	Statement() sqlparser.Statement
-	// StatementString "stringifies" the this diff's Statement(). It returns an empty string if the diff is empty
+	// StatementString "stringifies" this diff's Statement(). It returns an empty string if the diff is empty
 	StatementString() string
-	// CanonicalStatementString "stringifies" the this diff's Statement() to a canonical string. It returns an empty string if the diff is empty
+	// CanonicalStatementString "stringifies" this diff's Statement() to a canonical string. It returns an empty string if the diff is empty
 	CanonicalStatementString() string
 	// SubsequentDiff returns a followup diff to this one, if exists
 	SubsequentDiff() EntityDiff
diff --git a/go/vt/schemadiff/view.go b/go/vt/schemadiff/view.go
index c3c0ce713b..5a85eabe4b 100644
--- a/go/vt/schemadiff/view.go
+++ b/go/vt/schemadiff/view.go
@@ -33,7 +33,7 @@ func (d *AlterViewEntityDiff) IsEmpty() bool {
 	return d.Statement() == nil
 }
 
-// IsEmpty implements EntityDiff
+// Entities implements EntityDiff
 func (d *AlterViewEntityDiff) Entities() (from Entity, to Entity) {
 	return d.from, d.to
 }
@@ -233,7 +233,7 @@ func (c *CreateViewEntity) Diff(other Entity, hints *DiffHints) (EntityDiff, err
 // change this view to look like the other view.
 // It returns an AlterView statement if changes are found, or nil if not.
 // the other view may be of different name; its name is ignored.
-func (c *CreateViewEntity) ViewDiff(other *CreateViewEntity, hints *DiffHints) (*AlterViewEntityDiff, error) {
+func (c *CreateViewEntity) ViewDiff(other *CreateViewEntity, _ *DiffHints) (*AlterViewEntityDiff, error) {
 	otherStmt := other.CreateView
 	otherStmt.ViewName = c.CreateView.ViewName
 
@@ -244,9 +244,7 @@ func (c *CreateViewEntity) ViewDiff(other *CreateViewEntity, hints *DiffHints) (
 		return nil, &NotFullyParsedError{Entity: c.Name(), Statement: sqlparser.CanonicalString(&otherStmt)}
 	}
 
-	format := sqlparser.CanonicalString(&c.CreateView)
-	otherFormat := sqlparser.CanonicalString(&otherStmt)
-	if format == otherFormat {
+	if sqlparser.EqualsRefOfCreateView(&c.CreateView, &otherStmt) {
 		return nil, nil
 	}
 
diff --git a/go/vt/vtgate/engine/plan_description.go b/go/vt/vtgate/engine/plan_description.go
index 3c84f82776..4433dcd69d 100644
--- a/go/vt/vtgate/engine/plan_description.go
+++ b/go/vt/vtgate/engine/plan_description.go
@@ -19,8 +19,10 @@ package engine
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"sort"
 
+	"vitess.io/vitess/go/tools/graphviz"
 	"vitess.io/vitess/go/vt/key"
 	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
 	"vitess.io/vitess/go/vt/vtgate/vindexes"
@@ -91,6 +93,54 @@ func (pd PrimitiveDescription) MarshalJSON() ([]byte, error) {
 	return buf.Bytes(), nil
 }
 
+func (pd PrimitiveDescription) addToGraph(g *graphviz.Graph) (*graphviz.Node, error) {
+	var nodes []*graphviz.Node
+	for _, input := range pd.Inputs {
+		n, err := input.addToGraph(g)
+		if err != nil {
+			return nil, err
+		}
+		nodes = append(nodes, n)
+	}
+	name := pd.OperatorType + ":" + pd.Variant
+	if pd.Variant == "" {
+		name = pd.OperatorType
+	}
+	this := g.AddNode(name)
+	for k, v := range pd.Other {
+		switch k {
+		case "Query":
+			this.AddTooltip(fmt.Sprintf("%v", v))
+		case "FieldQuery":
+		// skip these
+		default:
+			slice, ok := v.([]string)
+			if ok {
+				this.AddAttribute(k)
+				for _, s := range slice {
+					this.AddAttribute(s)
+				}
+			} else {
+				this.AddAttribute(fmt.Sprintf("%s:%v", k, v))
+			}
+		}
+	}
+	for _, n := range nodes {
+		g.AddEdge(this, n)
+	}
+	return this, nil
+}
+
+func GraphViz(p Primitive) (*graphviz.Graph, error) {
+	g := graphviz.New()
+	description := PrimitiveToPlanDescription(p)
+	_, err := description.addToGraph(g)
+	if err != nil {
+		return nil, err
+	}
+	return g, nil
+}
+
 func addMap(input map[string]any, buf *bytes.Buffer) error {
 	var mk []string
 	for k, v := range input {
diff --git a/go/vt/vttablet/tabletmanager/vreplication/controller.go b/go/vt/vttablet/tabletmanager/vreplication/controller.go
index 6b9a2c4ddc..4a4ece3638 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/controller.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/controller.go
@@ -48,7 +48,7 @@ var (
 	_          = flag.Duration("vreplication_healthcheck_timeout", 1*time.Minute, "healthcheck retry delay")
 	retryDelay = flag.Duration("vreplication_retry_delay", 5*time.Second, "delay before retrying a failed workflow event in the replication phase")
 
-	maxTimeToRetryError = flag.Duration("vreplication_max_time_to_retry_on_error", 15*time.Minute, "stop automatically retrying when we've had consecutive failures with the same error for this long after the first occurrence")
+	maxTimeToRetryError = flag.Duration("vreplication_max_time_to_retry_on_error", 0, "stop automatically retrying when we've had consecutive failures with the same error for this long after the first occurrence")
 )
 
 // controller is created by Engine. Members are initialized upfront.
diff --git a/go/vt/vttablet/tabletmanager/vreplication/last_error.go b/go/vt/vttablet/tabletmanager/vreplication/last_error.go
index a3080973c9..e3a73d5b06 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/last_error.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/last_error.go
@@ -62,10 +62,18 @@ func (le *lastError) record(err error) {
 func (le *lastError) shouldRetry() bool {
 	le.mu.Lock()
 	defer le.mu.Unlock()
-	if !le.firstSeen.IsZero() && time.Since(le.firstSeen) > le.maxTimeInError {
-		log.Errorf("VReplication encountered the same error continuously since %s, we will assume this is a non-recoverable error and will not retry anymore; the workflow will need to be manually restarted once error '%s' has been addressed",
-			le.firstSeen.UTC(), le.err)
-		return false
+	if le.maxTimeInError == 0 {
+		// The value of 0 means "no time limit"
+		return true
 	}
-	return true
+	if le.firstSeen.IsZero() {
+		return true
+	}
+	if time.Since(le.firstSeen) <= le.maxTimeInError {
+		// within the max time range
+		return true
+	}
+	log.Errorf("VReplication encountered the same error continuously since %s, we will assume this is a non-recoverable error and will not retry anymore; the workflow will need to be manually restarted once error '%s' has been addressed",
+		le.firstSeen.UTC(), le.err)
+	return false
 }


Test patch:
diff --git a/go/tools/graphviz/graph_test.go b/go/tools/graphviz/graph_test.go
new file mode 100644
index 0000000000..3f9d7958e6
--- /dev/null
+++ b/go/tools/graphviz/graph_test.go
@@ -0,0 +1,38 @@
+/*
+Copyright 2022 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.
+*/
+
+package graphviz
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestSimple(t *testing.T) {
+	g := New()
+	n1 := g.AddNode("apa")
+	n1.AddAttribute("aåö**|")
+	n1.AddAttribute("value")
+	n2 := g.AddNode("banan")
+	g.AddEdge(n1, n2)
+
+	fmt.Println(g.produceDot())
+
+	err := g.Render()
+	require.NoError(t, err)
+}
diff --git a/go/vt/schemadiff/schema_test.go b/go/vt/schemadiff/schema_test.go
index 6412dfca4b..308333bf64 100644
--- a/go/vt/schemadiff/schema_test.go
+++ b/go/vt/schemadiff/schema_test.go
@@ -143,12 +143,12 @@ func TestToSQL(t *testing.T) {
 	assert.Equal(t, toSQL, sql)
 }
 
-func TestClone(t *testing.T) {
+func TestCopy(t *testing.T) {
 	schema, err := NewSchemaFromQueries(createQueries)
 	assert.NoError(t, err)
 	assert.NotNil(t, schema)
 
-	schemaClone := schema.Clone()
+	schemaClone := schema.copy()
 	assert.Equal(t, schema, schemaClone)
 	assert.Equal(t, schema.ToSQL(), schemaClone.ToSQL())
 	assert.False(t, schema == schemaClone)
diff --git a/go/vt/vtgate/planbuilder/plan_test.go b/go/vt/vtgate/planbuilder/plan_test.go
index 3d929e044b..faeca866f1 100644
--- a/go/vt/vtgate/planbuilder/plan_test.go
+++ b/go/vt/vtgate/planbuilder/plan_test.go
@@ -229,28 +229,28 @@ func TestPlan(t *testing.T) {
 	// the column is named as Id. This is to make sure that
 	// column names are case-preserved, but treated as
 	// case-insensitive even if they come from the vschema.
-	testFile(t, "aggr_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "dml_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "from_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "filter_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "postprocess_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "select_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "symtab_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "unsupported_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "vindex_func_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "wireup_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "memory_sort_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "use_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "set_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "union_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "transaction_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "lock_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "large_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "ddl_cases_no_default_keyspace.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "flush_cases_no_default_keyspace.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "show_cases_no_default_keyspace.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "stream_cases.txt", testOutputTempDir, vschemaWrapper)
-	testFile(t, "systemtables_cases.txt", testOutputTempDir, vschemaWrapper)
+	testFile(t, "aggr_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "dml_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "from_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "filter_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "postprocess_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "select_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "symtab_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "unsupported_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "vindex_func_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "wireup_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "memory_sort_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "use_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "set_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "union_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "transaction_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "lock_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "large_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "ddl_cases_no_default_keyspace.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "flush_cases_no_default_keyspace.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "show_cases_no_default_keyspace.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "stream_cases.txt", testOutputTempDir, vschemaWrapper, false)
+	testFile(t, "systemtables_cases.txt", testOutputTempDir, vschemaWrapper, false)
 }
 
 func TestSysVarSetDisabled(t *testing.T) {
@@ -259,7 +259,7 @@ func TestSysVarSetDisabled(t *testing.T) {
 		sysVarEnabled: false,
 	}
 
-	testFile(t, "set_sysvar_disabled_cases.txt", makeTestOutput(t), vschemaWrapper)
+	testFile(t, "set_sysvar_disabled_cases.txt", makeTestOutput(t), vschemaWrapper, false)
 }
 
 func TestOne(t *testing.T) {
@@ -267,7 +267,7 @@ func TestOne(t *testing.T) {
 		v: loadSchema(t, "schema_test.json", true),
 	}
 
-	testFile(t, "onecase.txt", "", vschema)
+	testFile(t, "onecase.txt", "", vschema, true)
 }
 
 func TestOneWithMainAsDefault(t *testing.T) {
@@ -279,7 +279,7 @@ func TestOneWithMainAsDefault(t *testing.T) {
 		},
 	}
 
-	testFile(t, "onecase.txt", "", vschema)
+	testFile(t, "onecase.txt", "", vschema, false)
 }
 
 func TestOneWithSecondUserAsDefault(t *testing.T) {
@@ -291,7 +291,7 @@ func TestOneWithSecondUserAsDefault(t *testing.T) {
 		},
 	}
 
-	testFile(t, "onecase.txt", "", vschema)
+	testFile(t, "onecase.txt", "", vschema, false)
 }
 
 func TestOneWithUserAsDefault(t *testing.T) {
@@ -303,7 +303,7 @@ func TestOneWithUserAsDefault(t *testing.T) {
 		},
 	}
 
-	testFile(t, "onecase.txt", "", vschema)
+	testFile(t, "onecase.txt", "", vschema, false)
 }
 
 func TestRubyOnRailsQueries(t *testing.T) {
@@ -312,7 +312,7 @@ func TestRubyOnRailsQueries(t *testing.T) {
 		sysVarEnabled: true,
 	}
 
-	testFile(t, "rails_cases.txt", makeTestOutput(t), vschemaWrapper)
+	testFile(t, "rails_cases.txt", makeTestOutput(t), vschemaWrapper, false)
 }
 
 func TestOLTP(t *testing.T) {
@@ -321,7 +321,7 @@ func TestOLTP(t *testing.T) {
 		sysVarEnabled: true,
 	}
 
-	testFile(t, "oltp_cases.txt", makeTestOutput(t), vschemaWrapper)
+	testFile(t, "oltp_cases.txt", makeTestOutput(t), vschemaWrapper, false)
 }
 
 func TestTPCC(t *testing.T) {
@@ -330,7 +330,7 @@ func TestTPCC(t *testing.T) {
 		sysVarEnabled: true,
 	}
 
-	testFile(t, "tpcc_cases.txt", makeTestOutput(t), vschemaWrapper)
+	testFile(t, "tpcc_cases.txt", makeTestOutput(t), vschemaWrapper, false)
 }
 
 func TestTPCH(t *testing.T) {
@@ -339,7 +339,7 @@ func TestTPCH(t *testing.T) {
 		sysVarEnabled: true,
 	}
 
-	testFile(t, "tpch_cases.txt", makeTestOutput(t), vschemaWrapper)
+	testFile(t, "tpch_cases.txt", makeTestOutput(t), vschemaWrapper, false)
 }
 
 func BenchmarkOLTP(b *testing.B) {
@@ -382,7 +382,7 @@ func TestBypassPlanningShardTargetFromFile(t *testing.T) {
 		tabletType: topodatapb.TabletType_PRIMARY,
 		dest:       key.DestinationShard("-80")}
 
-	testFile(t, "bypass_shard_cases.txt", makeTestOutput(t), vschema)
+	testFile(t, "bypass_shard_cases.txt", makeTestOutput(t), vschema, false)
 }
 func TestBypassPlanningKeyrangeTargetFromFile(t *testing.T) {
 	keyRange, _ := key.ParseShardingSpec("-")
@@ -397,7 +397,7 @@ func TestBypassPlanningKeyrangeTargetFromFile(t *testing.T) {
 		dest:       key.DestinationExactKeyRange{KeyRange: keyRange[0]},
 	}
 
-	testFile(t, "bypass_keyrange_cases.txt", makeTestOutput(t), vschema)
+	testFile(t, "bypass_keyrange_cases.txt", makeTestOutput(t), vschema, false)
 }
 
 func TestWithDefaultKeyspaceFromFile(t *testing.T) {
@@ -412,12 +412,12 @@ func TestWithDefaultKeyspaceFromFile(t *testing.T) {
 	}
 
 	testOutputTempDir := makeTestOutput(t)
-	testFile(t, "alterVschema_cases.txt", testOutputTempDir, vschema)
-	testFile(t, "ddl_cases.txt", testOutputTempDir, vschema)
-	testFile(t, "migration_cases.txt", testOutputTempDir, vschema)
-	testFile(t, "flush_cases.txt", testOutputTempDir, vschema)
-	testFile(t, "show_cases.txt", testOutputTempDir, vschema)
-	testFile(t, "call_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "alterVschema_cases.txt", testOutputTempDir, vschema, false)
+	testFile(t, "ddl_cases.txt", testOutputTempDir, vschema, false)
+	testFile(t, "migration_cases.txt", testOutputTempDir, vschema, false)
+	testFile(t, "flush_cases.txt", testOutputTempDir, vschema, false)
+	testFile(t, "show_cases.txt", testOutputTempDir, vschema, false)
+	testFile(t, "call_cases.txt", testOutputTempDir, vschema, false)
 }
 
 func TestWithDefaultKeyspaceFromFileSharded(t *testing.T) {
@@ -432,7 +432,7 @@ func TestWithDefaultKeyspaceFromFileSharded(t *testing.T) {
 	}
 
 	testOutputTempDir := makeTestOutput(t)
-	testFile(t, "select_cases_with_default.txt", testOutputTempDir, vschema)
+	testFile(t, "select_cases_with_default.txt", testOutputTempDir, vschema, false)
 }
 
 func TestWithUserDefaultKeyspaceFromFileSharded(t *testing.T) {
@@ -447,7 +447,7 @@ func TestWithUserDefaultKeyspaceFromFileSharded(t *testing.T) {
 	}
 
 	testOutputTempDir := makeTestOutput(t)
-	testFile(t, "select_cases_with_user_as_default.txt", testOutputTempDir, vschema)
+	testFile(t, "select_cases_with_user_as_default.txt", testOutputTempDir, vschema, false)
 }
 
 func TestWithSystemSchemaAsDefaultKeyspace(t *testing.T) {
@@ -458,7 +458,7 @@ func TestWithSystemSchemaAsDefaultKeyspace(t *testing.T) {
 		tabletType: topodatapb.TabletType_PRIMARY,
 	}
 
-	testFile(t, "sysschema_default.txt", makeTestOutput(t), vschema)
+	testFile(t, "sysschema_default.txt", makeTestOutput(t), vschema, false)
 }
 
 func TestOtherPlanningFromFile(t *testing.T) {
@@ -473,8 +473,8 @@ func TestOtherPlanningFromFile(t *testing.T) {
 	}
 
 	testOutputTempDir := makeTestOutput(t)
-	testFile(t, "other_read_cases.txt", testOutputTempDir, vschema)
-	testFile(t, "other_admin_cases.txt", testOutputTempDir, vschema)
+	testFile(t, "other_read_cases.txt", testOutputTempDir, vschema, false)
+	testFile(t, "other_admin_cases.txt", testOutputTempDir, vschema, false)
 }
 
 func loadSchema(t testing.TB, filename string, setCollation bool) *vindexes.VSchema {
@@ -692,7 +692,7 @@ func (vw *vschemaWrapper) currentDb() string {
 	return ksName
 }
 
-func testFile(t *testing.T, filename, tempDir string, vschema *vschemaWrapper) {
+func testFile(t *testing.T, filename, tempDir string, vschema *vschemaWrapper, render bool) {
 	t.Run(filename, func(t *testing.T) {
 		expected := &strings.Builder{}
 		var outFirstPlanner string
@@ -700,6 +700,12 @@ func testFile(t *testing.T, filename, tempDir string, vschema *vschemaWrapper) {
 			t.Run(fmt.Sprintf("%d V3: %s", tcase.lineno, tcase.comments), func(t *testing.T) {
 				vschema.version = V3
 				plan, err := TestBuilder(tcase.input, vschema, vschema.currentDb())
+				if render && plan != nil {
+					viz, err := engine.GraphViz(plan.Instructions)
+					if err == nil {
+						_ = viz.Render()
+					}
+				}
 				out := getPlanOrErrorOutput(err, plan)
 
 				if out != tcase.output {
@@ -717,7 +723,7 @@ func testFile(t *testing.T, filename, tempDir string, vschema *vschemaWrapper) {
 			})
 
 			vschema.version = Gen4
-			out, err := getPlanOutput(tcase, vschema)
+			out, err := getPlanOutput(tcase, vschema, render)
 			if err != nil && tcase.output2ndPlanner == "" && strings.HasPrefix(err.Error(), "gen4 does not yet support") {
 				expected.WriteString("\n")
 				continue
@@ -760,13 +766,19 @@ func testFile(t *testing.T, filename, tempDir string, vschema *vschemaWrapper) {
 	})
 }
 
-func getPlanOutput(tcase testCase, vschema *vschemaWrapper) (out string, err error) {
+func getPlanOutput(tcase testCase, vschema *vschemaWrapper, render bool) (out string, err error) {
 	defer func() {
 		if r := recover(); r != nil {
 			out = fmt.Sprintf("panicked: %v\n%s", r, string(debug.Stack()))
 		}
 	}()
 	plan, err := TestBuilder(tcase.input, vschema, vschema.currentDb())
+	if render && plan != nil {
+		viz, err := engine.GraphViz(plan.Instructions)
+		if err == nil {
+			_ = viz.Render()
+		}
+	}
 	out = getPlanOrErrorOutput(err, plan)
 	return out, err
 }