release-24.1: rowexec: fix oid handling in merge join and zigzag join

pkg/sql/logictest/testdata/logic_test/pgoidtype

@@ -710,3 +710,26 @@ query TT
 SELECT 0::REGROLE, 0::REGROLE::TEXT;
 ----
 -  -
+
+# Regression test for using the correct type when decoding EncDatum of Oid type
+# family (#123474).
+statement ok
+CREATE TABLE t123474 (
+  col_0 REGROLE, col_1 OID, col_2 INT,
+  INDEX (col_1 DESC) STORING (col_2),
+  INDEX (col_0) STORING (col_1)
+);
+INSERT INTO t123474 (col_0, col_1, col_2) VALUES (0, 0, 0);
+SET testing_optimizer_random_seed = 6047211422050928467;
+SET testing_optimizer_disable_rule_probability = 0.500000;
+
+query T
+SELECT t2.col_1
+  FROM t123474 AS t1 JOIN t123474 AS t2 ON (t1.col_0) = (t2.col_1)
+  ORDER BY t1.col_0;
+----
+0
+
+statement ok
+RESET testing_optimizer_random_seed;
+RESET testing_optimizer_disable_rule_probability;

pkg/sql/rowenc/encoded_datum.go

@@ -396,6 +396,15 @@ func (ed *EncDatum) Fingerprint(
 //	+1 if the receiver is greater than rhs.
 func (ed *EncDatum) Compare(
 	typ *types.T, a *tree.DatumAlloc, evalCtx *eval.Context, rhs *EncDatum,
+) (int, error) {
+	return ed.CompareEx(typ, a, evalCtx, rhs, typ)
+}
+
+// CompareEx is the same as Compare but allows specifying the type of RHS
+// EncDatum in case it's different from ed (e.g. we might be comparing Oid
+// family types with different Oids).
+func (ed *EncDatum) CompareEx(
+	typ *types.T, a *tree.DatumAlloc, evalCtx *eval.Context, rhs *EncDatum, rhsTyp *types.T,
 ) (int, error) {
 	// TODO(radu): if we have both the Datum and a key encoding available, which
 	// one would be faster to use?
@@ -410,7 +419,7 @@ func (ed *EncDatum) Compare(
 	if err := ed.EnsureDecoded(typ, a); err != nil {
 		return 0, err
 	}
-	if err := rhs.EnsureDecoded(typ, a); err != nil {
+	if err := rhs.EnsureDecoded(rhsTyp, a); err != nil {
 		return 0, err
 	}
 	return ed.Datum.CompareError(evalCtx, rhs.Datum)
@@ -553,11 +562,27 @@ func (r EncDatumRow) Compare(
 	evalCtx *eval.Context,
 	rhs EncDatumRow,
 ) (int, error) {
-	if len(r) != len(types) || len(rhs) != len(types) {
-		panic(errors.AssertionFailedf("length mismatch: %d types, %d lhs, %d rhs\n%+v\n%+v\n%+v", len(types), len(r), len(rhs), types, r, rhs))
+	return r.CompareEx(types, a, ordering, evalCtx, rhs, types)
+}
+
+// CompareEx is the same as Compare but allows specifying a different type
+// schema for RHS row.
+func (r EncDatumRow) CompareEx(
+	types []*types.T,
+	a *tree.DatumAlloc,
+	ordering colinfo.ColumnOrdering,
+	evalCtx *eval.Context,
+	rhs EncDatumRow,
+	rhsTypes []*types.T,
+) (int, error) {
+	if len(r) != len(types) || len(rhs) != len(rhsTypes) || len(r) != len(rhs) {
+		panic(errors.AssertionFailedf(
+			"length mismatch: %d types, %d rhs types, %d lhs, %d rhs\n%+v\n%+v\n%+v",
+			len(types), len(rhsTypes), len(r), len(rhs), types, r, rhs,
+		))
 	}
 	for _, c := range ordering {
-		cmp, err := r[c.ColIdx].Compare(types[c.ColIdx], a, evalCtx, &rhs[c.ColIdx])
+		cmp, err := r[c.ColIdx].CompareEx(types[c.ColIdx], a, evalCtx, &rhs[c.ColIdx], rhsTypes[c.ColIdx])
 		if err != nil {
 			return 0, err
 		}

pkg/sql/rowexec/stream_merger.go

@@ -79,7 +79,7 @@ func (sm *streamMerger) NextBatch(
 	}
 
 	cmp, err := CompareEncDatumRowForMerge(
-		sm.left.types, lrow, rrow, sm.left.ordering, sm.right.ordering,
+		sm.left.types, sm.right.types, lrow, rrow, sm.left.ordering, sm.right.ordering,
 		sm.nullEquality, &sm.datumAlloc, evalCtx,
 	)
 	if err != nil {
@@ -108,7 +108,7 @@ func (sm *streamMerger) NextBatch(
 // a DatumAlloc which is used for decoding if any underlying EncDatum is not
 // yet decoded.
 func CompareEncDatumRowForMerge(
-	lhsTypes []*types.T,
+	lhsTypes, rhsTypes []*types.T,
 	lhs, rhs rowenc.EncDatumRow,
 	leftOrdering, rightOrdering colinfo.ColumnOrdering,
 	nullEquality bool,
@@ -144,7 +144,7 @@ func CompareEncDatumRowForMerge(
 			}
 			continue
 		}
-		cmp, err := lhs[lIdx].Compare(lhsTypes[lIdx], da, evalCtx, &rhs[rIdx])
+		cmp, err := lhs[lIdx].CompareEx(lhsTypes[lIdx], da, evalCtx, &rhs[rIdx], rhsTypes[rIdx])
 		if err != nil {
 			return 0, err
 		}

pkg/sql/rowexec/zigzagjoiner.go

@@ -573,12 +573,14 @@ func (z *zigzagJoiner) matchBase(curRow rowenc.EncDatumRow, side int) (bool, err
 
 	prevEqDatums := z.extractEqDatums(z.baseRow, z.prevSide())
 	curEqDatums := z.extractEqDatums(curRow, side)
-
-	eqColTypes := z.infos[side].eqColTypes
+	prevEqColTypes := z.infos[z.prevSide()].eqColTypes
+	curEqColTypes := z.infos[side].eqColTypes
 	ordering := z.infos[side].eqColOrdering
 
 	// Compare the equality columns of the baseRow to that of the curRow.
-	cmp, err := prevEqDatums.Compare(eqColTypes, &z.infos[side].alloc, ordering, z.FlowCtx.EvalCtx, curEqDatums)
+	cmp, err := prevEqDatums.CompareEx(
+		prevEqColTypes, &z.infos[side].alloc, ordering, z.FlowCtx.EvalCtx, curEqDatums, curEqColTypes,
+	)
 	if err != nil {
 		return false, err
 	}
@@ -718,9 +720,13 @@ func (z *zigzagJoiner) nextRow(ctx context.Context) (rowenc.EncDatumRow, error)
 
 			prevEqCols := z.extractEqDatums(prevNext, prevSide)
 			currentEqCols := z.extractEqDatums(curNext, z.side)
-			eqColTypes := curInfo.eqColTypes
+			prevEqColTypes := z.infos[prevSide].eqColTypes
+			curEqColTypes := z.infos[z.side].eqColTypes
 			ordering := curInfo.eqColOrdering
-			cmp, err := prevEqCols.Compare(eqColTypes, &z.infos[z.side].alloc, ordering, z.FlowCtx.EvalCtx, currentEqCols)
+			cmp, err := prevEqCols.CompareEx(
+				prevEqColTypes, &z.infos[z.side].alloc, ordering,
+				z.FlowCtx.EvalCtx, currentEqCols, curEqColTypes,
+			)
 			if err != nil {
 				return nil, err
 			}