Repository: vitessio/vitess
Pull request #11587 (2022-10-26 20:01:26)

[cherry-pick]Adding deprecate message to backup hooks (#11491)
## Description
Cherry-pick https://github.com/vitessio/vitess/pull/11491

## Related Issue(s)

## Checklist

-   [ ] "Backport me!" label has been added if this change should be backported
-   [ ] Tests were added or are not required
-   [ ] Documentation was added or is not required

## Deployment Notes

<!-- Notes regarding deployment of the contained body of work. These should note any db migrations, etc. -->


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

#### General
- [ ] Ensure that the Pull Request has a descriptive title.
- [ ] If this is a change that users need to know about, please apply the `release notes (needs details)` label so that merging is blocked unless the summary release notes document is included.

#### If a new flag is being introduced:
- [ ] Is it really necessary to add this flag?
- [ ] Flag names should be clear and intuitive (as far as possible)
- [ ] Help text should be descriptive.
- [ ] Flag names should use dashes (`-`) as word separators rather than underscores (`_`).

#### If a workflow is added or modified:
- [ ] Each item in `Jobs` should be named in order to mark it as `required`.
- [ ] If the workflow should be required, the maintainer team should be notified.

#### Bug fixes
- [ ] There should be at least one unit or end-to-end test.
- [ ] The Pull Request description should include a link to an issue that describes the bug.

#### 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.
- [ ] RPC changes should be compatible with vitess-operator
- [ ] If a flag is removed, then it should also be removed from VTop, if used there.


Patch:
diff --git a/go/flags/endtoend/mysqlctl.txt b/go/flags/endtoend/mysqlctl.txt
index 9ab3a991eb..9dd71ce504 100644
--- a/go/flags/endtoend/mysqlctl.txt
+++ b/go/flags/endtoend/mysqlctl.txt
@@ -17,7 +17,6 @@ Global flags:
       --backup_engine_implementation string                              Specifies which implementation to use for creating new backups (builtin or xtrabackup). Restores will always be done with whichever engine created a given backup. (default "builtin")
       --backup_storage_block_size int                                    if backup_storage_compress is true, backup_storage_block_size sets the byte size for each block while compressing (default is 250000). (default 250000)
       --backup_storage_compress                                          if set, the backup files will be compressed (default is true). Set to false for instance if a backup_storage_hook is specified and it compresses the data. (default true)
-      --backup_storage_hook string                                       if set, we send the contents of the backup files through this hook.
       --backup_storage_number_blocks int                                 if backup_storage_compress is true, backup_storage_number_blocks sets the number of blocks that can be processed, at once, before the writer blocks, during compression (default is 2). It should be equal to the number of CPUs available for compression. (default 2)
       --builtinbackup_mysqld_timeout duration                            how long to wait for mysqld to shutdown at the start of the backup. (default 10m0s)
       --builtinbackup_progress duration                                  how often to send progress updates when backing up large files. (default 5s)
diff --git a/go/flags/endtoend/mysqlctld.txt b/go/flags/endtoend/mysqlctld.txt
index 9405bd323b..91b423ead2 100644
--- a/go/flags/endtoend/mysqlctld.txt
+++ b/go/flags/endtoend/mysqlctld.txt
@@ -5,7 +5,6 @@ Usage of mysqlctld:
       --backup_engine_implementation string                              Specifies which implementation to use for creating new backups (builtin or xtrabackup). Restores will always be done with whichever engine created a given backup. (default "builtin")
       --backup_storage_block_size int                                    if backup_storage_compress is true, backup_storage_block_size sets the byte size for each block while compressing (default is 250000). (default 250000)
       --backup_storage_compress                                          if set, the backup files will be compressed (default is true). Set to false for instance if a backup_storage_hook is specified and it compresses the data. (default true)
-      --backup_storage_hook string                                       if set, we send the contents of the backup files through this hook.
       --backup_storage_number_blocks int                                 if backup_storage_compress is true, backup_storage_number_blocks sets the number of blocks that can be processed, at once, before the writer blocks, during compression (default is 2). It should be equal to the number of CPUs available for compression. (default 2)
       --builtinbackup_mysqld_timeout duration                            how long to wait for mysqld to shutdown at the start of the backup. (default 10m0s)
       --builtinbackup_progress duration                                  how often to send progress updates when backing up large files. (default 5s)
diff --git a/go/flags/endtoend/vtbackup.txt b/go/flags/endtoend/vtbackup.txt
index bbbcd5cd20..aacde08d4d 100644
--- a/go/flags/endtoend/vtbackup.txt
+++ b/go/flags/endtoend/vtbackup.txt
@@ -9,7 +9,6 @@ Usage of vtbackup:
       --backup_engine_implementation string             Specifies which implementation to use for creating new backups (builtin or xtrabackup). Restores will always be done with whichever engine created a given backup. (default "builtin")
       --backup_storage_block_size int                   if backup_storage_compress is true, backup_storage_block_size sets the byte size for each block while compressing (default is 250000). (default 250000)
       --backup_storage_compress                         if set, the backup files will be compressed (default is true). Set to false for instance if a backup_storage_hook is specified and it compresses the data. (default true)
-      --backup_storage_hook string                      if set, we send the contents of the backup files through this hook.
       --backup_storage_implementation string            Which backup storage implementation to use for creating and restoring backups.
       --backup_storage_number_blocks int                if backup_storage_compress is true, backup_storage_number_blocks sets the number of blocks that can be processed, at once, before the writer blocks, during compression (default is 2). It should be equal to the number of CPUs available for compression. (default 2)
       --ceph_backup_storage_config string               Path to JSON config file for ceph backup storage. (default "ceph_backup_config.json")
diff --git a/go/flags/endtoend/vtctld.txt b/go/flags/endtoend/vtctld.txt
index 33c1df48cc..95136ae8ba 100644
--- a/go/flags/endtoend/vtctld.txt
+++ b/go/flags/endtoend/vtctld.txt
@@ -12,7 +12,6 @@ Usage of vtctld:
       --backup_engine_implementation string                              Specifies which implementation to use for creating new backups (builtin or xtrabackup). Restores will always be done with whichever engine created a given backup. (default "builtin")
       --backup_storage_block_size int                                    if backup_storage_compress is true, backup_storage_block_size sets the byte size for each block while compressing (default is 250000). (default 250000)
       --backup_storage_compress                                          if set, the backup files will be compressed (default is true). Set to false for instance if a backup_storage_hook is specified and it compresses the data. (default true)
-      --backup_storage_hook string                                       if set, we send the contents of the backup files through this hook.
       --backup_storage_implementation string                             Which backup storage implementation to use for creating and restoring backups.
       --backup_storage_number_blocks int                                 if backup_storage_compress is true, backup_storage_number_blocks sets the number of blocks that can be processed, at once, before the writer blocks, during compression (default is 2). It should be equal to the number of CPUs available for compression. (default 2)
       --builtinbackup_mysqld_timeout duration                            how long to wait for mysqld to shutdown at the start of the backup. (default 10m0s)
diff --git a/go/flags/endtoend/vtexplain.txt b/go/flags/endtoend/vtexplain.txt
index 50b7c741c4..f5df304a52 100644
--- a/go/flags/endtoend/vtexplain.txt
+++ b/go/flags/endtoend/vtexplain.txt
@@ -5,7 +5,6 @@ Usage of vtexplain:
       --backup_engine_implementation string          Specifies which implementation to use for creating new backups (builtin or xtrabackup). Restores will always be done with whichever engine created a given backup. (default "builtin")
       --backup_storage_block_size int                if backup_storage_compress is true, backup_storage_block_size sets the byte size for each block while compressing (default is 250000). (default 250000)
       --backup_storage_compress                      if set, the backup files will be compressed (default is true). Set to false for instance if a backup_storage_hook is specified and it compresses the data. (default true)
-      --backup_storage_hook string                   if set, we send the contents of the backup files through this hook.
       --backup_storage_number_blocks int             if backup_storage_compress is true, backup_storage_number_blocks sets the number of blocks that can be processed, at once, before the writer blocks, during compression (default is 2). It should be equal to the number of CPUs available for compression. (default 2)
       --batch-interval duration                      Interval between logical time slots. (default 10ms)
       --builtinbackup_mysqld_timeout duration        how long to wait for mysqld to shutdown at the start of the backup. (default 10m0s)
diff --git a/go/flags/endtoend/vttablet.txt b/go/flags/endtoend/vttablet.txt
index f2f68a8dcc..d74f79232b 100644
--- a/go/flags/endtoend/vttablet.txt
+++ b/go/flags/endtoend/vttablet.txt
@@ -10,7 +10,6 @@ Usage of vttablet:
       --backup_engine_implementation string                              Specifies which implementation to use for creating new backups (builtin or xtrabackup). Restores will always be done with whichever engine created a given backup. (default "builtin")
       --backup_storage_block_size int                                    if backup_storage_compress is true, backup_storage_block_size sets the byte size for each block while compressing (default is 250000). (default 250000)
       --backup_storage_compress                                          if set, the backup files will be compressed (default is true). Set to false for instance if a backup_storage_hook is specified and it compresses the data. (default true)
-      --backup_storage_hook string                                       if set, we send the contents of the backup files through this hook.
       --backup_storage_implementation string                             Which backup storage implementation to use for creating and restoring backups.
       --backup_storage_number_blocks int                                 if backup_storage_compress is true, backup_storage_number_blocks sets the number of blocks that can be processed, at once, before the writer blocks, during compression (default is 2). It should be equal to the number of CPUs available for compression. (default 2)
       --binlog_host string                                               PITR restore parameter: hostname/IP of binlog server.
diff --git a/go/vt/mysqlctl/backup.go b/go/vt/mysqlctl/backup.go
index 63868bf48a..a11ddb75ce 100644
--- a/go/vt/mysqlctl/backup.go
+++ b/go/vt/mysqlctl/backup.go
@@ -104,6 +104,7 @@ func init() {
 
 func registerBackupFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&backupStorageHook, "backup_storage_hook", backupStorageHook, "if set, we send the contents of the backup files through this hook.")
+	_ = fs.MarkDeprecated("backup_storage_hook", "consider using one of the builtin compression algorithms or --external-compressor and --external-decompressor instead.")
 	fs.BoolVar(&backupStorageCompress, "backup_storage_compress", backupStorageCompress, "if set, the backup files will be compressed (default is true). Set to false for instance if a backup_storage_hook is specified and it compresses the data.")
 	fs.IntVar(&backupCompressBlockSize, "backup_storage_block_size", backupCompressBlockSize, "if backup_storage_compress is true, backup_storage_block_size sets the byte size for each block while compressing (default is 250000).")
 	fs.IntVar(&backupCompressBlocks, "backup_storage_number_blocks", backupCompressBlocks, "if backup_storage_compress is true, backup_storage_number_blocks sets the number of blocks that can be processed, at once, before the writer blocks, during compression (default is 2). It should be equal to the number of CPUs available for compression.")
diff --git a/go/vt/mysqlctl/builtinbackupengine.go b/go/vt/mysqlctl/builtinbackupengine.go
index 09323c9387..45c6216646 100644
--- a/go/vt/mysqlctl/builtinbackupengine.go
+++ b/go/vt/mysqlctl/builtinbackupengine.go
@@ -160,9 +160,10 @@ func (fe *FileEntry) open(cnf *Mycnf, readOnly bool) (*os.File, error) {
 // ExecuteBackup returns a boolean that indicates if the backup is usable,
 // and an overall error.
 func (be *BuiltinBackupEngine) ExecuteBackup(ctx context.Context, params BackupParams, bh backupstorage.BackupHandle) (bool, error) {
-
 	params.Logger.Infof("Hook: %v, Compress: %v", backupStorageHook, backupStorageCompress)
-
+	if backupStorageHook != "" {
+		log.Warning("Flag --backup_storage_hook has been deprecated, consider using one of the builtin compression algorithms or --external-compressor and --external-decompressor instead.")
+	}
 	// Save initial state so we can restore.
 	replicaStartRequired := false
 	sourceIsPrimary := false
@@ -300,7 +301,6 @@ func (be *BuiltinBackupEngine) ExecuteBackup(ctx context.Context, params BackupP
 
 // backupFiles finds the list of files to backup, and creates the backup.
 func (be *BuiltinBackupEngine) backupFiles(ctx context.Context, params BackupParams, bh backupstorage.BackupHandle, replicationPosition mysql.Position) (finalErr error) {
-
 	// Get the files to backup.
 	// We don't care about totalSize because we add each file separately.
 	fes, _, err := findFilesToBackup(params.Cnf)
@@ -591,6 +591,9 @@ func (be *BuiltinBackupEngine) ExecuteRestore(ctx context.Context, params Restor
 	if err := prepareToRestore(ctx, params.Cnf, params.Mysqld, params.Logger); err != nil {
 		return nil, err
 	}
+	if bm.TransformHook != "" {
+		log.Warning("Flag --backup_storage_hook has been deprecated, consider using one of the builtin compression algorithms or --external-compressor and --external-decompressor instead.")
+	}
 
 	params.Logger.Infof("Restore: copying %v files", len(bm.FileEntries))
 
@@ -606,6 +609,16 @@ func (be *BuiltinBackupEngine) ExecuteRestore(ctx context.Context, params Restor
 // restoreFiles will copy all the files from the BackupStorage to the
 // right place.
 func (be *BuiltinBackupEngine) restoreFiles(ctx context.Context, params RestoreParams, bh backupstorage.BackupHandle, bm builtinBackupManifest) error {
+	// For optimization, we are replacing pargzip with pgzip, so newBuiltinDecompressor doesn't have to compare and print warning for every file
+	// since newBuiltinDecompressor is helper method and does not hold any state, it was hard to do it in that method itself.
+	if bm.CompressionEngine == PargzipCompressor {
+		params.Logger.Warningf(`engine "pargzip" doesn't support decompression, using "pgzip" instead`)
+		bm.CompressionEngine = PgzipCompressor
+		defer func() {
+			bm.CompressionEngine = PargzipCompressor
+		}()
+	}
+
 	fes := bm.FileEntries
 	sema := sync2.NewSemaphore(params.Concurrency, 0)
 	rec := concurrency.AllErrorRecorder{}
diff --git a/go/vt/mysqlctl/compression.go b/go/vt/mysqlctl/compression.go
index 40c4dc344a..3d6b017a70 100644
--- a/go/vt/mysqlctl/compression.go
+++ b/go/vt/mysqlctl/compression.go
@@ -193,7 +193,7 @@ func newExternalDecompressor(ctx context.Context, cmdStr string, reader io.Reade
 // This returns a reader that will decompress the underlying provided reader and will use the specified supported engine.
 func newBuiltinDecompressor(engine string, reader io.Reader, logger logutil.Logger) (decompressor io.ReadCloser, err error) {
 	if engine == PargzipCompressor {
-		logger.Warningf("engine \"pargzip\" doesn't support decompression, using \"pgzip\" instead")
+		logger.Warningf(`engine "pargzip" doesn't support decompression, using "pgzip" instead`)
 		engine = PgzipCompressor
 	}
 
diff --git a/go/vt/mysqlctl/xtrabackupengine.go b/go/vt/mysqlctl/xtrabackupengine.go
index d0d131e3ce..112bdf4179 100644
--- a/go/vt/mysqlctl/xtrabackupengine.go
+++ b/go/vt/mysqlctl/xtrabackupengine.go
@@ -451,7 +451,6 @@ func (be *XtrabackupEngine) ExecuteRestore(ctx context.Context, params RestorePa
 func (be *XtrabackupEngine) restoreFromBackup(ctx context.Context, cnf *Mycnf, bh backupstorage.BackupHandle, bm xtraBackupManifest, logger logutil.Logger) error {
 	// first download the file into a tmp dir
 	// and extract all the files
-
 	tempDir := fmt.Sprintf("%v/%v", cnf.TmpDir, time.Now().UTC().Format("xtrabackup-2006-01-02.150405"))
 	// create tempDir
 	if err := os.MkdirAll(tempDir, os.ModePerm); err != nil {
@@ -465,6 +464,16 @@ func (be *XtrabackupEngine) restoreFromBackup(ctx context.Context, cnf *Mycnf, b
 		}
 	}(tempDir, logger)
 
+	// For optimization, we are replacing pargzip with pgzip, so newBuiltinDecompressor doesn't have to compare and print warning for every file
+	// since newBuiltinDecompressor is helper method and does not hold any state, it was hard to do it in that method itself.
+	if bm.CompressionEngine == PargzipCompressor {
+		logger.Warningf(`engine "pargzip" doesn't support decompression, using "pgzip" instead`)
+		bm.CompressionEngine = PgzipCompressor
+		defer func() {
+			bm.CompressionEngine = PargzipCompressor
+		}()
+	}
+
 	if err := be.extractFiles(ctx, logger, bh, bm, tempDir); err != nil {
 		logger.Errorf("error extracting backup files: %v", err)
 		return err
diff --git a/go/vt/vtgate/planbuilder/gen4_planner.go b/go/vt/vtgate/planbuilder/gen4_planner.go
index 56b5efe6a3..dd2cd89cee 100644
--- a/go/vt/vtgate/planbuilder/gen4_planner.go
+++ b/go/vt/vtgate/planbuilder/gen4_planner.go
@@ -206,14 +206,6 @@ func newBuildSelectPlan(
 	if err != nil {
 		return nil, nil, err
 	}
-	logical, err = operators.Compact(ctx, logical)
-	if err != nil {
-		return nil, nil, err
-	}
-	err = operators.CheckValid(logical)
-	if err != nil {
-		return nil, nil, err
-	}
 
 	physOp, err := operators.TransformToPhysical(ctx, logical)
 	if err != nil {
@@ -324,10 +316,6 @@ func gen4UpdateStmtPlanner(
 	if err != nil {
 		return nil, err
 	}
-	err = operators.CheckValid(logical)
-	if err != nil {
-		return nil, err
-	}
 
 	physOp, err := operators.TransformToPhysical(ctx, logical)
 	if err != nil {
@@ -411,10 +399,6 @@ func gen4DeleteStmtPlanner(
 	if err != nil {
 		return nil, err
 	}
-	err = operators.CheckValid(logical)
-	if err != nil {
-		return nil, err
-	}
 
 	physOp, err := operators.TransformToPhysical(ctx, logical)
 	if err != nil {
diff --git a/go/vt/vtgate/planbuilder/operators/apply_join.go b/go/vt/vtgate/planbuilder/operators/apply_join.go
index 68219d0569..d9927bd88e 100644
--- a/go/vt/vtgate/planbuilder/operators/apply_join.go
+++ b/go/vt/vtgate/planbuilder/operators/apply_join.go
@@ -18,6 +18,8 @@ package operators
 
 import (
 	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
+	"vitess.io/vitess/go/vt/vtgate/semantics"
 )
 
 // ApplyJoin is a nested loop join - for each row on the LHS,
@@ -73,3 +75,61 @@ func (a *ApplyJoin) Clone(inputs []Operator) Operator {
 func (a *ApplyJoin) Inputs() []Operator {
 	return []Operator{a.LHS, a.RHS}
 }
+
+var _ joinOperator = (*ApplyJoin)(nil)
+
+func (a *ApplyJoin) tableID() semantics.TableSet {
+	return TableID(a)
+}
+
+func (a *ApplyJoin) getLHS() Operator {
+	return a.LHS
+}
+
+func (a *ApplyJoin) getRHS() Operator {
+	return a.RHS
+}
+
+func (a *ApplyJoin) setLHS(operator Operator) {
+	a.LHS = operator
+}
+
+func (a *ApplyJoin) setRHS(operator Operator) {
+	a.RHS = operator
+}
+
+func (a *ApplyJoin) makeInner() {
+	a.LeftJoin = false
+}
+
+func (a *ApplyJoin) isInner() bool {
+	return !a.LeftJoin
+}
+
+func (a *ApplyJoin) addJoinPredicate(ctx *plancontext.PlanningContext, expr sqlparser.Expr) error {
+	bvName, cols, predicate, err := BreakExpressionInLHSandRHS(ctx, expr, TableID(a.LHS))
+	if err != nil {
+		return err
+	}
+	lhs, idxs, err := PushOutputColumns(ctx, a.LHS, cols...)
+	if err != nil {
+		return err
+	}
+	a.LHSColumns = append(a.LHSColumns, cols...)
+
+	a.LHS = lhs
+	if a.Vars == nil {
+		a.Vars = map[string]int{}
+	}
+	for i, idx := range idxs {
+		a.Vars[bvName[i]] = idx
+	}
+	rhs, err := PushPredicate(ctx, predicate, a.RHS)
+	if err != nil {
+		return err
+	}
+	a.RHS = rhs
+
+	a.Predicate = sqlparser.AndExpressions(expr, a.Predicate)
+	return nil
+}
diff --git a/go/vt/vtgate/planbuilder/operators/compact.go b/go/vt/vtgate/planbuilder/operators/compact.go
index b451dce632..6f312194f9 100644
--- a/go/vt/vtgate/planbuilder/operators/compact.go
+++ b/go/vt/vtgate/planbuilder/operators/compact.go
@@ -21,8 +21,8 @@ import (
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
 )
 
-// Compact will optimise the operator tree into a smaller but equivalent version
-func Compact(ctx *plancontext.PlanningContext, op Operator) (Operator, error) {
+// compact will optimise the operator tree into a smaller but equivalent version
+func compact(ctx *plancontext.PlanningContext, op Operator) (Operator, error) {
 	newOp, _, err := rewriteBottomUp(ctx, op, func(ctx *plancontext.PlanningContext, op Operator) (Operator, bool, error) {
 		newOp, ok := op.(compactable)
 		if !ok {
@@ -37,7 +37,14 @@ func (f *Filter) compact(*plancontext.PlanningContext) (Operator, bool, error) {
 	if len(f.Predicates) == 0 {
 		return f.Source, true, nil
 	}
-	return f, false, nil
+
+	other, isFilter := f.Source.(*Filter)
+	if !isFilter {
+		return f, false, nil
+	}
+	f.Source = other.Source
+	f.Predicates = append(f.Predicates, other.Predicates...)
+	return f, true, nil
 }
 
 func (u *Union) compact(*plancontext.PlanningContext) (Operator, bool, error) {
@@ -72,7 +79,7 @@ func (u *Union) compact(*plancontext.PlanningContext) (Operator, bool, error) {
 	return u, anythingChanged, nil
 }
 
-func (j *Join) compactJoin(ctx *plancontext.PlanningContext) (Operator, bool, error) {
+func (j *Join) compact(ctx *plancontext.PlanningContext) (Operator, bool, error) {
 	if j.LeftJoin {
 		// we can't merge outer joins into a single QG
 		return j, false, nil
@@ -89,9 +96,11 @@ func (j *Join) compactJoin(ctx *plancontext.PlanningContext) (Operator, bool, er
 		innerJoins: append(lqg.innerJoins, rqg.innerJoins...),
 		NoDeps:     sqlparser.AndExpressions(lqg.NoDeps, rqg.NoDeps),
 	}
-	err := newOp.collectPredicate(ctx, j.Predicate)
-	if err != nil {
-		return nil, false, err
+	if j.Predicate != nil {
+		err := newOp.collectPredicate(ctx, j.Predicate)
+		if err != nil {
+			return nil, false, err
+		}
 	}
 	return newOp, true, nil
 }
diff --git a/go/vt/vtgate/planbuilder/operators/correlated_subquery.go b/go/vt/vtgate/planbuilder/operators/correlated_subquery.go
index ecbdd2cd7e..9a98c80314 100644
--- a/go/vt/vtgate/planbuilder/operators/correlated_subquery.go
+++ b/go/vt/vtgate/planbuilder/operators/correlated_subquery.go
@@ -69,11 +69,17 @@ func (c *CorrelatedSubQueryOp) Clone(inputs []Operator) Operator {
 	checkSize(inputs, 2)
 	columns := make([]*sqlparser.ColName, len(c.LHSColumns))
 	copy(columns, c.LHSColumns)
+	vars := make(map[string]int, len(c.Vars))
+	for k, v := range c.Vars {
+		vars[k] = v
+	}
+
 	result := &CorrelatedSubQueryOp{
 		Outer:      inputs[0],
 		Inner:      inputs[1],
 		Extracted:  c.Extracted,
 		LHSColumns: columns,
+		Vars:       vars,
 	}
 	return result
 }
diff --git a/go/vt/vtgate/planbuilder/operators/derived.go b/go/vt/vtgate/planbuilder/operators/derived.go
index 3e80efe79a..51b68f45d9 100644
--- a/go/vt/vtgate/planbuilder/operators/derived.go
+++ b/go/vt/vtgate/planbuilder/operators/derived.go
@@ -21,6 +21,7 @@ import (
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
+	"vitess.io/vitess/go/vt/vtgate/semantics"
 )
 
 type Derived struct {
@@ -101,3 +102,26 @@ func (d *Derived) IsMergeable(ctx *plancontext.PlanningContext) bool {
 func (d *Derived) Inputs() []Operator {
 	return []Operator{d.Source}
 }
+
+func (d *Derived) addPredicate(ctx *plancontext.PlanningContext, expr sqlparser.Expr) error {
+	if _, isUNion := d.Source.(*Union); isUNion {
+		// If we have a derived table on top of a UNION, we can let the UNION do the expression rewriting
+		var err error
+		d.Source, err = PushPredicate(ctx, expr, d.Source)
+		return err
+	}
+	tableInfo, err := ctx.SemTable.TableInfoForExpr(expr)
+	if err != nil {
+		if err == semantics.ErrMultipleTables {
+			return semantics.ProjError{Inner: vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: unable to split predicates to derived table: %s", sqlparser.String(expr))}
+		}
+		return err
+	}
+
+	newExpr, err := semantics.RewriteDerivedTableExpression(expr, tableInfo)
+	if err != nil {
+		return err
+	}
+	d.Source, err = PushPredicate(ctx, newExpr, d.Source)
+	return err
+}
diff --git a/go/vt/vtgate/planbuilder/operators/filter.go b/go/vt/vtgate/planbuilder/operators/filter.go
index 10de7665e3..e8d44c65fb 100644
--- a/go/vt/vtgate/planbuilder/operators/filter.go
+++ b/go/vt/vtgate/planbuilder/operators/filter.go
@@ -59,3 +59,9 @@ func (f *Filter) UnsolvedPredicates(st *semantics.SemTable) []sqlparser.Expr {
 	}
 	return result
 }
+
+func newFilter(op Operator, expr ...sqlparser.Expr) Operator {
+	return &Filter{
+		Source: op, Predicates: expr,
+	}
+}
diff --git a/go/vt/vtgate/planbuilder/operators/join.go b/go/vt/vtgate/planbuilder/operators/join.go
index 722ec6d663..76048d59e8 100644
--- a/go/vt/vtgate/planbuilder/operators/join.go
+++ b/go/vt/vtgate/planbuilder/operators/join.go
@@ -19,6 +19,7 @@ package operators
 import (
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
+	"vitess.io/vitess/go/vt/vtgate/semantics"
 )
 
 // Join represents a join. If we have a predicate, this is an inner join. If no predicate exists, it is a cross join
@@ -30,43 +31,6 @@ type Join struct {
 
 var _ Operator = (*Join)(nil)
 
-// 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(ctx *plancontext.PlanningContext, expr sqlparser.Expr) {
-	if !j.LeftJoin {
-		return
-	}
-
-	switch expr := expr.(type) {
-	case *sqlparser.ComparisonExpr:
-		if expr.Operator == sqlparser.NullSafeEqualOp {
-			return
-		}
-
-		if sqlparser.IsColName(expr.Left) && ctx.SemTable.RecursiveDeps(expr.Left).IsSolvedBy(TableID(j.RHS)) ||
-			sqlparser.IsColName(expr.Right) && ctx.SemTable.RecursiveDeps(expr.Right).IsSolvedBy(TableID(j.RHS)) {
-			j.LeftJoin = false
-		}
-
-	case *sqlparser.IsExpr:
-		if expr.Right != sqlparser.IsNotNullOp {
-			return
-		}
-
-		if sqlparser.IsColName(expr.Left) && ctx.SemTable.RecursiveDeps(expr.Left).IsSolvedBy(TableID(j.RHS)) {
-			j.LeftJoin = false
-		}
-	}
-}
-
 // Clone implements the Operator interface
 func (j *Join) Clone(inputs []Operator) Operator {
 	checkSize(inputs, 2)
@@ -85,3 +49,71 @@ func (j *Join) Clone(inputs []Operator) Operator {
 func (j *Join) Inputs() []Operator {
 	return []Operator{j.LHS, j.RHS}
 }
+
+func createOuterJoin(tableExpr *sqlparser.JoinTableExpr, lhs, rhs Operator) (Operator, error) {
+	if tableExpr.Join == sqlparser.RightJoinType {
+		lhs, rhs = rhs, lhs
+	}
+	return &Join{LHS: lhs, RHS: rhs, LeftJoin: true, Predicate: sqlparser.RemoveKeyspaceFromColName(tableExpr.Condition.On)}, nil
+}
+
+func createJoin(LHS, RHS Operator) Operator {
+	lqg, lok := LHS.(*QueryGraph)
+	rqg, rok := RHS.(*QueryGraph)
+	if lok && rok {
+		op := &QueryGraph{
+			Tables:     append(lqg.Tables, rqg.Tables...),
+			innerJoins: append(lqg.innerJoins, rqg.innerJoins...),
+			NoDeps:     sqlparser.AndExpressions(lqg.NoDeps, rqg.NoDeps),
+		}
+		return op
+	}
+	return &Join{LHS: LHS, RHS: RHS}
+}
+
+func createInnerJoin(ctx *plancontext.PlanningContext, tableExpr *sqlparser.JoinTableExpr, lhs, rhs Operator) (Operator, error) {
+	op := createJoin(lhs, rhs)
+	if tableExpr.Condition.On != nil {
+		var err error
+		op, err = PushPredicate(ctx, sqlparser.RemoveKeyspaceFromColName(tableExpr.Condition.On), op)
+		if err != nil {
+			return nil, err
+		}
+	}
+	return op, nil
+}
+
+var _ joinOperator = (*Join)(nil)
+
+func (j *Join) tableID() semantics.TableSet {
+	return TableID(j)
+}
+
+func (j *Join) getLHS() Operator {
+	return j.LHS
+}
+
+func (j *Join) getRHS() Operator {
+	return j.RHS
+}
+
+func (j *Join) setLHS(operator Operator) {
+	j.LHS = operator
+}
+
+func (j *Join) setRHS(operator Operator) {
+	j.RHS = operator
+}
+
+func (j *Join) makeInner() {
+	j.LeftJoin = false
+}
+
+func (j *Join) isInner() bool {
+	return !j.LeftJoin
+}
+
+func (j *Join) addJoinPredicate(_ *plancontext.PlanningContext, expr sqlparser.Expr) error {
+	j.Predicate = sqlparser.AndExpressions(j.Predicate, expr)
+	return nil
+}
diff --git a/go/vt/vtgate/planbuilder/operators/node.go b/go/vt/vtgate/planbuilder/operators/node.go
index 95e2ec8d7a..280ab07f79 100644
--- a/go/vt/vtgate/planbuilder/operators/node.go
+++ b/go/vt/vtgate/planbuilder/operators/node.go
@@ -58,9 +58,9 @@ func unresolvedPredicates(op Operator, st *semantics.SemTable) (result []sqlpars
 	return
 }
 
-func CheckValid(op Operator) error {
+func checkValid(op Operator) error {
 	return VisitTopDown(op, func(this Operator) error {
-		if chk, ok := this.(checked); ok {
+		if chk, ok := this.(checkable); ok {
 			return chk.CheckValid()
 		}
 		return nil
diff --git a/go/vt/vtgate/planbuilder/operators/operator.go b/go/vt/vtgate/planbuilder/operators/operator.go
index 539dd1139b..36894e7798 100644
--- a/go/vt/vtgate/planbuilder/operators/operator.go
+++ b/go/vt/vtgate/planbuilder/operators/operator.go
@@ -67,8 +67,8 @@ type (
 		Cost() int
 	}
 
-	checked interface {
-		// CheckValid allows operators that need a final check before being used, to make sure that
+	checkable interface {
+		// checkValid allows operators that need a final check before being used, to make sure that
 		// all the necessary information is in the operator
 		CheckValid() error
 	}
@@ -120,25 +120,6 @@ func getOperatorFromJoinTableExpr(ctx *plancontext.PlanningContext, tableExpr *s
 	}
 }
 
-func createOuterJoin(tableExpr *sqlparser.JoinTableExpr, lhs, rhs Operator) (Operator, error) {
-	if tableExpr.Join == sqlparser.RightJoinType {
-		lhs, rhs = rhs, lhs
-	}
-	return &Join{LHS: lhs, RHS: rhs, LeftJoin: true, Predicate: sqlparser.RemoveKeyspaceFromColName(tableExpr.Condition.On)}, nil
-}
-
-func createInnerJoin(ctx *plancontext.PlanningContext, tableExpr *sqlparser.JoinTableExpr, lhs, rhs Operator) (Operator, error) {
-	op := createJoin(lhs, rhs)
-	if tableExpr.Condition.On != nil {
-		var err error
-		op, err = LogicalPushPredicate(ctx, op, sqlparser.RemoveKeyspaceFromColName(tableExpr.Condition.On))
-		if err != nil {
-			return nil, err
-		}
-	}
-	return op, nil
-}
-
 func getOperatorFromAliasedTableExpr(ctx *plancontext.PlanningContext, tableExpr *sqlparser.AliasedTableExpr) (Operator, error) {
 	switch tbl := tableExpr.Expr.(type) {
 	case sqlparser.TableName:
@@ -219,6 +200,15 @@ func CreateLogicalOperatorFromAST(ctx *plancontext.PlanningContext, selStmt sqlp
 	if err != nil {
 		return nil, err
 	}
+
+	if op, err = compact(ctx, op); err != nil {
+		return nil, err
+	}
+
+	if err = checkValid(op); err != nil {
+		return nil, err
+	}
+
 	return op, nil
 }
 
@@ -258,7 +248,7 @@ func createOperatorFromSelect(ctx *plancontext.PlanningContext, sel *sqlparser.S
 	if sel.Where != nil {
 		exprs := sqlparser.SplitAndExpression(nil, sel.Where.Expr)
 		for _, expr := range exprs {
-			op, err = LogicalPushPredicate(ctx, op, sqlparser.RemoveKeyspaceFromColName(expr))
+			op, err = PushPredicate(ctx, sqlparser.RemoveKeyspaceFromColName(expr), op)
 			if err != nil {
 				return nil, err
 			}
@@ -430,24 +420,6 @@ func createOperatorFromDelete(ctx *plancontext.PlanningContext, deleteStmt *sqlp
 	return subq, nil
 }
 
-func createSubqueryFromStatement(ctx *plancontext.PlanningContext, stmt sqlparser.Statement) (*SubQuery, error) {
-	if len(ctx.SemTable.SubqueryMap[stmt]) == 0 {
-		return nil, nil
-	}
-	subq := &SubQuery{}
-	for _, sq := range ctx.SemTable.SubqueryMap[stmt] {
-		opInner, err := CreateLogicalOperatorFromAST(ctx, sq.Subquery.Select)
-		if err != nil {
-			return nil, err
-		}
-		subq.Inner = append(subq.Inner, &SubQueryInner{
-			ExtractedSubquery: sq,
-			Inner:             opInner,
-		})
-	}
-	return subq, nil
-}
-
 func addColumnEquality(ctx *plancontext.PlanningContext, expr sqlparser.Expr) {
 	switch expr := expr.(type) {
 	case *sqlparser.ComparisonExpr:
@@ -463,17 +435,3 @@ func addColumnEquality(ctx *plancontext.PlanningContext, expr sqlparser.Expr) {
 		}
 	}
 }
-
-func createJoin(LHS, RHS Operator) Operator {
-	lqg, lok := LHS.(*QueryGraph)
-	rqg, rok := RHS.(*QueryGraph)
-	if lok && rok {
-		op := &QueryGraph{
-			Tables:     append(lqg.Tables, rqg.Tables...),
-			innerJoins: append(lqg.innerJoins, rqg.innerJoins...),
-			NoDeps:     sqlparser.AndExpressions(lqg.NoDeps, rqg.NoDeps),
-		}
-		return op
-	}
-	return &Join{LHS: LHS, RHS: RHS}
-}
diff --git a/go/vt/vtgate/planbuilder/operators/operator_funcs.go b/go/vt/vtgate/planbuilder/operators/operator_funcs.go
index ea631f1d98..26f74bb2fb 100644
--- a/go/vt/vtgate/planbuilder/operators/operator_funcs.go
+++ b/go/vt/vtgate/planbuilder/operators/operator_funcs.go
@@ -30,115 +30,46 @@ import (
 // where data is fetched from the LHS of the join to be used in the evaluation on the RHS
 func PushPredicate(ctx *plancontext.PlanningContext, expr sqlparser.Expr, op Operator) (Operator, error) {
 	switch op := op.(type) {
-	case *Route:
-		err := op.UpdateRoutingLogic(ctx, expr)
+	case *Filter:
+		var err error
+		op.Source, err = PushPredicate(ctx, expr, op.Source)
 		if err != nil {
 			return nil, err
 		}
-		newSrc, err := PushPredicate(ctx, expr, op.Source)
+		return op, nil
+	case *QueryGraph:
+		err := op.addPredicate(ctx, expr)
 		if err != nil {
 			return nil, err
 		}
-		op.Source = newSrc
-		return op, err
-	case *ApplyJoin:
-		deps := ctx.SemTable.RecursiveDeps(expr)
-		switch {
-		case deps.IsSolvedBy(TableID(op.LHS)):
-			newSrc, err := PushPredicate(ctx, expr, op.LHS)
-			if err != nil {
-				return nil, err
-			}
-			op.LHS = newSrc
-			return op, err
-		case deps.IsSolvedBy(TableID(op.RHS)):
-			if !op.LeftJoin {
-				newSrc, err := PushPredicate(ctx, expr, op.RHS)
-				if err != nil {
-					return nil, err
-				}
-				op.RHS = newSrc
-				return op, err
-			}
-
-			// 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) && ctx.SemTable.RecursiveDeps(cmp.Left).IsSolvedBy(TableID(op.RHS)) ||
-					sqlparser.IsColName(cmp.Right) && ctx.SemTable.RecursiveDeps(cmp.Right).IsSolvedBy(TableID(op.RHS))) {
-				// 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
-				op.LeftJoin = false
-				newSrc, err := PushPredicate(ctx, expr, op.RHS)
-				if err != nil {
-					return nil, err
-				}
-				op.RHS = newSrc
-				return op, err
-			}
-
-			// finally, if we can't turn the outer join into an inner,
-			// we need to filter after the join has been evaluated
-			return &Filter{
-				Source:     op,
-				Predicates: []sqlparser.Expr{expr},
-			}, nil
-		case deps.IsSolvedBy(TableID(op)):
-			bvName, cols, predicate, err := BreakExpressionInLHSandRHS(ctx, expr, TableID(op))
-			if err != nil {
-				return nil, err
-			}
-			out, idxs, err := PushOutputColumns(ctx, op.LHS, cols...)
-			if err != nil {
-				return nil, err
-			}
-			op.LHS = out
-			for i, idx := range idxs {
-				op.Vars[bvName[i]] = idx
-			}
-			newSrc, err := PushPredicate(ctx, predicate, op.RHS)
-			if err != nil {
-				return nil, err
-			}
-			op.RHS = newSrc
-			op.Predicate = sqlparser.AndExpressions(op.Predicate, expr)
-			return op, err
-		}
-		return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "Cannot push predicate: %s", sqlparser.String(expr))
+		return op, nil
+	case *Route:
+		return op.addPredicate(ctx, expr)
+	case *ApplyJoin, *Join:
+		join := op.(joinOperator) // stupid golang doesn't understand this without an explicit cast
+		return addPredicate(join, ctx, expr, false)
 	case *Table:
 		// We do not add the predicate to op.qtable because that is an immutable struct that should not be
 		// changed by physical operators.
-		return &Filter{
-			Source:     op,
-			Predicates: []sqlparser.Expr{expr},
-		}, nil
-	case *Filter:
-		op.Predicates = append(op.Predicates, expr)
-		return op, nil
+		return newFilter(op, expr), nil
 	case *Derived:
-		tableInfo, err := ctx.SemTable.TableInfoForExpr(expr)
+		err := op.addPredicate(ctx, expr)
 		if err != nil {
-			if err == semantics.ErrMultipleTables {
-				return nil, semantics.ProjError{Inner: vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: unable to split predicates to derived table: %s", sqlparser.String(expr))}
-			}
 			return nil, err
 		}
-		newExpr, err := semantics.RewriteDerivedTableExpression(expr, tableInfo)
+		return op, nil
+	case *Vindex:
+		err := op.addPredicate(ctx, expr)
 		if err != nil {
 			return nil, err
 		}
-		newSrc, err := PushPredicate(ctx, newExpr, op.Source)
+		return op, nil
+	case *Union:
+		err := op.addPredicate(ctx, expr)
 		if err != nil {
 			return nil, err
 		}
-		op.Source = newSrc
-		return op, err
+		return op, nil
 	default:
 		return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "we cannot push predicates into %T", op)
 	}
@@ -368,3 +299,104 @@ func BreakExpressionInLHSandRHS(
 	ctx.JoinPredicates[expr] = append(ctx.JoinPredicates[expr], rewrittenExpr)
 	return
 }
+
+type joinOperator interface {
+	Operator
+	getLHS() Operator
+	getRHS() Operator
+	setLHS(Operator)
+	setRHS(Operator)
+	makeInner()
+	isInner() bool
+	addJoinPredicate(ctx *plancontext.PlanningContext, expr sqlparser.Expr) error
+}
+
+func addPredicate(join joinOperator, ctx *plancontext.PlanningContext, expr sqlparser.Expr, joinPredicates bool) (Operator, error) {
+	deps := ctx.SemTable.RecursiveDeps(expr)
+	switch {
+	case deps.IsSolvedBy(TableID(join.getLHS())):
+		// predicates can always safely be pushed down to the lhs if that is all they depend on
+		lhs, err := PushPredicate(ctx, expr, join.getLHS())
+		if err != nil {
+			return nil, err
+		}
+		join.setLHS(lhs)
+		return join, err
+	case deps.IsSolvedBy(TableID(join.getRHS())):
+		// if we are dealing with an outer join, always start by checking if this predicate can turn
+		// the join into an inner join
+		if !join.isInner() && canConvertToInner(ctx, expr, TableID(join.getRHS())) {
+			join.makeInner()
+		}
+
+		if !joinPredicates && !join.isInner() {
+			// if we still are dealing with an outer join
+			// we need to filter after the join has been evaluated
+			return newFilter(join, expr), nil
+		}
+
+		// For inner joins, we can just push the filtering on the RHS
+		rhs, err := PushPredicate(ctx, expr, join.getRHS())
+		if err != nil {
+			return nil, err
+		}
+		join.setRHS(rhs)
+		return join, err
+
+	case deps.IsSolvedBy(TableID(join)):
+		// if we are dealing with an outer join, always start by checking if this predicate can turn
+		// the join into an inner join
+		if !joinPredicates && !join.isInner() && canConvertToInner(ctx, expr, TableID(join.getRHS())) {
+			join.makeInner()
+		}
+
+		if !joinPredicates && !join.isInner() {
+			// if we still are dealing with an outer join
+			// we need to filter after the join has been evaluated
+			return newFilter(join, expr), nil
+		}
+
+		err := join.addJoinPredicate(ctx, expr)
+		if err != nil {
+			return nil, err
+		}
+
+		return join, nil
+	}
+	return nil, nil
+}
+
+// we are looking for predicates like `tbl.col = <>` or `<> = tbl.col`,
+// where tbl is on the rhs of the left outer join
+// 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 canConvertToInner(ctx *plancontext.PlanningContext, expr sqlparser.Expr, rhs semantics.TableSet) bool {
+	isColNameFromRHS := func(e sqlparser.Expr) bool {
+		return sqlparser.IsColName(e) && ctx.SemTable.RecursiveDeps(e).IsSolvedBy(rhs)
+	}
+	switch expr := expr.(type) {
+	case *sqlparser.ComparisonExpr:
+		if expr.Operator == sqlparser.NullSafeEqualOp {
+			return false
+		}
+
+		return isColNameFromRHS(expr.Left) || isColNameFromRHS(expr.Right)
+
+	case *sqlparser.IsExpr:
+		if expr.Right != sqlparser.IsNotNullOp {
+			return false
+		}
+
+		return isColNameFromRHS(expr.Left)
+	default:
+		return false
+	}
+}
diff --git a/go/vt/vtgate/planbuilder/operators/predicate_push_down.go b/go/vt/vtgate/planbuilder/operators/predicate_push_down.go
deleted file mode 100644
index 07249b6f32..0000000000
--- a/go/vt/vtgate/planbuilder/operators/predicate_push_down.go
+++ /dev/null
@@ -1,200 +0,0 @@
-/*
-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 operators
-
-import (
-	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
-	"vitess.io/vitess/go/vt/sqlparser"
-	"vitess.io/vitess/go/vt/vterrors"
-	"vitess.io/vitess/go/vt/vtgate/engine"
-	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
-	"vitess.io/vitess/go/vt/vtgate/semantics"
-)
-
-func LogicalPushPredicate(ctx *plancontext.PlanningContext, op Operator, expr sqlparser.Expr) (Operator, error) {
-	switch op := op.(type) {
-	case *Union:
-		return pushPredicateOnConcatenate(ctx, expr, op)
-	case *Derived:
-		return pushPredicateOnDerived(ctx, expr, op)
-	case *Filter:
-		return pushPredicateOnFilter(ctx, expr, op)
-	case *Join:
-		return pushPredicateOnJoin(ctx, expr, op)
-	case *QueryGraph:
-		return pushPredicateOnQG(ctx, expr, op)
-	case *Vindex:
-		return pushPredicateOnVindex(ctx, expr, op)
-	default:
-		return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "[%T] can't accept predicates", op)
-	}
-}
-
-func pushPredicateOnConcatenate(ctx *plancontext.PlanningContext, expr sqlparser.Expr, c *Union) (Operator, error) {
-	newSources := make([]Operator, 0, len(c.Sources))
-	for index, source := range c.Sources {
-		if len(c.SelectStmts[index].SelectExprs) != 1 {
-			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "can't push predicates on concatenate")
-		}
-		if _, isStarExpr := c.SelectStmts[index].SelectExprs[0].(*sqlparser.StarExpr); !isStarExpr {
-			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "can't push predicates on concatenate")
-		}
-
-		newSrc, err := LogicalPushPredicate(ctx, source, expr)
-		if err != nil {
-			return nil, err
-		}
-		newSources = append(newSources, newSrc)
-	}
-	c.Sources = newSources
-	return c, nil
-}
-
-func pushPredicateOnDerived(ctx *plancontext.PlanningContext, expr sqlparser.Expr, d *Derived) (Operator, error) {
-	tableInfo, err := ctx.SemTable.TableInfoForExpr(expr)
-	if err != nil {
-		if err == semantics.ErrMultipleTables {
-			return nil, semantics.ProjError{Inner: vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: unable to split predicates to derived table: %s", sqlparser.String(expr))}
-		}
-		return nil, err
-	}
-
-	newExpr, err := semantics.RewriteDerivedTableExpression(expr, tableInfo)
-	if err != nil {
-		return nil, err
-	}
-	newSrc, err := LogicalPushPredicate(ctx, d.Source, newExpr)
-	d.Source = newSrc
-	return d, err
-}
-
-func pushPredicateOnFilter(ctx *plancontext.PlanningContext, expr sqlparser.Expr, f *Filter) (Operator, error) {
-	op, err := LogicalPushPredicate(ctx, f.Source, expr)
-	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
-}
-
-func pushPredicateOnJoin(ctx *plancontext.PlanningContext, expr sqlparser.Expr, j *Join) (Operator, error) {
-	deps := ctx.SemTable.RecursiveDeps(expr)
-	switch {
-	case deps.IsSolvedBy(TableID(j.LHS)):
-		lhs, err := LogicalPushPredicate(ctx, j.LHS, expr)
-		if err != nil {
-			return nil, err
-		}
-		j.LHS = lhs
-		return j, nil
-
-	case deps.IsSolvedBy(TableID(j.RHS)):
-		j.tryConvertToInnerJoin(ctx, expr)
-
-		if !j.LeftJoin {
-			rhs, err := LogicalPushPredicate(ctx, j.RHS, expr)
-			if err != nil {
-				return nil, err
-			}
-			j.RHS = rhs
-			return j, err
-		}
-
-		op := &Filter{
-			Source:     j,
-			Predicates: []sqlparser.Expr{expr},
-		}
-		return op, nil
-
-	case deps.IsSolvedBy(TableID(j)):
-		j.tryConvertToInnerJoin(ctx, expr)
-
-		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))
-}
-
-func pushPredicateOnQG(ctx *plancontext.PlanningContext, expr sqlparser.Expr, qg *QueryGraph) (Operator, error) {
-	for _, e := range sqlparser.SplitAndExpression(nil, expr) {
-		err := qg.collectPredicate(ctx, e)
-		if err != nil {
-			return nil, err
-		}
-	}
-	return qg, nil
-}
-
-func pushPredicateOnVindex(ctx *plancontext.PlanningContext, expr sqlparser.Expr, v *Vindex) (Operator, error) {
-	for _, e := range sqlparser.SplitAndExpression(nil, expr) {
-		deps := ctx.SemTable.RecursiveDeps(e)
-		if deps.NumberOfTables() > 1 {
-			return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, vindexUnsupported+" (multiple tables involved)")
-		}
-		// check if we already have a predicate
-		if v.OpCode != engine.VindexNone {
-			return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (multiple filters)")
-		}
-
-		// check LHS
-		comparison, ok := e.(*sqlparser.ComparisonExpr)
-		if !ok {
-			return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (not a comparison)")
-		}
-		if comparison.Operator != sqlparser.EqualOp && comparison.Operator != sqlparser.InOp {
-			return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (not equality)")
-		}
-		colname, ok := comparison.Left.(*sqlparser.ColName)
-		if !ok {
-			return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (lhs is not a column)")
-		}
-		if !colname.Name.EqualString("id") {
-			return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (lhs is not id)")
-		}
-
-		// check RHS
-		var err error
-		if sqlparser.IsValue(comparison.Right) || sqlparser.IsSimpleTuple(comparison.Right) {
-			v.Value = comparison.Right
-		} else {
-			return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (rhs is not a value)")
-		}
-		if err != nil {
-			return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+": %v", err)
-		}
-		v.OpCode = engine.VindexMap
-		v.Table.Predicates = append(v.Table.Predicates, e)
-	}
-	return v, nil
-}
diff --git a/go/vt/vtgate/planbuilder/operators/querygraph.go b/go/vt/vtgate/planbuilder/operators/querygraph.go
index b74354b849..c302b4658e 100644
--- a/go/vt/vtgate/planbuilder/operators/querygraph.go
+++ b/go/vt/vtgate/planbuilder/operators/querygraph.go
@@ -187,3 +187,13 @@ func (qg *QueryGraph) Clone(inputs []Operator) Operator {
 	result.NoDeps = qg.NoDeps
 	return result
 }
+
+func (qg *QueryGraph) addPredicate(ctx *plancontext.PlanningContext, expr sqlparser.Expr) error {
+	for _, e := range sqlparser.SplitAndExpression(nil, expr) {
+		err := qg.collectPredicate(ctx, e)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/go/vt/vtgate/planbuilder/operators/route.go b/go/vt/vtgate/planbuilder/operators/route.go
index 49dc352d05..869766fdc9 100644
--- a/go/vt/vtgate/planbuilder/operators/route.go
+++ b/go/vt/vtgate/planbuilder/operators/route.go
@@ -17,8 +17,11 @@ limitations under the License.
 package operators
 
 import (
+	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/vt/key"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vtgate/engine"
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
@@ -713,3 +716,104 @@ func (r *Route) planIsExpr(ctx *plancontext.PlanningContext, node *sqlparser.IsE
 
 	return r.haveMatchingVindex(ctx, node, vdValue, column, val, opcodeF, justTheVindex)
 }
+
+func createRoute(ctx *plancontext.PlanningContext, table *QueryTable, solves semantics.TableSet) (*Route, error) {
+	if table.IsInfSchema {
+		return createInfSchemaRoute(ctx, table)
+	}
+	vschemaTable, _, _, _, target, err := ctx.VSchema.FindTableOrVindex(table.Table)
+	if target != nil {
+		return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: SELECT with a target destination")
+	}
+	if err != nil {
+		return nil, err
+	}
+	if vschemaTable.Name.String() != table.Table.Name.String() {
+		// we are dealing with a routed table
+		name := table.Table.Name
+		table.Table.Name = vschemaTable.Name
+		astTable, ok := table.Alias.Expr.(sqlparser.TableName)
+		if !ok {
+			return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "[BUG] a derived table should never be a routed table")
+		}
+		realTableName := sqlparser.NewIdentifierCS(vschemaTable.Name.String())
+		astTable.Name = realTableName
+		if table.Alias.As.IsEmpty() {
+			// if the user hasn't specified an alias, we'll insert one here so the old table name still works
+			table.Alias.As = sqlparser.NewIdentifierCS(name.String())
+		}
+	}
+	plan := &Route{
+		Source: &Table{
+			QTable: table,
+			VTable: vschemaTable,
+		},
+		Keyspace: vschemaTable.Keyspace,
+	}
+
+	for _, columnVindex := range vschemaTable.ColumnVindexes {
+		plan.VindexPreds = append(plan.VindexPreds, &VindexPlusPredicates{ColVindex: columnVindex, TableID: solves})
+	}
+
+	switch {
+	case vschemaTable.Type == vindexes.TypeSequence:
+		plan.RouteOpCode = engine.Next
+	case vschemaTable.Type == vindexes.TypeReference:
+		plan.RouteOpCode = engine.Reference
+	case !vschemaTable.Keyspace.Sharded:
+		plan.RouteOpCode = engine.Unsharded
+	case vschemaTable.Pinned != nil:
+		// Pinned tables have their keyspace ids already assigned.
+		// Use the Binary vindex, which is the identity function
+		// for keyspace id.
+		plan.RouteOpCode = engine.EqualUnique
+		vindex, _ := vindexes.NewBinary("binary", nil)
+		plan.Selected = &VindexOption{
+			Ready:       true,
+			Values:      []evalengine.Expr{evalengine.NewLiteralString(vschemaTable.Pinned, collations.TypedCollation{})},
+			ValueExprs:  nil,
+			Predicates:  nil,
+			OpCode:      engine.EqualUnique,
+			FoundVindex: vindex,
+			Cost: Cost{
+				OpCode: engine.EqualUnique,
+			},
+		}
+	default:
+		plan.RouteOpCode = engine.Scatter
+	}
+	for _, predicate := range table.Predicates {
+		err = plan.UpdateRoutingLogic(ctx, predicate)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	if plan.RouteOpCode == engine.Scatter && len(table.Predicates) > 0 {
+		// If we have a scatter query, it's worth spending a little extra time seeing if we can't improve it
+		for _, pred := range table.Predicates {
+			rewritten := tryRewriteOrToIn(pred)
+			if rewritten != nil {
+				err = plan.UpdateRoutingLogic(ctx, rewritten)
+				if err != nil {
+					return nil, err
+				}
+			}
+		}
+	}
+
+	return plan, nil
+}
+
+func (r *Route) addPredicate(ctx *plancontext.PlanningContext, expr sqlparser.Expr) (Operator, error) {
+	err := r.UpdateRoutingLogic(ctx, expr)
+	if err != nil {
+		return nil, err
+	}
+	newSrc, err := PushPredicate(ctx, expr, r.Source)
+	if err != nil {
+		return nil, err
+	}
+	r.Source = newSrc
+	return r, err
+}
diff --git a/go/vt/vtgate/planbuilder/operators/route_planning.go b/go/vt/vtgate/planbuilder/operators/route_planning.go
index 69eff7bc5c..4135542cb5 100644
--- a/go/vt/vtgate/planbuilder/operators/route_planning.go
+++ b/go/vt/vtgate/planbuilder/operators/route_planning.go
@@ -21,7 +21,6 @@ import (
 	"fmt"
 	"io"
 
-	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/vt/key"
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vterrors"
@@ -80,7 +79,7 @@ func TransformToPhysical(ctx *plancontext.PlanningContext, in Operator) (Operato
 		return nil, err
 	}
 
-	return op, nil
+	return compact(ctx, op)
 }
 
 func optimizeFilter(op *Filter) (Operator, bool, error) {
@@ -100,9 +99,8 @@ func optimizeDerived(ctx *plancontext.PlanningContext, op *Derived) (Operator, b
 		return op, false, nil
 	}
 
-	if innerRoute.RouteOpCode == engine.EqualUnique {
+	if !(innerRoute.RouteOpCode == engine.EqualUnique) && !op.IsMergeable(ctx) {
 		// no need to check anything if we are sure that we will only hit a single shard
-	} else if !op.IsMergeable(ctx) {
 		return op, false, nil
 	}
 
@@ -133,7 +131,7 @@ func optimizeQueryGraph(ctx *plancontext.PlanningContext, op *QueryGraph) (resul
 	if len(unresolved) > 0 {
 		// if we have any predicates that none of the joins or tables took care of,
 		// we add a single filter on top, so we don't lose it. This is used for sub-query planning
-		result = &Filter{Source: result, Predicates: unresolved}
+		result = newFilter(result, unresolved...)
 	}
 
 	return
@@ -260,9 +258,9 @@ func seedOperatorList(ctx *plancontext.PlanningContext, qg *QueryGraph) ([]Opera
 			return nil, err
 		}
 		if qg.NoDeps != nil {
-			plan.Source = &Filter{
-				Source:     plan.Source,
-				Predicates: []sqlparser.Expr{qg.NoDeps},
+			plan.Source, err = PushPredicate(ctx, qg.NoDeps, plan.Source)
+			if err != nil {
+				return nil, err
 			}
 		}
 		plans[i] = plan
@@ -270,94 +268,6 @@ func seedOperatorList(ctx *plancontext.PlanningContext, qg *QueryGraph) ([]Opera
 	return plans, nil
 }
 
-func createRoute(ctx *plancontext.PlanningContext, table *QueryTable, solves semantics.TableSet) (*Route, error) {
-	if table.IsInfSchema {
-		return createInfSchemaRoute(ctx, table)
-	}
-	vschemaTable, _, _, _, target, err := ctx.VSchema.FindTableOrVindex(table.Table)
-	if target != nil {
-		return nil, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: SELECT with a target destination")
-	}
-	if err != nil {
-		return nil, err
-	}
-	if vschemaTable.Name.String() != table.Table.Name.String() {
-		// we are dealing with a routed table
-		name := table.Table.Name
-		table.Table.Name = vschemaTable.Name
-		astTable, ok := table.Alias.Expr.(sqlparser.TableName)
-		if !ok {
-			return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "[BUG] a derived table should never be a routed table")
-		}
-		realTableName := sqlparser.NewIdentifierCS(vschemaTable.Name.String())
-		astTable.Name = realTableName
-		if table.Alias.As.IsEmpty() {
-			// if the user hasn't specified an alias, we'll insert one here so the old table name still works
-			table.Alias.As = sqlparser.NewIdentifierCS(name.String())
-		}
-	}
-	plan := &Route{
-		Source: &Table{
-			QTable: table,
-			VTable: vschemaTable,
-		},
-		Keyspace: vschemaTable.Keyspace,
-	}
-
-	for _, columnVindex := range vschemaTable.ColumnVindexes {
-		plan.VindexPreds = append(plan.VindexPreds, &VindexPlusPredicates{ColVindex: columnVindex, TableID: solves})
-	}
-
-	switch {
-	case vschemaTable.Type == vindexes.TypeSequence:
-		plan.RouteOpCode = engine.Next
-	case vschemaTable.Type == vindexes.TypeReference:
-		plan.RouteOpCode = engine.Reference
-	case !vschemaTable.Keyspace.Sharded:
-		plan.RouteOpCode = engine.Unsharded
-	case vschemaTable.Pinned != nil:
-		// Pinned tables have their keyspace ids already assigned.
-		// Use the Binary vindex, which is the identity function
-		// for keyspace id.
-		plan.RouteOpCode = engine.EqualUnique
-		vindex, _ := vindexes.NewBinary("binary", nil)
-		plan.Selected = &VindexOption{
-			Ready:       true,
-			Values:      []evalengine.Expr{evalengine.NewLiteralString(vschemaTable.Pinned, collations.TypedCollation{})},
-			ValueExprs:  nil,
-			Predicates:  nil,
-			OpCode:      engine.EqualUnique,
-			FoundVindex: vindex,
-			Cost: Cost{
-				OpCode: engine.EqualUnique,
-			},
-		}
-	default:
-		plan.RouteOpCode = engine.Scatter
-	}
-	for _, predicate := range table.Predicates {
-		err = plan.UpdateRoutingLogic(ctx, predicate)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	if plan.RouteOpCode == engine.Scatter && len(table.Predicates) > 0 {
-		// If we have a scatter query, it's worth spending a little extra time seeing if we can't improve it
-		for _, pred := range table.Predicates {
-			rewritten := tryRewriteOrToIn(pred)
-			if rewritten != nil {
-				err = plan.UpdateRoutingLogic(ctx, rewritten)
-				if err != nil {
-					return nil, err
-				}
-			}
-		}
-	}
-
-	return plan, nil
-}
-
 func tryRewriteOrToIn(expr sqlparser.Expr) sqlparser.Expr {
 	rewrote := false
 	newPred := sqlparser.Rewrite(sqlparser.CloneExpr(expr), func(cursor *sqlparser.Cursor) bool {
@@ -997,130 +907,17 @@ func hexEqual(a, b *sqlparser.Literal) bool {
 	return false
 }
 
-func pushJoinPredicates(
-	ctx *plancontext.PlanningContext,
-	exprs []sqlparser.Expr,
-	op Operator,
-) (Operator, error) {
+func pushJoinPredicates(ctx *plancontext.PlanningContext, exprs []sqlparser.Expr, op *ApplyJoin) (Operator, error) {
 	if len(exprs) == 0 {
 		return op, nil
 	}
 
-	switch op := op.(type) {
-	case *ApplyJoin:
-		return pushJoinPredicateOnJoin(ctx, exprs, op)
-	case *Route:
-		return pushJoinPredicateOnRoute(ctx, exprs, op)
-	case *Table:
-		return PushPredicate(ctx, sqlparser.AndExpressions(exprs...), op)
-	case *Derived:
-		return pushJoinPredicateOnDerived(ctx, exprs, op)
-	case *Filter:
-		op.Predicates = append(op.Predicates, exprs...)
-		return op, nil
-	default:
-		return nil, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "unknown type %T pushJoinPredicates", op)
-	}
-}
-
-func pushJoinPredicateOnRoute(ctx *plancontext.PlanningContext, exprs []sqlparser.Expr, op *Route) (Operator, error) {
-	for _, expr := range exprs {
-		err := op.UpdateRoutingLogic(ctx, expr)
-		if err != nil {
-			return nil, err
-		}
-	}
-	newSrc, err := pushJoinPredicates(ctx, exprs, op.Source)
-	op.Source = newSrc
-	return op, err
-}
-
-func pushJoinPredicateOnJoin(ctx *plancontext.PlanningContext, exprs []sqlparser.Expr, node *ApplyJoin) (Operator, error) {
-	node = Clone(node).(*ApplyJoin)
-	var rhsPreds []sqlparser.Expr
-	var lhsPreds []sqlparser.Expr
-	var lhsVarsName []string
-	for _, expr := range exprs {
-		// We find the dependencies for the given expression and if they are solved entirely by one
-		// side of the join tree, then we push the predicate there and do not break it into parts.
-		// In case a predicate has no dependencies, then it is pushed to both sides so that we can filter
-		// rows as early as possible making join cheaper on the vtgate level.
-		depsForExpr := ctx.SemTable.RecursiveDeps(expr)
-		singleSideDeps := false
-		lhsTables := TableID(node.LHS)
-		if depsForExpr.IsSolvedBy(lhsTables) {
-			lhsPreds = append(lhsPreds, expr)
-			singleSideDeps = true
-		}
-		if depsForExpr.IsSolvedBy(TableID(node.RHS)) {
-			rhsPreds = append(rhsPreds, expr)
-			singleSideDeps = true
-		}
-
-		if singleSideDeps {
-			continue
-		}
-
-		bvName, cols, predicate, err := BreakExpressionInLHSandRHS(ctx, expr, lhsTables)
-		if err != nil {
-			return nil, err
-		}
-		node.LHSColumns = append(node.LHSColumns, cols...)
-		lhsVarsName = append(lhsVarsName, bvName...)
-		rhsPreds = append(rhsPreds, predicate)
-	}
-	if node.LHSColumns != nil && lhsVarsName != nil {
-		newNode, offsets, err := PushOutputColumns(ctx, node.LHS, node.LHSColumns...)
-		if err != nil {
-			return nil, err
-		}
-		node.LHS = newNode
-		for i, idx := range offsets {
-			node.Vars[lhsVarsName[i]] = idx
-		}
-	}
-	lhsPlan, err := pushJoinPredicates(ctx, lhsPreds, node.LHS)
-	if err != nil {
-		return nil, err
-	}
-
-	rhsPlan, err := pushJoinPredicates(ctx, rhsPreds, node.RHS)
-	if err != nil {
-		return nil, err
-	}
-
-	node.LHS = lhsPlan
-	node.RHS = rhsPlan
-	// If the predicate field is previously non-empty
-	// keep that predicate too
-	if node.Predicate != nil {
-		exprs = append(exprs, node.Predicate)
-	}
-	node.Predicate = sqlparser.AndExpressions(exprs...)
-	return node, nil
-}
-
-func pushJoinPredicateOnDerived(ctx *plancontext.PlanningContext, exprs []sqlparser.Expr, node *Derived) (Operator, error) {
-	node = Clone(node).(*Derived)
-
-	newExpressions := make([]sqlparser.Expr, 0, len(exprs))
 	for _, expr := range exprs {
-		tblInfo, err := ctx.SemTable.TableInfoForExpr(expr)
-		if err != nil {
-			return nil, err
-		}
-		rewritten, err := semantics.RewriteDerivedTableExpression(expr, tblInfo)
+		_, err := addPredicate(op, ctx, expr, true)
 		if err != nil {
 			return nil, err
 		}
-		newExpressions = append(newExpressions, rewritten)
 	}
 
-	newInner, err := pushJoinPredicates(ctx, newExpressions, node.Source)
-	if err != nil {
-		return nil, err
-	}
-
-	node.Source = newInner
-	return node, nil
+	return op, nil
 }
diff --git a/go/vt/vtgate/planbuilder/operators/subquery.go b/go/vt/vtgate/planbuilder/operators/subquery.go
index f43292b3ca..0509ebc219 100644
--- a/go/vt/vtgate/planbuilder/operators/subquery.go
+++ b/go/vt/vtgate/planbuilder/operators/subquery.go
@@ -18,6 +18,7 @@ package operators
 
 import (
 	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
 )
 
 // SubQuery stores the information about subquery
@@ -78,3 +79,21 @@ func (s *SubQuery) Inputs() []Operator {
 	}
 	return operators
 }
+
+func createSubqueryFromStatement(ctx *plancontext.PlanningContext, stmt sqlparser.Statement) (*SubQuery, error) {
+	if len(ctx.SemTable.SubqueryMap[stmt]) == 0 {
+		return nil, nil
+	}
+	subq := &SubQuery{}
+	for _, sq := range ctx.SemTable.SubqueryMap[stmt] {
+		opInner, err := CreateLogicalOperatorFromAST(ctx, sq.Subquery.Select)
+		if err != nil {
+			return nil, err
+		}
+		subq.Inner = append(subq.Inner, &SubQueryInner{
+			ExtractedSubquery: sq,
+			Inner:             opInner,
+		})
+	}
+	return subq, nil
+}
diff --git a/go/vt/vtgate/planbuilder/operators/subquery_planning.go b/go/vt/vtgate/planbuilder/operators/subquery_planning.go
index cf6d71409b..83b6a4eb8a 100644
--- a/go/vt/vtgate/planbuilder/operators/subquery_planning.go
+++ b/go/vt/vtgate/planbuilder/operators/subquery_planning.go
@@ -404,12 +404,14 @@ func createCorrelatedSubqueryOp(
 		sqlparser.Rewrite(pred, func(cursor *sqlparser.Cursor) bool {
 			switch node := cursor.Node().(type) {
 			case *sqlparser.ColName:
-				if ctx.SemTable.RecursiveDeps(node).IsSolvedBy(TableID(resultOuterOp)) {
+				nodeDeps := ctx.SemTable.RecursiveDeps(node)
+				if nodeDeps.IsSolvedBy(TableID(resultOuterOp)) {
 					// check whether the bindVariable already exists in the map
 					// we do so by checking that the column names are the same and their recursive dependencies are the same
 					// so if the column names user.a and a would also be equal if the latter is also referencing the user table
 					for colName, bindVar := range bindVars {
-						if node.Name.Equal(colName.Name) && ctx.SemTable.RecursiveDeps(node).Equals(ctx.SemTable.RecursiveDeps(colName)) {
+						colNameDeps := ctx.SemTable.RecursiveDeps(colName)
+						if node.Name.Equal(colName.Name) && nodeDeps.Equals(colNameDeps) {
 							cursor.Replace(sqlparser.NewArgument(bindVar))
 							return false
 						}
diff --git a/go/vt/vtgate/planbuilder/operators/system_tables.go b/go/vt/vtgate/planbuilder/operators/system_tables.go
index c299354c50..8bcf2c5709 100644
--- a/go/vt/vtgate/planbuilder/operators/system_tables.go
+++ b/go/vt/vtgate/planbuilder/operators/system_tables.go
@@ -27,7 +27,7 @@ import (
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 )
 
-func (rp *Route) findSysInfoRoutingPredicatesGen4(predicates []sqlparser.Expr, reservedVars *sqlparser.ReservedVars) error {
+func (r *Route) findSysInfoRoutingPredicatesGen4(predicates []sqlparser.Expr, reservedVars *sqlparser.ReservedVars) error {
 	for _, pred := range predicates {
 		isTableSchema, bvName, out, err := extractInfoSchemaRoutingPredicate(pred, reservedVars)
 		if err != nil {
@@ -39,12 +39,12 @@ func (rp *Route) findSysInfoRoutingPredicatesGen4(predicates []sqlparser.Expr, r
 		}
 
 		if isTableSchema {
-			rp.SysTableTableSchema = append(rp.SysTableTableSchema, out)
+			r.SysTableTableSchema = append(r.SysTableTableSchema, out)
 		} else {
-			if rp.SysTableTableName == nil {
-				rp.SysTableTableName = map[string]evalengine.Expr{}
+			if r.SysTableTableName == nil {
+				r.SysTableTableName = map[string]evalengine.Expr{}
 			}
-			rp.SysTableTableName[bvName] = out
+			r.SysTableTableName[bvName] = out
 		}
 	}
 	return nil
diff --git a/go/vt/vtgate/planbuilder/operators/union.go b/go/vt/vtgate/planbuilder/operators/union.go
index 2c5df1a4af..42b9e18760 100644
--- a/go/vt/vtgate/planbuilder/operators/union.go
+++ b/go/vt/vtgate/planbuilder/operators/union.go
@@ -17,7 +17,10 @@ limitations under the License.
 package operators
 
 import (
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/sqlparser"
+	"vitess.io/vitess/go/vt/vterrors"
+	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
 )
 
 type Union struct {
@@ -46,3 +49,73 @@ func (u *Union) Clone(inputs []Operator) Operator {
 func (u *Union) Inputs() []Operator {
 	return u.Sources
 }
+
+// addPredicate adds a predicate a UNION by pushing the predicate to all sources of the UNION.
+/* this is done by offset and expression rewriting. Say we have a query like so:
+select * (
+ 	select foo as col, bar from tbl1
+	union
+	select id, baz from tbl2
+) as X where X.col = 42
+
+We want to push down the `X.col = 42` as far down the operator tree as possible. We want
+to end up with an operator tree that looks something like this:
+
+select * (
+ 	select foo as col, bar from tbl1 where foo = 42
+	union
+	select id, baz from tbl2 where id = 42
+) as X
+
+Notice how `X.col = 42` has been translated to `foo = 42` and `id = 42` on respective WHERE clause.
+The first SELECT of the union dictates the column names, and the second is whatever expression
+can be found on the same offset. The names of the RHS are discarded.
+*/
+func (u *Union) addPredicate(ctx *plancontext.PlanningContext, expr sqlparser.Expr) error {
+	offsets := make(map[string]int)
+	for i, selectExpr := range u.SelectStmts[0].SelectExprs {
+		ae, ok := selectExpr.(*sqlparser.AliasedExpr)
+		if !ok {
+			return vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "can't push predicates on UNION where the first SELECT contains star or next")
+		}
+		if !ae.As.IsEmpty() {
+			offsets[ae.As.String()] = i
+			continue
+		}
+		col, ok := ae.Expr.(*sqlparser.ColName)
+		if ok {
+			offsets[col.Name.Lowered()] = i
+		}
+	}
+
+	for i := range u.Sources {
+		predicate := sqlparser.CloneExpr(expr)
+		var err error
+		predicate = sqlparser.Rewrite(predicate, func(cursor *sqlparser.Cursor) bool {
+			col, ok := cursor.Node().(*sqlparser.ColName)
+			if !ok {
+				return err == nil
+			}
+
+			idx, ok := offsets[col.Name.Lowered()]
+			if !ok {
+				err = vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "can't push predicates on concatenate")
+				return false
+			}
+
+			ae, ok := u.SelectStmts[i].SelectExprs[idx].(*sqlparser.AliasedExpr)
+			if !ok {
+				err = vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "can't push predicates on concatenate")
+				return false
+			}
+			cursor.Replace(ae.Expr)
+			return false
+		}, nil).(sqlparser.Expr)
+		if err != nil {
+			return err
+		}
+		u.Sources[i], err = PushPredicate(ctx, predicate, u.Sources[i])
+	}
+
+	return nil
+}
diff --git a/go/vt/vtgate/planbuilder/operators/vindex.go b/go/vt/vtgate/planbuilder/operators/vindex.go
index c3df2ac488..35786afe68 100644
--- a/go/vt/vtgate/planbuilder/operators/vindex.go
+++ b/go/vt/vtgate/planbuilder/operators/vindex.go
@@ -21,6 +21,7 @@ import (
 	"vitess.io/vitess/go/vt/sqlparser"
 	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vtgate/engine"
+	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
 	"vitess.io/vitess/go/vt/vtgate/semantics"
 	"vitess.io/vitess/go/vt/vtgate/vindexes"
 )
@@ -82,7 +83,7 @@ outer:
 	return idxs, nil
 }
 
-// CheckValid implements the Operator interface
+// checkValid implements the Operator interface
 func (v *Vindex) CheckValid() error {
 	if len(v.Table.Predicates) == 0 {
 		return vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: where clause for vindex function must be of the form id = <val> or id in(<val>,...) (where clause missing)")
@@ -90,3 +91,46 @@ func (v *Vindex) CheckValid() error {
 
 	return nil
 }
+
+func (v *Vindex) addPredicate(ctx *plancontext.PlanningContext, expr sqlparser.Expr) error {
+	for _, e := range sqlparser.SplitAndExpression(nil, expr) {
+		deps := ctx.SemTable.RecursiveDeps(e)
+		if deps.NumberOfTables() > 1 {
+			return vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, vindexUnsupported+" (multiple tables involved)")
+		}
+		// check if we already have a predicate
+		if v.OpCode != engine.VindexNone {
+			return vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (multiple filters)")
+		}
+
+		// check LHS
+		comparison, ok := e.(*sqlparser.ComparisonExpr)
+		if !ok {
+			return vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (not a comparison)")
+		}
+		if comparison.Operator != sqlparser.EqualOp && comparison.Operator != sqlparser.InOp {
+			return vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (not equality)")
+		}
+		colname, ok := comparison.Left.(*sqlparser.ColName)
+		if !ok {
+			return vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (lhs is not a column)")
+		}
+		if !colname.Name.EqualString("id") {
+			return vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (lhs is not id)")
+		}
+
+		// check RHS
+		var err error
+		if sqlparser.IsValue(comparison.Right) || sqlparser.IsSimpleTuple(comparison.Right) {
+			v.Value = comparison.Right
+		} else {
+			return vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+" (rhs is not a value)")
+		}
+		if err != nil {
+			return vterrors.Errorf(vtrpcpb.Code_INTERNAL, vindexUnsupported+": %v", err)
+		}
+		v.OpCode = engine.VindexMap
+		v.Table.Predicates = append(v.Table.Predicates, e)
+	}
+	return nil
+}
diff --git a/go/vt/vttablet/tabletmanager/rpc_replication.go b/go/vt/vttablet/tabletmanager/rpc_replication.go
index 49dd010f28..3bee3383b3 100644
--- a/go/vt/vttablet/tabletmanager/rpc_replication.go
+++ b/go/vt/vttablet/tabletmanager/rpc_replication.go
@@ -43,7 +43,7 @@ var disableReplicationManager bool
 
 func registerReplicationFlags(fs *pflag.FlagSet) {
 	fs.Bool("enable_semi_sync", false, "")
-	fs.MarkDeprecated("enable_semi_sync", "--enable_semi_sync is deprecated; please set the correct durability policy on the keyspace instead.")
+	fs.MarkDeprecated("enable_semi_sync", "please set the correct durability policy on the keyspace instead.")
 
 	fs.BoolVar(&setSuperReadOnly, "use_super_read_only", setSuperReadOnly, "Set super_read_only flag when performing planned failover.")
 	fs.BoolVar(&disableReplicationManager, "disable-replication-manager", disableReplicationManager, "Disable replication manager to prevent replication repairs.")


Test patch:
diff --git a/go/flags/endtoend/vttestserver.txt b/go/flags/endtoend/vttestserver.txt
index 9113e6917a..31f1df7f3c 100644
--- a/go/flags/endtoend/vttestserver.txt
+++ b/go/flags/endtoend/vttestserver.txt
@@ -5,7 +5,6 @@ Usage of vttestserver:
       --backup_engine_implementation string                              Specifies which implementation to use for creating new backups (builtin or xtrabackup). Restores will always be done with whichever engine created a given backup. (default "builtin")
       --backup_storage_block_size int                                    if backup_storage_compress is true, backup_storage_block_size sets the byte size for each block while compressing (default is 250000). (default 250000)
       --backup_storage_compress                                          if set, the backup files will be compressed (default is true). Set to false for instance if a backup_storage_hook is specified and it compresses the data. (default true)
-      --backup_storage_hook string                                       if set, we send the contents of the backup files through this hook.
       --backup_storage_number_blocks int                                 if backup_storage_compress is true, backup_storage_number_blocks sets the number of blocks that can be processed, at once, before the writer blocks, during compression (default is 2). It should be equal to the number of CPUs available for compression. (default 2)
       --builtinbackup_mysqld_timeout duration                            how long to wait for mysqld to shutdown at the start of the backup. (default 10m0s)
       --builtinbackup_progress duration                                  how often to send progress updates when backing up large files. (default 5s)
diff --git a/go/vt/vtgate/planbuilder/operators/operator_test.go b/go/vt/vtgate/planbuilder/operators/operator_test.go
index 449a33a196..fed3fa8a81 100644
--- a/go/vt/vtgate/planbuilder/operators/operator_test.go
+++ b/go/vt/vtgate/planbuilder/operators/operator_test.go
@@ -105,7 +105,7 @@ func TestOperator(t *testing.T) {
 			ctx := plancontext.NewPlanningContext(nil, semTable, nil, 0)
 			optree, err := CreateLogicalOperatorFromAST(ctx, stmt)
 			require.NoError(t, err)
-			optree, err = Compact(ctx, optree)
+			optree, err = compact(ctx, optree)
 			require.NoError(t, err)
 			output := testString(optree)
 			assert.Equal(t, tc.expected, output)
diff --git a/go/vt/vtgate/planbuilder/plan_test.go b/go/vt/vtgate/planbuilder/plan_test.go
index 287a2b8043..70f89f88ec 100644
--- a/go/vt/vtgate/planbuilder/plan_test.go
+++ b/go/vt/vtgate/planbuilder/plan_test.go
@@ -781,8 +781,11 @@ func testFile(t *testing.T, filename, tempDir string, vschema *vschemaWrapper, r
 			//       with this last expectation, it is an error if the Gen4 planner
 			//       produces the same plan as the V3 planner does
 			t.Run(fmt.Sprintf("Gen4: %s", testName), func(t *testing.T) {
-				if compacted(out) != compacted(string(tcase.Gen4Plan)) {
-					t.Errorf("Gen4 - %s\nDiff:\n%s\n[%s] \n[%s]", filename, cmp.Diff(tcase.Gen4Plan, out), tcase.Gen4Plan, out)
+				x := string(tcase.Gen4Plan)
+				s := compacted(out)
+				s2 := compacted(x)
+				if s != s2 {
+					t.Errorf("Gen4 - %s\nDiff:\n%s\n[%s] \n[%s]", filename, cmp.Diff(s, s2), tcase.Gen4Plan, out)
 				}
 
 				if outFirstPlanner == out {
diff --git a/go/vt/vtgate/planbuilder/testdata/systemtables_cases.json b/go/vt/vtgate/planbuilder/testdata/systemtables_cases.json
index 235fc1c398..a880fec7fd 100644
--- a/go/vt/vtgate/planbuilder/testdata/systemtables_cases.json
+++ b/go/vt/vtgate/planbuilder/testdata/systemtables_cases.json
@@ -1434,28 +1434,6 @@
       }
     }
   },
-  {
-    "comment": "systable union query in derived table with constraint on outside (star projection)",
-    "query": "select * from (select * from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'user_extra' union select * from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'music') `kcu` where `constraint_name` = 'primary'",
-    "v3-plan": "symbol constraint_name not found in table or subquery",
-    "gen4-plan": {
-      "QueryType": "SELECT",
-      "Original": "select * from (select * from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'user_extra' union select * from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'music') `kcu` where `constraint_name` = 'primary'",
-      "Instructions": {
-        "OperatorType": "Route",
-        "Variant": "DBA",
-        "Keyspace": {
-          "Name": "main",
-          "Sharded": false
-        },
-        "FieldQuery": "select * from (select * from information_schema.key_column_usage as kcu where 1 != 1 union select * from information_schema.key_column_usage as kcu where 1 != 1) as kcu where 1 != 1",
-        "Query": "select * from (select * from information_schema.key_column_usage as kcu where kcu.table_schema = :__vtschemaname and kcu.table_name = :kcu_table_name and constraint_name = 'primary' union select * from information_schema.key_column_usage as kcu where kcu.table_schema = :__vtschemaname and kcu.table_name = :kcu_table_name1 and constraint_name = 'primary') as kcu",
-        "SysTableTableName": "[kcu_table_name1:VARCHAR(\"music\"), kcu_table_name:VARCHAR(\"user_extra\")]",
-        "SysTableTableSchema": "[VARCHAR(\"user\"), VARCHAR(\"user\")]",
-        "Table": "information_schema.key_column_usage"
-      }
-    }
-  },
   {
     "comment": "table_schema OR predicate\n# It is unsupported because we do not route queries to multiple keyspaces right now",
     "query": "SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = 'ks' OR TABLE_SCHEMA = 'main'",
diff --git a/go/vt/vtgate/planbuilder/testdata/union_cases.json b/go/vt/vtgate/planbuilder/testdata/union_cases.json
index e32d28c862..e1693a44e8 100644
--- a/go/vt/vtgate/planbuilder/testdata/union_cases.json
+++ b/go/vt/vtgate/planbuilder/testdata/union_cases.json
@@ -2286,6 +2286,7 @@
     "plan": "can't do ORDER BY on top of UNION"
   },
   {
+    "comment": "select 1 from (select id+42 as foo from user union select 1+id as foo from unsharded) as t",
     "query": "select 1 from (select id+42 as foo from user union select 1+id as foo from unsharded) as t",
     "v3-plan": "unsupported: expression on results of a cross-shard subquery",
     "gen4-plan": {
@@ -2339,5 +2340,55 @@
         "user.user"
       ]
     }
+  },
+  {
+    "comment": "systable union query in derived table with constraint on outside (without star projection)",
+    "query": "select id from (select id from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'user_extra' union select id from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'music') `kcu` where `id` = 'primary'",
+    "v3-plan": "unsupported: filtering on results of cross-shard subquery",
+    "gen4-plan": {
+      "QueryType": "SELECT",
+      "Original": "select id from (select id from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'user_extra' union select id from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'music') `kcu` where `id` = 'primary'",
+      "Instructions": {
+        "OperatorType": "Route",
+        "Variant": "DBA",
+        "Keyspace": {
+          "Name": "main",
+          "Sharded": false
+        },
+        "FieldQuery": "select id from (select id from information_schema.key_column_usage as kcu where 1 != 1 union select id from information_schema.key_column_usage as kcu where 1 != 1) as kcu where 1 != 1",
+        "Query": "select id from (select id from information_schema.key_column_usage as kcu where kcu.table_schema = :__vtschemaname and kcu.table_name = :kcu_table_name and id = 'primary' union select id from information_schema.key_column_usage as kcu where kcu.table_schema = :__vtschemaname and kcu.table_name = :kcu_table_name1 and id = 'primary') as kcu",
+        "SysTableTableName": "[kcu_table_name1:VARCHAR(\"music\"), kcu_table_name:VARCHAR(\"user_extra\")]",
+        "SysTableTableSchema": "[VARCHAR(\"user\"), VARCHAR(\"user\")]",
+        "Table": "information_schema.key_column_usage"
+      }
+    }
+  },
+  {
+    "comment": "pushes predicate on both sides of UNION",
+    "query": "select * from (select name, id as foo from user union select 'extra', user_id from user_extra) X where X.foo = 3",
+    "v3-plan": "unsupported: filtering on results of cross-shard subquery",
+    "gen4-plan": {
+      "QueryType": "SELECT",
+      "Original": "select * from (select name, id as foo from user union select 'extra', user_id from user_extra) X where X.foo = 3",
+      "Instructions": {
+        "OperatorType": "Route",
+        "Variant": "EqualUnique",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select X.`name`, X.foo from (select `name`, id as foo from `user` where 1 != 1 union select 'extra', user_id from user_extra where 1 != 1) as X where 1 != 1",
+        "Query": "select X.`name`, X.foo from (select `name`, id as foo from `user` where id = 3 union select 'extra', user_id from user_extra where user_id = 3) as X",
+        "Table": "`user`",
+        "Values": [
+          "INT64(3)"
+        ],
+        "Vindex": "user_index"
+      },
+      "TablesUsed": [
+        "user.user",
+        "user.user_extra"
+      ]
+    }
   }
-]
\ No newline at end of file
+]
diff --git a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.json b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.json
index 80fdce4e4a..421d6900bb 100644
--- a/go/vt/vtgate/planbuilder/testdata/unsupported_cases.json
+++ b/go/vt/vtgate/planbuilder/testdata/unsupported_cases.json
@@ -446,12 +446,6 @@
     "v3-plan": "unsupported: cross-shard query with aggregates",
     "gen4-plan": "using aggregation on top of a *planbuilder.concatenateGen4 plan is not yet supported"
   },
-  {
-    "comment": "systable union query in derived table with constraint on outside (without star projection)",
-    "query": "select id from (select id from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'user_extra' union select id from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'music') `kcu` where `id` = 'primary'",
-    "v3-plan": "unsupported: filtering on results of cross-shard subquery",
-    "gen4-plan": "can't push predicates on concatenate"
-  },
   {
     "comment": "insert having subquery in row values",
     "query": "insert into user(id, name) values ((select 1 from user where id = 1), 'A')",
@@ -471,5 +465,11 @@
     "comment": "mix lock with other expr",
     "query": "select get_lock('xyz', 10), 1 from dual",
     "plan": "unsupported: lock function and other expression in same select query"
+  },
+  {
+    "comment": "systable union query in derived table with constraint on outside (star projection)",
+    "query": "select * from (select * from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'user_extra' union select * from `information_schema`.`key_column_usage` `kcu` where `kcu`.`table_schema` = 'user' and `kcu`.`table_name` = 'music') `kcu` where `constraint_name` = 'primary'",
+    "v3-plan": "symbol constraint_name not found in table or subquery",
+    "gen4-plan": "can't push predicates on UNION where the first SELECT contains star or next"
   }
-]
\ No newline at end of file
+]