I wonder whether this whole boundCondition block can be simplified to

Predicate.create(condition.get, left.output ++ right.output).eval _

since the left side is always on the left and the right side is always on the right.

(Edit: the proposed simplification in ⬆️ is not correct (the bound condition input schema is not necessarily the join's output schema); see later comments for discussion of an alternative simplification).

Further down in this file, we have

showing that buildPlan and streamPlan are just re-mappings of left and right depending on the build side.

Several years ago, it looks like we used to do something similar to my proposal:

but we switched to the current use of buildPlan and streamPlan in a refactoring in #12102 (I'm not fully clear on why).

+1, It's fragile to keep 2 pieces of code in sync.

Oh I guess it's wrong for inner join + build left?

I found this

  protected def createResultProjection(): (InternalRow) => InternalRow = joinType match {
    case LeftExistence(_) =>
      UnsafeProjection.create(output, output)
    case _ =>
      // Always put the stream side on left to simplify implementation
      // both of left and right side could be null
      UnsafeProjection.create(
        output, (streamedPlan.output ++ buildPlan.output).map(_.withNullability(true)))
  }

I think stream side left and build side right is the common case, but we have some special cases for outer joins.

Actually I don't know why outer join needs to be a special case. Can't we always put streaming side left and build side right?

HashJoin.scala

override def output: Seq[Attribute] = {
    joinType match {
      case _: InnerLike =>
        left.output ++ right.output
      case LeftOuter =>
        left.output ++ right.output.map(_.withNullability(true))
      case RightOuter =>
        left.output.map(_.withNullability(true)) ++ right.output
      case j: ExistenceJoin =>
        left.output :+ j.exists
      case LeftExistence(_) =>
        left.output
      case x =>
        throw new IllegalArgumentException(s"HashJoin should not take $x as the JoinType")
    }
  }

ShuffledJoin.scala

override def output: Seq[Attribute] = {
    joinType match {
      case _: InnerLike =>
        left.output ++ right.output
      case LeftOuter =>
        left.output ++ right.output.map(_.withNullability(true))
      case RightOuter =>
        left.output.map(_.withNullability(true)) ++ right.output
      case FullOuter =>
        (left.output ++ right.output).map(_.withNullability(true))
      case j: ExistenceJoin =>
        left.output :+ j.exists
      case LeftExistence(_) =>
        left.output
      case x =>
        throw new IllegalArgumentException(
          s"${getClass.getSimpleName} not take $x as the JoinType")
    }
  }

@cloud-fan Output determines that we cannot simply put streaming side left and build side right.
Am I understanding correctly?

Yea that's why we have a createResultProjection to reorder the columns.

Ah, I think I understand now:

In all of the *Join methods in this file (which doesn't include full outer join), the streamed side is always the left side of the joinRow that is passed to boundCondition: in HashJoin.scala, the input to boundCondition is not necessarily the same as the output of the join operator itself (for that, we have resultProj which comes from createResultProjection which remains in sync with boundCondition (source).

However, boundCondition is inherited in ShuffledHashJoinExec and there we have a buildSideOrFullOuterJoin method (source) and there the resultProjection operates over the same input schema as the join output schema (rather than assuming that the joined row always has the streamed side on the left). In that code, boundCondition is evaluated over an input that matches the join output schema and there the streamed side could be on either side rather than only the left.

I think this inheritance is confusing and hard to reason about.

It seems like HashJoin.scala has an invariant of "streamed side always on left" which gets violated in ShuffleHashJoin.scala's separate implementation of outer joins.

I wonder whether we can address this bug by modifying ShuffledHashJoinExec.buildSideOrFullOuterJoin so that it always unconditionally places the streamed side on the left (the same as HashJoin.scala's default). (Edit: I'm basically agreeing with @cloud-fan's suggestion above).

In the golden file test outer-join.sql, we run it multiple times to test different code paths: broadcast join, shuffle join, sort merge join. code on/off etc., can we move the test there?