Merge #54808

54808: sql: introduce RIGHT SEMI and RIGHT ANTI joins r=yuzefovich a=yuzefovich

**sql: introduce RIGHT SEMI and RIGHT ANTI joins**

This commit introduces right semi and right anti joins which will give
an ability to the optimizer to control on which relation we're building
the hash table (in case of the hash join). This will allow us to bake in
the assumption that we're building the hash table on the right side into
the row-execution (we already have this assumption in the vectorized
hash join).

Additionally, this commit adds several assertions to different types of
joiners about which join types they support as well as fixes the
`joinerBase` code to handle the join types that emit only columns from
one side (previously, it just happened to work because the left side
columns were always in the output).

Release note: None

**rowexec: move hash joiner unit tests**

This commit renames "joinerTestCase" to "hashJoinerTestCase" since those
tests are only used by the hash joiner and moves the corresponding code
to hashjoiner_test.go. I believe that originally the intention was to
generalize them to use in several joiners types, but that intention was
never implemented, and since we're moving towards deprecating these
processors, I don't think it ever will.

Release note: None

**rowexec: support right semi and right anti hash joins**

This commit adds support for right semi and right anti hash joins which
turned out to be pretty easy to do. This paves the way for refactoring
the hash joiner code to hard-code that we're always storing the right
side.

Addresses: #54707.

Release note: None

**rowexec: clean up the hash joiner**

This commit cleans up the hash joiner by baking in the assumption that
we're always building the hash table on the right side. The assumption
allows us to remove the buffering logic (which attempted to find
a shorter relation, until a certain threshold), and we will now rely on
the optimizer to request an appropriate join type with the right
relation having smaller cardinality.

Release note: None

**colexec: support right semi and right anti in merge joiner**

This commit adds the support for right semi and right anti merge joins.
This turned out to be relatively easy when following the examples of
left semi and left anti (with the new right cases being simpler because
of not having to worry about set-operation joins which are intertwined
with left semi and left anti joins). The testing is done by "mirroring"
existing test cases for left semi and left anti joins.

Addresses: #54707.

Release note: None

**colexec: support right semi and right anti in the hash joiner**

The support of right semi and right anti in the hash joiner required
extending the idea of tracking whether build rows matched which was
already being done in case of full/right outer joins. Notable difference
is that right semi/anti joins don't emit any output during the probing
phase, so the "collection" simply tracks the matched rows whereas the
output is fully emitted in `hjEmittingRight` phase of the algorithm.
Note that because the external hash joiner falls back to the disk-backed
merge joiner in some cases it is important to point out that the
vectorized merge joiner already supports all join types.

Addresses: #54707.

Release note: None

**rowexec: support right semi and right anti in the merge joiner**

The support of right semi and right anti merge joiner is added via
tracking which rows have been matched on the right side. In case of
right semi we remember that the right row is matched, emit it, and then
make sure to ignore it (to not emit for the second time). In case of
right anti, all unmatched right rows will be emitted (similar to
right/full outer joins). This support was mostly added in order to test
the correctness of the vectorized merge joiner (I don't expect the
optimizer to ever plan right semi/anti merge joins).

Addresses: #54707.

Release note: None

**opt: plan right semi and right anti hash joins**

This commit utilizes the recently added right semi and right anti joins
in the optimizer. Internally, those join types are represented as the
same opt operator with the only changes being switching two input
relations if the left has smaller cardinality when execbuilding the hash
join (with the same change to the costing).

Fixes: #54707.

Release note: None

Co-authored-by: Yahor Yuzefovich <yahor@cockroachlabs.com>

Author: craig[bot]

Makefile

@@ -846,7 +846,9 @@ EXECGEN_TARGETS = \
   pkg/sql/colexec/mergejoiner_leftanti.eg.go \
   pkg/sql/colexec/mergejoiner_leftouter.eg.go \
   pkg/sql/colexec/mergejoiner_leftsemi.eg.go \
+  pkg/sql/colexec/mergejoiner_rightanti.eg.go \
   pkg/sql/colexec/mergejoiner_rightouter.eg.go \
+  pkg/sql/colexec/mergejoiner_rightsemi.eg.go \
   pkg/sql/colexec/ordered_synchronizer.eg.go \
   pkg/sql/colexec/proj_const_left_ops.eg.go \
   pkg/sql/colexec/proj_const_right_ops.eg.go \

pkg/sql/apply_join.go

@@ -86,6 +86,8 @@ func newApplyJoinNode(
 		return nil, errors.AssertionFailedf("unsupported right outer apply join: %d", log.Safe(joinType))
 	case descpb.ExceptAllJoin, descpb.IntersectAllJoin:
 		return nil, errors.AssertionFailedf("unsupported apply set op: %d", log.Safe(joinType))
+	case descpb.RightSemiJoin, descpb.RightAntiJoin:
+		return nil, errors.AssertionFailedf("unsupported right semi/anti apply join: %d", log.Safe(joinType))
 	}
 
 	return &applyJoinNode{

pkg/sql/catalog/descpb/join_type.go

@@ -26,6 +26,8 @@ const (
 	LeftAntiJoin     = JoinType_LEFT_ANTI
 	IntersectAllJoin = JoinType_INTERSECT_ALL
 	ExceptAllJoin    = JoinType_EXCEPT_ALL
+	RightSemiJoin    = JoinType_RIGHT_SEMI
+	RightAntiJoin    = JoinType_RIGHT_ANTI
 )
 
 // JoinTypeFromAstString takes a join string as found in a SQL
@@ -54,6 +56,17 @@ func (j JoinType) IsSetOpJoin() bool {
 	return j == IntersectAllJoin || j == ExceptAllJoin
 }
 
+// ShouldIncludeLeftColsInOutput returns true if this join should include
+// the columns from the left side into the output.
+func (j JoinType) ShouldIncludeLeftColsInOutput() bool {
+	switch j {
+	case RightSemiJoin, RightAntiJoin:
+		return false
+	default:
+		return true
+	}
+}
+
 // ShouldIncludeRightColsInOutput returns true if this join should include
 // the columns from the right side into the output.
 func (j JoinType) ShouldIncludeRightColsInOutput() bool {
@@ -64,3 +77,15 @@ func (j JoinType) ShouldIncludeRightColsInOutput() bool {
 		return true
 	}
 }
+
+// IsEmptyOutputWhenRightIsEmpty returns whether this join type will always
+// produce an empty output when the right relation is empty.
+func (j JoinType) IsEmptyOutputWhenRightIsEmpty() bool {
+	switch j {
+	case InnerJoin, RightOuterJoin, LeftSemiJoin,
+		RightSemiJoin, IntersectAllJoin, RightAntiJoin:
+		return true
+	default:
+		return false
+	}
+}

pkg/sql/catalog/descpb/join_type.pb.go

@@ -24,8 +24,8 @@ enum JoinType {
   // one row from the right side (as per equality columns and ON condition).
   LEFT_SEMI = 4;
 
-  // A left anti join is an "inverted" semi join: it returns the rows from the
-  // left side that don't match any columns on the right side (as per equality
+  // A left anti join is an "inverted" left semi join: it returns the rows from
+  // the left side that don't match any rows on the right side (as per equality
   // columns and ON condition).
   LEFT_ANTI = 5;
 
@@ -75,4 +75,15 @@ enum JoinType {
   // In practice, there is a one-to-one mapping between the left and right
   // columns (they are all equality columns).
   EXCEPT_ALL = 7;
+
+  // A right semi join returns the rows from the right side that match at least
+  // one row from the left side (as per equality columns and ON condition). It
+  // is a commuted version of the left semi join.
+  RIGHT_SEMI = 8;
+
+  // A right anti join is an "inverted" right semi join: it returns the rows
+  // from the right side that don't match any rows on the left side (as per
+  // equality columns and ON condition). It is a commuted version of the left
+  // anti join.
+  RIGHT_ANTI = 9;
 }

pkg/sql/catalog/descpb/join_type.proto

@@ -225,7 +225,9 @@ targets = [
     "mergejoiner_leftanti.eg.go",
     "mergejoiner_leftouter.eg.go",
     "mergejoiner_leftsemi.eg.go",
+    "mergejoiner_rightanti.eg.go",
     "mergejoiner_rightouter.eg.go",
+    "mergejoiner_rightsemi.eg.go",
     "ordered_synchronizer.eg.go",
     "proj_const_left_ops.eg.go",
     "proj_const_right_ops.eg.go",

pkg/sql/colexec/BUILD.bazel

@@ -227,8 +227,7 @@ func supportedNatively(spec *execinfrapb.ProcessorSpec) error {
 		return nil
 
 	case spec.Core.MergeJoiner != nil:
-		if !spec.Core.MergeJoiner.OnExpr.Empty() &&
-			spec.Core.MergeJoiner.Type != descpb.InnerJoin {
+		if !spec.Core.MergeJoiner.OnExpr.Empty() && spec.Core.MergeJoiner.Type != descpb.InnerJoin {
 			return errors.Errorf("can't plan non-inner merge join with ON expressions")
 		}
 		return nil
@@ -923,12 +922,12 @@ func NewColOperator(
 					args.TestingKnobs.SpillingCallbackFn,
 				)
 			}
-			result.ColumnTypes = make([]*types.T, len(leftTypes)+len(rightTypes))
-			copy(result.ColumnTypes, leftTypes)
-			if !core.HashJoiner.Type.ShouldIncludeRightColsInOutput() {
-				result.ColumnTypes = result.ColumnTypes[:len(leftTypes):len(leftTypes)]
-			} else {
-				copy(result.ColumnTypes[len(leftTypes):], rightTypes)
+			result.ColumnTypes = make([]*types.T, 0, len(leftTypes)+len(rightTypes))
+			if core.HashJoiner.Type.ShouldIncludeLeftColsInOutput() {
+				result.ColumnTypes = append(result.ColumnTypes, leftTypes...)
+			}
+			if core.HashJoiner.Type.ShouldIncludeRightColsInOutput() {
+				result.ColumnTypes = append(result.ColumnTypes, rightTypes...)
 			}
 
 			if !core.HashJoiner.OnExpr.Empty() && core.HashJoiner.Type == descpb.InnerJoin {
@@ -984,12 +983,12 @@ func NewColOperator(
 
 			result.Op = mj
 			result.ToClose = append(result.ToClose, mj.(colexecbase.Closer))
-			result.ColumnTypes = make([]*types.T, len(leftTypes)+len(rightTypes))
-			copy(result.ColumnTypes, leftTypes)
-			if !core.MergeJoiner.Type.ShouldIncludeRightColsInOutput() {
-				result.ColumnTypes = result.ColumnTypes[:len(leftTypes):len(leftTypes)]
-			} else {
-				copy(result.ColumnTypes[len(leftTypes):], rightTypes)
+			result.ColumnTypes = make([]*types.T, 0, len(leftTypes)+len(rightTypes))
+			if core.MergeJoiner.Type.ShouldIncludeLeftColsInOutput() {
+				result.ColumnTypes = append(result.ColumnTypes, leftTypes...)
+			}
+			if core.MergeJoiner.Type.ShouldIncludeRightColsInOutput() {
+				result.ColumnTypes = append(result.ColumnTypes, rightTypes...)
 			}
 
 			if onExpr != nil {

pkg/sql/colexec/colbuilder/execplan.go

@@ -34,7 +34,9 @@ type joinTypeInfo struct {
 	IsLeftOuter  bool
 	IsRightOuter bool
 	IsLeftSemi   bool
+	IsRightSemi  bool
 	IsLeftAnti   bool
+	IsRightAnti  bool
 	IsSetOp      bool
 
 	String string
@@ -173,10 +175,18 @@ func init() {
 			IsLeftSemi: true,
 			String:     "LeftSemi",
 		},
+		{
+			IsRightSemi: true,
+			String:      "RightSemi",
+		},
 		{
 			IsLeftAnti: true,
 			String:     "LeftAnti",
 		},
+		{
+			IsRightAnti: true,
+			String:      "RightAnti",
+		},
 		{
 			IsLeftSemi: true,
 			IsSetOp:    true,

pkg/sql/colexec/execgen/cmd/execgen/mergejoiner_gen.go

@@ -61,7 +61,7 @@ func TestExternalHashJoiner(t *testing.T) {
 	// which the joiner spills to disk.
 	for _, spillForced := range []bool{false, true} {
 		flowCtx.Cfg.TestingKnobs.ForceDiskSpill = spillForced
-		for _, tcs := range [][]*joinTestCase{hjTestCases, mjTestCases} {
+		for _, tcs := range [][]*joinTestCase{getHJTestCases(), getMJTestCases()} {
 			for _, tc := range tcs {
 				delegateFDAcquisitions := rng.Float64() < 0.5
 				log.Infof(ctx, "spillForced=%t/%s/delegateFDAcquisitions=%t", spillForced, tc.description, delegateFDAcquisitions)