diff --git a/pkg/sql/distsql_physical_planner.go b/pkg/sql/distsql_physical_planner.go
index 4b03185e2cbc..061a1823ba13 100644
--- a/pkg/sql/distsql_physical_planner.go
+++ b/pkg/sql/distsql_physical_planner.go
@@ -2049,14 +2049,18 @@ func (dsp *DistSQLPlanner) createPlanForLookupJoin(
 	}
 
 	joinReaderSpec := execinfrapb.JoinReaderSpec{
-		Table:                    *n.table.desc.TableDesc(),
-		Type:                     n.joinType,
-		Visibility:               n.table.colCfg.visibility,
-		LockingStrength:          n.table.lockingStrength,
-		LockingWaitPolicy:        n.table.lockingWaitPolicy,
-		MaintainOrdering:         len(n.reqOrdering) > 0,
-		HasSystemColumns:         n.table.containsSystemColumns,
-		LeftJoinWithPairedJoiner: n.isSecondJoinInPairedJoiner,
+		Table:             *n.table.desc.TableDesc(),
+		Type:              n.joinType,
+		Visibility:        n.table.colCfg.visibility,
+		LockingStrength:   n.table.lockingStrength,
+		LockingWaitPolicy: n.table.lockingWaitPolicy,
+		// TODO(sumeer): specifying ordering here using isFirstJoinInPairedJoiner
+		// is late in the sense that the cost of this has not been taken into
+		// account. Make this decision earlier in CustomFuncs.GenerateLookupJoins.
+		MaintainOrdering:                  len(n.reqOrdering) > 0 || n.isFirstJoinInPairedJoiner,
+		HasSystemColumns:                  n.table.containsSystemColumns,
+		LeftJoinWithPairedJoiner:          n.isSecondJoinInPairedJoiner,
+		OutputGroupContinuationForLeftRow: n.isFirstJoinInPairedJoiner,
 	}
 	joinReaderSpec.IndexIdx, err = getIndexIdx(n.table.index, n.table.desc)
 	if err != nil {
@@ -2072,7 +2076,7 @@ func (dsp *DistSQLPlanner) createPlanForLookupJoin(
 	joinReaderSpec.LookupColumnsAreKey = n.eqColsAreKey
 
 	numInputNodeCols, planToStreamColMap, post, types :=
-		mappingHelperForLookupJoins(plan, n.input, n.table, false /* addContinuationCol */)
+		mappingHelperForLookupJoins(plan, n.input, n.table, n.isFirstJoinInPairedJoiner)
 
 	// Set the ON condition.
 	if n.onCond != nil {
diff --git a/pkg/sql/distsql_spec_exec_factory.go b/pkg/sql/distsql_spec_exec_factory.go
index 262723f81d88..49286e57f8e7 100644
--- a/pkg/sql/distsql_spec_exec_factory.go
+++ b/pkg/sql/distsql_spec_exec_factory.go
@@ -625,6 +625,7 @@ func (e *distSQLSpecExecFactory) ConstructLookupJoin(
 	eqColsAreKey bool,
 	lookupCols exec.TableColumnOrdinalSet,
 	onCond tree.TypedExpr,
+	isFirstJoinInPairedJoiner bool,
 	isSecondJoinInPairedJoiner bool,
 	reqOrdering exec.OutputOrdering,
 	locking *tree.LockingItem,
diff --git a/pkg/sql/execinfrapb/flow_diagram.go b/pkg/sql/execinfrapb/flow_diagram.go
index 08b77da81f8e..4e24b44a4fed 100644
--- a/pkg/sql/execinfrapb/flow_diagram.go
+++ b/pkg/sql/execinfrapb/flow_diagram.go
@@ -185,6 +185,9 @@ func (jr *JoinReaderSpec) summary() (string, []string) {
 	if jr.LeftJoinWithPairedJoiner {
 		details = append(details, "second join in paired-join")
 	}
+	if jr.OutputGroupContinuationForLeftRow {
+		details = append(details, "first join in paired-join")
+	}
 	return "JoinReader", details
 }
 
diff --git a/pkg/sql/logictest/testdata/logic_test/lookup_join b/pkg/sql/logictest/testdata/logic_test/lookup_join
index a5d79285be46..5b5e2fec518b 100644
--- a/pkg/sql/logictest/testdata/logic_test/lookup_join
+++ b/pkg/sql/logictest/testdata/logic_test/lookup_join
@@ -378,7 +378,7 @@ SELECT small.c, large.c FROM small LEFT JOIN large ON small.c = large.b AND larg
 27  NULL
 30  NULL
 
-## Left join with ON filter on non-covering index
+## Left join with ON filter on non-covering index. Will execute as paired-joins.
 query II rowsort
 SELECT small.c, large.d FROM small LEFT JOIN large ON small.c = large.b AND large.d < 30
 ----
@@ -393,6 +393,26 @@ SELECT small.c, large.d FROM small LEFT JOIN large ON small.c = large.b AND larg
 27  NULL
 30  NULL
 
+## Left semi join with ON filter on non-covering index. Will execute as paired-joins.
+query I rowsort
+SELECT small.c FROM small WHERE EXISTS(SELECT 1 FROM large WHERE small.c = large.b AND large.d < 30)
+----
+6
+12
+
+## Left anti join with ON filter on non-covering index. Will execute as paired-joins.
+query I rowsort
+SELECT small.c FROM small WHERE NOT EXISTS(SELECT 1 FROM large WHERE small.c = large.b AND large.d < 30)
+----
+3
+9
+15
+18
+21
+24
+27
+30
+
 # Lookup joins against interleaved tables. Regression test for #28981.
 # This is now tested more thoroughly by joinreader_test.go.
 
diff --git a/pkg/sql/lookup_join.go b/pkg/sql/lookup_join.go
index d206af467caa..90c700d32f61 100644
--- a/pkg/sql/lookup_join.go
+++ b/pkg/sql/lookup_join.go
@@ -34,13 +34,20 @@ type lookupJoinNode struct {
 	eqColsAreKey bool
 
 	// columns are the produced columns, namely the input columns and (unless the
-	// join type is semi or anti join) the columns in the table scanNode.
+	// join type is semi or anti join) the columns in the table scanNode. It
+	// includes an additional continuation column when IsFirstJoinInPairedJoin
+	// is true.
 	columns colinfo.ResultColumns
 
 	// onCond is any ON condition to be used in conjunction with the implicit
 	// equality condition on eqCols.
 	onCond tree.TypedExpr
 
+	// At most one of is{First,Second}JoinInPairedJoiner can be true.
+	// IsFirstJoinInPairedJoiner can be true only if reqOrdering asks the join
+	// to preserve ordering (currently a non-empty reqOrdering is interpreted
+	// as a bool to preserve the row ordering of the input).
+	isFirstJoinInPairedJoiner  bool
 	isSecondJoinInPairedJoiner bool
 
 	reqOrdering ReqOrdering
diff --git a/pkg/sql/opt/exec/execbuilder/relational.go b/pkg/sql/opt/exec/execbuilder/relational.go
index 5695566b18cf..5bf7d1f69d14 100644
--- a/pkg/sql/opt/exec/execbuilder/relational.go
+++ b/pkg/sql/opt/exec/execbuilder/relational.go
@@ -1457,10 +1457,25 @@ func (b *Builder) buildLookupJoin(join *memo.LookupJoinExpr) (execPlan, error) {
 
 	inputCols := join.Input.Relational().OutputCols
 	lookupCols := join.Cols.Difference(inputCols)
+	if join.IsFirstJoinInPairedJoiner {
+		lookupCols.Remove(join.ContinuationCol)
+	}
 
 	lookupOrdinals, lookupColMap := b.getColumns(lookupCols, join.Table)
-	allCols := joinOutputMap(input.outputCols, lookupColMap)
-
+	// allExprCols are the columns used in expressions evaluated by this join.
+	allExprCols := joinOutputMap(input.outputCols, lookupColMap)
+	allCols := allExprCols
+	if join.IsFirstJoinInPairedJoiner {
+		// allCols needs to include the continuation column since it will be
+		// in the result output by this join.
+		allCols = allExprCols.Copy()
+		maxValue, ok := allCols.MaxValue()
+		if !ok {
+			return execPlan{}, errors.AssertionFailedf("allCols should not be empty")
+		}
+		// Assign the continuation column the next unused value in the map.
+		allCols.Set(int(join.ContinuationCol), maxValue+1)
+	}
 	res := execPlan{outputCols: allCols}
 	if join.JoinType == opt.SemiJoinOp || join.JoinType == opt.AntiJoinOp {
 		// For semi and anti join, only the left columns are output.
@@ -1468,8 +1483,8 @@ func (b *Builder) buildLookupJoin(join *memo.LookupJoinExpr) (execPlan, error) {
 	}
 
 	ctx := buildScalarCtx{
-		ivh:     tree.MakeIndexedVarHelper(nil /* container */, allCols.Len()),
-		ivarMap: allCols,
+		ivh:     tree.MakeIndexedVarHelper(nil /* container */, allExprCols.Len()),
+		ivarMap: allExprCols,
 	}
 	var onExpr tree.TypedExpr
 	if len(join.On) > 0 {
@@ -1497,6 +1512,7 @@ func (b *Builder) buildLookupJoin(join *memo.LookupJoinExpr) (execPlan, error) {
 		join.LookupColsAreTableKey,
 		lookupOrdinals,
 		onExpr,
+		join.IsFirstJoinInPairedJoiner,
 		join.IsSecondJoinInPairedJoiner,
 		res.reqOrdering(join),
 		locking,
diff --git a/pkg/sql/opt/exec/execbuilder/testdata/lookup_join b/pkg/sql/opt/exec/execbuilder/testdata/lookup_join
index 3565c2a0f911..de144b17961d 100644
--- a/pkg/sql/opt/exec/execbuilder/testdata/lookup_join
+++ b/pkg/sql/opt/exec/execbuilder/testdata/lookup_join
@@ -700,8 +700,6 @@ vectorized: true
           spans: FULL SCAN
 
 # Left join with ON filter on non-covering index
-# TODO(radu): this doesn't use lookup join yet, the current rules don't cover
-# left join with ON condition on columns that are not covered by the index.
 query T
 EXPLAIN (VERBOSE) SELECT small.c, large.d FROM small LEFT JOIN large ON small.c = large.b AND large.d < 30
 ----
@@ -712,27 +710,100 @@ vectorized: true
 │ columns: (c, d)
 │ estimated row count: 336
 │
-└── • hash join (right outer)
-    │ columns: (b, d, c)
+└── • project
+    │ columns: (c, b, d)
     │ estimated row count: 336
-    │ equality: (b) = (c)
     │
-    ├── • filter
-    │   │ columns: (b, d)
-    │   │ estimated row count: 3,303
-    │   │ filter: d < 30
-    │   │
-    │   └── • scan
-    │         columns: (b, d)
-    │         estimated row count: 10,000
-    │         table: large@primary
-    │         spans: FULL SCAN
+    └── • lookup join (left outer)
+        │ columns: (c, a, b, cont, d)
+        │ table: large@primary
+        │ equality: (a, b) = (a,b)
+        │ equality cols are key
+        │ pred: d < 30
+        │
+        └── • lookup join (left outer)
+            │ columns: (c, a, b, cont)
+            │ estimated row count: 1,000
+            │ table: large@bc
+            │ equality: (c) = (b)
+            │
+            └── • scan
+                  columns: (c)
+                  estimated row count: 100
+                  table: small@primary
+                  spans: FULL SCAN
+
+# Left semi-join with ON filter on non-covering index
+query T
+EXPLAIN (VERBOSE) SELECT small.c FROM small WHERE EXISTS(SELECT 1 FROM large WHERE small.c = large.b AND large.d < 30)
+----
+distribution: full
+vectorized: true
+·
+• project
+│ columns: (c)
+│ estimated row count: 100
+│
+└── • project
+    │ columns: (c, b)
+    │ estimated row count: 336
     │
-    └── • scan
-          columns: (c)
-          estimated row count: 100
-          table: small@primary
-          spans: FULL SCAN
+    └── • lookup join (semi)
+        │ columns: (c, a, b, cont)
+        │ table: large@primary
+        │ equality: (a, b) = (a,b)
+        │ equality cols are key
+        │ pred: d < 30
+        │
+        └── • lookup join (inner)
+            │ columns: (c, a, b, cont)
+            │ estimated row count: 990
+            │ table: large@bc
+            │ equality: (c) = (b)
+            │
+            └── • scan
+                  columns: (c)
+                  estimated row count: 100
+                  table: small@primary
+                  spans: FULL SCAN
+
+# Left anti-join with ON filter on non-covering index
+query T
+EXPLAIN (VERBOSE) SELECT small.c FROM small WHERE NOT EXISTS(SELECT 1 FROM large WHERE small.c = large.b AND large.d < small.a)
+----
+distribution: full
+vectorized: true
+·
+• project
+│ columns: (c)
+│ estimated row count: 67
+│
+└── • project
+    │ columns: (a, c)
+    │ estimated row count: 67
+    │
+    └── • project
+        │ columns: (a, c, b)
+        │ estimated row count: 667
+        │
+        └── • lookup join (anti)
+            │ columns: (a, c, a, b, cont)
+            │ table: large@primary
+            │ equality: (a, b) = (a,b)
+            │ equality cols are key
+            │ pred: d < a
+            │
+            └── • lookup join (left outer)
+                │ columns: (a, c, a, b, cont)
+                │ estimated row count: 1,000
+                │ table: large@bc
+                │ equality: (c) = (b)
+                │
+                └── • scan
+                      columns: (a, c)
+                      estimated row count: 100
+                      table: small@primary
+                      spans: FULL SCAN
 
 ###########################################################
 #  LOOKUP JOINS ON IMPLICIT INDEX KEY COLUMNS             #
diff --git a/pkg/sql/opt/exec/explain/result_columns.go b/pkg/sql/opt/exec/explain/result_columns.go
index 1912c16c740a..92e21e06f1d1 100644
--- a/pkg/sql/opt/exec/explain/result_columns.go
+++ b/pkg/sql/opt/exec/explain/result_columns.go
@@ -81,7 +81,12 @@ func getResultColumns(
 
 	case lookupJoinOp:
 		a := args.(*lookupJoinArgs)
-		return joinColumns(a.JoinType, inputs[0], tableColumns(a.Table, a.LookupCols)), nil
+		cols := joinColumns(a.JoinType, inputs[0], tableColumns(a.Table, a.LookupCols))
+		// The following matches the behavior of execFactory.ConstructLookupJoin.
+		if a.IsFirstJoinInPairedJoiner {
+			cols = append(cols, colinfo.ResultColumn{Name: "cont", Typ: types.Bool})
+		}
+		return cols, nil
 
 	case ordinalityOp:
 		return appendColumns(inputs[0], colinfo.ResultColumn{
diff --git a/pkg/sql/opt/exec/factory.opt b/pkg/sql/opt/exec/factory.opt
index e73b319d3b12..e7be6b976760 100644
--- a/pkg/sql/opt/exec/factory.opt
+++ b/pkg/sql/opt/exec/factory.opt
@@ -218,6 +218,7 @@ define LookupJoin {
     EqColsAreKey bool
     LookupCols exec.TableColumnOrdinalSet
     OnCond tree.TypedExpr
+    IsFirstJoinInPairedJoiner bool
     IsSecondJoinInPairedJoiner bool
     ReqOrdering exec.OutputOrdering
     Locking *tree.LockingItem
diff --git a/pkg/sql/opt/memo/check_expr.go b/pkg/sql/opt/memo/check_expr.go
index c45f26a2c1c4..20d1cce1f434 100644
--- a/pkg/sql/opt/memo/check_expr.go
+++ b/pkg/sql/opt/memo/check_expr.go
@@ -185,14 +185,19 @@ func (m *Memo) CheckExpr(e opt.Expr) {
 			panic(errors.AssertionFailedf("lookup join with columns that are not required"))
 		}
 		if t.IsSecondJoinInPairedJoiner {
-			ij, ok := t.Input.(*InvertedJoinExpr)
-			if !ok {
-				panic(errors.AssertionFailedf(
-					"lookup paired-join is paired with %T instead of inverted join", t.Input))
-			}
-			if !ij.IsFirstJoinInPairedJoiner {
-				panic(errors.AssertionFailedf(
-					"lookup paired-join is paired with inverted join that thinks it is unpaired"))
+			switch firstJoin := t.Input.(type) {
+			case *InvertedJoinExpr:
+				if !firstJoin.IsFirstJoinInPairedJoiner {
+					panic(errors.AssertionFailedf(
+						"lookup paired-join is paired with inverted join that thinks it is unpaired"))
+				}
+			case *LookupJoinExpr:
+				if !firstJoin.IsFirstJoinInPairedJoiner {
+					panic(errors.AssertionFailedf(
+						"lookup paired-join is paired with lookup join that thinks it is unpaired"))
+				}
+			default:
+				panic(errors.AssertionFailedf("lookup paired-join is paired with %T", t.Input))
 			}
 		}
 
diff --git a/pkg/sql/opt/ops/relational.opt b/pkg/sql/opt/ops/relational.opt
index 135ede8a09f6..a7cd0bd498d9 100644
--- a/pkg/sql/opt/ops/relational.opt
+++ b/pkg/sql/opt/ops/relational.opt
@@ -336,10 +336,20 @@ define LookupJoinPrivate {
     # table (and thus each left row matches with at most one table row).
     LookupColsAreTableKey bool
 
+    # At most one of Is{First,Second}JoinInPairedJoiner can be true.
+    #
+    # IsFirstJoinInPairedJoiner is true if this is the first join of a
+    # paired-joiner used for left joins.
+    IsFirstJoinInPairedJoiner bool
+
     # IsSecondJoinInPairedJoiner is true if this is the second join of a
     # paired-joiner used for left joins.
     IsSecondJoinInPairedJoiner bool
 
+    # ContinuationCol is the column ID of the continuation column when
+    # IsFirstJoinInPairedJoiner is true.
+    ContinuationCol ColumnID
+
     # ConstFilters contains the constant filters that are represented as equality
     # conditions on the KeyCols. These filters are needed by the statistics code to
     # correctly estimate selectivity.
diff --git a/pkg/sql/opt/xform/join_funcs.go b/pkg/sql/opt/xform/join_funcs.go
index ea9b1cee016a..43e74170dd88 100644
--- a/pkg/sql/opt/xform/join_funcs.go
+++ b/pkg/sql/opt/xform/join_funcs.go
@@ -135,10 +135,12 @@ func (c *CustomFuncs) GenerateMergeJoins(
 //      Input  Scan(t)                   Input
 //
 //
-//  2. The index is not covering. We have to generate an index join above the
-//     lookup join. Note that this index join is also implemented as a
-//     LookupJoin, because an IndexJoin can only output columns from one table,
-//     whereas we also need to output columns from Input.
+//  2. The index is not covering, but we can fully evaluate the ON condition
+//     using the index, or we are doing an InnerJoin. We have to generate
+//     an index join above the lookup join. Note that this index join is also
+//     implemented as a LookupJoin, because an IndexJoin can only output
+//     columns from one table, whereas we also need to output columns from
+//     Input.
 //
 //         Join                       LookupJoin(t@primary)
 //         /   \                           |
@@ -155,13 +157,30 @@ func (c *CustomFuncs) GenerateMergeJoins(
 //
 //     We want to first join abc with the index on y (which provides columns y, x)
 //     and then use a lookup join to retrieve column z. The "index join" (top
-//     LookupJoin) will produce columns a,b,c,x,y; the lookup columns are just z
-//     (the original index join produced x,y,z).
+//     LookupJoin) will produce columns a,b,c,x,y,z; the lookup columns are just x
+//     (the original index join produced a,b,c,x,y).
 //
 //     Note that the top LookupJoin "sees" column IDs from the table on both
 //     "sides" (in this example x,y on the left and z on the right) but there is
 //     no overlap.
 //
+//  3. The index is not covering and we cannot fully evaluate the ON condition
+//     using the index, and we are doing a LeftJoin/SemiJoin/AntiJoin. This is
+//     handled using a lower-upper pair of joins that are further specialized
+//     as paired-joins. The first (lower) join outputs a continuation column
+//     that is used by the second (upper) join. Like case 2, both are lookup
+//     joins, but paired-joins explicitly know their role in the pair and
+//     behave accordingly.
+//     For example, using the same tables in the example for case 2:
+//      SELECT * FROM abc JOIN xyz ON a=y AND b=z
+//
+//     The first join will evaluate a=y and produce columns a,b,c,x,y,cont
+//     where cont is the continuation column used to group together rows that
+//     correspond to the same original a,b,c. The second join will evaluate
+//     b=z and produce columns a,b,c,x,y,z. A similar approach works for
+//     anti-joins and semi-joins.
+//
+//
 // A lookup join can be created when the ON condition or implicit filters from
 // CHECK constraints and computed columns constrain a prefix of the index
 // columns to non-ranging constant values. To support this, the constant values
@@ -356,21 +375,27 @@ func (c *CustomFuncs) GenerateLookupJoins(
 			}
 		}
 
-		// All code that follows is for case 2 (see function comment).
+		// All code that follows is for cases 2 and 3 (see function comment).
+		// We need to generate two joins: a lower join followed by an upper join.
+		// In case 3, this lower-upper pair of joins is further specialized into
+		// paired-joins where we refer to the lower as first and upper as second.
 
 		if scanPrivate.Flags.NoIndexJoin {
 			return
 		}
-		if joinType == opt.SemiJoinOp || joinType == opt.AntiJoinOp {
-			// We cannot use a non-covering index for semi and anti join. Note that
-			// since the semi/anti join doesn't pass through any columns, "non
-			// covering" here means that not all columns in the ON condition are
-			// available.
-			//
-			// TODO(radu): We could create a semi/anti join on top of an inner join if
-			// the lookup columns form a key (to guarantee that input rows are not
-			// duplicated by the inner join).
-			return
+		pairedJoins := false
+		continuationCol := opt.ColumnID(0)
+		lowerJoinType := joinType
+		if joinType == opt.SemiJoinOp {
+			// Case 3: Semi joins are converted to a pair consisting of an inner
+			// lookup join and semi lookup join.
+			pairedJoins = true
+			lowerJoinType = opt.InnerJoinOp
+		} else if joinType == opt.AntiJoinOp {
+			// Case 3: Anti joins are converted to a pair consisting of a left
+			// lookup join and anti lookup join.
+			pairedJoins = true
+			lowerJoinType = opt.LeftJoinOp
 		}
 
 		if pkCols == nil {
@@ -395,6 +420,7 @@ func (c *CustomFuncs) GenerateLookupJoins(
 		// can refer to: input columns, or columns available in the index.
 		onCols := indexCols.Union(inputProps.OutputCols)
 		if c.FiltersBoundBy(lookupJoin.On, onCols) {
+			// Case 2.
 			// The ON condition refers only to the columns available in the index.
 			//
 			// For LeftJoin, both LookupJoins perform a LeftJoin. A null-extended row
@@ -408,21 +434,33 @@ func (c *CustomFuncs) GenerateLookupJoins(
 			// conditions that refer to other columns. We can put the former in the
 			// lower LookupJoin and the latter in the index join.
 			//
-			// This works for InnerJoin but not for LeftJoin because of a
-			// technicality: if an input (left) row has matches in the lower
-			// LookupJoin but has no matches in the index join, only the columns
-			// looked up by the top index join get NULL-extended.
+			// This works in a straightforward manner for InnerJoin but not for
+			// LeftJoin because of a technicality: if an input (left) row has
+			// matches in the lower LookupJoin but has no matches in the index join,
+			// only the columns looked up by the top index join get NULL-extended.
+			// Additionally if none of the lower matches are matches in the index
+			// join, we want to output only one NULL-extended row. To accomplish
+			// this, we need to use paired-joins.
 			if joinType == opt.LeftJoinOp {
-				// TODO(radu): support LeftJoin, perhaps by looking up all columns and
-				// discarding columns that are already available from the lower
-				// LookupJoin. This requires a projection to avoid having the same
-				// ColumnIDs on both sides of the index join.
-				return
+				// Case 3.
+				pairedJoins = true
+				// The lowerJoinType continues to be LeftJoinOp.
 			}
+			// We have already set pairedJoins=true for SemiJoin,AntiJoin earlier,
+			// and we don't need to do that for InnerJoin. The following sets up the
+			// ON conditions for both Case 2 and Case 3, when doing 2 joins that
+			// will each evaluate part of the ON condition.
 			conditions := lookupJoin.On
 			lookupJoin.On = c.ExtractBoundConditions(conditions, onCols)
 			indexJoin.On = c.ExtractUnboundConditions(conditions, onCols)
 		}
+		if pairedJoins {
+			lookupJoin.JoinType = lowerJoinType
+			continuationCol = c.constructContinuationColumnForPairedJoin()
+			lookupJoin.IsFirstJoinInPairedJoiner = true
+			lookupJoin.ContinuationCol = continuationCol
+			lookupJoin.Cols.Add(continuationCol)
+		}
 
 		indexJoin.Input = c.e.f.ConstructLookupJoin(
 			lookupJoin.Input,
@@ -435,12 +473,55 @@ func (c *CustomFuncs) GenerateLookupJoins(
 		indexJoin.KeyCols = pkCols
 		indexJoin.Cols = scanPrivate.Cols.Union(inputProps.OutputCols)
 		indexJoin.LookupColsAreTableKey = true
+		if pairedJoins {
+			indexJoin.IsSecondJoinInPairedJoiner = true
+		}
 
-		// Create the LookupJoin for the index join in the same group.
-		c.e.mem.AddLookupJoinToGroup(&indexJoin, grp)
+		c.addIndexJoinAsSecondJoinToMemo(grp, inputProps.OutputCols, &indexJoin)
 	})
 }
 
+// addIndexJoinAsSecondJoin is a helper function used when constructing two
+// joins to accomplish a join in the query. The second join is over the
+// primary index, and is added here.
+func (c *CustomFuncs) addIndexJoinAsSecondJoinToMemo(
+	grp memo.RelExpr, inputColsOfFirstJoin opt.ColSet, indexJoin *memo.LookupJoinExpr,
+) {
+	joinType := indexJoin.JoinType
+
+	// If this is not a semi- or anti-join, create the LookupJoin for the index
+	// join in the same group.
+	if joinType != opt.SemiJoinOp && joinType != opt.AntiJoinOp {
+		c.e.mem.AddLookupJoinToGroup(indexJoin, grp)
+		return
+	}
+
+	// Semi and anti joins here are always using paired-joins. Some of these
+	// require a project on top (see below). Avoid adding that projection if it
+	// will be a no-op (i.e., we already have the correct output columns from
+	// the lookup join).
+	outputCols := indexJoin.Cols.Intersection(indexJoin.Input.Relational().OutputCols)
+	if outputCols.SubsetOf(inputColsOfFirstJoin) {
+		c.e.mem.AddLookupJoinToGroup(indexJoin, grp)
+		return
+	}
+
+	// For some semi and anti joins, we need to add a project on top to ensure
+	// that only the original left-side columns are output. Normally, the
+	// LookupJoin would be able to perform the necessary projection for semi
+	// and anti joins by intersecting Cols with the OutputCols of its input,
+	// but that doesn't work for paired joins since the input to the second
+	// join may include more columns than the original input.
+	var project memo.ProjectExpr
+	project.Input = c.e.f.ConstructLookupJoin(
+		indexJoin.Input,
+		indexJoin.On,
+		&indexJoin.LookupJoinPrivate,
+	)
+	project.Passthrough = grp.Relational().OutputCols
+	c.e.mem.AddProjectToGroup(&project, grp)
+}
+
 // constructContinuationColumnForPairedJoin constructs a continuation column
 // ID for the paired-joiners used for left outer/semi/anti joins when the
 // first join generates false positives (due to an inverted index or
@@ -626,36 +707,7 @@ func (c *CustomFuncs) GenerateInvertedJoins(
 			indexJoin.IsSecondJoinInPairedJoiner = true
 		}
 
-		// If this is not a semi- or anti-join, create the LookupJoin for the index
-		// join in the same group.
-		if joinType != opt.SemiJoinOp && joinType != opt.AntiJoinOp {
-			c.e.mem.AddLookupJoinToGroup(&indexJoin, grp)
-			return
-		}
-
-		// Some semi and anti joins require a project on top (see below). Avoid
-		// adding that projection if it will be a no-op (i.e., we already have
-		// the correct output columns from the lookup join).
-		outputCols := indexJoin.Cols.Intersection(indexJoin.Input.Relational().OutputCols)
-		if outputCols.SubsetOf(inputCols) {
-			c.e.mem.AddLookupJoinToGroup(&indexJoin, grp)
-			return
-		}
-
-		// For some semi and anti joins, we need to add a project on top to ensure
-		// that only the original left-side columns are output. Normally, the
-		// LookupJoin would be able to perform the necessary projection for semi
-		// and anti joins by intersecting Cols with the OutputCols of its input,
-		// but that doesn't work for paired joins since the input to the second
-		// join may include more columns than the original input.
-		var project memo.ProjectExpr
-		project.Input = c.e.f.ConstructLookupJoin(
-			indexJoin.Input,
-			indexJoin.On,
-			&indexJoin.LookupJoinPrivate,
-		)
-		project.Passthrough = grp.Relational().OutputCols
-		c.e.mem.AddProjectToGroup(&project, grp)
+		c.addIndexJoinAsSecondJoinToMemo(grp, inputCols, &indexJoin)
 	})
 }
 
diff --git a/pkg/sql/opt/xform/testdata/external/liquibase b/pkg/sql/opt/xform/testdata/external/liquibase
index 039fbebde068..131514decf55 100644
--- a/pkg/sql/opt/xform/testdata/external/liquibase
+++ b/pkg/sql/opt/xform/testdata/external/liquibase
@@ -204,66 +204,64 @@ project
  │    │    │         │    │    ├── lookup columns are key
  │    │    │         │    │    ├── key: (1,78)
  │    │    │         │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (78)-->(79), (1,78)-->(34,35,85,98,99), (98)-->(99), (34)-->(35), (35)-->(34), (3)==(29), (29)==(3)
- │    │    │         │    │    ├── right-join (hash)
+ │    │    │         │    │    ├── left-join (lookup pg_index [as=ind])
  │    │    │         │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null t.oid:34 spcname:35 indexrelid:78 indrelid:79 indisclustered:85 ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131
+ │    │    │         │    │    │    ├── key columns: [78] = [78]
+ │    │    │         │    │    │    ├── lookup columns are key
  │    │    │         │    │    │    ├── key: (1,78)
  │    │    │         │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (78)-->(79), (1,78)-->(34,35,85), (34)-->(35), (35)-->(34), (3)==(29), (29)==(3)
- │    │    │         │    │    │    ├── select
- │    │    │         │    │    │    │    ├── columns: indexrelid:78!null indrelid:79!null indisclustered:85!null
- │    │    │         │    │    │    │    ├── key: (78)
- │    │    │         │    │    │    │    ├── fd: ()-->(85), (78)-->(79)
- │    │    │         │    │    │    │    ├── scan pg_index [as=ind]
- │    │    │         │    │    │    │    │    ├── columns: indexrelid:78!null indrelid:79!null indisclustered:85!null
- │    │    │         │    │    │    │    │    ├── key: (78)
- │    │    │         │    │    │    │    │    └── fd: (78)-->(79,85)
- │    │    │         │    │    │    │    └── filters
- │    │    │         │    │    │    │         └── indisclustered:85 = true [outer=(85), constraints=(/85: [/true - /true]; tight), fd=()-->(85)]
- │    │    │         │    │    │    ├── left-join (lookup pg_tablespace [as=t])
- │    │    │         │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null t.oid:34 spcname:35 ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131
- │    │    │         │    │    │    │    ├── key columns: [8] = [34]
- │    │    │         │    │    │    │    ├── lookup columns are key
- │    │    │         │    │    │    │    ├── key: (1)
- │    │    │         │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,34,35,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (34)-->(35), (35)-->(34), (3)==(29), (29)==(3)
- │    │    │         │    │    │    │    ├── left-join (lookup pg_foreign_server [as=fs])
- │    │    │         │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131
- │    │    │         │    │    │    │    │    ├── key columns: [127] = [130]
+ │    │    │         │    │    │    ├── left-join (lookup pg_index@pg_index_indrelid_index [as=ind])
+ │    │    │         │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null t.oid:34 spcname:35 indexrelid:78 indrelid:79 ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131 continuation:228
+ │    │    │         │    │    │    │    ├── key columns: [1] = [79]
+ │    │    │         │    │    │    │    ├── key: (1,78)
+ │    │    │         │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,34,35,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (34)-->(35), (35)-->(34), (3)==(29), (29)==(3), (78)-->(79,228)
+ │    │    │         │    │    │    │    ├── left-join (lookup pg_tablespace [as=t])
+ │    │    │         │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null t.oid:34 spcname:35 ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131
+ │    │    │         │    │    │    │    │    ├── key columns: [8] = [34]
  │    │    │         │    │    │    │    │    ├── lookup columns are key
  │    │    │         │    │    │    │    │    ├── key: (1)
- │    │    │         │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (3)==(29), (29)==(3)
- │    │    │         │    │    │    │    │    ├── left-join (lookup pg_foreign_table [as=ft])
- │    │    │         │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null ftrelid:126 ftserver:127 ftoptions:128
- │    │    │         │    │    │    │    │    │    ├── key columns: [1] = [126]
+ │    │    │         │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,34,35,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (34)-->(35), (35)-->(34), (3)==(29), (29)==(3)
+ │    │    │         │    │    │    │    │    ├── left-join (lookup pg_foreign_server [as=fs])
+ │    │    │         │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131
+ │    │    │         │    │    │    │    │    │    ├── key columns: [127] = [130]
  │    │    │         │    │    │    │    │    │    ├── lookup columns are key
  │    │    │         │    │    │    │    │    │    ├── key: (1)
- │    │    │         │    │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128), (3)==(29), (29)==(3)
- │    │    │         │    │    │    │    │    │    ├── inner-join (hash)
- │    │    │         │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null
- │    │    │         │    │    │    │    │    │    │    ├── multiplicity: left-rows(zero-or-one), right-rows(zero-or-more)
+ │    │    │         │    │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (3)==(29), (29)==(3)
+ │    │    │         │    │    │    │    │    │    ├── left-join (lookup pg_foreign_table [as=ft])
+ │    │    │         │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null ftrelid:126 ftserver:127 ftoptions:128
+ │    │    │         │    │    │    │    │    │    │    ├── key columns: [1] = [126]
+ │    │    │         │    │    │    │    │    │    │    ├── lookup columns are key
  │    │    │         │    │    │    │    │    │    │    ├── key: (1)
- │    │    │         │    │    │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (3)==(29), (29)==(3)
- │    │    │         │    │    │    │    │    │    │    ├── select
- │    │    │         │    │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27
+ │    │    │         │    │    │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128), (3)==(29), (29)==(3)
+ │    │    │         │    │    │    │    │    │    │    ├── inner-join (hash)
+ │    │    │         │    │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null
+ │    │    │         │    │    │    │    │    │    │    │    ├── multiplicity: left-rows(zero-or-one), right-rows(zero-or-more)
  │    │    │         │    │    │    │    │    │    │    │    ├── key: (1)
- │    │    │         │    │    │    │    │    │    │    │    ├── fd: (1)-->(2,3,5,8,10,13,15,17,20,22,23,26,27), (2,3)-->(1,5,8,10,13,15,17,20,22,23,26,27)
- │    │    │         │    │    │    │    │    │    │    │    ├── scan pg_class [as=c]
+ │    │    │         │    │    │    │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (3)==(29), (29)==(3)
+ │    │    │         │    │    │    │    │    │    │    │    ├── select
  │    │    │         │    │    │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27
  │    │    │         │    │    │    │    │    │    │    │    │    ├── key: (1)
- │    │    │         │    │    │    │    │    │    │    │    │    └── fd: (1)-->(2,3,5,8,10,13,15,17,20,22,23,26,27), (2,3)-->(1,5,8,10,13,15,17,20,22,23,26,27)
+ │    │    │         │    │    │    │    │    │    │    │    │    ├── fd: (1)-->(2,3,5,8,10,13,15,17,20,22,23,26,27), (2,3)-->(1,5,8,10,13,15,17,20,22,23,26,27)
+ │    │    │         │    │    │    │    │    │    │    │    │    ├── scan pg_class [as=c]
+ │    │    │         │    │    │    │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27
+ │    │    │         │    │    │    │    │    │    │    │    │    │    ├── key: (1)
+ │    │    │         │    │    │    │    │    │    │    │    │    │    └── fd: (1)-->(2,3,5,8,10,13,15,17,20,22,23,26,27), (2,3)-->(1,5,8,10,13,15,17,20,22,23,26,27)
+ │    │    │         │    │    │    │    │    │    │    │    │    └── filters
+ │    │    │         │    │    │    │    │    │    │    │    │         └── (c.relkind:17 = 'r') OR (c.relkind:17 = 'f') [outer=(17), constraints=(/17: [/'f' - /'f'] [/'r' - /'r']; tight)]
+ │    │    │         │    │    │    │    │    │    │    │    ├── scan pg_namespace@pg_namespace_nspname_index [as=n]
+ │    │    │         │    │    │    │    │    │    │    │    │    ├── columns: n.oid:29!null n.nspname:30!null
+ │    │    │         │    │    │    │    │    │    │    │    │    ├── constraint: /30: [/'public' - /'public']
+ │    │    │         │    │    │    │    │    │    │    │    │    ├── cardinality: [0 - 1]
+ │    │    │         │    │    │    │    │    │    │    │    │    ├── key: ()
+ │    │    │         │    │    │    │    │    │    │    │    │    └── fd: ()-->(29,30)
  │    │    │         │    │    │    │    │    │    │    │    └── filters
- │    │    │         │    │    │    │    │    │    │    │         └── (c.relkind:17 = 'r') OR (c.relkind:17 = 'f') [outer=(17), constraints=(/17: [/'f' - /'f'] [/'r' - /'r']; tight)]
- │    │    │         │    │    │    │    │    │    │    ├── scan pg_namespace@pg_namespace_nspname_index [as=n]
- │    │    │         │    │    │    │    │    │    │    │    ├── columns: n.oid:29!null n.nspname:30!null
- │    │    │         │    │    │    │    │    │    │    │    ├── constraint: /30: [/'public' - /'public']
- │    │    │         │    │    │    │    │    │    │    │    ├── cardinality: [0 - 1]
- │    │    │         │    │    │    │    │    │    │    │    ├── key: ()
- │    │    │         │    │    │    │    │    │    │    │    └── fd: ()-->(29,30)
- │    │    │         │    │    │    │    │    │    │    └── filters
- │    │    │         │    │    │    │    │    │    │         └── n.oid:29 = c.relnamespace:3 [outer=(3,29), constraints=(/3: (/NULL - ]; /29: (/NULL - ]), fd=(3)==(29), (29)==(3)]
+ │    │    │         │    │    │    │    │    │    │    │         └── n.oid:29 = c.relnamespace:3 [outer=(3,29), constraints=(/3: (/NULL - ]; /29: (/NULL - ]), fd=(3)==(29), (29)==(3)]
+ │    │    │         │    │    │    │    │    │    │    └── filters (true)
  │    │    │         │    │    │    │    │    │    └── filters (true)
  │    │    │         │    │    │    │    │    └── filters (true)
  │    │    │         │    │    │    │    └── filters (true)
  │    │    │         │    │    │    └── filters
- │    │    │         │    │    │         └── indrelid:79 = c.oid:1 [outer=(1,79), constraints=(/1: (/NULL - ]; /79: (/NULL - ]), fd=(1)==(79), (79)==(1)]
+ │    │    │         │    │    │         └── indisclustered:85 = true [outer=(85), constraints=(/85: [/true - /true]; tight), fd=()-->(85)]
  │    │    │         │    │    └── filters (true)
  │    │    │         │    └── filters
  │    │    │         │         └── i.inhrelid:41 = c.oid:1 [outer=(1,41), constraints=(/1: (/NULL - ]; /41: (/NULL - ]), fd=(1)==(41), (41)==(1)]
diff --git a/pkg/sql/opt/xform/testdata/external/navicat b/pkg/sql/opt/xform/testdata/external/navicat
index c3a6bb8176a9..bccd0f688baa 100644
--- a/pkg/sql/opt/xform/testdata/external/navicat
+++ b/pkg/sql/opt/xform/testdata/external/navicat
@@ -208,66 +208,64 @@ sort
       │    │    │         │    │    ├── lookup columns are key
       │    │    │         │    │    ├── key: (1,78)
       │    │    │         │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (78)-->(79), (1,78)-->(34,35,85,98,99), (98)-->(99), (34)-->(35), (35)-->(34), (3)==(29), (29)==(3)
-      │    │    │         │    │    ├── right-join (hash)
+      │    │    │         │    │    ├── left-join (lookup pg_index [as=ind])
       │    │    │         │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null t.oid:34 spcname:35 indexrelid:78 indrelid:79 indisclustered:85 ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131
+      │    │    │         │    │    │    ├── key columns: [78] = [78]
+      │    │    │         │    │    │    ├── lookup columns are key
       │    │    │         │    │    │    ├── key: (1,78)
       │    │    │         │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (78)-->(79), (1,78)-->(34,35,85), (34)-->(35), (35)-->(34), (3)==(29), (29)==(3)
-      │    │    │         │    │    │    ├── select
-      │    │    │         │    │    │    │    ├── columns: indexrelid:78!null indrelid:79!null indisclustered:85!null
-      │    │    │         │    │    │    │    ├── key: (78)
-      │    │    │         │    │    │    │    ├── fd: ()-->(85), (78)-->(79)
-      │    │    │         │    │    │    │    ├── scan pg_index [as=ind]
-      │    │    │         │    │    │    │    │    ├── columns: indexrelid:78!null indrelid:79!null indisclustered:85!null
-      │    │    │         │    │    │    │    │    ├── key: (78)
-      │    │    │         │    │    │    │    │    └── fd: (78)-->(79,85)
-      │    │    │         │    │    │    │    └── filters
-      │    │    │         │    │    │    │         └── indisclustered:85 = true [outer=(85), constraints=(/85: [/true - /true]; tight), fd=()-->(85)]
-      │    │    │         │    │    │    ├── left-join (lookup pg_tablespace [as=t])
-      │    │    │         │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null t.oid:34 spcname:35 ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131
-      │    │    │         │    │    │    │    ├── key columns: [8] = [34]
-      │    │    │         │    │    │    │    ├── lookup columns are key
-      │    │    │         │    │    │    │    ├── key: (1)
-      │    │    │         │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,34,35,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (34)-->(35), (35)-->(34), (3)==(29), (29)==(3)
-      │    │    │         │    │    │    │    ├── left-join (lookup pg_foreign_server [as=fs])
-      │    │    │         │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131
-      │    │    │         │    │    │    │    │    ├── key columns: [127] = [130]
+      │    │    │         │    │    │    ├── left-join (lookup pg_index@pg_index_indrelid_index [as=ind])
+      │    │    │         │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null t.oid:34 spcname:35 indexrelid:78 indrelid:79 ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131 continuation:228
+      │    │    │         │    │    │    │    ├── key columns: [1] = [79]
+      │    │    │         │    │    │    │    ├── key: (1,78)
+      │    │    │         │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,34,35,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (34)-->(35), (35)-->(34), (3)==(29), (29)==(3), (78)-->(79,228)
+      │    │    │         │    │    │    │    ├── left-join (lookup pg_tablespace [as=t])
+      │    │    │         │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null t.oid:34 spcname:35 ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131
+      │    │    │         │    │    │    │    │    ├── key columns: [8] = [34]
       │    │    │         │    │    │    │    │    ├── lookup columns are key
       │    │    │         │    │    │    │    │    ├── key: (1)
-      │    │    │         │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (3)==(29), (29)==(3)
-      │    │    │         │    │    │    │    │    ├── left-join (lookup pg_foreign_table [as=ft])
-      │    │    │         │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null ftrelid:126 ftserver:127 ftoptions:128
-      │    │    │         │    │    │    │    │    │    ├── key columns: [1] = [126]
+      │    │    │         │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,34,35,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (34)-->(35), (35)-->(34), (3)==(29), (29)==(3)
+      │    │    │         │    │    │    │    │    ├── left-join (lookup pg_foreign_server [as=fs])
+      │    │    │         │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null ftrelid:126 ftserver:127 ftoptions:128 fs.oid:130 srvname:131
+      │    │    │         │    │    │    │    │    │    ├── key columns: [127] = [130]
       │    │    │         │    │    │    │    │    │    ├── lookup columns are key
       │    │    │         │    │    │    │    │    │    ├── key: (1)
-      │    │    │         │    │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128), (3)==(29), (29)==(3)
-      │    │    │         │    │    │    │    │    │    ├── inner-join (hash)
-      │    │    │         │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null
-      │    │    │         │    │    │    │    │    │    │    ├── multiplicity: left-rows(zero-or-one), right-rows(zero-or-more)
+      │    │    │         │    │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128,130,131), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128,130,131), (130)~~>(131), (131)~~>(130), (3)==(29), (29)==(3)
+      │    │    │         │    │    │    │    │    │    ├── left-join (lookup pg_foreign_table [as=ft])
+      │    │    │         │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null ftrelid:126 ftserver:127 ftoptions:128
+      │    │    │         │    │    │    │    │    │    │    ├── key columns: [1] = [126]
+      │    │    │         │    │    │    │    │    │    │    ├── lookup columns are key
       │    │    │         │    │    │    │    │    │    │    ├── key: (1)
-      │    │    │         │    │    │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (3)==(29), (29)==(3)
-      │    │    │         │    │    │    │    │    │    │    ├── select
-      │    │    │         │    │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27
+      │    │    │         │    │    │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27,126-128), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (126)-->(127,128), (3)==(29), (29)==(3)
+      │    │    │         │    │    │    │    │    │    │    ├── inner-join (hash)
+      │    │    │         │    │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27 n.oid:29!null n.nspname:30!null
+      │    │    │         │    │    │    │    │    │    │    │    ├── multiplicity: left-rows(zero-or-one), right-rows(zero-or-more)
       │    │    │         │    │    │    │    │    │    │    │    ├── key: (1)
-      │    │    │         │    │    │    │    │    │    │    │    ├── fd: (1)-->(2,3,5,8,10,13,15,17,20,22,23,26,27), (2,3)-->(1,5,8,10,13,15,17,20,22,23,26,27)
-      │    │    │         │    │    │    │    │    │    │    │    ├── scan pg_class [as=c]
+      │    │    │         │    │    │    │    │    │    │    │    ├── fd: ()-->(3,29,30), (1)-->(2,5,8,10,13,15,17,20,22,23,26,27), (2)-->(1,5,8,10,13,15,17,20,22,23,26,27), (3)==(29), (29)==(3)
+      │    │    │         │    │    │    │    │    │    │    │    ├── select
       │    │    │         │    │    │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27
       │    │    │         │    │    │    │    │    │    │    │    │    ├── key: (1)
-      │    │    │         │    │    │    │    │    │    │    │    │    └── fd: (1)-->(2,3,5,8,10,13,15,17,20,22,23,26,27), (2,3)-->(1,5,8,10,13,15,17,20,22,23,26,27)
+      │    │    │         │    │    │    │    │    │    │    │    │    ├── fd: (1)-->(2,3,5,8,10,13,15,17,20,22,23,26,27), (2,3)-->(1,5,8,10,13,15,17,20,22,23,26,27)
+      │    │    │         │    │    │    │    │    │    │    │    │    ├── scan pg_class [as=c]
+      │    │    │         │    │    │    │    │    │    │    │    │    │    ├── columns: c.oid:1!null c.relname:2!null c.relnamespace:3!null c.relowner:5!null c.reltablespace:8!null c.reltuples:10!null c.relhasindex:13!null c.relpersistence:15!null c.relkind:17!null c.relhasoids:20!null c.relhasrules:22!null c.relhastriggers:23!null c.relacl:26 c.reloptions:27
+      │    │    │         │    │    │    │    │    │    │    │    │    │    ├── key: (1)
+      │    │    │         │    │    │    │    │    │    │    │    │    │    └── fd: (1)-->(2,3,5,8,10,13,15,17,20,22,23,26,27), (2,3)-->(1,5,8,10,13,15,17,20,22,23,26,27)
+      │    │    │         │    │    │    │    │    │    │    │    │    └── filters
+      │    │    │         │    │    │    │    │    │    │    │    │         └── (c.relkind:17 = 'r') OR (c.relkind:17 = 'f') [outer=(17), constraints=(/17: [/'f' - /'f'] [/'r' - /'r']; tight)]
+      │    │    │         │    │    │    │    │    │    │    │    ├── scan pg_namespace@pg_namespace_nspname_index [as=n]
+      │    │    │         │    │    │    │    │    │    │    │    │    ├── columns: n.oid:29!null n.nspname:30!null
+      │    │    │         │    │    │    │    │    │    │    │    │    ├── constraint: /30: [/'public' - /'public']
+      │    │    │         │    │    │    │    │    │    │    │    │    ├── cardinality: [0 - 1]
+      │    │    │         │    │    │    │    │    │    │    │    │    ├── key: ()
+      │    │    │         │    │    │    │    │    │    │    │    │    └── fd: ()-->(29,30)
       │    │    │         │    │    │    │    │    │    │    │    └── filters
-      │    │    │         │    │    │    │    │    │    │    │         └── (c.relkind:17 = 'r') OR (c.relkind:17 = 'f') [outer=(17), constraints=(/17: [/'f' - /'f'] [/'r' - /'r']; tight)]
-      │    │    │         │    │    │    │    │    │    │    ├── scan pg_namespace@pg_namespace_nspname_index [as=n]
-      │    │    │         │    │    │    │    │    │    │    │    ├── columns: n.oid:29!null n.nspname:30!null
-      │    │    │         │    │    │    │    │    │    │    │    ├── constraint: /30: [/'public' - /'public']
-      │    │    │         │    │    │    │    │    │    │    │    ├── cardinality: [0 - 1]
-      │    │    │         │    │    │    │    │    │    │    │    ├── key: ()
-      │    │    │         │    │    │    │    │    │    │    │    └── fd: ()-->(29,30)
-      │    │    │         │    │    │    │    │    │    │    └── filters
-      │    │    │         │    │    │    │    │    │    │         └── n.oid:29 = c.relnamespace:3 [outer=(3,29), constraints=(/3: (/NULL - ]; /29: (/NULL - ]), fd=(3)==(29), (29)==(3)]
+      │    │    │         │    │    │    │    │    │    │    │         └── n.oid:29 = c.relnamespace:3 [outer=(3,29), constraints=(/3: (/NULL - ]; /29: (/NULL - ]), fd=(3)==(29), (29)==(3)]
+      │    │    │         │    │    │    │    │    │    │    └── filters (true)
       │    │    │         │    │    │    │    │    │    └── filters (true)
       │    │    │         │    │    │    │    │    └── filters (true)
       │    │    │         │    │    │    │    └── filters (true)
       │    │    │         │    │    │    └── filters
-      │    │    │         │    │    │         └── indrelid:79 = c.oid:1 [outer=(1,79), constraints=(/1: (/NULL - ]; /79: (/NULL - ]), fd=(1)==(79), (79)==(1)]
+      │    │    │         │    │    │         └── indisclustered:85 = true [outer=(85), constraints=(/85: [/true - /true]; tight), fd=()-->(85)]
       │    │    │         │    │    └── filters (true)
       │    │    │         │    └── filters
       │    │    │         │         └── i.inhrelid:41 = c.oid:1 [outer=(1,41), constraints=(/1: (/NULL - ]; /41: (/NULL - ]), fd=(1)==(41), (41)==(1)]
diff --git a/pkg/sql/opt/xform/testdata/rules/join b/pkg/sql/opt/xform/testdata/rules/join
index 99c4ca94acc4..a2385bcff773 100644
--- a/pkg/sql/opt/xform/testdata/rules/join
+++ b/pkg/sql/opt/xform/testdata/rules/join
@@ -2470,19 +2470,22 @@ inner-join (lookup abcd)
       └── c:7 > n:2 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ])]
 
 # Non-covering case, extra filter not bound by index, left join.
-# In this case, we can't yet convert to a lookup join (see
-# the GenerateLookupJoins custom func).
-opt expect-not=GenerateLookupJoins
+# In this case, we can generate lookup joins as paired-joins.
+opt expect=GenerateLookupJoins
 SELECT * FROM small LEFT JOIN abcd ON a=m AND c>n
 ----
-right-join (hash)
+left-join (lookup abcd)
  ├── columns: m:1 n:2 a:5 b:6 c:7
- ├── scan abcd
- │    └── columns: a:5 b:6 c:7
- ├── scan small
- │    └── columns: m:1 n:2
+ ├── key columns: [8] = [8]
+ ├── lookup columns are key
+ ├── left-join (lookup abcd@secondary)
+ │    ├── columns: m:1 n:2 a:5 b:6 abcd.rowid:8 continuation:10
+ │    ├── key columns: [1] = [5]
+ │    ├── fd: (8)-->(5,6,10)
+ │    ├── scan small
+ │    │    └── columns: m:1 n:2
+ │    └── filters (true)
  └── filters
-      ├── a:5 = m:1 [outer=(1,5), constraints=(/1: (/NULL - ]; /5: (/NULL - ]), fd=(1)==(5), (5)==(1)]
       └── c:7 > n:2 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ])]
 
 
@@ -2729,20 +2732,27 @@ semi-join (lookup abcd@secondary)
  │    └── columns: m:1 n:2
  └── filters (true)
 
-# We should not generate a lookup semi-join when the index doesn't contain all
-# columns in the join condition.
-opt expect-not=GenerateLookupJoins
+# We can generate a lookup semi-join when the index doesn't contain all
+# columns in the join condition, using paired-joins.
+opt expect=GenerateLookupJoins
 SELECT m, n FROM small WHERE EXISTS(SELECT 1 FROM abcd WHERE m = a AND n = c)
 ----
-semi-join (hash)
+project
  ├── columns: m:1 n:2
- ├── scan small
- │    └── columns: m:1 n:2
- ├── scan abcd
- │    └── columns: a:5 c:7
- └── filters
-      ├── m:1 = a:5 [outer=(1,5), constraints=(/1: (/NULL - ]; /5: (/NULL - ]), fd=(1)==(5), (5)==(1)]
-      └── n:2 = c:7 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ]), fd=(2)==(7), (7)==(2)]
+ └── semi-join (lookup abcd)
+      ├── columns: m:1!null n:2 a:5!null
+      ├── key columns: [8] = [8]
+      ├── lookup columns are key
+      ├── fd: (1)==(5), (5)==(1)
+      ├── inner-join (lookup abcd@secondary)
+      │    ├── columns: m:1!null n:2 a:5!null abcd.rowid:8!null continuation:11
+      │    ├── key columns: [1] = [5]
+      │    ├── fd: (8)-->(5,11), (1)==(5), (5)==(1)
+      │    ├── scan small
+      │    │    └── columns: m:1 n:2
+      │    └── filters (true)
+      └── filters
+           └── n:2 = c:7 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ]), fd=(2)==(7), (7)==(2)]
 
 # Lookup anti-join with index that contains all columns in the join condition.
 opt expect=GenerateLookupJoins
@@ -2755,20 +2765,27 @@ anti-join (lookup abcd@secondary)
  │    └── columns: m:1 n:2
  └── filters (true)
 
-# We should not generate a lookup anti-join when the index doesn't contain all
-# columns in the join condition.
-opt expect-not=GenerateLookupJoins
+# We can generate a lookup anti-join when the index doesn't contain all
+# columns in the join condition, using paired-joins.
+opt expect=GenerateLookupJoins
 SELECT m, n FROM small WHERE NOT EXISTS(SELECT 1 FROM abcd WHERE m = a AND n = c)
 ----
-anti-join (hash)
+project
  ├── columns: m:1 n:2
- ├── scan small
- │    └── columns: m:1 n:2
- ├── scan abcd
- │    └── columns: a:5 c:7
- └── filters
-      ├── m:1 = a:5 [outer=(1,5), constraints=(/1: (/NULL - ]; /5: (/NULL - ]), fd=(1)==(5), (5)==(1)]
-      └── n:2 = c:7 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ]), fd=(2)==(7), (7)==(2)]
+ └── anti-join (lookup abcd)
+      ├── columns: m:1 n:2 a:5
+      ├── key columns: [8] = [8]
+      ├── lookup columns are key
+      ├── fd: (8)-->(5,11)
+      ├── left-join (lookup abcd@secondary)
+      │    ├── columns: m:1 n:2 a:5 abcd.rowid:8 continuation:11
+      │    ├── key columns: [1] = [5]
+      │    ├── fd: (8)-->(5,11)
+      │    ├── scan small
+      │    │    └── columns: m:1 n:2
+      │    └── filters (true)
+      └── filters
+           └── n:2 = c:7 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ]), fd=(2)==(7), (7)==(2)]
 
 # --------------------------------------------------
 # GenerateLookupJoinsWithFilter
@@ -2900,23 +2917,23 @@ inner-join (lookup abcd)
       └── c:7 > n:2 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ])]
 
 # Non-covering case, extra filter not bound by index, left join.
-# In this case, we can't yet convert to a lookup join (see
-# the GenerateLookupJoins custom func).
-opt expect-not=GenerateLookupJoinsWithFilter
+# In this case, we can generate lookup joins as paired-joins.
+opt expect=GenerateLookupJoinsWithFilter
 SELECT * FROM small LEFT JOIN abcd ON a=m AND c>n AND b>1
 ----
-right-join (hash)
+left-join (lookup abcd)
  ├── columns: m:1 n:2 a:5 b:6 c:7
- ├── select
- │    ├── columns: a:5 b:6!null c:7
- │    ├── scan abcd
- │    │    └── columns: a:5 b:6 c:7
+ ├── key columns: [8] = [8]
+ ├── lookup columns are key
+ ├── left-join (lookup abcd@secondary)
+ │    ├── columns: m:1 n:2 a:5 b:6 abcd.rowid:8 continuation:10
+ │    ├── key columns: [1] = [5]
+ │    ├── fd: (8)-->(5,6,10)
+ │    ├── scan small
+ │    │    └── columns: m:1 n:2
  │    └── filters
  │         └── b:6 > 1 [outer=(6), constraints=(/6: [/2 - ]; tight)]
- ├── scan small
- │    └── columns: m:1 n:2
  └── filters
-      ├── a:5 = m:1 [outer=(1,5), constraints=(/1: (/NULL - ]; /5: (/NULL - ]), fd=(1)==(5), (5)==(1)]
       └── c:7 > n:2 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ])]
 
 # Constant columns are projected and used by lookup joiner.
@@ -3436,23 +3453,28 @@ semi-join (lookup abcd@secondary)
  └── filters
       └── a:5 > b:6 [outer=(5,6), constraints=(/5: (/NULL - ]; /6: (/NULL - ])]
 
-# We should not generate a lookup semi-join when the index is non-covering.
-opt expect-not=GenerateLookupJoinsWithFilter
+# We can generate a lookup semi-join when the index is non-covering using
+# paired-joins.
+opt expect=GenerateLookupJoinsWithFilter
 SELECT m, n FROM small WHERE EXISTS(SELECT 1 FROM abcd WHERE m = a AND n = c AND a > b)
 ----
-semi-join (hash)
+project
  ├── columns: m:1 n:2
- ├── scan small
- │    └── columns: m:1 n:2
- ├── select
- │    ├── columns: a:5!null b:6!null c:7
- │    ├── scan abcd
- │    │    └── columns: a:5 b:6 c:7
- │    └── filters
- │         └── a:5 > b:6 [outer=(5,6), constraints=(/5: (/NULL - ]; /6: (/NULL - ])]
- └── filters
-      ├── m:1 = a:5 [outer=(1,5), constraints=(/1: (/NULL - ]; /5: (/NULL - ]), fd=(1)==(5), (5)==(1)]
-      └── n:2 = c:7 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ]), fd=(2)==(7), (7)==(2)]
+ └── semi-join (lookup abcd)
+      ├── columns: m:1!null n:2 a:5!null b:6!null
+      ├── key columns: [8] = [8]
+      ├── lookup columns are key
+      ├── fd: (1)==(5), (5)==(1)
+      ├── inner-join (lookup abcd@secondary)
+      │    ├── columns: m:1!null n:2 a:5!null b:6!null abcd.rowid:8!null continuation:11
+      │    ├── key columns: [1] = [5]
+      │    ├── fd: (8)-->(5,6,11), (1)==(5), (5)==(1)
+      │    ├── scan small
+      │    │    └── columns: m:1 n:2
+      │    └── filters
+      │         └── a:5 > b:6 [outer=(5,6), constraints=(/5: (/NULL - ]; /6: (/NULL - ])]
+      └── filters
+           └── n:2 = c:7 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ]), fd=(2)==(7), (7)==(2)]
 
 # Lookup anti-join with covering index.
 opt expect=GenerateLookupJoinsWithFilter
@@ -3466,23 +3488,28 @@ anti-join (lookup abcd@secondary)
  └── filters
       └── a:5 > b:6 [outer=(5,6), constraints=(/5: (/NULL - ]; /6: (/NULL - ])]
 
-# We should not generate a lookup semi-join when the index is non-covering.
-opt expect-not=GenerateLookupJoinsWithFilter
+# We can generate a lookup semi-join when the index is non-covering using
+# paired-joins.
+opt expect=GenerateLookupJoinsWithFilter
 SELECT m, n FROM small WHERE NOT EXISTS(SELECT 1 FROM abcd WHERE m = a AND n = c AND a > b)
 ----
-anti-join (hash)
+project
  ├── columns: m:1 n:2
- ├── scan small
- │    └── columns: m:1 n:2
- ├── select
- │    ├── columns: a:5!null b:6!null c:7
- │    ├── scan abcd
- │    │    └── columns: a:5 b:6 c:7
- │    └── filters
- │         └── a:5 > b:6 [outer=(5,6), constraints=(/5: (/NULL - ]; /6: (/NULL - ])]
- └── filters
-      ├── m:1 = a:5 [outer=(1,5), constraints=(/1: (/NULL - ]; /5: (/NULL - ]), fd=(1)==(5), (5)==(1)]
-      └── n:2 = c:7 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ]), fd=(2)==(7), (7)==(2)]
+ └── anti-join (lookup abcd)
+      ├── columns: m:1 n:2 a:5 b:6
+      ├── key columns: [8] = [8]
+      ├── lookup columns are key
+      ├── fd: (8)-->(5,6,11)
+      ├── left-join (lookup abcd@secondary)
+      │    ├── columns: m:1 n:2 a:5 b:6 abcd.rowid:8 continuation:11
+      │    ├── key columns: [1] = [5]
+      │    ├── fd: (8)-->(5,6,11)
+      │    ├── scan small
+      │    │    └── columns: m:1 n:2
+      │    └── filters
+      │         └── a:5 > b:6 [outer=(5,6), constraints=(/5: (/NULL - ]; /6: (/NULL - ])]
+      └── filters
+           └── n:2 = c:7 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ]), fd=(2)==(7), (7)==(2)]
 
 # --------------------------------------------------
 # GenerateLookupJoinsWithFilter + Partial Indexes
@@ -3670,30 +3697,29 @@ project
       │    └── columns: m:1 n:2
       └── filters (true)
 
-# We should not generate a lookup semi-join when the index does not cover "s"
-# which is referenced in the remaining filter.
-opt expect-not=GenerateLookupJoinsWithFilter
+# We can generate a lookup semi-join when the index does not cover "s",
+# which is referenced in the remaining filter, by using paired-joins.
+opt expect=GenerateLookupJoinsWithFilter
 SELECT m FROM small WHERE EXISTS (SELECT 1 FROM partial_tab WHERE s = 'foo' AND n = i)
 ----
 project
  ├── columns: m:1
- └── semi-join (hash)
+ └── project
       ├── columns: m:1 n:2
-      ├── scan small
-      │    └── columns: m:1 n:2
-      ├── select
-      │    ├── columns: i:6 s:7!null
-      │    ├── fd: ()-->(7)
-      │    ├── index-join partial_tab
-      │    │    ├── columns: i:6 s:7
-      │    │    └── scan partial_tab@partial_idx,partial
-      │    │         ├── columns: k:5!null i:6
-      │    │         ├── key: (5)
-      │    │         └── fd: (5)-->(6)
-      │    └── filters
-      │         └── s:7 = 'foo' [outer=(7), constraints=(/7: [/'foo' - /'foo']; tight), fd=()-->(7)]
-      └── filters
-           └── n:2 = i:6 [outer=(2,6), constraints=(/2: (/NULL - ]; /6: (/NULL - ]), fd=(2)==(6), (6)==(2)]
+      └── semi-join (lookup partial_tab)
+           ├── columns: m:1 n:2!null i:6!null
+           ├── key columns: [5] = [5]
+           ├── lookup columns are key
+           ├── fd: (2)==(6), (6)==(2)
+           ├── inner-join (lookup partial_tab@partial_idx,partial)
+           │    ├── columns: m:1 n:2!null k:5!null i:6!null continuation:10
+           │    ├── key columns: [2] = [6]
+           │    ├── fd: (5)-->(6,10), (2)==(6), (6)==(2)
+           │    ├── scan small
+           │    │    └── columns: m:1 n:2
+           │    └── filters (true)
+           └── filters
+                └── s:7 = 'foo' [outer=(7), constraints=(/7: [/'foo' - /'foo']; tight), fd=()-->(7)]
 
 # Generate a lookup anti-join when the index does not cover "s", but the
 # reference to "s" no longer exists in the filters.
@@ -3709,30 +3735,29 @@ project
       │    └── columns: m:1 n:2
       └── filters (true)
 
-# We should not generate a lookup anti-join when the index does not cover "s"
-# which is referenced in the remaining filter.
-opt expect-not=GenerateLookupJoinsWithFilter
+# We can generate a lookup anti-join when the index does not cover "s",
+# which is referenced in the remaining filter, by using paired-joins.
+opt expect=GenerateLookupJoinsWithFilter
 SELECT m FROM small WHERE NOT EXISTS (SELECT 1 FROM partial_tab WHERE s = 'foo' AND n = i)
 ----
 project
  ├── columns: m:1
- └── anti-join (hash)
+ └── project
       ├── columns: m:1 n:2
-      ├── scan small
-      │    └── columns: m:1 n:2
-      ├── select
-      │    ├── columns: i:6 s:7!null
-      │    ├── fd: ()-->(7)
-      │    ├── index-join partial_tab
-      │    │    ├── columns: i:6 s:7
-      │    │    └── scan partial_tab@partial_idx,partial
-      │    │         ├── columns: k:5!null i:6
-      │    │         ├── key: (5)
-      │    │         └── fd: (5)-->(6)
-      │    └── filters
-      │         └── s:7 = 'foo' [outer=(7), constraints=(/7: [/'foo' - /'foo']; tight), fd=()-->(7)]
-      └── filters
-           └── n:2 = i:6 [outer=(2,6), constraints=(/2: (/NULL - ]; /6: (/NULL - ]), fd=(2)==(6), (6)==(2)]
+      └── anti-join (lookup partial_tab)
+           ├── columns: m:1 n:2 i:6
+           ├── key columns: [5] = [5]
+           ├── lookup columns are key
+           ├── fd: (5)-->(6,10)
+           ├── left-join (lookup partial_tab@partial_idx,partial)
+           │    ├── columns: m:1 n:2 k:5 i:6 continuation:10
+           │    ├── key columns: [2] = [6]
+           │    ├── fd: (5)-->(6,10)
+           │    ├── scan small
+           │    │    └── columns: m:1 n:2
+           │    └── filters (true)
+           └── filters
+                └── s:7 = 'foo' [outer=(7), constraints=(/7: [/'foo' - /'foo']; tight), fd=()-->(7)]
 
 # A lookup semi-join on a partial index should have the same cost as a lookup
 # semi-join on a non-partial index.
diff --git a/pkg/sql/opt_exec_factory.go b/pkg/sql/opt_exec_factory.go
index 7597727b9507..b3979fd17c91 100644
--- a/pkg/sql/opt_exec_factory.go
+++ b/pkg/sql/opt_exec_factory.go
@@ -577,6 +577,7 @@ func (ef *execFactory) ConstructLookupJoin(
 	eqColsAreKey bool,
 	lookupCols exec.TableColumnOrdinalSet,
 	onCond tree.TypedExpr,
+	isFirstJoinInPairedJoiner bool,
 	isSecondJoinInPairedJoiner bool,
 	reqOrdering exec.OutputOrdering,
 	locking *tree.LockingItem,
@@ -604,6 +605,7 @@ func (ef *execFactory) ConstructLookupJoin(
 		table:                      tableScan,
 		joinType:                   joinType,
 		eqColsAreKey:               eqColsAreKey,
+		isFirstJoinInPairedJoiner:  isFirstJoinInPairedJoiner,
 		isSecondJoinInPairedJoiner: isSecondJoinInPairedJoiner,
 		reqOrdering:                ReqOrdering(reqOrdering),
 	}
@@ -616,6 +618,9 @@ func (ef *execFactory) ConstructLookupJoin(
 		n.onCond = pred.iVarHelper.Rebind(onCond)
 	}
 	n.columns = pred.cols
+	if isFirstJoinInPairedJoiner {
+		n.columns = append(n.columns, colinfo.ResultColumn{Name: "cont", Typ: types.Bool})
+	}
 
 	return n, nil
 }