diff --git a/pkg/sql/logictest/testdata/logic_test/lookup_join_spans b/pkg/sql/logictest/testdata/logic_test/lookup_join_spans
new file mode 100644
index 000000000000..13197c0ea43a
--- /dev/null
+++ b/pkg/sql/logictest/testdata/logic_test/lookup_join_spans
@@ -0,0 +1,232 @@
+statement ok
+CREATE TABLE metrics (
+  id   SERIAL PRIMARY KEY,
+  name STRING,
+  INDEX name_index (name)
+)
+
+statement ok
+insert into metrics (id,name) values (1,'cpu'), (2,'cpu'), (3,'mem'), (4,'disk')
+
+statement ok
+CREATE TABLE metric_values (
+  metric_id INT8,
+  time      TIMESTAMPTZ,
+  value     INT8,
+  PRIMARY KEY (metric_id, time)
+)
+
+statement ok
+insert into metric_values (metric_id, time, value) values
+ (1,'2020-01-01 00:00:00+00:00',0),
+ (1,'2020-01-01 00:00:01+00:00',1),
+ (2,'2020-01-01 00:00:00+00:00',3),
+ (2,'2020-01-01 00:00:01+00:00',4)
+
+statement ok
+CREATE TABLE metric_valuesd (
+  metric_id INT8,
+  time      TIMESTAMPTZ,
+  value     INT8,
+  PRIMARY KEY (metric_id, time DESC)
+)
+
+statement ok
+insert into metric_valuesd (metric_id, time, value) values
+ (1,'2020-01-01 00:00:00+00:00',0),
+ (1,'2020-01-01 00:00:01+00:00',1),
+ (2,'2020-01-01 00:00:00+00:00',3),
+ (2,'2020-01-01 00:00:01+00:00',4)
+
+query ITIIT
+SELECT *
+FROM metric_values
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time > '2020-01-01 00:00:00+00:00' AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:01 +0000 UTC  1  1  cpu
+2  2020-01-01 00:00:01 +0000 UTC  4  2  cpu
+
+query ITIIT
+SELECT *
+FROM metric_valuesd
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time > '2020-01-01 00:00:00+00:00' AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:01 +0000 UTC  1  1  cpu
+2  2020-01-01 00:00:01 +0000 UTC  4  2  cpu
+
+query ITIIT
+SELECT *
+FROM metric_values
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time >= '2020-01-01 00:00:00+00:00' AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:00 +0000 UTC  0  1  cpu
+1  2020-01-01 00:00:01 +0000 UTC  1  1  cpu
+2  2020-01-01 00:00:00 +0000 UTC  3  2  cpu
+2  2020-01-01 00:00:01 +0000 UTC  4  2  cpu
+
+query ITIIT
+SELECT *
+FROM metric_valuesd
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time >= '2020-01-01 00:00:00+00:00' AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:00 +0000 UTC  0  1  cpu
+1  2020-01-01 00:00:01 +0000 UTC  1  1  cpu
+2  2020-01-01 00:00:00 +0000 UTC  3  2  cpu
+2  2020-01-01 00:00:01 +0000 UTC  4  2  cpu
+
+query ITIIT
+SELECT *
+FROM metric_values
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time < '2020-01-01 00:00:00+00:00' AND
+  name='cpu'
+----
+
+query ITIIT
+SELECT *
+FROM metric_valuesd
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time < '2020-01-01 00:00:00+00:00' AND
+  name='cpu'
+----
+
+query ITIIT
+SELECT *
+FROM metric_values
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time <= '2020-01-01 00:00:00+00:00' AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:00 +0000 UTC  0  1  cpu
+2  2020-01-01 00:00:00 +0000 UTC  3  2  cpu
+
+query ITIIT
+SELECT *
+FROM metric_valuesd
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time <= '2020-01-01 00:00:00+00:00' AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:00 +0000 UTC  0  1  cpu
+2  2020-01-01 00:00:00 +0000 UTC  3  2  cpu
+
+query ITIIT
+SELECT *
+FROM metric_values
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time in ('2020-01-01 00:00:00+00:00','2020-01-01 00:00:01+00:00') AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:00 +0000 UTC  0  1  cpu
+1  2020-01-01 00:00:01 +0000 UTC  1  1  cpu
+2  2020-01-01 00:00:00 +0000 UTC  3  2  cpu
+2  2020-01-01 00:00:01 +0000 UTC  4  2  cpu
+
+query ITIIT
+SELECT *
+FROM metric_valuesd
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time in ('2020-01-01 00:00:00+00:00','2020-01-01 00:00:01+00:00') AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:00 +0000 UTC  0  1  cpu
+1  2020-01-01 00:00:01 +0000 UTC  1  1  cpu
+2  2020-01-01 00:00:00 +0000 UTC  3  2  cpu
+2  2020-01-01 00:00:01 +0000 UTC  4  2  cpu
+
+query ITIIT
+SELECT *
+FROM metric_values
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time < '2020-01-01 00:00:10+00:00' AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:00 +0000 UTC  0  1  cpu
+1  2020-01-01 00:00:01 +0000 UTC  1  1  cpu
+2  2020-01-01 00:00:00 +0000 UTC  3  2  cpu
+2  2020-01-01 00:00:01 +0000 UTC  4  2  cpu
+
+query ITIIT
+SELECT *
+FROM metric_valuesd
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time < '2020-01-01 00:00:10+00:00' AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:00 +0000 UTC  0  1  cpu
+1  2020-01-01 00:00:01 +0000 UTC  1  1  cpu
+2  2020-01-01 00:00:00 +0000 UTC  3  2  cpu
+2  2020-01-01 00:00:01 +0000 UTC  4  2  cpu
+
+query ITIIT
+SELECT *
+FROM metric_values
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time BETWEEN '2020-01-01 00:00:00+00:00' AND '2020-01-01 00:10:00+00:00' AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:00 +0000 UTC  0  1  cpu
+1  2020-01-01 00:00:01 +0000 UTC  1  1  cpu
+2  2020-01-01 00:00:00 +0000 UTC  3  2  cpu
+2  2020-01-01 00:00:01 +0000 UTC  4  2  cpu
+
+query ITIIT
+SELECT *
+FROM metric_valuesd
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time BETWEEN '2020-01-01 00:00:00+00:00' AND '2020-01-01 00:10:00+00:00' AND
+  name='cpu'
+ORDER BY value
+----
+1  2020-01-01 00:00:00 +0000 UTC  0  1  cpu
+1  2020-01-01 00:00:01 +0000 UTC  1  1  cpu
+2  2020-01-01 00:00:00 +0000 UTC  3  2  cpu
+2  2020-01-01 00:00:01 +0000 UTC  4  2  cpu
+
diff --git a/pkg/sql/opt/exec/execbuilder/testdata/lookup_join b/pkg/sql/opt/exec/execbuilder/testdata/lookup_join
index 2b683114a0ba..160f05ce5d77 100644
--- a/pkg/sql/opt/exec/execbuilder/testdata/lookup_join
+++ b/pkg/sql/opt/exec/execbuilder/testdata/lookup_join
@@ -75,8 +75,7 @@ vectorized: true
 │ columns: (a, b, c, d, e, f)
 │ estimated row count: 33
 │ table: def@primary
-│ equality: (b) = (f)
-│ pred: e > 1
+│ lookup condition: (f = b) AND (e > 1)
 │
 └── • scan
       columns: (a, b, c)
diff --git a/pkg/sql/opt/memo/testdata/logprops/lookup-join b/pkg/sql/opt/memo/testdata/logprops/lookup-join
index 61df384a6fe0..ad3779b024f0 100644
--- a/pkg/sql/opt/memo/testdata/logprops/lookup-join
+++ b/pkg/sql/opt/memo/testdata/logprops/lookup-join
@@ -84,17 +84,21 @@ inner-join (lookup abcd)
  ├── prune: (2,8)
  ├── interesting orderings: (+6,+7)
  ├── inner-join (lookup abcd@secondary)
- │    ├── columns: m:1(int!null) n:2(int) a:6(int!null) b:7(int!null) abcd.rowid:9(int!null)
- │    ├── key columns: [1] = [6]
- │    ├── fd: (9)-->(6,7), (1)==(6), (6)==(1)
+ │    ├── columns: m:1(int!null) n:2(int) a:5(int!null) b:6(int!null) abcd.rowid:8(int!null)
+ │    ├── lookup expression
+ │    │    └── filters
+ │    │         ├── eq [type=bool, outer=(1,5), constraints=(/1: (/NULL - ]; /5: (/NULL - ]), fd=(1)==(5), (5)==(1)]
+ │    │         │    ├── variable: a:5 [type=int]
+ │    │         │    └── variable: m:1 [type=int]
+ │    │         └── gt [type=bool, outer=(6), constraints=(/6: [/3 - ]; tight)]
+ │    │              ├── variable: b:6 [type=int]
+ │    │              └── const: 2 [type=int]
+ │    ├── fd: (8)-->(5,6), (1)==(5), (5)==(1)
  │    ├── scan small
  │    │    ├── columns: m:1(int) n:2(int)
  │    │    ├── prune: (1,2)
- │    │    └── unfiltered-cols: (1-5)
- │    └── filters
- │         └── gt [type=bool, outer=(7), constraints=(/7: [/3 - ]; tight)]
- │              ├── variable: b:7 [type=int]
- │              └── const: 2 [type=int]
+ │    │    └── unfiltered-cols: (1-4)
+ │    └── filters (true)
  └── filters (true)
 
 # Filter that can only be applied after the primary index join.
diff --git a/pkg/sql/opt/memo/testdata/stats/lookup-join b/pkg/sql/opt/memo/testdata/stats/lookup-join
index 189677947c32..db012c022fa4 100644
--- a/pkg/sql/opt/memo/testdata/stats/lookup-join
+++ b/pkg/sql/opt/memo/testdata/stats/lookup-join
@@ -80,15 +80,17 @@ inner-join (lookup abcd)
  ├── stats: [rows=33.5823697, distinct(1)=10, null(1)=0, distinct(6)=10, null(6)=0]
  ├── fd: (1)==(6), (6)==(1)
  ├── inner-join (lookup abcd@secondary)
- │    ├── columns: m:1(int!null) n:2(int) a:6(int!null) b:7(int!null) abcd.rowid:9(int!null)
- │    ├── key columns: [1] = [6]
- │    ├── stats: [rows=33, distinct(1)=10, null(1)=0, distinct(6)=10, null(6)=0, distinct(7)=33, null(7)=0]
- │    ├── fd: (9)-->(6,7), (1)==(6), (6)==(1)
+ │    ├── columns: m:1(int!null) n:2(int) a:5(int!null) b:6(int!null) abcd.rowid:8(int!null)
+ │    ├── lookup expression
+ │    │    └── filters
+ │    │         ├── a:5 = m:1 [type=bool, outer=(1,5), constraints=(/1: (/NULL - ]; /5: (/NULL - ]), fd=(1)==(5), (5)==(1)]
+ │    │         └── b:6 > 2 [type=bool, outer=(6), constraints=(/6: [/3 - ]; tight)]
+ │    ├── stats: [rows=33, distinct(1)=10, null(1)=0, distinct(5)=10, null(5)=0, distinct(6)=33, null(6)=0]
+ │    ├── fd: (8)-->(5,6), (1)==(5), (5)==(1)
  │    ├── scan small
  │    │    ├── columns: m:1(int) n:2(int)
  │    │    └── stats: [rows=10, distinct(1)=10, null(1)=0]
- │    └── filters
- │         └── b:7 > 2 [type=bool, outer=(7), constraints=(/7: [/3 - ]; tight)]
+ │    └── filters (true)
  └── filters (true)
 
 # Filter that can only be applied after the primary index join.
diff --git a/pkg/sql/opt/xform/join_funcs.go b/pkg/sql/opt/xform/join_funcs.go
index 084c17e38304..229835e45929 100644
--- a/pkg/sql/opt/xform/join_funcs.go
+++ b/pkg/sql/opt/xform/join_funcs.go
@@ -297,12 +297,17 @@ func (c *CustomFuncs) GenerateLookupJoins(
 			// join implements logic equivalent to simple equality between
 			// columns (where NULL never equals anything).
 			foundVals, allIdx, ok := c.findJoinFilterConstants(allFilters, idxCol)
+			var foundRange bool = false
 			if !ok {
-				break
+				// Also allow a limited form of range condition filters.
+				allIdx, foundRange = c.findJoinFilterRange(allFilters, idxCol)
+				if !foundRange {
+					break
+				}
 			}
 
-			if len(foundVals) > 1 {
-				if joinType == opt.LeftJoinOp || joinType == opt.AntiJoinOp {
+			if len(foundVals) > 1 || foundRange {
+				if joinType == opt.LeftJoinOp || joinType == opt.AntiJoinOp || foundRange {
 					// We cannot use the method constructJoinWithConstants to create a cross
 					// join for left or anti joins, because constructing a cross join with
 					// foundVals will increase the size of the input. As a result,
@@ -343,11 +348,13 @@ func (c *CustomFuncs) GenerateLookupJoins(
 		}
 
 		if shouldBuildMultiSpanLookupJoin {
-			// Some of the index columns were constrained to multiple constant values,
-			// and we did not use the method constructJoinWithConstants to create a
-			// cross join as the input (either because it would have been incorrect or
-			// because it would have eliminated the opportunity to apply other
-			// optimizations such as locality optimized search; see above).
+
+			// Some of the index columns were constrained to multiple constant values
+			// or a range expression, and we did not use the method
+			// constructJoinWithConstants to create a cross join as the input (either
+			// because it would have been incorrect or because it would have
+			// eliminated the opportunity to apply other optimizations such as
+			// locality optimized search; see above).
 			//
 			// As an alternative, we store all the filters needed for the lookup in
 			// LookupExpr, which will be used to construct spans at execution time.
@@ -566,27 +573,31 @@ func (c *CustomFuncs) findFiltersForIndexLookup(
 			continue
 		}
 
+		var foundRange bool = false
 		// Try to find a filter that constrains this column to non-NULL
 		// constant values. We cannot use a NULL value because the lookup
 		// join implements logic equivalent to simple equality between
 		// columns (where NULL never equals anything).
-		values, allIdx, ok := c.findJoinFilterConstants(filters, idxCol)
+		_, allIdx, ok := c.findJoinFilterConstants(filters, idxCol)
 		if !ok {
-			break
+			// If there's no const filters look for an inequality range.
+			allIdx, foundRange = c.findJoinFilterRange(filters, idxCol)
+			if !foundRange {
+				break
+			}
 		}
 
 		if constFilters == nil {
 			constFilters = make(memo.FiltersExpr, 0, numIndexKeyCols-j)
 		}
 
-		// Ensure that the constant filter is either an equality or an IN expression.
-		// These are the only two types of expressions currently supported by the
-		// lookupJoiner for building lookup spans.
 		constFilter := filters[allIdx]
-		if !c.isCanonicalConstFilter(constFilter) {
-			constFilter = c.makeConstFilter(idxCol, values)
-		}
 		constFilters = append(constFilters, constFilter)
+
+		// Generating additional columns after a range isn't helpful so stop here.
+		if foundRange {
+			break
+		}
 	}
 
 	if len(eqFilters) == 0 {
@@ -597,26 +608,6 @@ func (c *CustomFuncs) findFiltersForIndexLookup(
 	return eqFilters, constFilters, rightSideCols
 }
 
-// isCanonicalConstFilter checks that the given filter is a constant filter in
-// one of two possible canonical formats:
-//  1. It is an equality between a variable and a constant.
-//  2. It is an IN expression between a variable and a tuple of constants.
-// Returns true if the filter matches one of these two formats. Otherwise
-// returns false.
-func (c *CustomFuncs) isCanonicalConstFilter(filter memo.FiltersItem) bool {
-	switch t := filter.Condition.(type) {
-	case *memo.EqExpr:
-		if t.Left.Op() == opt.VariableOp && opt.IsConstValueOp(t.Right) {
-			return true
-		}
-	case *memo.InExpr:
-		if t.Left.Op() == opt.VariableOp && memo.CanExtractConstTuple(t.Right) {
-			return true
-		}
-	}
-	return false
-}
-
 // makeConstFilter builds a filter that constrains the given column to the given
 // set of constant values. This is performed by either constructing an equality
 // expression or an IN expression.
@@ -975,6 +966,47 @@ func (c *CustomFuncs) findJoinFilterConstants(
 	return bestValues, bestFilterIdx, true
 }
 
+// See if there exists an inequality range for this column.  Currently we
+// only support <,>,<=,>=.
+func (c *CustomFuncs) findJoinFilterRange(
+	filters memo.FiltersExpr, col opt.ColumnID,
+) (filterIdx int, ok bool) {
+	for filterIdx := range filters {
+		props := filters[filterIdx].ScalarProps()
+		if props.TightConstraints && !props.Constraints.IsUnconstrained() {
+			constraintCol := props.Constraints.Constraint(0).Columns.Get(0).ID()
+			if constraintCol != col {
+				continue
+			}
+			// Make sure we support the expression type
+			var checkExpr func(expr opt.Expr) bool
+			checkExpr = func(expr opt.Expr) bool {
+				switch t := expr.(type) {
+				case *memo.RangeExpr:
+					return checkExpr(t.And)
+				case *memo.AndExpr:
+					return checkExpr(t.Left) && checkExpr(t.Right)
+				case *memo.GeExpr:
+					return checkExpr(t.Left) && checkExpr(t.Right)
+				case *memo.GtExpr:
+					return checkExpr(t.Left) && checkExpr(t.Right)
+				case *memo.LeExpr:
+					return checkExpr(t.Left) && checkExpr(t.Right)
+				case *memo.LtExpr:
+					return checkExpr(t.Left) && checkExpr(t.Right)
+				case *memo.VariableExpr, *memo.ConstExpr:
+					return true
+				}
+				return false
+			}
+			if checkExpr(filters[filterIdx].Condition) {
+				return filterIdx, true
+			}
+		}
+	}
+	return 0, false
+}
+
 // constructJoinWithConstants constructs a cross join that joins every row in
 // the input with every value in vals. The cross join will be converted into a
 // projection by inlining normalization rules if vals contains only a single
diff --git a/pkg/sql/opt/xform/testdata/external/lookup_join_spans b/pkg/sql/opt/xform/testdata/external/lookup_join_spans
new file mode 100644
index 000000000000..3b9f4bfa5021
--- /dev/null
+++ b/pkg/sql/opt/xform/testdata/external/lookup_join_spans
@@ -0,0 +1,65 @@
+# illustrative examples from GH #51576, see rules/join for normative tests
+exec-ddl
+CREATE TABLE metrics (
+  id   SERIAL PRIMARY KEY,
+  name STRING,
+  INDEX name_index (name)
+)
+----
+
+exec-ddl
+CREATE TABLE metric_values (
+  metric_id INT8,
+  time      TIMESTAMPTZ,
+  value     INT8,
+  PRIMARY KEY (metric_id, time)
+)
+----
+
+opt expect=GenerateLookupJoins
+SELECT *
+FROM metric_values
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time BETWEEN '2020-01-01 00:00:00+00:00' AND '2020-01-01 00:10:00+00:00' AND
+  name='cpu'
+----
+inner-join (lookup metric_values)
+ ├── columns: metric_id:1!null time:2!null value:3 id:5!null name:6!null
+ ├── lookup expression
+ │    └── filters
+ │         ├── metric_id:1 = id:5 [outer=(1,5), constraints=(/1: (/NULL - ]; /5: (/NULL - ]), fd=(1)==(5), (5)==(1)]
+ │         └── (time:2 >= '2020-01-01 00:00:00+00:00') AND (time:2 <= '2020-01-01 00:10:00+00:00') [outer=(2), constraints=(/2: [/'2020-01-01 00:00:00+00:00' - /'2020-01-01 00:10:00+00:00']; tight)]
+ ├── key: (2,5)
+ ├── fd: ()-->(6), (1,2)-->(3), (1)==(5), (5)==(1)
+ ├── scan metrics@name_index
+ │    ├── columns: id:5!null name:6!null
+ │    ├── constraint: /6/5: [/'cpu' - /'cpu']
+ │    ├── key: (5)
+ │    └── fd: ()-->(6)
+ └── filters (true)
+
+# we don't support turning LIKE into spans yet, test that we properly fall back to filters.
+opt not-expect=GenerateLookupJoins
+SELECT *
+FROM metric_values
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time::STRING LIKE '202%' AND
+  name='cpu'
+----
+inner-join (lookup metric_values)
+ ├── columns: metric_id:1!null time:2!null value:3 id:5!null name:6!null
+ ├── key columns: [5] = [1]
+ ├── stable
+ ├── key: (2,5)
+ ├── fd: ()-->(6), (1,2)-->(3), (1)==(5), (5)==(1)
+ ├── scan metrics@name_index
+ │    ├── columns: id:5!null name:6!null
+ │    ├── constraint: /6/5: [/'cpu' - /'cpu']
+ │    ├── key: (5)
+ │    └── fd: ()-->(6)
+ └── filters
+      └── time:2::STRING LIKE '202%' [outer=(2), stable]
diff --git a/pkg/sql/opt/xform/testdata/external/tpce b/pkg/sql/opt/xform/testdata/external/tpce
index 8e19660c2ff5..dace24b450df 100644
--- a/pkg/sql/opt/xform/testdata/external/tpce
+++ b/pkg/sql/opt/xform/testdata/external/tpce
@@ -4113,7 +4113,10 @@ limit
  │    │    ├── ordering: +20 opt(24) [actual: +20]
  │    │    └── inner-join (lookup watch_item)
  │    │         ├── columns: wi_wl_id:19!null wi_s_symb:20!null wl_id:23!null wl_c_id:24!null
- │    │         ├── key columns: [23] = [19]
+ │    │         ├── lookup expression
+ │    │         │    └── filters
+ │    │         │         ├── wi_wl_id:19 = wl_id:23 [outer=(19,23), constraints=(/19: (/NULL - ]; /23: (/NULL - ]), fd=(19)==(23), (23)==(19)]
+ │    │         │         └── wi_s_symb:20 > 'SYMB' [outer=(20), constraints=(/20: [/e'SYMB\x00' - ]; tight)]
  │    │         ├── key: (20,23)
  │    │         ├── fd: ()-->(24), (19)==(23), (23)==(19)
  │    │         ├── select
@@ -4126,8 +4129,7 @@ limit
  │    │         │    │    └── fd: (23)-->(24)
  │    │         │    └── filters
  │    │         │         └── wl_c_id:24 = 0 [outer=(24), constraints=(/24: [/0 - /0]; tight), fd=()-->(24)]
- │    │         └── filters
- │    │              └── wi_s_symb:20 > 'SYMB' [outer=(20), constraints=(/20: [/e'SYMB\x00' - ]; tight)]
+ │    │         └── filters (true)
  │    └── filters (true)
  └── 1
 
diff --git a/pkg/sql/opt/xform/testdata/external/tpce-no-stats b/pkg/sql/opt/xform/testdata/external/tpce-no-stats
index b53395de1c80..bbd0fc1b1eb2 100644
--- a/pkg/sql/opt/xform/testdata/external/tpce-no-stats
+++ b/pkg/sql/opt/xform/testdata/external/tpce-no-stats
@@ -4133,7 +4133,10 @@ limit
  │    │    ├── ordering: +20 opt(24) [actual: +20]
  │    │    └── inner-join (lookup watch_item)
  │    │         ├── columns: wi_wl_id:19!null wi_s_symb:20!null wl_id:23!null wl_c_id:24!null
- │    │         ├── key columns: [23] = [19]
+ │    │         ├── lookup expression
+ │    │         │    └── filters
+ │    │         │         ├── wi_wl_id:19 = wl_id:23 [outer=(19,23), constraints=(/19: (/NULL - ]; /23: (/NULL - ]), fd=(19)==(23), (23)==(19)]
+ │    │         │         └── wi_s_symb:20 > 'SYMB' [outer=(20), constraints=(/20: [/e'SYMB\x00' - ]; tight)]
  │    │         ├── key: (20,23)
  │    │         ├── fd: ()-->(24), (19)==(23), (23)==(19)
  │    │         ├── select
@@ -4146,8 +4149,7 @@ limit
  │    │         │    │    └── fd: (23)-->(24)
  │    │         │    └── filters
  │    │         │         └── wl_c_id:24 = 0 [outer=(24), constraints=(/24: [/0 - /0]; tight), fd=()-->(24)]
- │    │         └── filters
- │    │              └── wi_s_symb:20 > 'SYMB' [outer=(20), constraints=(/20: [/e'SYMB\x00' - ]; tight)]
+ │    │         └── filters (true)
  │    └── filters (true)
  └── 1
 
diff --git a/pkg/sql/opt/xform/testdata/external/trading b/pkg/sql/opt/xform/testdata/external/trading
index 47823f8b24b0..0076de7353d5 100644
--- a/pkg/sql/opt/xform/testdata/external/trading
+++ b/pkg/sql/opt/xform/testdata/external/trading
@@ -845,7 +845,12 @@ project
  │    │         │    │    └── ordering: +10 opt(9) [actual: +10]
  │    │         │    ├── left-join (lookup transactiondetails@detailscardidindex)
  │    │         │    │    ├── columns: id:1!null name:2!null rarity:3 setname:4 number:5!null isfoil:6!null transactiondetails.dealerid:20 isbuy:21 transactiondate:22 transactiondetails.cardid:23 quantity:24
- │    │         │    │    ├── key columns: [42 43 1] = [20 21 23]
+ │    │         │    │    ├── lookup expression
+ │    │         │    │    │    └── filters
+ │    │         │    │    │         ├── transactiondetails.cardid:23 = id:1 [outer=(1,23), constraints=(/1: (/NULL - ]; /23: (/NULL - ]), fd=(1)==(23), (23)==(1)]
+ │    │         │    │    │         ├── transactiondetails.dealerid:20 = 1 [outer=(20), constraints=(/20: [/1 - /1]; tight), fd=()-->(20)]
+ │    │         │    │    │         ├── NOT isbuy:21 [outer=(21), constraints=(/21: [/false - /false]; tight), fd=()-->(21)]
+ │    │         │    │    │         └── (transactiondate:22 >= '2020-02-28 00:00:00+00:00') AND (transactiondate:22 <= '2020-03-01 00:00:00+00:00') [outer=(22), constraints=(/22: [/'2020-02-28 00:00:00+00:00' - /'2020-03-01 00:00:00+00:00']; tight)]
  │    │         │    │    ├── immutable
  │    │         │    │    ├── stats: [rows=3543333.33, distinct(1)=19000, null(1)=0, distinct(2)=13000, null(2)=0, distinct(5)=829, null(5)=0, distinct(6)=5601.15328, null(6)=0, distinct(23)=19000, null(23)=0]
  │    │         │    │    ├── key: (1,22-24)
@@ -854,7 +859,7 @@ project
  │    │         │    │    ├── project
  │    │         │    │    │    ├── columns: "lookup_join_const_col_@21":43!null "lookup_join_const_col_@20":42!null id:1!null name:2!null rarity:3 setname:4 number:5!null isfoil:6!null
  │    │         │    │    │    ├── immutable
- │    │         │    │    │    ├── stats: [rows=19000, distinct(1)=19000, null(1)=0, distinct(2)=13000, null(2)=0, distinct(5)=829, null(5)=0, distinct(6)=5601.15328, null(6)=0, distinct(42)=1, null(42)=0, distinct(43)=1, null(43)=0]
+ │    │         │    │    │    ├── stats: [rows=19000]
  │    │         │    │    │    ├── key: (1)
  │    │         │    │    │    ├── fd: ()-->(42,43), (1)-->(2-6), (2,4,5)~~>(1,3,6)
  │    │         │    │    │    ├── ordering: +1
@@ -876,8 +881,7 @@ project
  │    │         │    │    │    └── projections
  │    │         │    │    │         ├── false [as="lookup_join_const_col_@21":43]
  │    │         │    │    │         └── 1 [as="lookup_join_const_col_@20":42]
- │    │         │    │    └── filters
- │    │         │    │         └── (transactiondate:22 >= '2020-02-28 00:00:00+00:00') AND (transactiondate:22 <= '2020-03-01 00:00:00+00:00') [outer=(22), constraints=(/22: [/'2020-02-28 00:00:00+00:00' - /'2020-03-01 00:00:00+00:00']; tight)]
+ │    │         │    │    └── filters (true)
  │    │         │    └── filters (true)
  │    │         └── aggregations
  │    │              ├── sum [as=sum:30, outer=(24)]
diff --git a/pkg/sql/opt/xform/testdata/external/trading-mutation b/pkg/sql/opt/xform/testdata/external/trading-mutation
index 95702925936d..29c7d7794d5b 100644
--- a/pkg/sql/opt/xform/testdata/external/trading-mutation
+++ b/pkg/sql/opt/xform/testdata/external/trading-mutation
@@ -849,7 +849,12 @@ project
  │    │         │    │    └── ordering: +10 opt(9) [actual: +10]
  │    │         │    ├── left-join (lookup transactiondetails@detailscardidindex)
  │    │         │    │    ├── columns: id:1!null name:2!null rarity:3 setname:4 number:5!null isfoil:6!null transactiondetails.dealerid:24 isbuy:25 transactiondate:26 transactiondetails.cardid:27 quantity:28
- │    │         │    │    ├── key columns: [48 49 1] = [24 25 27]
+ │    │         │    │    ├── lookup expression
+ │    │         │    │    │    └── filters
+ │    │         │    │    │         ├── transactiondetails.cardid:27 = id:1 [outer=(1,27), constraints=(/1: (/NULL - ]; /27: (/NULL - ]), fd=(1)==(27), (27)==(1)]
+ │    │         │    │    │         ├── transactiondetails.dealerid:24 = 1 [outer=(24), constraints=(/24: [/1 - /1]; tight), fd=()-->(24)]
+ │    │         │    │    │         ├── NOT isbuy:25 [outer=(25), constraints=(/25: [/false - /false]; tight), fd=()-->(25)]
+ │    │         │    │    │         └── (transactiondate:26 >= '2020-02-28 00:00:00+00:00') AND (transactiondate:26 <= '2020-03-01 00:00:00+00:00') [outer=(26), constraints=(/26: [/'2020-02-28 00:00:00+00:00' - /'2020-03-01 00:00:00+00:00']; tight)]
  │    │         │    │    ├── immutable
  │    │         │    │    ├── stats: [rows=3543333.33, distinct(1)=19000, null(1)=0, distinct(2)=13000, null(2)=0, distinct(5)=829, null(5)=0, distinct(6)=5601.15328, null(6)=0, distinct(27)=19000, null(27)=0]
  │    │         │    │    ├── key: (1,26-28)
@@ -858,7 +863,7 @@ project
  │    │         │    │    ├── project
  │    │         │    │    │    ├── columns: "lookup_join_const_col_@25":49!null "lookup_join_const_col_@24":48!null id:1!null name:2!null rarity:3 setname:4 number:5!null isfoil:6!null
  │    │         │    │    │    ├── immutable
- │    │         │    │    │    ├── stats: [rows=19000, distinct(1)=19000, null(1)=0, distinct(2)=13000, null(2)=0, distinct(5)=829, null(5)=0, distinct(6)=5601.15328, null(6)=0, distinct(48)=1, null(48)=0, distinct(49)=1, null(49)=0]
+ │    │         │    │    │    ├── stats: [rows=19000]
  │    │         │    │    │    ├── key: (1)
  │    │         │    │    │    ├── fd: ()-->(48,49), (1)-->(2-6), (2,4,5)~~>(1,3,6)
  │    │         │    │    │    ├── ordering: +1
@@ -880,8 +885,7 @@ project
  │    │         │    │    │    └── projections
  │    │         │    │    │         ├── false [as="lookup_join_const_col_@25":49]
  │    │         │    │    │         └── 1 [as="lookup_join_const_col_@24":48]
- │    │         │    │    └── filters
- │    │         │    │         └── (transactiondate:26 >= '2020-02-28 00:00:00+00:00') AND (transactiondate:26 <= '2020-03-01 00:00:00+00:00') [outer=(26), constraints=(/26: [/'2020-02-28 00:00:00+00:00' - /'2020-03-01 00:00:00+00:00']; tight)]
+ │    │         │    │    └── filters (true)
  │    │         │    └── filters (true)
  │    │         └── aggregations
  │    │              ├── sum [as=sum:36, outer=(28)]
diff --git a/pkg/sql/opt/xform/testdata/rules/join b/pkg/sql/opt/xform/testdata/rules/join
index dd5524ca559e..376019d43a7c 100644
--- a/pkg/sql/opt/xform/testdata/rules/join
+++ b/pkg/sql/opt/xform/testdata/rules/join
@@ -2997,7 +2997,7 @@ left-join (lookup lookup_expr [as=t])
  │    │    └── filters
  │    │         ├── column2:2 = v:6 [outer=(2,6), constraints=(/2: (/NULL - ]; /6: (/NULL - ]), fd=(2)==(6), (6)==(2)]
  │    │         ├── r:3 IN ('east', 'west') [outer=(3), constraints=(/3: [/'east' - /'east'] [/'west' - /'west']; tight)]
- │    │         ├── u:5 IN (1, 2) [outer=(5), constraints=(/5: [/1 - /1] [/2 - /2]; tight)]
+ │    │         ├── (u:5 = 1) OR (u:5 = 2) [outer=(5), constraints=(/5: [/1 - /1] [/2 - /2]; tight)]
  │    │         ├── y:9 IN (10, 20) [outer=(9), constraints=(/9: [/10 - /10] [/20 - /20]; tight)]
  │    │         └── z:10 = 5 [outer=(10), constraints=(/10: [/5 - /5]; tight), fd=()-->(10)]
  │    ├── cardinality: [3 - ]
@@ -3008,8 +3008,7 @@ left-join (lookup lookup_expr [as=t])
  │    │    ├── (1, 10)
  │    │    ├── (2, 20)
  │    │    └── (3, NULL)
- │    └── filters
- │         └── (u:5 = 1) OR (u:5 = 2) [outer=(5), constraints=(/5: [/1 - /1] [/2 - /2]; tight)]
+ │    └── filters (true)
  └── filters (true)
 
 # We can't build a lookup join with any of the indexes.
@@ -3082,12 +3081,14 @@ SELECT a,b,n,m FROM small JOIN abcd ON a=m AND b>1
 ----
 inner-join (lookup abcd@secondary)
  ├── columns: a:6!null b:7!null n:2 m:1!null
- ├── key columns: [1] = [6]
+ ├── lookup expression
+ │    └── filters
+ │         ├── a:6 = m:1 [outer=(1,6), constraints=(/1: (/NULL - ]; /6: (/NULL - ]), fd=(1)==(6), (6)==(1)]
+ │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
  ├── fd: (1)==(6), (6)==(1)
  ├── scan small
  │    └── columns: m:1 n:2
- └── filters
-      └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
+ └── filters (true)
 
 # Covering case, left-join.
 opt expect=GenerateLookupJoinsWithFilter
@@ -3095,11 +3096,13 @@ SELECT a,b,n,m FROM small LEFT JOIN abcd ON a=m AND b>1
 ----
 left-join (lookup abcd@secondary)
  ├── columns: a:6 b:7 n:2 m:1
- ├── key columns: [1] = [6]
+ ├── lookup expression
+ │    └── filters
+ │         ├── a:6 = m:1 [outer=(1,6), constraints=(/1: (/NULL - ]; /6: (/NULL - ]), fd=(1)==(6), (6)==(1)]
+ │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
  ├── scan small
  │    └── columns: m:1 n:2
- └── filters
-      └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
+ └── filters (true)
 
 # Non-covering case.
 opt expect=GenerateLookupJoinsWithFilter
@@ -3112,12 +3115,14 @@ inner-join (lookup abcd)
  ├── fd: (1)==(6), (6)==(1)
  ├── inner-join (lookup abcd@secondary)
  │    ├── columns: m:1!null n:2 a:6!null b:7!null abcd.rowid:9!null
- │    ├── key columns: [1] = [6]
+ │    ├── lookup expression
+ │    │    └── filters
+ │    │         ├── a:6 = m:1 [outer=(1,6), constraints=(/1: (/NULL - ]; /6: (/NULL - ]), fd=(1)==(6), (6)==(1)]
+ │    │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
  │    ├── fd: (9)-->(6,7), (1)==(6), (6)==(1)
  │    ├── scan small
  │    │    └── columns: m:1 n:2
- │    └── filters
- │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
+ │    └── filters (true)
  └── filters (true)
 
 # Non-covering case, left join.
@@ -3130,12 +3135,14 @@ left-join (lookup abcd)
  ├── lookup columns are key
  ├── left-join (lookup abcd@secondary)
  │    ├── columns: m:1 n:2 a:6 b:7 abcd.rowid:9
- │    ├── key columns: [1] = [6]
+ │    ├── lookup expression
+ │    │    └── filters
+ │    │         ├── a:6 = m:1 [outer=(1,6), constraints=(/1: (/NULL - ]; /6: (/NULL - ]), fd=(1)==(6), (6)==(1)]
+ │    │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
  │    ├── fd: (9)-->(6,7)
  │    ├── scan small
  │    │    └── columns: m:1 n:2
- │    └── filters
- │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
+ │    └── filters (true)
  └── filters (true)
 
 # Non-covering case, extra filter bound by index.
@@ -3149,13 +3156,15 @@ inner-join (lookup abcd)
  ├── fd: (1)==(6), (6)==(1)
  ├── inner-join (lookup abcd@secondary)
  │    ├── columns: m:1!null n:2!null a:6!null b:7!null abcd.rowid:9!null
- │    ├── key columns: [1] = [6]
+ │    ├── lookup expression
+ │    │    └── filters
+ │    │         ├── a:6 = m:1 [outer=(1,6), constraints=(/1: (/NULL - ]; /6: (/NULL - ]), fd=(1)==(6), (6)==(1)]
+ │    │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
  │    ├── fd: (9)-->(6,7), (1)==(6), (6)==(1)
  │    ├── scan small
  │    │    └── columns: m:1 n:2
  │    └── filters
- │         ├── b:7 > n:2 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ])]
- │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
+ │         └── b:7 > n:2 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ])]
  └── filters (true)
 
 # Non-covering case, extra filter bound by index, left join.
@@ -3168,13 +3177,15 @@ left-join (lookup abcd)
  ├── lookup columns are key
  ├── left-join (lookup abcd@secondary)
  │    ├── columns: m:1 n:2 a:6 b:7 abcd.rowid:9
- │    ├── key columns: [1] = [6]
+ │    ├── lookup expression
+ │    │    └── filters
+ │    │         ├── a:6 = m:1 [outer=(1,6), constraints=(/1: (/NULL - ]; /6: (/NULL - ]), fd=(1)==(6), (6)==(1)]
+ │    │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
  │    ├── fd: (9)-->(6,7)
  │    ├── scan small
  │    │    └── columns: m:1 n:2
  │    └── filters
- │         ├── b:7 > n:2 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ])]
- │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
+ │         └── b:7 > n:2 [outer=(2,7), constraints=(/2: (/NULL - ]; /7: (/NULL - ])]
  └── filters (true)
 
 # Non-covering case, extra filter not bound by index.
@@ -3188,12 +3199,14 @@ inner-join (lookup abcd)
  ├── fd: (1)==(6), (6)==(1)
  ├── inner-join (lookup abcd@secondary)
  │    ├── columns: m:1!null n:2 a:6!null b:7!null abcd.rowid:9!null
- │    ├── key columns: [1] = [6]
+ │    ├── lookup expression
+ │    │    └── filters
+ │    │         ├── a:6 = m:1 [outer=(1,6), constraints=(/1: (/NULL - ]; /6: (/NULL - ]), fd=(1)==(6), (6)==(1)]
+ │    │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
  │    ├── fd: (9)-->(6,7), (1)==(6), (6)==(1)
  │    ├── scan small
  │    │    └── columns: m:1 n:2
- │    └── filters
- │         └── b:7 > 1 [outer=(7), constraints=(/7: [/2 - ]; tight)]
+ │    └── filters (true)
  └── filters
       └── c:8 > n:2 [outer=(2,8), constraints=(/2: (/NULL - ]; /8: (/NULL - ])]
 
@@ -3448,12 +3461,14 @@ inner-join (lookup abcde)
  ├── fd: (1)==(6), (6)==(1)
  ├── inner-join (lookup abcde@secondary)
  │    ├── columns: m:1!null n:2 a:6!null b:7!null c:8 abcde.rowid:11!null
- │    ├── key columns: [1] = [6]
+ │    ├── lookup expression
+ │    │    └── filters
+ │    │         ├── a:6 = m:1 [outer=(1,6), constraints=(/1: (/NULL - ]; /6: (/NULL - ]), fd=(1)==(6), (6)==(1)]
+ │    │         └── b:7 < 10 [outer=(7), constraints=(/7: (/NULL - /9]; tight)]
  │    ├── fd: (11)-->(6-8), (1)==(6), (6)==(1)
  │    ├── scan small
  │    │    └── columns: m:1 n:2
- │    └── filters
- │         └── b:7 < 10 [outer=(7), constraints=(/7: (/NULL - /9]; tight)]
+ │    └── filters (true)
  └── filters (true)
 
 # Lookup Joiner uses the constant equality columns at the same time as the explicit
diff --git a/pkg/sql/opt/xform/testdata/rules/lookup_join_spans b/pkg/sql/opt/xform/testdata/rules/lookup_join_spans
new file mode 100644
index 000000000000..f7c2449cb55f
--- /dev/null
+++ b/pkg/sql/opt/xform/testdata/rules/lookup_join_spans
@@ -0,0 +1,65 @@
+# illustrative examples from GH #51576, see rules/join for normative tests
+exec-ddl
+CREATE TABLE metrics (
+  id   SERIAL PRIMARY KEY,
+  name STRING,
+  INDEX name_index (name)
+)
+----
+
+exec-ddl
+CREATE TABLE metric_values (
+  metric_id INT8,
+  time      TIMESTAMPTZ,
+  value     INT8,
+  PRIMARY KEY (metric_id, time)
+)
+----
+
+opt expect=GenerateLookupJoins
+SELECT *
+FROM metric_values
+INNER JOIN metrics
+ON metric_id=id
+WHERE
+  time BETWEEN '2020-01-01 00:00:00+00:00' AND '2020-01-01 00:10:00+00:00' AND
+  name='cpu'
+----
+inner-join (lookup metric_values)
+ ├── columns: metric_id:1!null time:2!null value:3 id:5!null name:6!null
+ ├── lookup expression
+ │    └── filters
+ │         ├── metric_id:1 = id:5 [outer=(1,5), constraints=(/1: (/NULL - ]; /5: (/NULL - ]), fd=(1)==(5), (5)==(1)]
+ │         └── (time:2 >= '2020-01-01 00:00:00+00:00') AND (time:2 <= '2020-01-01 00:10:00+00:00') [outer=(2), constraints=(/2: [/'2020-01-01 00:00:00+00:00' - /'2020-01-01 00:10:00+00:00']; tight)]
+ ├── key: (2,5)
+ ├── fd: ()-->(6), (1,2)-->(3), (1)==(5), (5)==(1)
+ ├── scan metrics@name_index
+ │    ├── columns: id:5!null name:6!null
+ │    ├── constraint: /6/5: [/'cpu' - /'cpu']
+ │    ├── key: (5)
+ │    └── fd: ()-->(6)
+ └── filters (true)
+
+# we don't support turning LIKE into scans yet, test that we fall back to filters
+ opt not-expect=GenerateLookupJoins
+ SELECT *
+ FROM metric_values
+ INNER JOIN metrics
+ ON metric_id=id
+ WHERE
+   time::STRING LIKE '202%' AND
+   name='cpu'
+----
+inner-join (lookup metric_values)
+ ├── columns: metric_id:1!null time:2!null value:3 id:5!null name:6!null
+ ├── key columns: [5] = [1]
+ ├── stable
+ ├── key: (2,5)
+ ├── fd: ()-->(6), (1,2)-->(3), (1)==(5), (5)==(1)
+ ├── scan metrics@name_index
+ │    ├── columns: id:5!null name:6!null
+ │    ├── constraint: /6/5: [/'cpu' - /'cpu']
+ │    ├── key: (5)
+ │    └── fd: ()-->(6)
+ └── filters
+      └── time:2::STRING LIKE '202%' [outer=(2), stable]
diff --git a/pkg/sql/rowexec/joinreader.go b/pkg/sql/rowexec/joinreader.go
index a7d8e9836a16..ebb443191f30 100644
--- a/pkg/sql/rowexec/joinreader.go
+++ b/pkg/sql/rowexec/joinreader.go
@@ -371,14 +371,14 @@ func (jr *joinReader) initJoinReaderStrategy(
 	spanBuilder.SetNeededColumns(neededRightCols)
 
 	var generator joinReaderSpanGenerator
-	var keyToInputRowIndices map[string][]int
-	if readerType != indexJoinReaderType {
-		keyToInputRowIndices = make(map[string][]int)
-	}
-	// Else: see the comment in defaultSpanGenerator on why we don't need
-	// this map for index joins.
 
 	if jr.lookupExpr.Expr == nil {
+		// See the comment in defaultSpanGenerator on why we don't need
+		// this map for index joins.
+		var keyToInputRowIndices map[string][]int
+		if readerType != indexJoinReaderType {
+			keyToInputRowIndices = make(map[string][]int)
+		}
 		generator = &defaultSpanGenerator{
 			spanBuilder:          spanBuilder,
 			keyToInputRowIndices: keyToInputRowIndices,
@@ -404,7 +404,6 @@ func (jr *joinReader) initJoinReaderStrategy(
 				spanBuilder,
 				numKeyCols,
 				len(jr.input.OutputTypes()),
-				keyToInputRowIndices,
 				&jr.lookupExpr,
 				tableOrdToIndexOrd,
 			); err != nil {
@@ -417,7 +416,6 @@ func (jr *joinReader) initJoinReaderStrategy(
 				spanBuilder,
 				numKeyCols,
 				len(jr.input.OutputTypes()),
-				keyToInputRowIndices,
 				&jr.lookupExpr,
 				&jr.remoteLookupExpr,
 				tableOrdToIndexOrd,
diff --git a/pkg/sql/rowexec/joinreader_span_generator.go b/pkg/sql/rowexec/joinreader_span_generator.go
index 4433979ab2d5..0d1c9986df53 100644
--- a/pkg/sql/rowexec/joinreader_span_generator.go
+++ b/pkg/sql/rowexec/joinreader_span_generator.go
@@ -12,6 +12,7 @@ package rowexec
 
 import (
 	"fmt"
+	"sort"
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/execinfrapb"
@@ -134,6 +135,19 @@ func (g *defaultSpanGenerator) maxLookupCols() int {
 	return len(g.lookupCols)
 }
 
+type spanRowIndex struct {
+	span       roachpb.Span
+	rowIndices []int
+}
+
+type spanRowIndices []spanRowIndex
+
+func (a spanRowIndices) Len() int           { return len(a) }
+func (a spanRowIndices) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
+func (a spanRowIndices) Less(i, j int) bool { return a[i].span.Key.Compare(a[j].span.Key) < 0 }
+
+var _ sort.Interface = &spanRowIndices{}
+
 // multiSpanGenerator is the joinReaderSpanGenerator used when each lookup will
 // scan multiple spans in the index. This is the case when some of the index
 // columns can take on multiple constant values. For example, the
@@ -150,17 +164,17 @@ type multiSpanGenerator struct {
 	// indexColInfos stores info about the values that each index column can
 	// take on in the spans produced by the multiSpanGenerator. See the comment
 	// above multiSpanGeneratorIndexColInfo for more details.
-	indexColInfos []multiSpanGeneratorIndexColInfo
+	indexColInfos []interface{ multiSpanGeneratorColInfo }
 
 	// indexKeyRows and indexKeySpans are used to generate the spans for a single
 	// input row. They are allocated once in init(), and then reused for every row.
 	indexKeyRows  []rowenc.EncDatumRow
 	indexKeySpans roachpb.Spans
 
-	// keyToInputRowIndices maps a lookup span key to the input row indices that
+	// spanToInputRowIndices maps a lookup span to the input row indices that
 	// desire that span. This is used for de-duping spans, and to map the fetched
 	// rows to the input rows that need to join with them.
-	keyToInputRowIndices map[string][]int
+	spanToInputRowIndices spanRowIndices
 
 	// spansCount is the number of spans generated for each input row.
 	spansCount int
@@ -177,32 +191,73 @@ type multiSpanGenerator struct {
 	// numInputCols is the number of columns in the input to the joinReader.
 	numInputCols int
 
+	// index of inequality colinfo (there can be only one), -1 otherwise
+	inequalityColIdx int
+
 	scratchSpans roachpb.Spans
 }
 
-// multiSpanGeneratorIndexColInfo contains info about the values that a specific
+// multiSpanGeneratorColInfo contains info about the values that a specific
 // index column can take on in the spans produced by the multiSpanGenerator. The
 // column ordinal is not contained in this struct, but depends on the location
 // of this struct in the indexColInfos slice; the position in the slice
 // corresponds to the position in the index.
-// - If len(constVals) > 0, the index column can equal any of the given
-//   constant values. This is the case when there is a join filter such as
-//   c IN ('a', 'b', 'c'), where c is a key column in the index.
-// - If constVals is empty, then inputRowIdx corresponds to an index into the
-//   input row. This is the case for join filters such as c = a, where c is a
-//   column in the index and a is a column in the input.
-type multiSpanGeneratorIndexColInfo struct {
-	constVals   tree.Datums
+type multiSpanGeneratorColInfo interface {
+	String() string
+}
+
+// multiSpanGeneratorValuesColInfo is used to represent a column constrained
+// by a set of constants (ie '=' or 'in' expressions).
+type multiSpanGeneratorValuesColInfo struct {
+	constVals tree.Datums
+}
+
+func (i multiSpanGeneratorValuesColInfo) String() string {
+	return fmt.Sprintf("[constVals: %s]", i.constVals.String())
+}
+
+// multiSpanGeneratorIndexVarColInfo represents a column that matches a column
+// in the input row.  inputRowIdx corresponds to an index into the input row.
+// This is the case for join filters such as c = a, where c is a column in the
+// index and a is a column in the input.
+type multiSpanGeneratorIndexVarColInfo struct {
 	inputRowIdx int
 }
 
-func (i multiSpanGeneratorIndexColInfo) String() string {
-	if len(i.constVals) > 0 {
-		return fmt.Sprintf("[constVals: %s]", i.constVals.String())
-	}
+func (i multiSpanGeneratorIndexVarColInfo) String() string {
 	return fmt.Sprintf("[inputRowIdx: %d]", i.inputRowIdx)
 }
 
+// multiSpanGeneratorInequalityColInfo represents a column that is bound by a
+// range expression.	If there are <,>, >= or <= inequalities we distill them
+// into a start and end datum.
+type multiSpanGeneratorInequalityColInfo struct {
+	start          tree.Datum
+	startInclusive bool
+	end            tree.Datum
+	endInclusive   bool
+}
+
+func (i multiSpanGeneratorInequalityColInfo) String() string {
+	var startBoundary rune
+	if i.startInclusive {
+		startBoundary = '['
+	} else {
+		startBoundary = '('
+	}
+	var endBoundary rune
+	if i.endInclusive {
+		endBoundary = ']'
+	} else {
+		endBoundary = ')'
+	}
+	return fmt.Sprintf("%c%v - %v%c", startBoundary, i.start, i.end, endBoundary)
+}
+
+var _ multiSpanGeneratorColInfo = &multiSpanGeneratorValuesColInfo{}
+var _ multiSpanGeneratorColInfo = &multiSpanGeneratorIndexVarColInfo{}
+var _ multiSpanGeneratorColInfo = &multiSpanGeneratorInequalityColInfo{}
+
 // maxLookupCols is part of the joinReaderSpanGenerator interface.
 func (g *multiSpanGenerator) maxLookupCols() int {
 	return len(g.indexColInfos)
@@ -214,14 +269,13 @@ func (g *multiSpanGenerator) init(
 	spanBuilder *span.Builder,
 	numKeyCols int,
 	numInputCols int,
-	keyToInputRowIndices map[string][]int,
 	exprHelper *execinfrapb.ExprHelper,
 	tableOrdToIndexOrd util.FastIntMap,
 ) error {
 	g.spanBuilder = spanBuilder
 	g.numInputCols = numInputCols
-	g.keyToInputRowIndices = keyToInputRowIndices
 	g.tableOrdToIndexOrd = tableOrdToIndexOrd
+	g.inequalityColIdx = -1
 
 	// Initialize the spansCount to 1, since we'll always have at least one span.
 	// This number may increase when we call fillInIndexColInfos() below.
@@ -229,7 +283,7 @@ func (g *multiSpanGenerator) init(
 
 	// Process the given expression to fill in g.indexColInfos with info from the
 	// join conditions. This info will be used later to generate the spans.
-	g.indexColInfos = make([]multiSpanGeneratorIndexColInfo, 0, numKeyCols)
+	g.indexColInfos = make([]interface{ multiSpanGeneratorColInfo }, 0, numKeyCols)
 	if err := g.fillInIndexColInfos(exprHelper.Expr); err != nil {
 		return err
 	}
@@ -269,19 +323,21 @@ func (g *multiSpanGenerator) init(
 	//   [ 'east'  -  2  - ]
 	//   [ 'west'  -  2  - ]
 	//
+
+	// Make first pass flushing out the structure with const values
 	g.indexKeyRows = make([]rowenc.EncDatumRow, 1, g.spansCount)
 	g.indexKeyRows[0] = make(rowenc.EncDatumRow, 0, lookupColsCount)
 	for _, info := range g.indexColInfos {
-		if len(info.constVals) > 0 {
+		if valuesInfo, ok := info.(multiSpanGeneratorValuesColInfo); ok {
 			for i, n := 0, len(g.indexKeyRows); i < n; i++ {
 				indexKeyRow := g.indexKeyRows[i]
-				for j := 1; j < len(info.constVals); j++ {
+				for j := 1; j < len(valuesInfo.constVals); j++ {
 					newIndexKeyRow := make(rowenc.EncDatumRow, len(indexKeyRow), lookupColsCount)
 					copy(newIndexKeyRow, indexKeyRow)
-					newIndexKeyRow = append(newIndexKeyRow, rowenc.EncDatum{Datum: info.constVals[j]})
+					newIndexKeyRow = append(newIndexKeyRow, rowenc.EncDatum{Datum: valuesInfo.constVals[j]})
 					g.indexKeyRows = append(g.indexKeyRows, newIndexKeyRow)
 				}
-				g.indexKeyRows[i] = append(indexKeyRow, rowenc.EncDatum{Datum: info.constVals[0]})
+				g.indexKeyRows[i] = append(indexKeyRow, rowenc.EncDatum{Datum: valuesInfo.constVals[0]})
 			}
 		} else {
 			for i := 0; i < len(g.indexKeyRows); i++ {
@@ -304,8 +360,11 @@ func (g *multiSpanGenerator) init(
 //  1. Equalities between input columns and index columns, such as c1 = c2.
 //  2. Equalities or IN conditions between index columns and constants, such
 //     as c = 5 or c IN ('a', 'b', 'c').
+//  3. Inequalities from (possibly AND'd) <,>,<=,>= exprs.
+//
 // The optimizer should have ensured that all conditions fall into one of
-// these two categories. Any other expression types will return an error.
+// these categories. Any other expression types will return an error.
+// TODO: We should probably be doing this at compile time, see #65773
 func (g *multiSpanGenerator) fillInIndexColInfos(expr tree.TypedExpr) error {
 	switch t := expr.(type) {
 	case *tree.AndExpr:
@@ -315,16 +374,27 @@ func (g *multiSpanGenerator) fillInIndexColInfos(expr tree.TypedExpr) error {
 		return g.fillInIndexColInfos(t.Right.(tree.TypedExpr))
 
 	case *tree.ComparisonExpr:
-		if t.Operator != tree.EQ && t.Operator != tree.In {
-			return errors.AssertionFailedf("comparison operator must be EQ or In. Found %s", t.Operator)
+
+		setOfVals := false
+		inequality := false
+		switch t.Operator {
+		case tree.EQ, tree.In:
+			setOfVals = true
+		case tree.GE, tree.LE, tree.GT, tree.LT:
+			inequality = true
+		default:
+			// This should never happen because of enforcement at opt time.
+			return errors.AssertionFailedf("comparison operator not supported. Found %s", t.Operator)
 		}
 
 		tabOrd := -1
-		info := multiSpanGeneratorIndexColInfo{inputRowIdx: -1}
 
-		// Since we only support EQ and In, we don't need to check anything other
-		// than the types of the arguments in order to extract the info.
-		getInfo := func(typedExpr tree.TypedExpr) error {
+		var info multiSpanGeneratorColInfo
+
+		// For EQ and In, we just need to check the types of the arguments in order
+		// to extract the info. For inequalities we return the const datums that
+		// will form the span boundaries.
+		getInfo := func(typedExpr tree.TypedExpr) (tree.Datum, error) {
 			switch t := typedExpr.(type) {
 			case *tree.IndexedVar:
 				// IndexedVars can either be from the input or the index. If the
@@ -333,31 +403,58 @@ func (g *multiSpanGenerator) fillInIndexColInfos(expr tree.TypedExpr) error {
 				if t.Idx >= g.numInputCols {
 					tabOrd = t.Idx - g.numInputCols
 				} else {
-					info.inputRowIdx = t.Idx
+					info = multiSpanGeneratorIndexVarColInfo{inputRowIdx: t.Idx}
 				}
 
 			case tree.Datum:
-				switch t.ResolvedType().Family() {
-				case types.TupleFamily:
-					info.constVals = t.(*tree.DTuple).D
-				default:
-					info.constVals = tree.Datums{t}
+				if setOfVals {
+					var values tree.Datums
+					switch t.ResolvedType().Family() {
+					case types.TupleFamily:
+						values = t.(*tree.DTuple).D
+					default:
+						values = tree.Datums{t}
+					}
+					// Every time there are multiple possible values, we multiply the
+					// spansCount by the number of possibilities. We will need to create
+					// spans representing the cartesian product of possible values for
+					// each column.
+					info = multiSpanGeneratorValuesColInfo{constVals: values}
+					g.spansCount *= len(values)
+				} else {
+					return t, nil
 				}
-				// Every time there are multiple possible values, we multiply the
-				// spansCount by the number of possibilities. We will need to create
-				// spans representing the cartesian product of possible values for
-				// each column.
-				g.spansCount *= len(info.constVals)
 
 			default:
-				return errors.AssertionFailedf("unhandled comparison argument type %T", t)
+				return nil, errors.AssertionFailedf("unhandled comparison argument type %T", t)
 			}
-			return nil
+			return nil, nil
 		}
-		if err := getInfo(t.Left.(tree.TypedExpr)); err != nil {
+
+		// NB: we make no attempt to deal with column direction here, that is sorted
+		// out later in the span builder.
+		var inequalityInfo multiSpanGeneratorInequalityColInfo
+		if lval, err := getInfo(t.Left.(tree.TypedExpr)); lval != nil {
+			if t.Operator == tree.LT || t.Operator == tree.LE {
+				inequalityInfo.start = lval
+				inequalityInfo.startInclusive = t.Operator == tree.LE
+			} else {
+				inequalityInfo.end = lval
+				inequalityInfo.endInclusive = t.Operator == tree.GE
+			}
+		} else if err != nil {
 			return err
 		}
-		if err := getInfo(t.Right.(tree.TypedExpr)); err != nil {
+
+		if rval, err := getInfo(t.Right.(tree.TypedExpr)); rval != nil {
+			if t.Operator == tree.LT || t.Operator == tree.LE {
+				inequalityInfo.end = rval
+				inequalityInfo.endInclusive = t.Operator == tree.LE
+			} else {
+				inequalityInfo.start = rval
+				inequalityInfo.startInclusive = t.Operator == tree.GE
+			}
+		} else if err != nil {
 			return err
 		}
 
@@ -365,6 +462,12 @@ func (g *multiSpanGenerator) fillInIndexColInfos(expr tree.TypedExpr) error {
 		if !ok {
 			return errors.AssertionFailedf("table column %d not found in index", tabOrd)
 		}
+
+		if inequality {
+			info = inequalityInfo
+			g.inequalityColIdx = idxOrd
+		}
+
 		if len(g.indexColInfos) <= idxOrd {
 			g.indexColInfos = g.indexColInfos[:idxOrd+1]
 		}
@@ -380,36 +483,95 @@ func (g *multiSpanGenerator) fillInIndexColInfos(expr tree.TypedExpr) error {
 // generateNonNullSpans generates spans for a given row. It does not include
 // null values, since those values would not match the lookup condition anyway.
 func (g *multiSpanGenerator) generateNonNullSpans(row rowenc.EncDatumRow) (roachpb.Spans, error) {
-	// Fill in the holes in g.indexKeyRows that correspond to input row
-	// values.
-	for j, info := range g.indexColInfos {
-		if len(info.constVals) == 0 {
-			for i := 0; i < len(g.indexKeyRows); i++ {
-				g.indexKeyRows[i][j] = row[info.inputRowIdx]
+	// Fill in the holes in g.indexKeyRows that correspond to input row values.
+	for i := 0; i < len(g.indexKeyRows); i++ {
+		for j, info := range g.indexColInfos {
+			if inf, ok := info.(multiSpanGeneratorIndexVarColInfo); ok {
+				g.indexKeyRows[i][j] = row[inf.inputRowIdx]
 			}
 		}
 	}
 
 	// Convert the index key rows to spans.
 	g.indexKeySpans = g.indexKeySpans[:0]
+
+	var inequalityInfo multiSpanGeneratorInequalityColInfo
+	if g.inequalityColIdx != -1 {
+		inequalityInfo = g.indexColInfos[g.inequalityColIdx].(multiSpanGeneratorInequalityColInfo)
+	}
+
+	//build spans for each row
 	for _, indexKeyRow := range g.indexKeyRows {
-		span, containsNull, err := g.spanBuilder.SpanFromEncDatums(indexKeyRow, len(g.indexColInfos))
+		var s roachpb.Span
+		var err error
+		var containsNull bool
+		if g.inequalityColIdx == -1 {
+			s, containsNull, err = g.spanBuilder.SpanFromEncDatums(indexKeyRow, len(g.indexColInfos))
+		} else {
+			s, containsNull, err = g.spanBuilder.SpanFromEncDatumsWithRange(indexKeyRow, len(g.indexColInfos),
+				inequalityInfo.start, inequalityInfo.startInclusive, inequalityInfo.end, inequalityInfo.endInclusive)
+		}
+
 		if err != nil {
 			return roachpb.Spans{}, err
 		}
+
 		if !containsNull {
-			g.indexKeySpans = append(g.indexKeySpans, span)
+			g.indexKeySpans = append(g.indexKeySpans, s)
 		}
 	}
+
 	return g.indexKeySpans, nil
 }
 
+// findInputRowIndicesByKey does a binary search to find the span that contains
+// the given key.
+func (g *multiSpanGenerator) findInputRowIndicesByKey(key roachpb.Key) []int {
+	i, j := 0, len(g.spanToInputRowIndices)
+	for i < j {
+		h := (i + j) >> 1
+		sp := g.spanToInputRowIndices[h]
+		if sp.span.ContainsKey(key) {
+			return sp.rowIndices
+		}
+		if key.Compare(sp.span.Key) < 0 {
+			j = h
+		} else {
+			i = h + 1
+		}
+	}
+
+	return nil
+}
+
+// findInputRowIndices does a linear search for a given span, this is during
+// rowexec.generateSpans before spanToInputRowIndices is sorted.
+func (g *multiSpanGenerator) findInputRowIndices(span roachpb.Span) []int {
+	for _, s := range g.spanToInputRowIndices {
+		if span.Equal(s.span) {
+			return s.rowIndices
+		}
+	}
+	return nil
+}
+
+// setInputRowIndices installs a slice of row indices for a given span, this is during
+// rowexec.generateSpans before spanToInputRowIndices is sorted.
+func (g *multiSpanGenerator) setInputRowIndices(span roachpb.Span, indices []int) {
+	for i, s := range g.spanToInputRowIndices {
+		if span.Equal(s.span) {
+			g.spanToInputRowIndices[i].rowIndices = indices
+			return
+		}
+	}
+	// not found, add it
+	g.spanToInputRowIndices = append(g.spanToInputRowIndices, spanRowIndex{span: span, rowIndices: indices})
+}
+
 // generateSpans is part of the joinReaderSpanGenerator interface.
 func (g *multiSpanGenerator) generateSpans(rows []rowenc.EncDatumRow) (roachpb.Spans, error) {
-	// This loop gets optimized to a runtime.mapclear call.
-	for k := range g.keyToInputRowIndices {
-		delete(g.keyToInputRowIndices, k)
-	}
+	g.spanToInputRowIndices = g.spanToInputRowIndices[:0]
+
 	// We maintain a map from index key to the corresponding input rows so we can
 	// join the index results to the inputs.
 	g.scratchSpans = g.scratchSpans[:0]
@@ -420,21 +582,32 @@ func (g *multiSpanGenerator) generateSpans(rows []rowenc.EncDatumRow) (roachpb.S
 		}
 		for j := range generatedSpans {
 			generatedSpan := &generatedSpans[j]
-			inputRowIndices := g.keyToInputRowIndices[string(generatedSpan.Key)]
+			inputRowIndices := g.findInputRowIndices(*generatedSpan)
 			if inputRowIndices == nil {
-				g.scratchSpans = g.spanBuilder.MaybeSplitSpanIntoSeparateFamilies(
-					g.scratchSpans, *generatedSpan, len(g.indexColInfos), false /* containsNull */)
+				// RFC: should doing range spans affects how/whether we should call this?
+				// it seems to break when calling it with inequality spans
+				if g.inequalityColIdx != -1 {
+					g.scratchSpans = append(g.scratchSpans, *generatedSpan)
+				} else {
+					g.scratchSpans = g.spanBuilder.MaybeSplitSpanIntoSeparateFamilies(
+						g.scratchSpans, *generatedSpan, len(g.indexColInfos), false /* containsNull */)
+				}
 			}
-			g.keyToInputRowIndices[string(generatedSpan.Key)] = append(inputRowIndices, i)
+			inputRowIndices = append(inputRowIndices, i)
+			g.setInputRowIndices(*generatedSpan, inputRowIndices)
 		}
 	}
 
+	if len(g.spanToInputRowIndices) > 1 {
+		sort.Sort(g.spanToInputRowIndices)
+	}
+
 	return g.scratchSpans, nil
 }
 
 // getMatchingRowIndices is part of the joinReaderSpanGenerator interface.
 func (g *multiSpanGenerator) getMatchingRowIndices(key roachpb.Key) []int {
-	return g.keyToInputRowIndices[string(key)]
+	return g.findInputRowIndicesByKey(key)
 }
 
 // localityOptimizedSpanGenerator is the span generator for locality optimized
@@ -452,18 +625,17 @@ func (g *localityOptimizedSpanGenerator) init(
 	spanBuilder *span.Builder,
 	numKeyCols int,
 	numInputCols int,
-	keyToInputRowIndices map[string][]int,
 	localExprHelper *execinfrapb.ExprHelper,
 	remoteExprHelper *execinfrapb.ExprHelper,
 	tableOrdToIndexOrd util.FastIntMap,
 ) error {
 	if err := g.localSpanGen.init(
-		spanBuilder, numKeyCols, numInputCols, keyToInputRowIndices, localExprHelper, tableOrdToIndexOrd,
+		spanBuilder, numKeyCols, numInputCols, localExprHelper, tableOrdToIndexOrd,
 	); err != nil {
 		return err
 	}
 	if err := g.remoteSpanGen.init(
-		spanBuilder, numKeyCols, numInputCols, keyToInputRowIndices, remoteExprHelper, tableOrdToIndexOrd,
+		spanBuilder, numKeyCols, numInputCols, remoteExprHelper, tableOrdToIndexOrd,
 	); err != nil {
 		return err
 	}
diff --git a/pkg/sql/span/span_builder.go b/pkg/sql/span/span_builder.go
index e4619bd3e4b4..4b1f5b00fb34 100644
--- a/pkg/sql/span/span_builder.go
+++ b/pkg/sql/span/span_builder.go
@@ -145,6 +145,66 @@ func (s *Builder) SpanFromEncDatums(
 		values[:prefixLen], s.indexColTypes[:prefixLen], s.indexColDirs[:prefixLen], s.table, s.index, &s.alloc, s.KeyPrefix)
 }
 
+// SpanFromEncDatumsWithRange encodes a range span. The inequality is assumed to
+// be the end of the span and the start/end keys are generated by putting them
+// in the values row at the prefixLen - 1 position.  Only one of start or end
+// need be non-nil, omitted one causing an open ended range span to be
+// generated.  Since the exec code knows nothing about index column sorting
+// direction we assume ascending if they are descending we deal with that here.
+func (s *Builder) SpanFromEncDatumsWithRange(
+	values rowenc.EncDatumRow,
+	prefixLen int,
+	startDatum tree.Datum,
+	startInclusive bool,
+	endDatum tree.Datum,
+	endInclusive bool,
+) (_ roachpb.Span, containsNull bool, err error) {
+
+	if s.indexColDirs[prefixLen-1] == descpb.IndexDescriptor_DESC {
+		startDatum, endDatum = endDatum, startDatum
+		startInclusive, endInclusive = endInclusive, startInclusive
+	}
+
+	makeKeyFromRow := func(r rowenc.EncDatumRow, l int) (k roachpb.Key, cn bool, e error) {
+		k, _, cn, e = rowenc.MakeKeyFromEncDatums(r[:l], s.indexColTypes[:l], s.indexColDirs[:l],
+			s.table, s.index, &s.alloc, s.KeyPrefix)
+		return
+	}
+
+	var startKey, endKey roachpb.Key
+	var startContainsNull, endContainsNull bool
+	if startDatum != nil {
+		values[prefixLen-1] = rowenc.EncDatum{Datum: startDatum}
+		startKey, startContainsNull, err = makeKeyFromRow(values, prefixLen)
+		if !startInclusive {
+			startKey = startKey.Next()
+		}
+	} else {
+		startKey, startContainsNull, err = makeKeyFromRow(values, prefixLen-1)
+	}
+
+	if err != nil {
+		return roachpb.Span{}, false, err
+	}
+
+	if endDatum != nil {
+		values[prefixLen-1] = rowenc.EncDatum{Datum: endDatum}
+		endKey, endContainsNull, err = makeKeyFromRow(values, prefixLen)
+		if endInclusive {
+			endKey = endKey.PrefixEnd()
+		}
+	} else {
+		endKey, endContainsNull, err = makeKeyFromRow(values, prefixLen-1)
+		endKey = endKey.PrefixEnd()
+	}
+
+	if err != nil {
+		return roachpb.Span{}, false, err
+	}
+
+	return roachpb.Span{Key: startKey, EndKey: endKey}, startContainsNull || endContainsNull, nil
+}
+
 // SpanFromDatumRow generates an index span with prefixLen constraint columns from the index.
 // SpanFromDatumRow assumes that values is a valid table row for the Builder's table.
 // It also returns whether or not the input values contain a null value or not, which can be