colexec: short-circuit hash join when build side is empty

Previously, we would always fully consume both sides of the join, even
when the build (right) side is empty. We can, however, short-circuit in
such case for INNER, RIGHT OUTER, and LEFT SEMI joins and skip probing
phase altogether in such scenarios. For example, this helps query 93 of
TPC-DS benchmark with scale factor 1.

Release note: None

pkg/sql/colexec/hashjoiner.go

@@ -26,19 +26,25 @@ import (
 type hashJoinerState int
 
 const (
-	// hjBuilding represents the state the hashJoiner when it is in the build
-	// phase. Output columns from the build table are stored and a hash map is
-	// constructed from its equality columns.
+	// hjBuilding represents the state the hashJoiner is in when it is in the
+	// build phase. Output columns from the build table are stored and a hash
+	// map is constructed from its equality columns.
 	hjBuilding = iota
 
-	// hjProbing represents the state the hashJoiner is in when it is in the probe
-	// phase. Probing is done in batches against the stored hash map.
+	// hjProbing represents the state the hashJoiner is in when it is in the
+	// probe phase. Probing is done in batches against the stored hash map.
 	hjProbing
 
 	// hjEmittingUnmatched represents the state the hashJoiner is in when it is
 	// emitting unmatched rows from its build table after having consumed the
 	// probe table. This happens in the case of an outer join on the build side.
 	hjEmittingUnmatched
+
+	// hjDone represents the state the hashJoiner is in when it has finished
+	// emitting all output rows. Note that the build side will have been fully
+	// consumed in this state, but the probe side *might* have not been fully
+	// consumed.
+	hjDone
 )
 
 // hashJoinerSpec is the specification for a hash join operator. The hash
@@ -236,18 +242,38 @@ func (hj *hashJoiner) Next(ctx context.Context) coldata.Batch {
 		switch hj.state {
 		case hjBuilding:
 			hj.build(ctx)
+			if hj.ht.vals.Length() == 0 {
+				// The build side is empty, so we can short-circuit probing
+				// phase altogether for INNER, RIGHT OUTER, and LEFT SEMI
+				// joins.
+				if hj.spec.joinType == sqlbase.JoinType_INNER ||
+					hj.spec.joinType == sqlbase.JoinType_RIGHT_OUTER ||
+					hj.spec.joinType == sqlbase.JoinType_LEFT_SEMI {
+					hj.state = hjDone
+				}
+			}
 			continue
 		case hjProbing:
 			hj.exec(ctx)
 
-			if hj.output.Length() == 0 && hj.spec.right.outer {
-				hj.state = hjEmittingUnmatched
+			if hj.output.Length() == 0 {
+				if hj.spec.right.outer {
+					hj.state = hjEmittingUnmatched
+				} else {
+					hj.state = hjDone
+				}
 				continue
 			}
 			return hj.output
 		case hjEmittingUnmatched:
+			if hj.emittingUnmatchedState.rowIdx == hj.ht.vals.Length() {
+				hj.state = hjDone
+				continue
+			}
 			hj.emitUnmatched()
 			return hj.output
+		case hjDone:
+			return coldata.ZeroBatch
 		default:
 			colexecerror.InternalError("hash joiner in unhandled state")
 			// This code is unreachable, but the compiler cannot infer that.

pkg/sql/colexec/hashjoiner_test.go

@@ -1168,8 +1168,7 @@ func TestHashJoinerProjection(t *testing.T) {
 	hjOp, err := NewColOperator(ctx, flowCtx, args)
 	require.NoError(t, err)
 	hjOp.Op.Init()
-	for {
-		b := hjOp.Op.Next(ctx)
+	for b := hjOp.Op.Next(ctx); b.Length() > 0; b = hjOp.Op.Next(ctx) {
 		// The output types should be {Int64, Int64, Bool, Decimal, Float64, Bytes}
 		// and we check this explicitly.
 		b.ColVec(0).Int64()
@@ -1178,8 +1177,5 @@ func TestHashJoinerProjection(t *testing.T) {
 		b.ColVec(3).Decimal()
 		b.ColVec(4).Float64()
 		b.ColVec(5).Bytes()
-		if b.Length() == 0 {
-			break
-		}
 	}
 }