[BUG] test_broadcast_nested_loop_join_special_case fails on databricks

[tgravescs]

Running integration tests on Databricks, I'm seeing test_broadcast_nested_loop_join_special_case fail:

FAILED src/main/python/join_test.py::test_broadcast_nested_loop_join_special_case[String][IGNORE_ORDER({'local': True})]
FAILED src/main/python/join_test.py::test_broadcast_nested_loop_join_special_case[Byte][IGNORE_ORDER({'local': True})]
FAILED src/main/python/join_test.py::test_broadcast_nested_loop_join_special_case[Short][IGNORE_ORDER({'local': True})]
FAILED src/main/python/join_test.py::test_broadcast_nested_loop_join_special_case[Integer][IGNORE_ORDER({'local': True})]
FAILED src/main/python/join_test.py::test_broadcast_nested_loop_join_special_case[Long][IGNORE_ORDER({'local': True})]
FAILED src/main/python/join_test.py::test_broadcast_nested_loop_join_special_case[Boolean][IGNORE_ORDER({'local': True})]
FAILED src/main/python/join_test.py::test_broadcast_nested_loop_join_special_case[Date][IGNORE_ORDER({'local': True})]
FAILED src/main/python/join_test.py::test_broadcast_nested_loop_join_special_case[Timestamp][IGNORE_ORDER({'local': True})]
FAILED src/main/python/join_test.py::test_broadcast_nested_loop_join_special_case[Float][IGNORE_ORDER({'local': True}), INCOMPAT]
FAILED src/main/python/join_test.py::test_broadcast_nested_loop_join_special_case[Double][IGNORE_ORDER({'local': True}), INCOMPAT]

[revans2]

The error is

20/07/29 18:58:27 WARN TaskSetManager: Lost task 0.0 in stage 58.0 (TID 49, ip-10-59-230-224.us-west-2.compute.internal, executor driver): java.lang.ArrayIndexOutOfBoundsException: 0
	at ai.rapids.cudf.Table.<init>(Table.java:52)
	at com.nvidia.spark.rapids.GpuColumnVector.from(GpuColumnVector.java:245)
	at org.apache.spark.sql.rapids.execution.GpuBroadcastNestedLoopJoinExecBase.$anonfun$doExecuteColumnar$5(GpuBroadcastNestedLoopJoinExec.scala:235)
	at com.nvidia.spark.rapids.Arm.withResource(Arm.scala:26)
	at com.nvidia.spark.rapids.Arm.withResource$(Arm.scala:24)
	at org.apache.spark.sql.rapids.execution.GpuBroadcastNestedLoopJoinExecBase.withResource(GpuBroadcastNestedLoopJoinExec.scala:134)
	at org.apache.spark.sql.rapids.execution.GpuBroadcastNestedLoopJoinExecBase.builtTable$lzycompute$1(GpuBroadcastNestedLoopJoinExec.scala:234)
	at org.apache.spark.sql.rapids.execution.GpuBroadcastNestedLoopJoinExecBase.builtTable$2(GpuBroadcastNestedLoopJoinExec.scala:233)
	at org.apache.spark.sql.rapids.execution.GpuBroadcastNestedLoopJoinExecBase.$anonfun$doExecuteColumnar$7(GpuBroadcastNestedLoopJoinExec.scala:249)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2(RDD.scala:844)
	at org.apache.spark.rdd.RDD.$anonfun$mapPartitions$2$adapted(RDD.scala:844)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:356)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:320)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:356)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:320)
	at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
	at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:356)
	at org.apache.spark.rdd.RDD.iterator(RDD.scala:320)
	at org.apache.spark.shuffle.ShuffleWriteProcessor.write(ShuffleWriteProcessor.scala:59)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:99)
	at org.apache.spark.scheduler.ShuffleMapTask.runTask(ShuffleMapTask.scala:52)
	at org.apache.spark.scheduler.Task.doRunTask(Task.scala:144)
	at org.apache.spark.scheduler.Task.run(Task.scala:117)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$9(Executor.scala:639)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1559)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:642)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

[revans2]

Looks like they are doing a join where one of the sides has no columns, just rows. In this case I think we need to duplicate the table that does have rows N times where N is the number of rows in the join table with no columns.

[revans2]

I think to do this properly we want to put in the cudf repeat API.