[BUG] data corruption in GpuBroadcastNestedLoopJoin with empty relations edge case

[abellina]

I am testing a patch for empty relations for the broadcast* joins and ran into an edge case that is a bug in 21.12 where we return nothing when the CPU returns results, or the other way around (we return a result where the CPU returns nothing), this is specific to unconditional joins.

This test can reproduce the issue:

@ignore_order(local=True)
@pytest.mark.parametrize('join_type', ['Left', 'Inner', 'LeftSemi', 'LeftAnti', 'Cross'], ids=idfn)
@pytest.mark.parametrize('batch_size', ['100', '1g'], ids=idfn)
def test_right_broadcast_nested_loop_join_without_condition_empty(join_type, batch_size):
    def do_join(spark):
        left, right = create_df(spark, long_gen, 50, 0)
        return left.join(broadcast(right), how=join_type)
    conf = copy_and_update(allow_negative_scale_of_decimal_conf, {'spark.rapids.sql.batchSizeBytes': batch_size})
    assert_gpu_and_cpu_are_equal_collect(do_join, conf=conf)

This will fail for BuildRight: Left, LeftSemi and LeftAnti. I think I have a good idea about where to fix for the LeftSemi case anyway. We currently just return the left relation, whereas spark has other logic here:

  private def leftExistenceJoin(
      relation: Broadcast[Array[InternalRow]],
      exists: Boolean): RDD[InternalRow] = {
    assert(buildSide == BuildRight)
    streamed.execute().mapPartitionsInternal { streamedIter =>
      val buildRows = relation.value
      val joinedRow = new JoinedRow

      if (condition.isDefined) {
        streamedIter.filter(l =>
          buildRows.exists(r => boundCondition(joinedRow(l, r))) == exists
        )
      } else if (buildRows.nonEmpty == exists) {
        streamedIter
      } else {
        Iterator.empty
      }
    }
  }

I think this also covers LeftAnti, but I haven't tested it.