diff --git a/go/test/endtoend/vtgate/queries/misc/misc_test.go b/go/test/endtoend/vtgate/queries/misc/misc_test.go
index e768524c2c9..5e991024db1 100644
--- a/go/test/endtoend/vtgate/queries/misc/misc_test.go
+++ b/go/test/endtoend/vtgate/queries/misc/misc_test.go
@@ -291,3 +291,177 @@ func TestAnalyze(t *testing.T) {
 		})
 	}
 }
+<<<<<<< HEAD
+=======
+
+// TestTransactionModeVar executes SELECT on `transaction_mode` variable
+func TestTransactionModeVar(t *testing.T) {
+	utils.SkipIfBinaryIsBelowVersion(t, 19, "vtgate")
+
+	mcmp, closer := start(t)
+	defer closer()
+
+	tcases := []struct {
+		setStmt string
+		expRes  string
+	}{{
+		expRes: `[[VARCHAR("MULTI")]]`,
+	}, {
+		setStmt: `set transaction_mode = single`,
+		expRes:  `[[VARCHAR("SINGLE")]]`,
+	}, {
+		setStmt: `set transaction_mode = multi`,
+		expRes:  `[[VARCHAR("MULTI")]]`,
+	}, {
+		setStmt: `set transaction_mode = twopc`,
+		expRes:  `[[VARCHAR("TWOPC")]]`,
+	}}
+
+	for _, tcase := range tcases {
+		mcmp.Run(tcase.setStmt, func(mcmp *utils.MySQLCompare) {
+			if tcase.setStmt != "" {
+				utils.Exec(t, mcmp.VtConn, tcase.setStmt)
+			}
+			utils.AssertMatches(t, mcmp.VtConn, "select @@transaction_mode", tcase.expRes)
+		})
+	}
+}
+
+// TestAliasesInOuterJoinQueries tests that aliases work in queries that have outer join clauses.
+func TestAliasesInOuterJoinQueries(t *testing.T) {
+	utils.SkipIfBinaryIsBelowVersion(t, 20, "vtgate")
+
+	mcmp, closer := start(t)
+	defer closer()
+
+	// Insert data into the 2 tables
+	mcmp.Exec("insert into t1(id1, id2) values (1,2), (42,5), (5, 42)")
+	mcmp.Exec("insert into tbl(id, unq_col, nonunq_col) values (1,2,3), (2,5,3), (3, 42, 2)")
+
+	// Check that the select query works as intended and verifying the column names as well.
+	mcmp.ExecWithColumnCompare("select t1.id1 as t0, t1.id1 as t1, tbl.unq_col as col from t1 left outer join tbl on t1.id2 = tbl.nonunq_col")
+	mcmp.ExecWithColumnCompare("select t1.id1 as t0, t1.id1 as t1, tbl.unq_col as col from t1 left outer join tbl on t1.id2 = tbl.nonunq_col order by t1.id2 limit 2")
+	mcmp.ExecWithColumnCompare("select t1.id1 as t0, t1.id1 as t1, tbl.unq_col as col from t1 left outer join tbl on t1.id2 = tbl.nonunq_col order by t1.id2 limit 2 offset 2")
+	mcmp.ExecWithColumnCompare("select t1.id1 as t0, t1.id1 as t1, count(*) as leCount from t1 left outer join tbl on t1.id2 = tbl.nonunq_col group by 1, t1")
+	mcmp.ExecWithColumnCompare("select t.id1, t.id2, derived.unq_col from t1 t join (select id, unq_col, nonunq_col from tbl) as derived on t.id2 = derived.nonunq_col")
+	if utils.BinaryIsAtLeastAtVersion(21, "vtgate") {
+		mcmp.ExecWithColumnCompare("select * from t1 t left join tbl on t.id1 = 666 and t.id2 = tbl.id")
+	}
+}
+
+func TestAlterTableWithView(t *testing.T) {
+	utils.SkipIfBinaryIsBelowVersion(t, 20, "vtgate")
+	mcmp, closer := start(t)
+	defer closer()
+
+	// Test that create/alter view works and the output is as expected
+	mcmp.Exec(`use ks_misc`)
+	mcmp.Exec(`create view v1 as select * from t1`)
+	var viewDef string
+	utils.WaitForVschemaCondition(t, clusterInstance.VtgateProcess, keyspaceName, func(t *testing.T, ksMap map[string]any) bool {
+		views, ok := ksMap["views"]
+		if !ok {
+			return false
+		}
+		viewsMap := views.(map[string]any)
+		view, ok := viewsMap["v1"]
+		if ok {
+			viewDef = view.(string)
+		}
+		return ok
+	}, "Waiting for view creation")
+	mcmp.Exec(`insert into t1(id1, id2) values (1, 1)`)
+	mcmp.AssertMatches("select * from v1", `[[INT64(1) INT64(1)]]`)
+
+	// alter table add column
+	mcmp.Exec(`alter table t1 add column test bigint`)
+	time.Sleep(10 * time.Second)
+	mcmp.Exec(`alter view v1 as select * from t1`)
+
+	waitForChange := func(t *testing.T, ksMap map[string]any) bool {
+		// wait for the view definition to change
+		views := ksMap["views"]
+		viewsMap := views.(map[string]any)
+		newView := viewsMap["v1"]
+		if newView.(string) == viewDef {
+			return false
+		}
+		viewDef = newView.(string)
+		return true
+	}
+	utils.WaitForVschemaCondition(t, clusterInstance.VtgateProcess, keyspaceName, waitForChange, "Waiting for alter view")
+
+	mcmp.AssertMatches("select * from v1", `[[INT64(1) INT64(1) NULL]]`)
+
+	// alter table remove column
+	mcmp.Exec(`alter table t1 drop column test`)
+	mcmp.Exec(`alter view v1 as select * from t1`)
+
+	utils.WaitForVschemaCondition(t, clusterInstance.VtgateProcess, keyspaceName, waitForChange, "Waiting for alter view")
+
+	mcmp.AssertMatches("select * from v1", `[[INT64(1) INT64(1)]]`)
+}
+
+// TestStraightJoin tests that Vitess respects the ordering of join in a STRAIGHT JOIN query.
+func TestStraightJoin(t *testing.T) {
+	utils.SkipIfBinaryIsBelowVersion(t, 20, "vtgate")
+	mcmp, closer := start(t)
+	defer closer()
+
+	mcmp.Exec("insert into tbl(id, unq_col, nonunq_col) values (1,0,10), (2,10,10), (3,4,20), (4,30,20), (5,40,10)")
+	mcmp.Exec(`insert into t1(id1, id2) values (10, 11), (20, 13)`)
+
+	mcmp.AssertMatchesNoOrder("select tbl.unq_col, tbl.nonunq_col, t1.id2 from t1 join tbl where t1.id1 = tbl.nonunq_col",
+		`[[INT64(0) INT64(10) INT64(11)] [INT64(10) INT64(10) INT64(11)] [INT64(4) INT64(20) INT64(13)] [INT64(40) INT64(10) INT64(11)] [INT64(30) INT64(20) INT64(13)]]`,
+	)
+	// Verify that in a normal join query, vitess joins tbl with t1.
+	res, err := mcmp.VtConn.ExecuteFetch("vexplain plan select tbl.unq_col, tbl.nonunq_col, t1.id2 from t1 join tbl where t1.id1 = tbl.nonunq_col", 100, false)
+	require.NoError(t, err)
+	require.Contains(t, fmt.Sprintf("%v", res.Rows), "tbl_t1")
+
+	// Test the same query with a straight join
+	mcmp.AssertMatchesNoOrder("select tbl.unq_col, tbl.nonunq_col, t1.id2 from t1 straight_join tbl where t1.id1 = tbl.nonunq_col",
+		`[[INT64(0) INT64(10) INT64(11)] [INT64(10) INT64(10) INT64(11)] [INT64(4) INT64(20) INT64(13)] [INT64(40) INT64(10) INT64(11)] [INT64(30) INT64(20) INT64(13)]]`,
+	)
+	// Verify that in a straight join query, vitess joins t1 with tbl.
+	res, err = mcmp.VtConn.ExecuteFetch("vexplain plan select tbl.unq_col, tbl.nonunq_col, t1.id2 from t1 straight_join tbl where t1.id1 = tbl.nonunq_col", 100, false)
+	require.NoError(t, err)
+	require.Contains(t, fmt.Sprintf("%v", res.Rows), "t1_tbl")
+}
+
+func TestColumnAliases(t *testing.T) {
+	utils.SkipIfBinaryIsBelowVersion(t, 20, "vtgate")
+	mcmp, closer := start(t)
+	defer closer()
+
+	mcmp.Exec("insert into t1(id1, id2) values (0,0), (1,1)")
+	mcmp.ExecWithColumnCompare(`select a as k from (select count(*) as a from t1) t`)
+}
+
+func TestHandleNullableColumn(t *testing.T) {
+	utils.SkipIfBinaryIsBelowVersion(t, 21, "vtgate")
+	require.NoError(t,
+		utils.WaitForAuthoritative(t, keyspaceName, "tbl", clusterInstance.VtgateProcess.ReadVSchema))
+	mcmp, closer := start(t)
+	defer closer()
+
+	mcmp.Exec("insert into t1(id1, id2) values (0,0), (1,1), (2,2)")
+	mcmp.Exec("insert into tbl(id, unq_col, nonunq_col) values (0,0,0), (1,1,6)")
+	// This query tests that we handle nullable columns correctly
+	// tbl.nonunq_col is not nullable according to the schema, but because of the left join, it can be NULL
+	mcmp.ExecWithColumnCompare(`select * from t1 left join tbl on t1.id2 = tbl.id where t1.id1 = 6 or tbl.nonunq_col = 6`)
+}
+
+func TestEnumSetVals(t *testing.T) {
+	utils.SkipIfBinaryIsBelowVersion(t, 20, "vtgate")
+
+	mcmp, closer := start(t)
+	defer closer()
+	require.NoError(t, utils.WaitForAuthoritative(t, keyspaceName, "tbl_enum_set", clusterInstance.VtgateProcess.ReadVSchema))
+
+	mcmp.Exec("insert into tbl_enum_set(id, enum_col, set_col) values (1, 'medium', 'a,b,e'), (2, 'small', 'e,f,g'), (3, 'large', 'c'), (4, 'xsmall', 'a,b'), (5, 'medium', 'a,d')")
+
+	mcmp.AssertMatches("select id, enum_col, cast(enum_col as signed) from tbl_enum_set order by enum_col, id", `[[INT64(4) ENUM("xsmall") INT64(1)] [INT64(2) ENUM("small") INT64(2)] [INT64(1) ENUM("medium") INT64(3)] [INT64(5) ENUM("medium") INT64(3)] [INT64(3) ENUM("large") INT64(4)]]`)
+	mcmp.AssertMatches("select id, set_col, cast(set_col as unsigned) from tbl_enum_set order by set_col, id", `[[INT64(4) SET("a,b") UINT64(3)] [INT64(3) SET("c") UINT64(4)] [INT64(5) SET("a,d") UINT64(9)] [INT64(1) SET("a,b,e") UINT64(19)] [INT64(2) SET("e,f,g") UINT64(112)]]`)
+}
+>>>>>>> 59408f95e1 (bugfix: don't treat join predicates as filter predicates (#16472))
diff --git a/go/vt/vtgate/planbuilder/operators/apply_join.go b/go/vt/vtgate/planbuilder/operators/apply_join.go
index 5e48fb4d5e3..692a7ecd705 100644
--- a/go/vt/vtgate/planbuilder/operators/apply_join.go
+++ b/go/vt/vtgate/planbuilder/operators/apply_join.go
@@ -35,8 +35,14 @@ type (
 	ApplyJoin struct {
 		LHS, RHS ops.Operator
 
+<<<<<<< HEAD
 		// LeftJoin will be true in the case of an outer join
 		LeftJoin bool
+=======
+		// JoinType is permitted to store only 3 of the possible values
+		// NormalJoinType, StraightJoinType and LeftJoinType.
+		JoinType sqlparser.JoinType
+>>>>>>> 59408f95e1 (bugfix: don't treat join predicates as filter predicates (#16472))
 
 		// Before offset planning
 		Predicate sqlparser.Expr
@@ -312,11 +318,26 @@ func (aj *ApplyJoin) addOffset(offset int) {
 }
 
 func (aj *ApplyJoin) ShortDescription() string {
+<<<<<<< HEAD
 	pred := sqlparser.String(aj.Predicate)
 	columns := slice.Map(aj.JoinColumns, func(from JoinColumn) string {
 		return sqlparser.String(from.Original)
 	})
 	firstPart := fmt.Sprintf("on %s columns: %s", pred, strings.Join(columns, ", "))
+=======
+	fn := func(cols *applyJoinColumns) string {
+		out := slice.Map(cols.columns, func(jc applyJoinColumn) string {
+			return jc.String()
+		})
+		return strings.Join(out, ", ")
+	}
+
+	firstPart := fmt.Sprintf("on %s columns: %s", fn(aj.JoinPredicates), fn(aj.JoinColumns))
+	if aj.JoinType == sqlparser.LeftJoinType {
+		firstPart = "LEFT JOIN " + firstPart
+	}
+
+>>>>>>> 59408f95e1 (bugfix: don't treat join predicates as filter predicates (#16472))
 	if len(aj.ExtraLHSVars) == 0 {
 		return firstPart
 	}
diff --git a/go/vt/vtgate/planbuilder/operators/route_planning.go b/go/vt/vtgate/planbuilder/operators/route_planning.go
index 9a4940173e4..8e221e43441 100644
--- a/go/vt/vtgate/planbuilder/operators/route_planning.go
+++ b/go/vt/vtgate/planbuilder/operators/route_planning.go
@@ -378,6 +378,7 @@ func mergeOrJoin(ctx *plancontext.PlanningContext, lhs, rhs ops.Operator, joinPr
 			return nil, nil, vterrors.VT12001("LEFT JOIN with LIMIT on the outer side")
 		}
 
+<<<<<<< HEAD
 		if requiresSwitchingSides(ctx, lhs) {
 			return nil, nil, vterrors.VT12001("JOIN between derived tables with LIMIT")
 		}
@@ -396,6 +397,21 @@ func mergeOrJoin(ctx *plancontext.PlanningContext, lhs, rhs ops.Operator, joinPr
 		return nil, nil, err
 	}
 	return newOp, rewrite.NewTree("logical join to applyJoin ", newOp), nil
+=======
+		join := NewApplyJoin(ctx, Clone(rhs), Clone(lhs), nil, joinType)
+		for _, pred := range joinPredicates {
+			join.AddJoinPredicate(ctx, pred)
+		}
+		return join, Rewrote("logical join to applyJoin, switching side because LIMIT")
+	}
+
+	join := NewApplyJoin(ctx, Clone(lhs), Clone(rhs), nil, joinType)
+	for _, pred := range joinPredicates {
+		join.AddJoinPredicate(ctx, pred)
+	}
+
+	return join, Rewrote("logical join to applyJoin ")
+>>>>>>> 59408f95e1 (bugfix: don't treat join predicates as filter predicates (#16472))
 }
 
 func operatorsToRoutes(a, b ops.Operator) (*Route, *Route) {
@@ -614,6 +630,7 @@ func hexEqual(a, b *sqlparser.Literal) bool {
 	}
 	return false
 }
+<<<<<<< HEAD
 
 func pushJoinPredicates(ctx *plancontext.PlanningContext, exprs []sqlparser.Expr, op *ApplyJoin) (ops.Operator, error) {
 	if len(exprs) == 0 {
@@ -629,3 +646,5 @@ func pushJoinPredicates(ctx *plancontext.PlanningContext, exprs []sqlparser.Expr
 
 	return op, nil
 }
+=======
+>>>>>>> 59408f95e1 (bugfix: don't treat join predicates as filter predicates (#16472))
diff --git a/go/vt/vtgate/planbuilder/testdata/from_cases.json b/go/vt/vtgate/planbuilder/testdata/from_cases.json
index 6b8e8eba70a..4febb58075b 100644
--- a/go/vt/vtgate/planbuilder/testdata/from_cases.json
+++ b/go/vt/vtgate/planbuilder/testdata/from_cases.json
@@ -808,6 +808,59 @@
       ]
     }
   },
+  {
+    "comment": "Outer join with join predicates that only depend on the inner side",
+    "query": "select 1 from user left join user_extra on user.foo = 42 and user.bar = user_extra.bar",
+    "plan": {
+      "QueryType": "SELECT",
+      "Original": "select 1 from user left join user_extra on user.foo = 42 and user.bar = user_extra.bar",
+      "Instructions": {
+        "OperatorType": "Projection",
+        "Expressions": [
+          "1 as 1"
+        ],
+        "Inputs": [
+          {
+            "OperatorType": "Join",
+            "Variant": "LeftJoin",
+            "JoinVars": {
+              "user_bar": 1,
+              "user_foo": 0
+            },
+            "TableName": "`user`_user_extra",
+            "Inputs": [
+              {
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select `user`.foo, `user`.bar from `user` where 1 != 1",
+                "Query": "select `user`.foo, `user`.bar from `user`",
+                "Table": "`user`"
+              },
+              {
+                "OperatorType": "Route",
+                "Variant": "Scatter",
+                "Keyspace": {
+                  "Name": "user",
+                  "Sharded": true
+                },
+                "FieldQuery": "select 1 from user_extra where 1 != 1",
+                "Query": "select 1 from user_extra where user_extra.bar = :user_bar and :user_foo = 42",
+                "Table": "user_extra"
+              }
+            ]
+          }
+        ]
+      },
+      "TablesUsed": [
+        "user.user",
+        "user.user_extra"
+      ]
+    }
+  },
   {
     "comment": "Parenthesized, single chunk",
     "query": "select user.col from user join (unsharded as m1 join unsharded as m2)",