Repository: vitessio/vitess
Pull request #11333 (2022-09-23 00:22:39)

[release-14.0] Fix complex predicates being pulled into `ON` conditions for `LEFT JOIN` statements
## Description
This is a backport of #11317 & #11310


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.


Patch:
diff --git a/.gitignore b/.gitignore
index 576562a691..6e48b17eca 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,6 +4,9 @@
 # For mac users
 .DS_Store
 
+# For direnv users
+.envrc
+
 # Produced by yacc
 *.output
 
diff --git a/doc/internal/ReleaseInstructions.md b/doc/internal/ReleaseInstructions.md
index 649eb8d411..014487568c 100644
--- a/doc/internal/ReleaseInstructions.md
+++ b/doc/internal/ReleaseInstructions.md
@@ -2,6 +2,17 @@
 
 This page describes the steps for cutting a new [open source release](https://github.com/vitessio/vitess/releases).
 
+### Summary
+
+- [Versioning](#versioning)
+- [Release Branches](#release-branches)
+- [Release Tags](#release-tags)
+- [Docker Images](#docker-images)
+- [Java Packages](#java-packages)
+- [Release Cutover](#release-cutover)
+
+-------
+
 ## Versioning
 
 Our versioning strategy is based on [VEP5](https://github.com/vitessio/enhancements/blob/main/veps/vep-5.md).
@@ -39,6 +50,8 @@ changes) and safe (isolated from active development on `main`).
 
 Pre-release versions should be labeled with a suffix like `-beta2` or `-rc1`.
 
+-------
+
 ## Release Branches
 
 Each major and minor releases (X.Y) should have a [release branch](https://github.com/vitessio/vitess/branches/all?query=release) named
@@ -57,6 +70,8 @@ git push upstream release-X.Y
 The branches are named `release-X.Y` to distinguish them from point-in-time
 tags, which are named `vX.Y.Z`.
 
+-------
+
 ## Release Tags
 
 While the release branch is a moving target, release tags mark point-in-time
@@ -64,10 +79,14 @@ snapshots of the repository. Essentially, a tag assigns a human-readable name to
 a specific Git commit hash. Although it's technically possible to reassign a tag
 name to a different hash, we must never do this.
 
+-------
+
 ## Docker Images
 
 Docker images built automatically on DockerHub and can be found [here](https://hub.docker.com/repository/docker/vitess/lite/).
 
+-------
+
 ## Java Packages
 
 We publish binary packages for our [JDBC driver and Java client on Maven Central](https://search.maven.org/#search|ga|1|g:"io.vitess").
@@ -95,129 +114,167 @@ Install `gpg-agent` (needed below) e.g. on Ubuntu via: `sudo apt-get install gnu
 
 Create the `settings.xml` in the `$HOME/.m2/` directory as described in their [instructions](https://central.sonatype.org/pages/apache-maven.html).
 
-## Release Cutover 
-
-The release cutover section is divided in 4 steps:
-
-1. Pre requisites for either Release or Release Candidate (`rc`).
-2. Creation of Release or Release Candidate (`rc`) on GitHub.
-3. Post release steps.
-4. Java Packages deploy.
-
-### Pre-Requisites for Release Candidates (`rc`)
-
-> In this example our current version is `v11` and we release the version `v12.0.0-rc1`.
-
-* Announce dates on Vitess slack `#release-planning`.
-
-* Fetch `github.com/vitessio/vitess`'s remote.
-    ```shell
-    git fetch <vitessio/vitess remote>
-    ```
-
-* Build a Release Notes document using the Makefile: 
-    ```shell
-    make VERSION="v12.0.0" FROM="<ref/SHA of the latest tag for v11>" TO="<ref/SHA of main>" SUMMARY="./doc/releasenotes/12_0_0_summary.md" release-notes
-    ```
-
-* Check to make sure all labels and categories set for each PR.
-
-    > The Release Notes document can be put aside for now, it will be used when creating the release tag in a future step.
-
-* Create a new release branch from main.
-    ```shell
-    git checkout -b release-12.0 upstream/main
-    ```
-
-* Run the release script using the Makefile:
-    ```shell
-    make RELEASE_VERSION="12.0.0-rc1" DEV_VERSION="12.0.0-SNAPSHOT" do_release
-    ```
-    
-    > The script will prompt you `Pausing so relase notes can be added. Press enter to continue`, execute the next step and press enter.
-
-* Put the Release Notes document in `./doc/releasenotes/12_0_0_release_notes.md`.
-
-* Push the current dev branch to upstream. No pull request required.
-    ```shell
-    git push upstream release-12.0
-    ```
-
-* As prompted in the `do_release` Makefile command's output, push the the `v12.0.0-rc1` tag.
-    ```shell
-    git push upstream v12.0.0-rc1
-    ```
-
-* Release the tag on GitHub UI as explained in the following section.
-
-------
-### Pre-Requisites for Releases
-
-> In this example our current version is `v11` and we release the version `v12.0.0`.
-
-
-* Announce dates on Vitess slack `#release-planning`.
-
-* Fetch `github.com/vitessio/vitess`'s remote.
-    ```shell
-    git fetch <vitessio/vitess remote>
-    ```
-
-* Build a Release Notes document using the Makefile:
-    ```shell
-    make VERSION="v12.0.0" FROM="<ref/SHA of the latest tag for v11>" TO="<ref/SHA of main>" SUMMARY="./doc/releasenotes/12_0_0_summary.md" release-notes
-    ```
+-------
 
-* Check to make sure all labels and categories set for each PR.
+## Release Cutover 
 
-    > The Release Notes document can be put aside for now, it will be used when creating the release tag in a future step.
+In this section we describe our current release process. We begin with a short [**overview**](#overview).
+The release process is divided into three parts: [**Pre-Release**](#pre-release), [**Release**](#release), [**Post-Release**](#post-release), which are detailed after the overview.
+
+### Overview
+
+#### Schedule
+
+A new major version of Vitess is released every four months. For each major version there is at least one release candidate, which we release three weeks before the GA version.
+We usually create the RC1 during the first week of the month, and the GA version three weeks later.
+
+#### Code Freeze
+
+Before creating RC1, there is a code freeze. Assuming the release of RC1 happens on a Tuesday, the release branch will be frozen Friday of the previous week.
+This allows us to test that the release branch can be released and avoid discovering unwanted events during the release day. Once the RC1 is released, there are three more weeks to backport bug fixes into the release branches. However, we also proceed to a code freeze the Friday before the GA release. (Assuming GA is on a Tuesday)
+Regarding patch releases, no code freeze is planned.
+
+#### Tracking Issue for each Release
+
+For each release, it is recommended to create an issue like [this one](https://github.com/vitessio/vitess/issues/10476) to track the current and past progress of a release.
+It also allows us to document what happened during a release.
+
+### Pre-Release
+
+This step happens a few weeks before the actual release (whether it is an RC, GA or a patch release).
+The main goal of this step is to make sure everything is ready to be released for the release day.
+That includes:
+- **Making sure Pull Requests are being reviewed and merged.**
+  > - All the Pull Requests that need to be in the release must be reviewed and merged before the code freeze.
+  > - The code freeze usually happens a few days before the release.
+- **Making sure the people doing the release have access to all the tools and infrastructure needed to do the release.**
+  > - This includes write access to the Vitess repository and to the Maven repository. 
+- **Preparing and cleaning the release notes summary.**
+  > - One or more Pull Requests have to be submitted in advance to create and update the release summary.
+  > - The summary files are located in: `./doc/releasenotes/*_*_*_summary.md`.
+  > - The summary file for a release candidate is the same as the one for the GA release.
+- **Finishing the blog post, and coordinating with the different organizations for cross-posting. Usually CNCF and PlanetScale. This step applies only for GA releases.**
+  > - The blog post must be finished and reviewed.
+  > - A Pull Request on the website repository of Vitess has to be created so we can easily publish the blog during the release day.
+- **Code freeze.**
+  > - As soon as we go into code freeze, if we are doing an RC, create the release branch.
+  > - If we are doing a GA release, do not merge any new Pull Requests.
+  > - The guide on how to do a code freeze is available in the [How To Code Freeze](#how-to-code-freeze) section.
+- **Preparing the Vitess Operator release.**
+  > - While the Vitess Operator is located in a different repository, we also need to do a release for it.
+  > - The Operator follows the same cycle: RC1 -> GA -> Patches.
+  > - Documentation for the pre-release of the Vitess Operator is available [here](https://github.com/planetscale/vitess-operator/blob/main/docs/release-process.md#prepare-for-release).
+
+### Release
+
+On the release day, there are several things to do:
+
+- **Create the Vitess release.**
+  > - A guide on how to create a Vitess release is available in the [How To Release Vitess](#how-to-release-vitess) section.
+- **Create the corresponding Vitess operator release.**
+  > - Applies only to versions greater or equal to `v14.0.0`.
+  > - If we are doing an RC release, then we will need to create the Vitess Operator RC too. If we are doing a GA release, we're also doing a GA release in the Operator.
+  > - The Vitess Operator release documentation is available [here](https://github.com/planetscale/vitess-operator/blob/main/docs/release-process.md#cut-release).
+- **Create the Java release.**
+  > - Applies only to GA releases.
+  > - This step is explained in the [Java Packages: Deploy & Release](#java-packages-deploy--release) section.
+- **Update the website documentation repository.**
+  > - Applies only to GA and RC releases.
+  > - There are two scripts in the website repository in `./tools/{ga|rc}_release.sh`, use them to update the website documentation. The scripts automate:
+  >   - For an RC, we need to create a new version in the sidebar and mark the current version as RC.
+  >   - For a GA, we need to mark the version we are releasing as "Stable" and the next one as "Development".
+- **Publish the blog post on the Vitess website.**
+  > - Applies only to GA releases.
+  > - The corresponding Pull Request was created beforehand during the pre-release. Merge it.
+- **Make sure _arewefastyet_ starts benchmarking the new release.**
+  > - This can be done by visiting [arewefastyet status page](https://benchmark.vitess.io/status).
+  > - New elements should be added to the execution queue.
+  > - After a while, those elements will finish their execution and their status will be green.
+  > - This step is even more important for GA releases as we often include a link to _arewefastyet_ in the blog post.
+  > - The benchmarks need to complete before announcing the blog posts or before they get cross-posted.
+- **Update the release notes on the release branch and on `main`.**
+  > - Two new Pull Requests have to be created.
+  > - One against `main`, it will contain only the new release notes.
+  > - And another against the release branch, this one contains the release notes and the release commit. (The commit on which we did `git tag`) 
+
+### Post-Release
+
+Once the release is over, we need to announce it on both Slack and Twitter. We also want to make sure the blog post was cross-posted, if applicable.
+We need to verify that _arewefastyet_ has finished the benchmark too.
+
+
+### How To Release Vitess
+This section is divided into two parts:
+- [Creation of the tags and release notes](#creation-of-the-tags-and-release-notes).
+- [Creating Release or Release Candidate on the GitHub UI](#creating-release-or-release-candidate-on-the-github-ui)
+
+#### Creation of the tags and release notes
+
+> In this example our current version is `v14` and we release the version `v15.0.0`.
+> Alongside Vitess' release, we also release a new version of the operator.
+> Since we are releasing a release candidate here, the new version of the operator will also be a release candidate.
+> In this example, the new operator version is `2.8.0`.
+> 
+> It is important to note that before the RC, there is a code freeze during which we create the release branch.
+>
+> The release branch in this example is `release-15.0`.
+> 
+> The example also assumes that `origin` is the `vitessio/vitess` remote.
 
-* Make sure to stand in the latest commit of the release branch (`upstream/release-12.0` in our example).
-  Then, checkout to a new branch that will be used to push our release commits.
+1. Fetch `github.com/vitessio/vitess`'s remote.
     ```shell
-    git checkout -b at-release-12.0.0 upstream/release-12.0
+    git fetch origin
     ```
 
-* Run the release script using the Makefile:
-    ```
-    make RELEASE_VERSION="12.0.0" GODOC_RELEASE_VERSION="0.12.0" DEV_VERSION="12.0.1-SNAPSHOT" do_release
-    ```
+2. Creation of the release notes and tags.
+   1. Run the release script using the Makefile:
+      1. Release Candidate:
+          ```shell
+          make BASE_BRANCH="release-15.0" BASE_REMOTE="origin" RELEASE_VERSION="15.0.0-rc1" DEV_VERSION="15.0.0-SNAPSHOT" VTOP_VERSION="2.8.0-rc1" do_release
+          ```
+      2. General Availability:
+         ```shell
+         make BASE_BRANCH="release-15.0" BASE_REMOTE="origin" RELEASE_VERSION="15.0.0" DEV_VERSION="15.0.1-SNAPSHOT" VTOP_VERSION="2.8.0" do_release
+         ```
 
-    > The script will prompt you `Pausing so relase notes can be added. Press enter to continue`, execute the next step and press enter.
+      The script will prompt you `Pausing so release notes can be added. Press enter to continue`. We are now going to generate the release notes, continue to the next sub-step.
 
-* Put the Release Notes document in `./doc/releasenotes/12_0_0_release_notes.md`.
+   2. Run the following command to generate the release notes:
+      1. Release Candidate:
+          ```shell
+          make VERSION="v15.0.0-rc1" FROM="v14.0.0" TO="HEAD" SUMMARY="./doc/releasenotes/15_0_0_summary.md" release-notes  
+          ```
+      2. General Availability:
+          ```shell
+          make VERSION="v15.0.0-rc1" FROM="v14.0.0" TO="HEAD" SUMMARY="./doc/releasenotes/15_0_0_summary.md" release-notes  
+          ```
+      This command will generate the release notes by looking at all the commits between the tag `v14.0.0` and the reference `HEAD`.
+      It will also use the file located in `./doc/releasenotes/15_0_0_summary.md` to prefix the release notes with a text that the maintainers wrote before the release.
 
-* Put the output in `./doc/releasenotes/12_0_0_release_notes.md`.
 
-* Push your current branch and create a pull request against the existing `release-12.0` branch.
-    ```shell
-    git push origin at-release-12.0.0
-    ```
+3. Follow the instruction prompted by the `do_release` Makefile command's output in order to push the tags, branches and create the Pull Requests.
 
-* As prompted in the `do_release` Makefile command's output, push the `v12.0.0` and `v0.12.0` tags.
-    ```shell
-    git push upstream v12.0.0 && git push upstream v0.12.0
-    ```
+4. Create a Pull Request against the `main` branch with the newly created release notes.
 
-* Release the tag on GitHub UI as explained in the following section.
+5. Release the tag on GitHub UI as explained in the following section.
 
-### Creating Release or Release Candidate
+#### Creating Release or Release Candidate on the GitHub UI
 
 > In the below steps, we use `v8.0.0` and `v9.0.0` as an example.
 
-#### 1. Open the releases page
+##### 1. Open the releases page
 
 On Vitess' GitHub repository main page, click on Code -> [Releases](https://github.com/vitessio/vitess/releases).
 
 ![alt text](.images/release-01.png)
 
-#### 2. Draft a new release
+##### 2. Draft a new release
 
 On the Releases page, click on `Draft a new release`.
 
 ![alt text](.images/release-02.png)
 
-#### 3. Tag a new release
+##### 3. Tag a new release
 
 When drafting a new release, we are asked to choose the release's tag and branch.
 We format the tag this way: `v9.0.0`. We append `-rcN` to the tag name for release candidates,
@@ -225,7 +282,7 @@ with `N` being the increment of the release candidate.
 
 ![alt text](.images/release-03.png)
 
-#### 4. Add release notes and release
+##### 4. Add release notes and release
 
 Copy/paste the previously built Release Notes into the description of the release.
 
@@ -235,21 +292,32 @@ And finally, click on `Publish release`.
 
 ![alt text](.images/release-04.png)
 
-### Post Release Steps
-* Announce new release in Vitess Slack `#general` channel. 
-* Create a new [PR](https://github.com/vitessio/website/pull/670) for Vitess Blog. 
-* Netlify -> Sites →vitess.io → Deploy ![GitHub Post Release01](/doc/internal/.images/post-release-01.png)
-* Coordinate CNCF cross-posting Vitess Blog. 
-* Schedule and publish Tweet on Vitess account.
-* Run following script to once the `base` Docker image is live. 
+### How To Code Freeze
+
+In this example we are going to do a code freeze on the `release-15.0` branch. If we are doing a release candidate, there won't be a branch yet, hence we need to create it.
+
+```
+git fetch --all
+git checkout -b release-15.0 origin/main
+```
+
+The new branch will be based on `origin/main`, here `origin` points to `vitessio/vitess`. If we are not doing a release candidate, then the branch already exists and we can checkout on it.
+
+Now, if we are doing a GA release, let's update the branch:
+
+```
+git pull origin release-15.0
+```
+
+Finally, let's run the code freeze script:
 
 ```
-https://github.com/vitessio/vitess/blob/main/docker/release.sh
+./tools/code_freeze.sh freeze release-15.0
 ```
 
-* Deploy and release Java packages by following the `Java Packages Deploy & Release` section below. 
+The script will prompt the command that will allow you to push the code freeze change. Once pushed, open a PR that will be merged on `release-15.0`.
 
-### Java Packages Deploy & Release
+### Java Packages: Deploy & Release
 
 > **Warning:** This section's steps need to be executed only when releasing a new major version of Vitess,
 > or if the Java packages changed from one minor/patch version to another.
diff --git a/go/mysql/binlog_event_rbr.go b/go/mysql/binlog_event_rbr.go
index 20c41acfe4..3403be2757 100644
--- a/go/mysql/binlog_event_rbr.go
+++ b/go/mysql/binlog_event_rbr.go
@@ -709,6 +709,14 @@ func CellValue(data []byte, pos int, typ byte, metadata uint16, field *querypb.F
 
 		// now see if we have a fraction
 		if scale == 0 {
+			// When the field is a DECIMAL using a scale of 0, e.g.
+			// DECIMAL(5,0), a binlogged value of 0 is almost treated
+			// like the NULL byte and we get a 0 byte length value.
+			// In this case let's return the correct value of 0.
+			if txt.Len() == 0 {
+				txt.WriteRune('0')
+			}
+
 			return sqltypes.MakeTrusted(querypb.Type_DECIMAL,
 				txt.Bytes()), l, nil
 		}
diff --git a/go/tools/release-notes/release_notes.go b/go/tools/release-notes/release_notes.go
index b7777c854a..2e1fd07e4a 100644
--- a/go/tools/release-notes/release_notes.go
+++ b/go/tools/release-notes/release_notes.go
@@ -116,7 +116,7 @@ The entire changelog for this release can be found [here]({{ .PathToChangeLogFil
 {{- range $component := $type.Components }} 
 #### {{ $component.Name }}
 {{- range $prInfo := $component.PrInfos }}
- * {{ $prInfo.Title }} #{{ $prInfo.Number }}
+ * {{ $prInfo.Title }} [#{{ $prInfo.Number }}](https://github.com/vitessio/vitess/pull/{{ $prInfo.Number }})
 {{- end }}
 {{- end }}
 {{- end }}
@@ -495,7 +495,7 @@ func main() {
 	flag.Parse()
 
 	// The -version flag must be of a valid format.
-	rx := regexp.MustCompile("v([0-9]+)\\.([0-9]+)\\.([0-9]+)")
+	rx := regexp.MustCompile(`v([0-9]+)\.([0-9]+)\.([0-9]+)`)
 	// There should be 4 sub-matches, input: "v14.0.0", output: ["v14.0.0", "14", "0", "0"].
 	versionMatch := rx.FindStringSubmatch(*versionName)
 	if len(versionMatch) != 4 {
diff --git a/go/vt/vtgate/engine/route.go b/go/vt/vtgate/engine/route.go
index 6d4484b9ba..053d0da45c 100644
--- a/go/vt/vtgate/engine/route.go
+++ b/go/vt/vtgate/engine/route.go
@@ -384,7 +384,7 @@ func (route *Route) sort(in *sqltypes.Result) (*sqltypes.Result, error) {
 		return true
 	})
 
-	return out, err
+	return out.Truncate(route.TruncateColumnCount), err
 }
 
 func (route *Route) description() PrimitiveDescription {
diff --git a/go/vt/vtgate/planbuilder/abstract/filter.go b/go/vt/vtgate/planbuilder/abstract/filter.go
index 0408f46bc1..46b564e7b9 100644
--- a/go/vt/vtgate/planbuilder/abstract/filter.go
+++ b/go/vt/vtgate/planbuilder/abstract/filter.go
@@ -48,8 +48,20 @@ func (f *Filter) CheckValid() error {
 
 // PushPredicate implements the LogicalOperator interface
 func (f *Filter) PushPredicate(expr sqlparser.Expr, semTable *semantics.SemTable) (LogicalOperator, error) {
-	f.Predicates = append(f.Predicates, expr)
-	return f, nil
+	op, err := f.Source.PushPredicate(expr, semTable)
+	if err != nil {
+		return nil, err
+	}
+
+	if filter, isFilter := op.(*Filter); isFilter {
+		filter.Predicates = append(f.Predicates, filter.Predicates...)
+		return filter, err
+	}
+
+	return &Filter{
+		Source:     op,
+		Predicates: f.Predicates,
+	}, nil
 }
 
 // Compact implements the LogicalOperator interface
diff --git a/go/vt/vtgate/planbuilder/abstract/join.go b/go/vt/vtgate/planbuilder/abstract/join.go
index 55ad992f3e..b15c0b780d 100644
--- a/go/vt/vtgate/planbuilder/abstract/join.go
+++ b/go/vt/vtgate/planbuilder/abstract/join.go
@@ -46,22 +46,10 @@ func (j *Join) PushPredicate(expr sqlparser.Expr, semTable *semantics.SemTable)
 		}
 		j.LHS = lhs
 		return j, nil
+
 	case deps.IsSolvedBy(j.RHS.TableID()):
-		// we are looking for predicates like `tbl.col = <>` or `<> = tbl.col`,
-		// where tbl is on the rhs of the left outer join
-		if cmp, isCmp := expr.(*sqlparser.ComparisonExpr); isCmp && cmp.Operator != sqlparser.NullSafeEqualOp &&
-			(sqlparser.IsColName(cmp.Left) && semTable.RecursiveDeps(cmp.Left).IsSolvedBy(j.RHS.TableID()) ||
-				sqlparser.IsColName(cmp.Right) && semTable.RecursiveDeps(cmp.Right).IsSolvedBy(j.RHS.TableID())) {
-			// When the predicate we are pushing is using information from an outer table, we can
-			// check whether the predicate is "null-intolerant" or not. Null-intolerant in this context means that
-			// the predicate will not return true if the table columns are null.
-			// Since an outer join is an inner join with the addition of all the rows from the left-hand side that
-			// matched no rows on the right-hand, if we are later going to remove all the rows where the right-hand
-			// side did not match, we might as well turn the join into an inner join.
-
-			// This is based on the paper "Canonical Abstraction for Outerjoin Optimization" by J Rao et al
-			j.LeftJoin = false
-		}
+		j.tryConvertToInnerJoin(expr, semTable)
+
 		if !j.LeftJoin {
 			rhs, err := j.RHS.PushPredicate(expr, semTable)
 			if err != nil {
@@ -70,19 +58,68 @@ func (j *Join) PushPredicate(expr sqlparser.Expr, semTable *semantics.SemTable)
 			j.RHS = rhs
 			return j, err
 		}
+
 		op := &Filter{
 			Source:     j,
 			Predicates: []sqlparser.Expr{expr},
 		}
 		return op, nil
+
 	case deps.IsSolvedBy(j.LHS.TableID().Merge(j.RHS.TableID())):
-		j.Predicate = sqlparser.AndExpressions(j.Predicate, expr)
-		return j, nil
+		j.tryConvertToInnerJoin(expr, semTable)
+
+		if !j.LeftJoin {
+			j.Predicate = sqlparser.AndExpressions(j.Predicate, expr)
+			return j, nil
+		}
+
+		op := &Filter{
+			Source:     j,
+			Predicates: []sqlparser.Expr{expr},
+		}
+		return op, nil
 	}
 
 	return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "Cannot push predicate: %s", sqlparser.String(expr))
 }
 
+// When a predicate uses information from an outer table, we can convert from an outer join to an inner join
+// if the predicate is "null-intolerant".
+//
+// Null-intolerant in this context means that the predicate will not be true if the table columns are null.
+//
+// Since an outer join is an inner join with the addition of all the rows from the left-hand side that
+// matched no rows on the right-hand, if we are later going to remove all the rows where the right-hand
+// side did not match, we might as well turn the join into an inner join.
+//
+// This is based on the paper "Canonical Abstraction for Outerjoin Optimization" by J Rao et al
+func (j *Join) tryConvertToInnerJoin(expr sqlparser.Expr, semTable *semantics.SemTable) {
+	if !j.LeftJoin {
+		return
+	}
+
+	switch expr := expr.(type) {
+	case *sqlparser.ComparisonExpr:
+		if expr.Operator == sqlparser.NullSafeEqualOp {
+			return
+		}
+
+		if sqlparser.IsColName(expr.Left) && semTable.RecursiveDeps(expr.Left).IsSolvedBy(j.RHS.TableID()) ||
+			sqlparser.IsColName(expr.Right) && semTable.RecursiveDeps(expr.Right).IsSolvedBy(j.RHS.TableID()) {
+			j.LeftJoin = false
+		}
+
+	case *sqlparser.IsExpr:
+		if expr.Right != sqlparser.IsNotNullOp {
+			return
+		}
+
+		if sqlparser.IsColName(expr.Left) && semTable.RecursiveDeps(expr.Left).IsSolvedBy(j.RHS.TableID()) {
+			j.LeftJoin = false
+		}
+	}
+}
+
 // TableID implements the Operator interface
 func (j *Join) TableID() semantics.TableSet {
 	return j.RHS.TableID().Merge(j.LHS.TableID())


Test patch:
diff --git a/go/test/endtoend/vreplication/config_test.go b/go/test/endtoend/vreplication/config_test.go
index de36971e93..dd566e35f5 100644
--- a/go/test/endtoend/vreplication/config_test.go
+++ b/go/test/endtoend/vreplication/config_test.go
@@ -34,7 +34,7 @@ var (
 create table product(pid int, description varbinary(128), date1 datetime not null default '0000-00-00 00:00:00', date2 datetime not null default '2021-00-01 00:00:00', primary key(pid)) CHARSET=utf8mb4;
 create table customer(cid int, name varchar(128) collate utf8mb4_bin, meta json default null, typ enum('individual','soho','enterprise'), sport set('football','cricket','baseball'),
 	ts timestamp not null default current_timestamp, bits bit(2) default b'11', date1 datetime not null default '0000-00-00 00:00:00', 
-	date2 datetime not null default '2021-00-01 00:00:00', primary key(cid,typ)) CHARSET=utf8mb4;
+	date2 datetime not null default '2021-00-01 00:00:00', dec80 decimal(8,0), primary key(cid,typ)) CHARSET=utf8mb4;
 create table customer_seq(id int, next_id bigint, cache bigint, primary key(id)) comment 'vitess_sequence';
 create table merchant(mname varchar(128), category varchar(128), primary key(mname)) CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
 create table orders(oid int, cid int, pid int, mname varchar(128), price int, qty int, total int as (qty * price), total2 int as (qty * price) stored, primary key(oid)) CHARSET=utf8;
diff --git a/go/test/endtoend/vreplication/helper_test.go b/go/test/endtoend/vreplication/helper_test.go
index 9ecfa5b050..264c748a5b 100644
--- a/go/test/endtoend/vreplication/helper_test.go
+++ b/go/test/endtoend/vreplication/helper_test.go
@@ -33,6 +33,7 @@ import (
 	"github.com/buger/jsonparser"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/tidwall/gjson"
 
 	"vitess.io/vitess/go/test/endtoend/cluster"
 	"vitess.io/vitess/go/vt/schema"
@@ -43,6 +44,16 @@ import (
 	"vitess.io/vitess/go/sqltypes"
 )
 
+const (
+	defaultTick          = 1 * time.Second
+	defaultTimeout       = 30 * time.Second
+	workflowStateTimeout = 90 * time.Second
+	workflowStateCopying = "Copying" // nolint
+	workflowStateRunning = "Running" // nolint
+	workflowStateStopped = "Stopped" // nolint
+	workflowStateError   = "Error"   // nolint
+)
+
 func execMultipleQueries(t *testing.T, conn *mysql.Conn, database string, lines string) {
 	queries := strings.Split(lines, "\n")
 	for _, query := range queries {
@@ -92,22 +103,74 @@ func checkHealth(t *testing.T, url string) bool {
 	return true
 }
 
-func waitForQueryToExecute(t *testing.T, conn *mysql.Conn, database string, query string, want string) {
-	done := false
-	ticker := time.NewTicker(10 * time.Millisecond)
+func waitForQueryResult(t *testing.T, conn *mysql.Conn, database string, query string, want string) {
+	timer := time.NewTimer(defaultTimeout)
+	defer timer.Stop()
+	for {
+		qr := execVtgateQuery(t, conn, database, query)
+		require.NotNil(t, qr)
+		if want == fmt.Sprintf("%v", qr.Rows) {
+			return
+		}
+		select {
+		case <-timer.C:
+			require.FailNow(t, fmt.Sprintf("query %q on database %q did not return the expected result of %v before the timeout of %s; last seen result: %v",
+				query, database, want, defaultTimeout, qr.Rows))
+		default:
+			time.Sleep(defaultTick)
+		}
+	}
+}
+
+// waitForTabletThrottlingStatus waits for the tablet to return the provided HTTP code for
+// the provided app name in its self check.
+func waitForTabletThrottlingStatus(t *testing.T, tablet *cluster.VttabletProcess, appName string, wantCode int64) {
+	var gotCode int64
+	timer := time.NewTimer(defaultTimeout)
+	defer timer.Stop()
+	for {
+		_, output, err := throttlerCheckSelf(tablet, appName)
+		require.NoError(t, err)
+		require.NotNil(t, output)
+		gotCode, err = jsonparser.GetInt([]byte(output), "StatusCode")
+		require.NoError(t, err)
+		if wantCode == gotCode {
+			// Wait for any cached check values to be cleared and the new
+			// status value to be in effect everywhere before returning.
+			time.Sleep(500 * time.Millisecond)
+			return
+		}
+		select {
+		case <-timer.C:
+			require.FailNow(t, fmt.Sprintf("tablet %q did not return expected status of %d for application %q before the timeout of %s; last seen status: %d",
+				tablet.Name, wantCode, appName, defaultTimeout, gotCode))
+		default:
+			time.Sleep(defaultTick)
+		}
+	}
+}
+
+// waitForNoWorkflowLag waits for the VReplication workflow's MaxVReplicationTransactionLag
+// value to be 0.
+func waitForNoWorkflowLag(t *testing.T, vc *VitessCluster, keyspace, worfklow string) {
+	ksWorkflow := fmt.Sprintf("%s.%s", keyspace, worfklow)
+	lag := int64(0)
+	timer := time.NewTimer(defaultTimeout)
+	defer timer.Stop()
 	for {
+		output, err := vc.VtctlClient.ExecuteCommandWithOutput("Workflow", "--", ksWorkflow, "show")
+		require.NoError(t, err)
+		lag, err = jsonparser.GetInt([]byte(output), "MaxVReplicationTransactionLag")
+		require.NoError(t, err)
+		if lag == 0 {
+			return
+		}
 		select {
-		case <-ticker.C:
-			if done {
-				return
-			}
-			qr := execVtgateQuery(t, conn, database, query)
-			require.NotNil(t, qr)
-			if want == fmt.Sprintf("%v", qr.Rows) {
-				done = true
-			}
-		case <-time.After(5 * time.Second):
-			require.FailNow(t, "query %s.%s did not execute in time", database, query)
+		case <-timer.C:
+			require.FailNow(t, fmt.Sprintf("workflow %q did not eliminate VReplication lag before the timeout of %s; last seen MaxVReplicationTransactionLag: %d",
+				ksWorkflow, defaultTimeout, lag))
+		default:
+			time.Sleep(defaultTick)
 		}
 	}
 }
@@ -120,28 +183,51 @@ func verifyNoInternalTables(t *testing.T, conn *mysql.Conn, keyspaceShard string
 	require.NotNil(t, qr.Rows)
 	for _, row := range qr.Rows {
 		tableName := row[0].ToString()
-		assert.False(t, schema.IsInternalOperationTableName(tableName), "found internal table %s in shard %s", tableName, keyspaceShard)
+		assert.False(t, schema.IsInternalOperationTableName(tableName), "found internal table %q in shard %q", tableName, keyspaceShard)
 	}
 }
 
-func validateCount(t *testing.T, conn *mysql.Conn, database string, table string, want int) {
-	qr := execVtgateQuery(t, conn, database, fmt.Sprintf("select count(*) from %s", table))
-	require.NotNil(t, qr)
-	require.NotNil(t, qr.Rows)
-	require.Equal(t, fmt.Sprintf("[[INT64(%d)]]", want), fmt.Sprintf("%v", qr.Rows))
-}
-
-func validateQuery(t *testing.T, conn *mysql.Conn, database string, query string, want string) {
-	qr := execVtgateQuery(t, conn, database, query)
-	require.NotNil(t, qr)
-	require.Equal(t, want, fmt.Sprintf("%v", qr.Rows))
+func waitForRowCount(t *testing.T, conn *mysql.Conn, database string, table string, want int) {
+	query := fmt.Sprintf("select count(*) from %s", table)
+	wantRes := fmt.Sprintf("[[INT64(%d)]]", want)
+	timer := time.NewTimer(defaultTimeout)
+	defer timer.Stop()
+	for {
+		qr := execVtgateQuery(t, conn, database, query)
+		require.NotNil(t, qr)
+		if wantRes == fmt.Sprintf("%v", qr.Rows) {
+			return
+		}
+		select {
+		case <-timer.C:
+			require.FailNow(t, fmt.Sprintf("table %q did not reach the expected number of rows (%d) before the timeout of %s; last seen result: %v",
+				table, want, defaultTimeout, qr.Rows))
+		default:
+			time.Sleep(defaultTick)
+		}
+	}
 }
 
-func validateCountInTablet(t *testing.T, vttablet *cluster.VttabletProcess, database string, table string, want int) {
+func waitForRowCountInTablet(t *testing.T, vttablet *cluster.VttabletProcess, database string, table string, want int) {
 	query := fmt.Sprintf("select count(*) from %s", table)
-	qr, err := vttablet.QueryTablet(query, database, true)
-	require.NoError(t, err)
-	require.Equal(t, fmt.Sprintf("[[INT64(%d)]]", want), fmt.Sprintf("%v", qr.Rows))
+	wantRes := fmt.Sprintf("[[INT64(%d)]]", want)
+	timer := time.NewTimer(defaultTimeout)
+	defer timer.Stop()
+	for {
+		qr, err := vttablet.QueryTablet(query, database, true)
+		require.NoError(t, err)
+		require.NotNil(t, qr)
+		if wantRes == fmt.Sprintf("%v", qr.Rows) {
+			return
+		}
+		select {
+		case <-timer.C:
+			require.FailNow(t, fmt.Sprintf("table %q did not reach the expected number of rows (%d) on tablet %q before the timeout of %s; last seen result: %v",
+				table, want, vttablet.Name, defaultTimeout, qr.Rows))
+		default:
+			time.Sleep(defaultTick)
+		}
+	}
 }
 
 func validateThatQueryExecutesOnTablet(t *testing.T, conn *mysql.Conn, tablet *cluster.VttabletProcess, ksName string, query string, matchQuery string) bool {
@@ -152,6 +238,45 @@ func validateThatQueryExecutesOnTablet(t *testing.T, conn *mysql.Conn, tablet *c
 	return newCount == count+1
 }
 
+func waitForWorkflowState(t *testing.T, vc *VitessCluster, ksWorkflow string, wantState string) {
+	done := false
+	timer := time.NewTimer(workflowStateTimeout)
+	log.Infof("Waiting for workflow %q to fully reach %q state", ksWorkflow, wantState)
+	for {
+		output, err := vc.VtctlClient.ExecuteCommandWithOutput("Workflow", ksWorkflow, "show")
+		require.NoError(t, err)
+		done = true
+		state := ""
+		result := gjson.Get(output, "ShardStatuses")
+		result.ForEach(func(tabletId, tabletStreams gjson.Result) bool { // for each participating tablet
+			tabletStreams.ForEach(func(streamId, streamInfos gjson.Result) bool { // for each stream
+				if streamId.String() == "PrimaryReplicationStatuses" {
+					streamInfos.ForEach(func(attributeKey, attributeValue gjson.Result) bool { // for each attribute in the stream
+						state = attributeValue.Get("State").String()
+						if state != wantState {
+							done = false // we need to wait for all streams to have the desired state
+						}
+						return true
+					})
+				}
+				return true
+			})
+			return true
+		})
+		if done {
+			log.Infof("Workflow %q has fully reached the desired state of %q", ksWorkflow, wantState)
+			return
+		}
+		select {
+		case <-timer.C:
+			require.FailNowf(t, "workflow %q did not fully reach the expected state of %q before the timeout of %s; last seen output: %s",
+				ksWorkflow, wantState, workflowStateTimeout, output)
+		default:
+			time.Sleep(defaultTick)
+		}
+	}
+}
+
 func getHTTPBody(url string) string {
 	resp, err := http.Get(url)
 	if err != nil {
@@ -296,7 +421,7 @@ func checkIfDenyListExists(t *testing.T, vc *VitessCluster, ksShard string, tabl
 
 func expectNumberOfStreams(t *testing.T, vtgateConn *mysql.Conn, name string, workflow string, database string, want int) {
 	query := fmt.Sprintf("select count(*) from _vt.vreplication where workflow='%s';", workflow)
-	validateQuery(t, vtgateConn, database, query, fmt.Sprintf(`[[INT64(%d)]]`, want))
+	waitForQueryResult(t, vtgateConn, database, query, fmt.Sprintf(`[[INT64(%d)]]`, want))
 }
 
 func printShardPositions(vc *VitessCluster, ksShards []string) {
@@ -342,3 +467,39 @@ func getDebugVar(t *testing.T, port int, varPath []string) (string, error) {
 	require.NoError(t, err)
 	return string(val), nil
 }
+
+func confirmWorkflowHasCopiedNoData(t *testing.T, targetKS, workflow string) {
+	timer := time.NewTimer(defaultTimeout)
+	defer timer.Stop()
+	ksWorkflow := fmt.Sprintf("%s.%s", targetKS, workflow)
+	for {
+		output, err := vc.VtctlClient.ExecuteCommandWithOutput("Workflow", ksWorkflow, "show")
+		require.NoError(t, err)
+		result := gjson.Get(output, "ShardStatuses")
+		result.ForEach(func(tabletId, tabletStreams gjson.Result) bool { // for each source tablet
+			tabletStreams.ForEach(func(streamId, streamInfos gjson.Result) bool { // for each stream
+				if streamId.String() == "PrimaryReplicationStatuses" {
+					streamInfos.ForEach(func(attributeKey, attributeValue gjson.Result) bool { // for each attribute in the stream
+						state := attributeValue.Get("State").String()
+						pos := attributeValue.Get("Pos").String()
+						// If we've actually copied anything then we'll have a position in the stream
+						if (state == workflowStateRunning || state == workflowStateCopying) && pos != "" {
+							require.FailNowf(t, "Unexpected data copied in workflow",
+								"The MoveTables workflow %q copied data in less than %s when it should have been waiting. Show output: %s",
+								ksWorkflow, defaultTimeout, output)
+						}
+						return true // end attribute loop
+					})
+				}
+				return true // end stream loop
+			})
+			return true // end tablet loop
+		})
+		select {
+		case <-timer.C:
+			return
+		default:
+			time.Sleep(defaultTick)
+		}
+	}
+}
diff --git a/go/test/endtoend/vreplication/materialize_test.go b/go/test/endtoend/vreplication/materialize_test.go
index cd60214d38..0016a0771d 100644
--- a/go/test/endtoend/vreplication/materialize_test.go
+++ b/go/test/endtoend/vreplication/materialize_test.go
@@ -92,8 +92,8 @@ func testShardedMaterialize(t *testing.T) {
 	tab := vc.getPrimaryTablet(t, ks2, "0")
 	catchup(t, tab, "wf1", "Materialize")
 
-	validateCount(t, vtgateConn, ks2, "tx", 2)
-	validateQuery(t, vtgateConn, "ks2:0", "select id, val from tx",
+	waitForRowCount(t, vtgateConn, ks2, "tx", 2)
+	waitForQueryResult(t, vtgateConn, "ks2:0", "select id, val from tx",
 		`[[INT64(3) VARBINARY("def")] [INT64(5) VARBINARY("def")]]`)
 }
 
@@ -214,17 +214,17 @@ func testMaterialize(t *testing.T) {
 	catchup(t, ks2Primary, "wf1", "Materialize")
 
 	// validate data after the copy phase
-	validateCount(t, vtgateConn, targetKs, "mat2", 2)
+	waitForRowCount(t, vtgateConn, targetKs, "mat2", 2)
 	want := `[[INT64(1) VARBINARY("abc") TIMESTAMP("2021-10-09 16:17:36") INT32(9) INT32(10) INT32(3)] [INT64(2) VARBINARY("def") TIMESTAMP("2021-11-10 16:17:36") INT32(10) INT32(11) INT32(6)]]`
-	validateQuery(t, vtgateConn, targetKs, "select id, val, ts, day, month, x from mat2", want)
+	waitForQueryResult(t, vtgateConn, targetKs, "select id, val, ts, day, month, x from mat2", want)
 
 	// insert data to test the replication phase
 	execVtgateQuery(t, vtgateConn, sourceKs, "insert into mat(id, val, ts) values (3, 'ghi', '2021-12-11 16:17:36')")
 
 	// validate data after the replication phase
-	waitForQueryToExecute(t, vtgateConn, targetKs, "select count(*) from mat2", "[[INT64(3)]]")
+	waitForQueryResult(t, vtgateConn, targetKs, "select count(*) from mat2", "[[INT64(3)]]")
 	want = `[[INT64(1) VARBINARY("abc") TIMESTAMP("2021-10-09 16:17:36") INT32(9) INT32(10) INT32(3)] [INT64(2) VARBINARY("def") TIMESTAMP("2021-11-10 16:17:36") INT32(10) INT32(11) INT32(6)] [INT64(3) VARBINARY("ghi") TIMESTAMP("2021-12-11 16:17:36") INT32(11) INT32(12) INT32(9)]]`
-	validateQuery(t, vtgateConn, targetKs, "select id, val, ts, day, month, x from mat2", want)
+	waitForQueryResult(t, vtgateConn, targetKs, "select id, val, ts, day, month, x from mat2", want)
 }
 
 // TestMaterialize runs all the individual materialize tests defined above
diff --git a/go/test/endtoend/vreplication/migrate_test.go b/go/test/endtoend/vreplication/migrate_test.go
index 75d45bf28f..9701f26648 100644
--- a/go/test/endtoend/vreplication/migrate_test.go
+++ b/go/test/endtoend/vreplication/migrate_test.go
@@ -19,7 +19,6 @@ package vreplication
 import (
 	"fmt"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -117,15 +116,14 @@ func TestMigrate(t *testing.T) {
 			"--source=ext1.rating", "create", ksWorkflow); err != nil {
 			t.Fatalf("Migrate command failed with %+v : %s\n", err, output)
 		}
-		time.Sleep(1 * time.Second) // wait for migrate to run
+		waitForWorkflowState(t, vc, ksWorkflow, workflowStateRunning)
 		expectNumberOfStreams(t, vtgateConn, "migrate", "e1", "product:0", 1)
-		validateCount(t, vtgateConn, "product:0", "rating", 2)
-		validateCount(t, vtgateConn, "product:0", "review", 3)
+		waitForRowCount(t, vtgateConn, "product:0", "rating", 2)
+		waitForRowCount(t, vtgateConn, "product:0", "review", 3)
 		execVtgateQuery(t, extVtgateConn, "rating", "insert into review(rid, pid, review) values(4, 1, 'review4');")
 		execVtgateQuery(t, extVtgateConn, "rating", "insert into rating(gid, pid, rating) values(3, 1, 3);")
-		time.Sleep(1 * time.Second) // wait for stream to find row
-		validateCount(t, vtgateConn, "product:0", "rating", 3)
-		validateCount(t, vtgateConn, "product:0", "review", 4)
+		waitForRowCount(t, vtgateConn, "product:0", "rating", 3)
+		waitForRowCount(t, vtgateConn, "product:0", "review", 4)
 		vdiff1(t, ksWorkflow, "extcell1")
 
 		if output, err = vc.VtctlClient.ExecuteCommandWithOutput("Migrate", "complete", ksWorkflow); err != nil {
@@ -142,8 +140,8 @@ func TestMigrate(t *testing.T) {
 			t.Fatalf("Migrate command failed with %+v : %s\n", err, output)
 		}
 		expectNumberOfStreams(t, vtgateConn, "migrate", "e1", "product:0", 1)
-		validateCount(t, vtgateConn, "product:0", "rating", 0)
-		validateCount(t, vtgateConn, "product:0", "review", 0)
+		waitForRowCount(t, vtgateConn, "product:0", "rating", 0)
+		waitForRowCount(t, vtgateConn, "product:0", "review", 0)
 		if output, err = vc.VtctlClient.ExecuteCommandWithOutput("Migrate", "cancel", ksWorkflow); err != nil {
 			t.Fatalf("Migrate command failed with %+v : %s\n", err, output)
 		}
diff --git a/go/test/endtoend/vreplication/performance_test.go b/go/test/endtoend/vreplication/performance_test.go
index cfb708aa18..14d5d6c867 100644
--- a/go/test/endtoend/vreplication/performance_test.go
+++ b/go/test/endtoend/vreplication/performance_test.go
@@ -79,11 +79,11 @@ create table customer(cid int, name varbinary(128), meta json default null, typ
 		}
 	})
 
-	validateCount(t, vtgateConn, "stress_src:0", "largebin", insertCount)
+	waitForRowCount(t, vtgateConn, "stress_src:0", "largebin", insertCount)
 
 	t.Logf("creating new keysepace '%s'", targetKs)
 	vc.AddKeyspace(t, []*Cell{defaultCell}, targetKs, "0", initialStressVSchema, initialStressSchema, 0, 0, 200, nil)
-	validateCount(t, vtgateConn, "stress_tgt:0", "largebin", 0)
+	waitForRowCount(t, vtgateConn, "stress_tgt:0", "largebin", 0)
 
 	t.Logf("moving 'largebin' table...")
 	moveStart := time.Now()
@@ -115,5 +115,5 @@ create table customer(cid int, name varbinary(128), meta json default null, typ
 	}
 
 	t.Logf("finished catching up after MoveTables (%v)", time.Since(moveStart))
-	validateCount(t, vtgateConn, "stress_tgt:0", "largebin", insertCount)
+	waitForRowCount(t, vtgateConn, "stress_tgt:0", "largebin", insertCount)
 }
diff --git a/go/test/endtoend/vreplication/resharding_workflows_v2_test.go b/go/test/endtoend/vreplication/resharding_workflows_v2_test.go
index 0c78dc6231..94db57d8ab 100644
--- a/go/test/endtoend/vreplication/resharding_workflows_v2_test.go
+++ b/go/test/endtoend/vreplication/resharding_workflows_v2_test.go
@@ -20,11 +20,9 @@ import (
 	"fmt"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"github.com/tidwall/gjson"
 
 	"vitess.io/vitess/go/test/endtoend/cluster"
 	"vitess.io/vitess/go/vt/log"
@@ -62,7 +60,7 @@ func createReshardWorkflow(t *testing.T, sourceShards, targetShards string) erro
 	err := tstWorkflowExec(t, defaultCellName, workflowName, targetKs, targetKs,
 		"", workflowActionCreate, "", sourceShards, targetShards)
 	require.NoError(t, err)
-	time.Sleep(1 * time.Second)
+	waitForWorkflowState(t, vc, ksWorkflow, workflowStateRunning)
 	catchup(t, targetTab1, workflowName, "Reshard")
 	catchup(t, targetTab2, workflowName, "Reshard")
 	vdiff1(t, ksWorkflow, "")
@@ -76,9 +74,9 @@ func createMoveTablesWorkflow(t *testing.T, tables string) error {
 	err := tstWorkflowExec(t, defaultCellName, workflowName, sourceKs, targetKs,
 		tables, workflowActionCreate, "", "", "")
 	require.NoError(t, err)
+	waitForWorkflowState(t, vc, ksWorkflow, workflowStateRunning)
 	catchup(t, targetTab1, workflowName, "MoveTables")
 	catchup(t, targetTab2, workflowName, "MoveTables")
-	time.Sleep(1 * time.Second)
 	vdiff1(t, ksWorkflow, "")
 	return nil
 }
@@ -256,47 +254,6 @@ func TestBasicV2Workflows(t *testing.T) {
 	log.Flush()
 }
 
-const workflowStartTimeout = 5 * time.Second
-
-func waitForWorkflowToStart(t *testing.T, ksWorkflow string) {
-	done := false
-	ticker := time.NewTicker(100 * time.Millisecond)
-	timer := time.NewTimer(workflowStartTimeout)
-	log.Infof("Waiting for workflow %s to start", ksWorkflow)
-	for {
-		select {
-		case <-ticker.C:
-			if done {
-				log.Infof("Workflow %s has started", ksWorkflow)
-				return
-			}
-			output, err := vc.VtctlClient.ExecuteCommandWithOutput("Workflow", ksWorkflow, "show")
-			require.NoError(t, err)
-			done = true
-			state := ""
-			result := gjson.Get(output, "ShardStatuses")
-			result.ForEach(func(tabletId, tabletStreams gjson.Result) bool { // for each participating tablet
-				tabletStreams.ForEach(func(streamId, streamInfos gjson.Result) bool { // for each stream
-					if streamId.String() == "PrimaryReplicationStatuses" {
-						streamInfos.ForEach(func(attributeKey, attributeValue gjson.Result) bool { // for each attribute in the stream
-							state = attributeValue.Get("State").String()
-							if state != "Running" {
-								done = false // we need to wait for all streams to start
-							}
-							return true
-						})
-					}
-					return true
-				})
-				return true
-			})
-
-		case <-timer.C:
-			require.FailNowf(t, "workflow %s not yet started", ksWorkflow)
-		}
-	}
-}
-
 /*
 testVSchemaForSequenceAfterMoveTables checks that the related sequence tag is migrated correctly in the vschema
 while moving a table with an auto-increment from sharded to unsharded.
@@ -310,7 +267,7 @@ func testVSchemaForSequenceAfterMoveTables(t *testing.T) {
 		"customer2", workflowActionCreate, "", "", "")
 	require.NoError(t, err)
 
-	waitForWorkflowToStart(t, "customer.wf2")
+	waitForWorkflowState(t, vc, "customer.wf2", workflowStateRunning)
 	waitForLowLag(t, "customer", "wf2")
 
 	err = tstWorkflowExec(t, defaultCellName, "wf2", sourceKs, targetKs,
@@ -324,7 +281,7 @@ func testVSchemaForSequenceAfterMoveTables(t *testing.T) {
 	output, err := vc.VtctlClient.ExecuteCommandWithOutput("GetVSchema", "product")
 	require.NoError(t, err)
 	assert.NotContains(t, output, "customer2\"", "customer2 still found in keyspace product")
-	validateCount(t, vtgateConn, "customer", "customer2", 3)
+	waitForRowCount(t, vtgateConn, "customer", "customer2", 3)
 
 	// check that customer2 has the sequence tag
 	output, err = vc.VtctlClient.ExecuteCommandWithOutput("GetVSchema", "customer")
@@ -336,15 +293,15 @@ func testVSchemaForSequenceAfterMoveTables(t *testing.T) {
 	for i := 0; i < num; i++ {
 		execVtgateQuery(t, vtgateConn, "customer", "insert into customer2(name) values('a')")
 	}
-	validateCount(t, vtgateConn, "customer", "customer2", 3+num)
+	waitForRowCount(t, vtgateConn, "customer", "customer2", 3+num)
 	want := fmt.Sprintf("[[INT32(%d)]]", 100+num-1)
-	validateQuery(t, vtgateConn, "customer", "select max(cid) from customer2", want)
+	waitForQueryResult(t, vtgateConn, "customer", "select max(cid) from customer2", want)
 
 	// use MoveTables to move customer2 back to product. Note that now the table has an associated sequence
 	err = tstWorkflowExec(t, defaultCellName, "wf3", targetKs, sourceKs,
 		"customer2", workflowActionCreate, "", "", "")
 	require.NoError(t, err)
-	waitForWorkflowToStart(t, "product.wf3")
+	waitForWorkflowState(t, vc, "product.wf3", workflowStateRunning)
 
 	waitForLowLag(t, "product", "wf3")
 	err = tstWorkflowExec(t, defaultCellName, "wf3", targetKs, sourceKs,
@@ -368,9 +325,9 @@ func testVSchemaForSequenceAfterMoveTables(t *testing.T) {
 	for i := 0; i < num; i++ {
 		execVtgateQuery(t, vtgateConn, "product", "insert into customer2(name) values('a')")
 	}
-	validateCount(t, vtgateConn, "product", "customer2", 3+num+num)
+	waitForRowCount(t, vtgateConn, "product", "customer2", 3+num+num)
 	want = fmt.Sprintf("[[INT32(%d)]]", 100+num+num-1)
-	validateQuery(t, vtgateConn, "product", "select max(cid) from customer2", want)
+	waitForQueryResult(t, vtgateConn, "product", "select max(cid) from customer2", want)
 }
 
 // testReplicatingWithPKEnumCols ensures that we properly apply binlog events
@@ -387,11 +344,11 @@ func testReplicatingWithPKEnumCols(t *testing.T) {
 	// typ is an enum, with soho having a stored and binlogged value of 2
 	deleteQuery := "delete from customer where cid = 2 and typ = 'soho'"
 	insertQuery := "insert into customer(cid, name, typ, sport, meta) values(2, 'Paül','soho','cricket',convert(x'7b7d' using utf8mb4))"
-	execVtgateQuery(t, vtgateConn, "product", deleteQuery)
-	time.Sleep(2 * time.Second)
+	execVtgateQuery(t, vtgateConn, sourceKs, deleteQuery)
+	waitForNoWorkflowLag(t, vc, targetKs, workflowName)
 	vdiff1(t, ksWorkflow, "")
-	execVtgateQuery(t, vtgateConn, "product", insertQuery)
-	time.Sleep(2 * time.Second)
+	execVtgateQuery(t, vtgateConn, sourceKs, insertQuery)
+	waitForNoWorkflowLag(t, vc, targetKs, workflowName)
 	vdiff1(t, ksWorkflow, "")
 }
 
diff --git a/go/test/endtoend/vreplication/vreplication_test.go b/go/test/endtoend/vreplication/vreplication_test.go
index aa069dd9df..a748ecede5 100644
--- a/go/test/endtoend/vreplication/vreplication_test.go
+++ b/go/test/endtoend/vreplication/vreplication_test.go
@@ -36,7 +36,6 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/buger/jsonparser"
-	"github.com/tidwall/gjson"
 
 	"vitess.io/vitess/go/mysql"
 	"vitess.io/vitess/go/sqltypes"
@@ -62,15 +61,19 @@ var (
 	targetThrottlerAppName = "vreplication"
 )
 
-// for some tests we keep an open transaction during a SwitchWrites and commit it afterwards, to reproduce https://github.com/vitessio/vitess/issues/9400
-// we also then delete the extra row (if) added so that the row counts for the future count comparisons stay the same
 const (
+	// for some tests we keep an open transaction during a SwitchWrites and commit it afterwards, to reproduce https://github.com/vitessio/vitess/issues/9400
+	// we also then delete the extra row (if) added so that the row counts for the future count comparisons stay the same
 	openTxQuery       = "insert into customer(cid, name, typ, sport, meta) values(4, 'openTxQuery',1,'football,baseball','{}');"
 	deleteOpenTxQuery = "delete from customer where name = 'openTxQuery'"
 
-	merchantKeyspace = "merchant-type"
-	maxWait          = 60 * time.Second
-	BypassLagCheck   = true // temporary fix for flakiness seen only in CI when lag check is introduced
+	historyLenQuery = "select count as history_len from information_schema.INNODB_METRICS where name = 'trx_rseg_history_len'"
+
+	merchantKeyspace            = "merchant-type"
+	maxWait                     = 60 * time.Second
+	BypassLagCheck              = true                         // temporary fix for flakiness seen only in CI when lag check is introduced
+	throttlerStatusThrottled    = http.StatusExpectationFailed // 417
+	throttlerStatusNotThrottled = http.StatusOK                // 200
 )
 
 func init() {
@@ -117,8 +120,11 @@ func TestVreplicationCopyThrottling(t *testing.T) {
 	defer vc.TearDown(t)
 	defaultCell = vc.Cells[cell]
 	// To test vstreamer source throttling for the MoveTables operation
-	maxSourceTrxHistory := 5
+	maxSourceTrxHistory := int64(5)
 	extraVTTabletArgs = []string{
+		// We rely on holding open transactions to generate innodb history so extend the timeout
+		// to avoid flakiness when the CI is very slow.
+		fmt.Sprintf("--queryserver-config-transaction-timeout=%d", int64(defaultTimeout.Seconds())*3),
 		fmt.Sprintf("--vreplication_copy_phase_max_innodb_history_list_length=%d", maxSourceTrxHistory),
 	}
 
@@ -138,9 +144,14 @@ func TestVreplicationCopyThrottling(t *testing.T) {
 	// We update rows in a table not part of the MoveTables operation so that we're not blocking
 	// on the LOCK TABLE call but rather the InnoDB History List length.
 	trxConn := generateInnoDBRowHistory(t, sourceKs, maxSourceTrxHistory)
+	// History should have been generated on the source primary tablet
+	waitForInnoDBHistoryLength(t, vc.getPrimaryTablet(t, sourceKs, shard), maxSourceTrxHistory)
 	// We need to force primary tablet types as the history list has been increased on the source primary
 	moveTablesWithTabletTypes(t, defaultCell.Name, workflow, sourceKs, targetKs, table, "primary")
-	verifySourceTabletThrottling(t, targetKs, workflow)
+	// Wait for the copy phase to start
+	waitForWorkflowState(t, vc, fmt.Sprintf("%s.%s", targetKs, workflow), workflowStateCopying)
+	// The initial copy phase should be blocking on the history list
+	confirmWorkflowHasCopiedNoData(t, targetKs, workflow)
 	releaseInnoDBRowHistory(t, trxConn)
 	trxConn.Close()
 }
@@ -430,13 +441,15 @@ func insertInitialData(t *testing.T) {
 		execVtgateQuery(t, vtgateConn, "product:0", "insert into customer_seq2(id, next_id, cache) values(0, 100, 100);")
 		log.Infof("Done inserting initial data")
 
-		validateCount(t, vtgateConn, "product:0", "product", 2)
-		validateCount(t, vtgateConn, "product:0", "customer", 3)
-		validateQuery(t, vtgateConn, "product:0", "select * from merchant",
+		waitForRowCount(t, vtgateConn, "product:0", "product", 2)
+		waitForRowCount(t, vtgateConn, "product:0", "customer", 3)
+		waitForQueryResult(t, vtgateConn, "product:0", "select * from merchant",
 			`[[VARCHAR("Monoprice") VARCHAR("eléctronics")] [VARCHAR("newegg") VARCHAR("elec†ronics")]]`)
 	})
 }
 
+// insertMoreCustomers creates additional customers.
+// Note: this will only work when the customer sequence is in place.
 func insertMoreCustomers(t *testing.T, numCustomers int) {
 	sql := "insert into customer (name) values "
 	i := 0
@@ -492,9 +505,23 @@ func shardCustomer(t *testing.T, testReverse bool, cells []*Cell, sourceCellOrAl
 		customerTab2 := custKs.Shards["80-"].Tablets["zone1-300"].Vttablet
 		productTab := vc.Cells[defaultCell.Name].Keyspaces["product"].Shards["0"].Tablets["zone1-100"].Vttablet
 
+		// Wait to finish the copy phase for all tables
 		catchup(t, customerTab1, workflow, "MoveTables")
 		catchup(t, customerTab2, workflow, "MoveTables")
 
+		// Confirm that the 0 scale decimal field, dec80, is replicated correctly
+		dec80Replicated := false
+		execVtgateQuery(t, vtgateConn, sourceKs, "update customer set dec80 = 0")
+		waitForNoWorkflowLag(t, vc, targetKs, workflow)
+		for _, shard := range []string{"-80", "80-"} {
+			shardTarget := fmt.Sprintf("%s:%s", targetKs, shard)
+			if res := execVtgateQuery(t, vtgateConn, shardTarget, "select cid from customer"); len(res.Rows) > 0 {
+				waitForQueryResult(t, vtgateConn, shardTarget, "select distinct dec80 from customer", `[[DECIMAL(0)]]`)
+				dec80Replicated = true
+			}
+		}
+		require.Equal(t, true, dec80Replicated)
+
 		query := "select cid from customer"
 		require.True(t, validateThatQueryExecutesOnTablet(t, vtgateConn, productTab, "product", query, query))
 		insertQuery1 := "insert into customer(cid, name) values(1001, 'tempCustomer1')"
@@ -606,15 +633,15 @@ func shardCustomer(t *testing.T, testReverse bool, cells []*Cell, sourceCellOrAl
 			require.True(t, validateThatQueryExecutesOnTablet(t, vtgateConn, customerTab2, "customer", insertQuery2, matchInsertQuery2))
 
 			execVtgateQuery(t, vtgateConn, "customer", "delete from customer where name like 'tempCustomer%'")
-			validateCountInTablet(t, customerTab1, "customer", "customer", 1)
-			validateCountInTablet(t, customerTab2, "customer", "customer", 2)
-			validateCount(t, vtgateConn, "customer", "customer.customer", 3)
+			waitForRowCountInTablet(t, customerTab1, "customer", "customer", 1)
+			waitForRowCountInTablet(t, customerTab2, "customer", "customer", 2)
+			waitForRowCount(t, vtgateConn, "customer", "customer.customer", 3)
 
 			query = "insert into customer (name, cid) values('george', 5)"
 			execVtgateQuery(t, vtgateConn, "customer", query)
-			validateCountInTablet(t, customerTab1, "customer", "customer", 1)
-			validateCountInTablet(t, customerTab2, "customer", "customer", 3)
-			validateCount(t, vtgateConn, "customer", "customer.customer", 4)
+			waitForRowCountInTablet(t, customerTab1, "customer", "customer", 1)
+			waitForRowCountInTablet(t, customerTab2, "customer", "customer", 3)
+			waitForRowCount(t, vtgateConn, "customer", "customer.customer", 4)
 		}
 	})
 }
@@ -622,56 +649,21 @@ func shardCustomer(t *testing.T, testReverse bool, cells []*Cell, sourceCellOrAl
 func validateRollupReplicates(t *testing.T) {
 	t.Run("validateRollupReplicates", func(t *testing.T) {
 		insertMoreProducts(t)
-		time.Sleep(1 * time.Second)
-		validateCount(t, vtgateConn, "product", "rollup", 1)
-		validateQuery(t, vtgateConn, "product:0", "select rollupname, kount from rollup",
+		waitForRowCount(t, vtgateConn, "product", "rollup", 1)
+		waitForQueryResult(t, vtgateConn, "product:0", "select rollupname, kount from rollup",
 			`[[VARCHAR("total") INT32(5)]]`)
 	})
 }
 
-func verifySourceTabletThrottling(t *testing.T, targetKS, workflow string) {
-	tDuration := time.Duration(15 * time.Second)
-	ticker := time.NewTicker(5 * time.Second)
-	timer := time.NewTimer(tDuration)
-	ksWorkflow := fmt.Sprintf("%s.%s", targetKS, workflow)
-	for {
-		select {
-		case <-ticker.C:
-			output, err := vc.VtctlClient.ExecuteCommandWithOutput("Workflow", ksWorkflow, "show")
-			require.NoError(t, err)
-			result := gjson.Get(output, "ShardStatuses")
-			result.ForEach(func(tabletId, tabletStreams gjson.Result) bool { // for each source tablet
-				tabletStreams.ForEach(func(streamId, streamInfos gjson.Result) bool { // for each stream
-					if streamId.String() == "PrimaryReplicationStatuses" {
-						streamInfos.ForEach(func(attributeKey, attributeValue gjson.Result) bool { // for each attribute in the stream
-							state := attributeValue.Get("State").String()
-							if state != "Copying" {
-								require.FailNowf(t, "Unexpected running workflow stream",
-									"Initial copy phase for the MoveTables workflow %s started in less than %d seconds when it should have been waiting. Show output: %s",
-									ksWorkflow, int(tDuration.Seconds()), output)
-							}
-							return true // end attribute loop
-						})
-					}
-					return true // end stream loop
-				})
-				return true // end tablet loop
-			})
-		case <-timer.C:
-			return
-		}
-	}
-}
-
 func reshardCustomer2to4Split(t *testing.T, cells []*Cell, sourceCellOrAlias string) {
 	t.Run("reshardCustomer2to4Split", func(t *testing.T) {
 		ksName := "customer"
 		counts := map[string]int{"zone1-600": 4, "zone1-700": 5, "zone1-800": 6, "zone1-900": 5}
 		reshard(t, ksName, "customer", "c2c4", "-80,80-", "-40,40-80,80-c0,c0-", 600, counts, nil, cells, sourceCellOrAlias)
-		validateCount(t, vtgateConn, ksName, "customer", 20)
+		waitForRowCount(t, vtgateConn, ksName, "customer", 20)
 		query := "insert into customer (name) values('yoko')"
 		execVtgateQuery(t, vtgateConn, ksName, query)
-		validateCount(t, vtgateConn, ksName, "customer", 21)
+		waitForRowCount(t, vtgateConn, ksName, "customer", 21)
 	})
 }
 
@@ -680,10 +672,10 @@ func reshardMerchant2to3SplitMerge(t *testing.T) {
 		ksName := merchantKeyspace
 		counts := map[string]int{"zone1-1600": 0, "zone1-1700": 2, "zone1-1800": 0}
 		reshard(t, ksName, "merchant", "m2m3", "-80,80-", "-40,40-c0,c0-", 1600, counts, dryRunResultsSwitchWritesM2m3, nil, "")
-		validateCount(t, vtgateConn, ksName, "merchant", 2)
+		waitForRowCount(t, vtgateConn, ksName, "merchant", 2)
 		query := "insert into merchant (mname, category) values('amazon', 'electronics')"
 		execVtgateQuery(t, vtgateConn, ksName, query)
-		validateCount(t, vtgateConn, ksName, "merchant", 3)
+		waitForRowCount(t, vtgateConn, ksName, "merchant", 3)
 
 		var output string
 		var err error
@@ -726,10 +718,10 @@ func reshardMerchant3to1Merge(t *testing.T) {
 		ksName := merchantKeyspace
 		counts := map[string]int{"zone1-2000": 3}
 		reshard(t, ksName, "merchant", "m3m1", "-40,40-c0,c0-", "0", 2000, counts, nil, nil, "")
-		validateCount(t, vtgateConn, ksName, "merchant", 3)
+		waitForRowCount(t, vtgateConn, ksName, "merchant", 3)
 		query := "insert into merchant (mname, category) values('flipkart', 'electronics')"
 		execVtgateQuery(t, vtgateConn, ksName, query)
-		validateCount(t, vtgateConn, ksName, "merchant", 4)
+		waitForRowCount(t, vtgateConn, ksName, "merchant", 4)
 	})
 }
 
@@ -792,7 +784,7 @@ func reshard(t *testing.T, ksName string, tableName string, workflow string, sou
 			if tablets[tabletName] == nil {
 				continue
 			}
-			validateCountInTablet(t, tablets[tabletName], ksName, tableName, count)
+			waitForRowCountInTablet(t, tablets[tabletName], ksName, tableName, count)
 		}
 	})
 }
@@ -817,9 +809,9 @@ func shardOrders(t *testing.T) {
 		switchReads(t, allCellNames, ksWorkflow)
 		switchWrites(t, ksWorkflow, false)
 		dropSources(t, ksWorkflow)
-		validateCountInTablet(t, customerTab1, "customer", "orders", 1)
-		validateCountInTablet(t, customerTab2, "customer", "orders", 2)
-		validateCount(t, vtgateConn, "customer", "orders", 3)
+		waitForRowCountInTablet(t, customerTab1, "customer", "orders", 1)
+		waitForRowCountInTablet(t, customerTab2, "customer", "orders", 2)
+		waitForRowCount(t, vtgateConn, "customer", "orders", 3)
 	})
 }
 
@@ -860,9 +852,9 @@ func shardMerchant(t *testing.T) {
 		}
 		dropSources(t, ksWorkflow)
 
-		validateCountInTablet(t, merchantTab1, merchantKeyspace, "merchant", 1)
-		validateCountInTablet(t, merchantTab2, merchantKeyspace, "merchant", 1)
-		validateCount(t, vtgateConn, merchantKeyspace, "merchant", 2)
+		waitForRowCountInTablet(t, merchantTab1, merchantKeyspace, "merchant", 1)
+		waitForRowCountInTablet(t, merchantTab2, merchantKeyspace, "merchant", 1)
+		waitForRowCount(t, vtgateConn, merchantKeyspace, "merchant", 2)
 	})
 }
 
@@ -881,13 +873,9 @@ func materializeProduct(t *testing.T) {
 		applyVSchema(t, materializeProductVSchema, keyspace)
 		materialize(t, materializeProductSpec)
 		customerTablets := vc.getVttabletsInKeyspace(t, defaultCell, keyspace, "primary")
-		{
-			for _, tab := range customerTablets {
-				catchup(t, tab, workflow, "Materialize")
-			}
-			for _, tab := range customerTablets {
-				validateCountInTablet(t, tab, keyspace, workflow, 5)
-			}
+		for _, tab := range customerTablets {
+			catchup(t, tab, workflow, "Materialize")
+			waitForRowCountInTablet(t, tab, keyspace, workflow, 5)
 		}
 
 		productTablets := vc.getVttabletsInKeyspace(t, defaultCell, "product", "primary")
@@ -897,27 +885,16 @@ func materializeProduct(t *testing.T) {
 				_, body, err := throttleApp(tab, sourceThrottlerAppName)
 				assert.NoError(t, err)
 				assert.Contains(t, body, sourceThrottlerAppName)
-			}
-			// Wait for throttling to take effect (caching will expire by this time):
-			time.Sleep(1 * time.Second)
-			for _, tab := range productTablets {
-				{
-					_, body, err := throttlerCheckSelf(tab, sourceThrottlerAppName)
-					assert.NoError(t, err)
-					assert.Contains(t, body, "417")
-				}
-				{
-					_, body, err := throttlerCheckSelf(tab, targetThrottlerAppName)
-					assert.NoError(t, err)
-					assert.Contains(t, body, "200")
-				}
+
+				// Wait for throttling to take effect (caching will expire by this time):
+				waitForTabletThrottlingStatus(t, tab, sourceThrottlerAppName, throttlerStatusThrottled)
+				waitForTabletThrottlingStatus(t, tab, targetThrottlerAppName, throttlerStatusNotThrottled)
 			}
 			insertMoreProductsForSourceThrottler(t)
 			// To be fair to the test, we give the target time to apply the new changes. We expect it to NOT get them in the first place,
-			time.Sleep(1 * time.Second)
 			// we expect the additional rows to **not appear** in the materialized view
 			for _, tab := range customerTablets {
-				validateCountInTablet(t, tab, keyspace, workflow, 5)
+				waitForRowCountInTablet(t, tab, keyspace, workflow, 5)
 			}
 		})
 		t.Run("unthrottle-app-product", func(t *testing.T) {
@@ -926,68 +903,44 @@ func materializeProduct(t *testing.T) {
 				_, body, err := unthrottleApp(tab, sourceThrottlerAppName)
 				assert.NoError(t, err)
 				assert.Contains(t, body, sourceThrottlerAppName)
-			}
-			// give time for unthrottling to take effect and for target to fetch data
-			time.Sleep(3 * time.Second)
-			for _, tab := range productTablets {
-				{
-					_, body, err := throttlerCheckSelf(tab, sourceThrottlerAppName)
-					assert.NoError(t, err)
-					assert.Contains(t, body, "200")
-				}
+				// give time for unthrottling to take effect and for target to fetch data
+				waitForTabletThrottlingStatus(t, tab, sourceThrottlerAppName, throttlerStatusNotThrottled)
 			}
 			for _, tab := range customerTablets {
-				validateCountInTablet(t, tab, keyspace, workflow, 8)
+				waitForRowCountInTablet(t, tab, keyspace, workflow, 8)
 			}
 		})
 
 		t.Run("throttle-app-customer", func(t *testing.T) {
-			// Now, throttle the streamer on source tablets, insert some rows
+			// Now, throttle vreplication (vcopier/vapplier) on target tablets, and
+			// insert some more rows.
 			for _, tab := range customerTablets {
 				_, body, err := throttleApp(tab, targetThrottlerAppName)
 				assert.NoError(t, err)
 				assert.Contains(t, body, targetThrottlerAppName)
-			}
-			// Wait for throttling to take effect (caching will expire by this time):
-			time.Sleep(1 * time.Second)
-			for _, tab := range customerTablets {
-				{
-					_, body, err := throttlerCheckSelf(tab, targetThrottlerAppName)
-					assert.NoError(t, err)
-					assert.Contains(t, body, "417")
-				}
-				{
-					_, body, err := throttlerCheckSelf(tab, sourceThrottlerAppName)
-					assert.NoError(t, err)
-					assert.Contains(t, body, "200")
-				}
+				// Wait for throttling to take effect (caching will expire by this time):
+				waitForTabletThrottlingStatus(t, tab, targetThrottlerAppName, throttlerStatusThrottled)
+				waitForTabletThrottlingStatus(t, tab, sourceThrottlerAppName, throttlerStatusNotThrottled)
 			}
 			insertMoreProductsForTargetThrottler(t)
-			// To be fair to the test, we give the target time to apply the new changes. We expect it to NOT get them in the first place,
-			time.Sleep(1 * time.Second)
-			// we expect the additional rows to **not appear** in the materialized view
+			// To be fair to the test, we give the target time to apply the new changes.
+			// We expect it to NOT get them in the first place, we expect the additional
+			// rows to **not appear** in the materialized view.
 			for _, tab := range customerTablets {
-				validateCountInTablet(t, tab, keyspace, workflow, 8)
+				waitForRowCountInTablet(t, tab, keyspace, workflow, 8)
 			}
 		})
 		t.Run("unthrottle-app-customer", func(t *testing.T) {
-			// unthrottle on source tablets, and expect the rows to show up
+			// unthrottle on target tablets, and expect the rows to show up
 			for _, tab := range customerTablets {
 				_, body, err := unthrottleApp(tab, targetThrottlerAppName)
 				assert.NoError(t, err)
 				assert.Contains(t, body, targetThrottlerAppName)
 			}
 			// give time for unthrottling to take effect and for target to fetch data
-			time.Sleep(3 * time.Second)
 			for _, tab := range customerTablets {
-				{
-					_, body, err := throttlerCheckSelf(tab, targetThrottlerAppName)
-					assert.NoError(t, err)
-					assert.Contains(t, body, "200")
-				}
-			}
-			for _, tab := range customerTablets {
-				validateCountInTablet(t, tab, keyspace, workflow, 11)
+				waitForTabletThrottlingStatus(t, tab, targetThrottlerAppName, throttlerStatusNotThrottled)
+				waitForRowCountInTablet(t, tab, keyspace, workflow, 11)
 			}
 		})
 	})
@@ -1001,8 +954,8 @@ func materializeRollup(t *testing.T) {
 		productTab := vc.Cells[defaultCell.Name].Keyspaces["product"].Shards["0"].Tablets["zone1-100"].Vttablet
 		materialize(t, materializeRollupSpec)
 		catchup(t, productTab, workflow, "Materialize")
-		validateCount(t, vtgateConn, "product", "rollup", 1)
-		validateQuery(t, vtgateConn, "product:0", "select rollupname, kount from rollup",
+		waitForRowCount(t, vtgateConn, "product", "rollup", 1)
+		waitForQueryResult(t, vtgateConn, "product:0", "select rollupname, kount from rollup",
 			`[[VARCHAR("total") INT32(2)]]`)
 	})
 }
@@ -1014,8 +967,8 @@ func materializeSales(t *testing.T) {
 		materialize(t, materializeSalesSpec)
 		productTab := vc.Cells[defaultCell.Name].Keyspaces["product"].Shards["0"].Tablets["zone1-100"].Vttablet
 		catchup(t, productTab, "sales", "Materialize")
-		validateCount(t, vtgateConn, "product", "sales", 2)
-		validateQuery(t, vtgateConn, "product:0", "select kount, amount from sales",
+		waitForRowCount(t, vtgateConn, "product", "sales", 2)
+		waitForQueryResult(t, vtgateConn, "product:0", "select kount, amount from sales",
 			`[[INT32(1) INT32(10)] [INT32(2) INT32(35)]]`)
 	})
 }
@@ -1028,9 +981,9 @@ func materializeMerchantSales(t *testing.T) {
 		for _, tab := range merchantTablets {
 			catchup(t, tab, workflow, "Materialize")
 		}
-		validateCountInTablet(t, merchantTablets["zone1-400"], merchantKeyspace, "msales", 1)
-		validateCountInTablet(t, merchantTablets["zone1-500"], merchantKeyspace, "msales", 1)
-		validateCount(t, vtgateConn, merchantKeyspace, "msales", 2)
+		waitForRowCountInTablet(t, merchantTablets["zone1-400"], merchantKeyspace, "msales", 1)
+		waitForRowCountInTablet(t, merchantTablets["zone1-500"], merchantKeyspace, "msales", 1)
+		waitForRowCount(t, vtgateConn, merchantKeyspace, "msales", 2)
 	})
 }
 
@@ -1044,9 +997,9 @@ func materializeMerchantOrders(t *testing.T) {
 		for _, tab := range merchantTablets {
 			catchup(t, tab, workflow, "Materialize")
 		}
-		validateCountInTablet(t, merchantTablets["zone1-400"], merchantKeyspace, "morders", 2)
-		validateCountInTablet(t, merchantTablets["zone1-500"], merchantKeyspace, "morders", 1)
-		validateCount(t, vtgateConn, merchantKeyspace, "morders", 3)
+		waitForRowCountInTablet(t, merchantTablets["zone1-400"], merchantKeyspace, "morders", 2)
+		waitForRowCountInTablet(t, merchantTablets["zone1-500"], merchantKeyspace, "morders", 1)
+		waitForRowCount(t, vtgateConn, merchantKeyspace, "morders", 3)
 	})
 }
 
@@ -1225,14 +1178,15 @@ func dropSources(t *testing.T, ksWorkflow string) {
 //  2. All other workflows continue to work w/o issue with this MVCC history in place (not used yet)
 // Returns a db connection used for the transaction which you can use for follow-up
 // work, such as rolling it back directly or using the releaseInnoDBRowHistory call.
-func generateInnoDBRowHistory(t *testing.T, sourceKS string, neededTrxHistory int) *mysql.Conn {
+func generateInnoDBRowHistory(t *testing.T, sourceKS string, neededTrxHistory int64) *mysql.Conn {
 	dbConn1 := getConnection(t, vc.ClusterConfig.hostname, vc.ClusterConfig.vtgateMySQLPort)
 	dbConn2 := getConnection(t, vc.ClusterConfig.hostname, vc.ClusterConfig.vtgateMySQLPort)
 	execQuery(t, dbConn1, "use "+sourceKS)
 	execQuery(t, dbConn2, "use "+sourceKS)
-	offset := 1000
+	offset := int64(1000)
+	limit := int64(neededTrxHistory * 100)
 	insertStmt := strings.Builder{}
-	for i := offset; i <= (neededTrxHistory*10)+offset; i++ {
+	for i := offset; i <= offset+limit; i++ {
 		if i == offset {
 			insertStmt.WriteString(fmt.Sprintf("insert into product (pid, description) values (%d, 'test')", i))
 		} else {
@@ -1245,10 +1199,34 @@ func generateInnoDBRowHistory(t *testing.T, sourceKS string, neededTrxHistory in
 	execQuery(t, dbConn2, "rollback")
 	execQuery(t, dbConn2, "start transaction")
 	execQuery(t, dbConn2, "select count(*) from product")
-	execQuery(t, dbConn1, fmt.Sprintf("delete from product where pid >= %d and pid < %d", offset, offset+10000))
+	execQuery(t, dbConn1, fmt.Sprintf("delete from product where pid >= %d and pid <= %d", offset, offset+limit))
 	return dbConn2
 }
 
+// waitForInnoDBRowHistory waits for the history list length to be greater than the
+// expected length.
+func waitForInnoDBHistoryLength(t *testing.T, tablet *cluster.VttabletProcess, expectedLength int64) {
+	timer := time.NewTimer(defaultTimeout)
+	historyLen := int64(0)
+	for {
+		res, err := tablet.QueryTablet(historyLenQuery, tablet.Keyspace, false)
+		require.NoError(t, err)
+		require.NotNil(t, res)
+		require.Equal(t, 1, len(res.Rows))
+		historyLen, err = res.Rows[0][0].ToInt64()
+		require.NoError(t, err)
+		if historyLen > expectedLength {
+			return
+		}
+		select {
+		case <-timer.C:
+			t.Fatalf("Did not reach the expected InnoDB history length of %d before the timeout of %s; last seen value: %d", expectedLength, defaultTimeout, historyLen)
+		default:
+			time.Sleep(defaultTick)
+		}
+	}
+}
+
 func releaseInnoDBRowHistory(t *testing.T, dbConn *mysql.Conn) {
 	execQuery(t, dbConn, "rollback")
 }
diff --git a/go/test/endtoend/vtgate/gen4/gen4_test.go b/go/test/endtoend/vtgate/gen4/gen4_test.go
index a0f1bf8cf2..080204ee29 100644
--- a/go/test/endtoend/vtgate/gen4/gen4_test.go
+++ b/go/test/endtoend/vtgate/gen4/gen4_test.go
@@ -474,7 +474,5 @@ func TestFilterOnLeftOuterJoin(t *testing.T) {
 				  and team_fact.team >= 22
 				)`
 
-	// Ideally we should get `[[INT32(22)] [INT32(33)]]`
-	// Change the expectation when we support better conversion handling in numeric comparison.
-	utils.AssertContainsError(t, mcmp.VtConn, query, `unsupported: cannot compare INT32 and INT64`)
+	utils.AssertMatches(t, mcmp.VtConn, query, `[[INT32(22)] [INT32(33)]]`)
 }
diff --git a/go/test/endtoend/vtgate/queries/orderby/without_schematracker/main_test.go b/go/test/endtoend/vtgate/queries/orderby/without_schematracker/main_test.go
new file mode 100644
index 0000000000..00221e9c9f
--- /dev/null
+++ b/go/test/endtoend/vtgate/queries/orderby/without_schematracker/main_test.go
@@ -0,0 +1,95 @@
+/*
+Copyright 2021 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 orderby
+
+import (
+	_ "embed"
+	"flag"
+	"fmt"
+	"os"
+	"testing"
+
+	"vitess.io/vitess/go/test/endtoend/utils"
+
+	"vitess.io/vitess/go/mysql"
+	"vitess.io/vitess/go/test/endtoend/cluster"
+)
+
+var (
+	clusterInstance *cluster.LocalProcessCluster
+	vtParams        mysql.ConnParams
+	mysqlParams     mysql.ConnParams
+	keyspaceName    = "ks_orderby"
+	cell            = "test_orderby"
+
+	//go:embed schema.sql
+	schemaSQL string
+
+	//go:embed vschema.json
+	vschema string
+)
+
+func TestMain(m *testing.M) {
+	defer cluster.PanicHandler(nil)
+	flag.Parse()
+
+	exitCode := func() int {
+		clusterInstance = cluster.NewCluster(cell, "localhost")
+		defer clusterInstance.Teardown()
+
+		// Start topo server
+		err := clusterInstance.StartTopo()
+		if err != nil {
+			return 1
+		}
+
+		// Start keyspace
+		keyspace := &cluster.Keyspace{
+			Name:      keyspaceName,
+			SchemaSQL: schemaSQL,
+			VSchema:   vschema,
+		}
+		clusterInstance.VtGateExtraArgs = []string{"--schema_change_signal=false"}
+		err = clusterInstance.StartKeyspace(*keyspace, []string{"-80", "80-"}, 1, true)
+		if err != nil {
+			return 1
+		}
+
+		clusterInstance.VtGateExtraArgs = append(clusterInstance.VtGateExtraArgs, "--enable_system_settings=true")
+		// Start vtgate
+		err = clusterInstance.StartVtgate()
+		if err != nil {
+			return 1
+		}
+
+		vtParams = mysql.ConnParams{
+			Host: clusterInstance.Hostname,
+			Port: clusterInstance.VtgateMySQLPort,
+		}
+
+		// create mysql instance and connection parameters
+		conn, closer, err := utils.NewMySQL(clusterInstance, keyspaceName, schemaSQL)
+		if err != nil {
+			fmt.Println(err)
+			return 1
+		}
+		defer closer()
+		mysqlParams = conn
+		return m.Run()
+	}()
+	os.Exit(exitCode)
+}
diff --git a/go/test/endtoend/vtgate/queries/orderby/without_schematracker/orderby_test.go b/go/test/endtoend/vtgate/queries/orderby/without_schematracker/orderby_test.go
new file mode 100644
index 0000000000..77a2ceeb51
--- /dev/null
+++ b/go/test/endtoend/vtgate/queries/orderby/without_schematracker/orderby_test.go
@@ -0,0 +1,58 @@
+/*
+Copyright 2021 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 orderby
+
+import (
+	"testing"
+
+	"vitess.io/vitess/go/test/endtoend/utils"
+
+	"github.com/stretchr/testify/require"
+
+	"vitess.io/vitess/go/test/endtoend/cluster"
+)
+
+func start(t *testing.T) (utils.MySQLCompare, func()) {
+	mcmp, err := utils.NewMySQLCompare(t, vtParams, mysqlParams)
+	require.NoError(t, err)
+
+	deleteAll := func() {
+		_, _ = utils.ExecAllowError(t, mcmp.VtConn, "set workload = oltp")
+
+		tables := []string{"user"}
+		for _, table := range tables {
+			_, _ = mcmp.ExecAndIgnore("delete from " + table)
+		}
+	}
+
+	deleteAll()
+
+	return mcmp, func() {
+		deleteAll()
+		mcmp.Close()
+		cluster.PanicHandler(t)
+	}
+}
+
+func TestSimpleOrderBy(t *testing.T) {
+	mcmp, closer := start(t)
+	defer closer()
+
+	mcmp.Exec("insert into user(id, name) values (0,'Apa'),(1,'Banan'),(3,'Ceasar'),(4,'David')")
+	mcmp.AssertMatches(`SELECT name FROM user WHERE id in (0,4) ORDER BY name ASC`,
+		`[[VARCHAR("Apa")] [VARCHAR("David)]]`)
+}
diff --git a/go/test/endtoend/vtgate/queries/orderby/without_schematracker/schema.sql b/go/test/endtoend/vtgate/queries/orderby/without_schematracker/schema.sql
new file mode 100644
index 0000000000..5274203d7c
--- /dev/null
+++ b/go/test/endtoend/vtgate/queries/orderby/without_schematracker/schema.sql
@@ -0,0 +1,6 @@
+create table user
+(
+    id bigint,
+    name varchar(10),
+    primary key (id)
+) Engine = InnoDB;
\ No newline at end of file
diff --git a/go/test/endtoend/vtgate/queries/orderby/without_schematracker/vschema.json b/go/test/endtoend/vtgate/queries/orderby/without_schematracker/vschema.json
new file mode 100644
index 0000000000..8e040df5af
--- /dev/null
+++ b/go/test/endtoend/vtgate/queries/orderby/without_schematracker/vschema.json
@@ -0,0 +1,25 @@
+{
+  "sharded": true,
+  "vindexes": {
+    "user_index": {
+      "type": "hash",
+      "owner": "user"
+    }
+  },
+  "tables": {
+    "user": {
+      "column_vindexes": [
+        {
+          "column": "Id",
+          "name": "user_index"
+        }
+      ],
+      "columns": [
+        {
+          "name": "id",
+          "type": "INT64"
+        }
+      ]
+    }
+  }
+}
\ No newline at end of file
diff --git a/go/vt/vtgate/planbuilder/testdata/filter_cases.txt b/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
index 35b9d599af..c88a4360dd 100644
--- a/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/filter_cases.txt
@@ -3937,6 +3937,121 @@ Gen4 plan same as above
 }
 Gen4 plan same as above
 
+# For left joins, where conditions using both sides of the join are not pulled into the join conditions
+"SELECT music.id FROM music LEFT OUTER JOIN user ON music.user_id = user.id WHERE (user.name = 'Trent Reznor' OR music.genre = 'pop') AND music.user_id = 5"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT music.id FROM music LEFT OUTER JOIN user ON music.user_id = user.id WHERE (user.name = 'Trent Reznor' OR music.genre = 'pop') AND music.user_id = 5",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "FieldQuery": "select music.id from music left join `user` on music.user_id = `user`.id where 1 != 1",
+    "Query": "select music.id from music left join `user` on music.user_id = `user`.id where music.user_id = 5 and (`user`.`name` = 'Trent Reznor' or music.genre = 'pop')",
+    "Table": "`user`, music",
+    "Values": [
+      "INT64(5)"
+    ],
+    "Vindex": "user_index"
+  }
+}
+Gen4 plan same as above
+
+# For left joins, where conditions using both sides of the join are not pulled into the join conditions (swapped order)
+"SELECT music.id FROM music LEFT OUTER JOIN user ON music.user_id = user.id WHERE music.user_id = 5 AND (user.name = 'Trent Reznor' OR music.genre = 'pop')"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT music.id FROM music LEFT OUTER JOIN user ON music.user_id = user.id WHERE music.user_id = 5 AND (user.name = 'Trent Reznor' OR music.genre = 'pop')",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "FieldQuery": "select music.id from music left join `user` on music.user_id = `user`.id where 1 != 1",
+    "Query": "select music.id from music left join `user` on music.user_id = `user`.id where music.user_id = 5 and (`user`.`name` = 'Trent Reznor' or music.genre = 'pop')",
+    "Table": "`user`, music",
+    "Values": [
+      "INT64(5)"
+    ],
+    "Vindex": "user_index"
+  }
+}
+Gen4 plan same as above
+
+# For left joins, null intolerant where conditions using both sides of the join are transformed to inner joins
+"SELECT music.id FROM music LEFT OUTER JOIN user ON music.user_id = user.id WHERE music.user_id = 5 AND music.componist = user.name"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT music.id FROM music LEFT OUTER JOIN user ON music.user_id = user.id WHERE music.user_id = 5 AND music.componist = user.name",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "FieldQuery": "select music.id from music, `user` where 1 != 1",
+    "Query": "select music.id from music, `user` where music.user_id = 5 and music.user_id = `user`.id and music.componist = `user`.`name`",
+    "Table": "`user`, music",
+    "Values": [
+      "INT64(5)"
+    ],
+    "Vindex": "user_index"
+  }
+}
+Gen4 plan same as above
+
+# For left joins, null intolerant where conditions using `IS NOT NULL` allow outer join simplification
+"SELECT music.id FROM music LEFT OUTER JOIN user ON user.id = music.user_id WHERE music.user_id = 5 AND user.id IS NOT NULL"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT music.id FROM music LEFT OUTER JOIN user ON user.id = music.user_id WHERE music.user_id = 5 AND user.id IS NOT NULL",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "FieldQuery": "select music.id from music, `user` where 1 != 1",
+    "Query": "select music.id from music, `user` where music.user_id = 5 and `user`.id is not null and `user`.id = music.user_id",
+    "Table": "`user`, music",
+    "Values": [
+      "INT64(5)"
+    ],
+    "Vindex": "user_index"
+  }
+}
+Gen4 plan same as above
+
+# conditions following a null safe comparison operator can be used for routing
+"SELECT music.id FROM music LEFT OUTER JOIN user ON music.user_id = user.id WHERE user.id <=> NULL AND music.user_id = 10"
+{
+  "QueryType": "SELECT",
+  "Original": "SELECT music.id FROM music LEFT OUTER JOIN user ON music.user_id = user.id WHERE user.id \u003c=\u003e NULL AND music.user_id = 10",
+  "Instructions": {
+    "OperatorType": "Route",
+    "Variant": "EqualUnique",
+    "Keyspace": {
+      "Name": "user",
+      "Sharded": true
+    },
+    "FieldQuery": "select music.id from music left join `user` on music.user_id = `user`.id where 1 != 1",
+    "Query": "select music.id from music left join `user` on music.user_id = `user`.id where music.user_id = 10 and `user`.id \u003c=\u003e null",
+    "Table": "`user`, music",
+    "Values": [
+      "INT64(10)"
+    ],
+    "Vindex": "user_index"
+  }
+}
+Gen4 plan same as above
+
 # optimize ORs to IN route op codes #1
 "select col from user where id = 1 or id = 2"
 {
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go b/go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go
index 5bc26372c7..cf70af6612 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vcopier_test.go
@@ -449,9 +449,9 @@ func TestPlayerCopyTables(t *testing.T) {
 	defer deleteTablet(addTablet(100))
 
 	execStatements(t, []string{
-		"create table src1(id int, val varbinary(128), primary key(id))",
-		"insert into src1 values(2, 'bbb'), (1, 'aaa')",
-		fmt.Sprintf("create table %s.dst1(id int, val varbinary(128), val2 varbinary(128), primary key(id))", vrepldb),
+		"create table src1(id int, val varbinary(128), d decimal(8,0), primary key(id))",
+		"insert into src1 values(2, 'bbb', 1), (1, 'aaa', 0)",
+		fmt.Sprintf("create table %s.dst1(id int, val varbinary(128), val2 varbinary(128), d decimal(8,0), primary key(id))", vrepldb),
 		"create table yes(id int, val varbinary(128), primary key(id))",
 		fmt.Sprintf("create table %s.yes(id int, val varbinary(128), primary key(id))", vrepldb),
 		"create table no(id int, val varbinary(128), primary key(id))",
@@ -468,7 +468,7 @@ func TestPlayerCopyTables(t *testing.T) {
 	filter := &binlogdatapb.Filter{
 		Rules: []*binlogdatapb.Rule{{
 			Match:  "dst1",
-			Filter: "select id, val, val as val2 from src1",
+			Filter: "select id, val, val as val2, d from src1",
 		}, {
 			Match: "/yes",
 		}},
@@ -504,7 +504,7 @@ func TestPlayerCopyTables(t *testing.T) {
 		// The first fast-forward has no starting point. So, it just saves the current position.
 		"/update _vt.vreplication set pos=",
 		"begin",
-		"insert into dst1(id,val,val2) values (1,'aaa','aaa'), (2,'bbb','bbb')",
+		"insert into dst1(id,val,val2,d) values (1,'aaa','aaa',0), (2,'bbb','bbb',1)",
 		`/update _vt.copy_state set lastpk='fields:{name:\\"id\\" type:INT32} rows:{lengths:1 values:\\"2\\"}' where vrepl_id=.*`,
 		"commit",
 		// copy of dst1 is done: delete from copy_state.
@@ -519,8 +519,8 @@ func TestPlayerCopyTables(t *testing.T) {
 		"/update _vt.vreplication set state='Running'",
 	})
 	expectData(t, "dst1", [][]string{
-		{"1", "aaa", "aaa"},
-		{"2", "bbb", "bbb"},
+		{"1", "aaa", "aaa", "0"},
+		{"2", "bbb", "bbb", "1"},
 	})
 	expectData(t, "yes", [][]string{})
 	validateCopyRowCountStat(t, 2)
diff --git a/go/vt/vttablet/tabletmanager/vreplication/vplayer_flaky_test.go b/go/vt/vttablet/tabletmanager/vreplication/vplayer_flaky_test.go
index 5a01b58882..3e55658e39 100644
--- a/go/vt/vttablet/tabletmanager/vreplication/vplayer_flaky_test.go
+++ b/go/vt/vttablet/tabletmanager/vreplication/vplayer_flaky_test.go
@@ -1409,6 +1409,8 @@ func TestPlayerTypes(t *testing.T) {
 		fmt.Sprintf("create table %s.src1(id int, val varbinary(128), primary key(id))", vrepldb),
 		"create table binary_pk(b binary(4), val varbinary(4), primary key(b))",
 		fmt.Sprintf("create table %s.binary_pk(b binary(4), val varbinary(4), primary key(b))", vrepldb),
+		"create table vitess_decimal(id int, d1 decimal(8,0) default null, d2 decimal(8,0) default null, d3 decimal(8,0) default null, d4 decimal(8, 1), d5 decimal(8, 1), d6 decimal(8, 1), primary key(id))",
+		fmt.Sprintf("create table %s.vitess_decimal(id int, d1 decimal(8,0) default null, d2 decimal(8,0) default null, d3 decimal(8,0) default null, d4 decimal(8, 1), d5 decimal(8, 1), d6 decimal(8, 1), primary key(id))", vrepldb),
 	})
 	defer execStatements(t, []string{
 		"drop table vitess_ints",
@@ -1425,6 +1427,8 @@ func TestPlayerTypes(t *testing.T) {
 		fmt.Sprintf("drop table %s.src1", vrepldb),
 		"drop table binary_pk",
 		fmt.Sprintf("drop table %s.binary_pk", vrepldb),
+		"drop table vitess_decimal",
+		fmt.Sprintf("drop table %s.vitess_decimal", vrepldb),
 	})
 	if enableJSONColumnTesting {
 		execStatements(t, []string{
@@ -1500,6 +1504,13 @@ func TestPlayerTypes(t *testing.T) {
 		data: [][]string{
 			{"a\000\000\000", "aaa"},
 		},
+	}, {
+		input:  "insert into vitess_decimal values(1, 0, 1, null, 0, 1.1, 1)",
+		output: "insert into vitess_decimal(id,d1,d2,d3,d4,d5,d6) values (1,0,1,null,.0,1.1,1.0)",
+		table:  "vitess_decimal",
+		data: [][]string{
+			{"1", "0", "1", "", "0.0", "1.1", "1.0"},
+		},
 	}, {
 		// Binary pk is a special case: https://github.com/vitessio/vitess/issues/3984
 		input:  "update binary_pk set val='bbb' where b='a\\0\\0\\0'",