[FEA][Follow On] Support non AST rewrite as join condition

[winningsix]

Is your feature request related to a problem? Please describe.
There're some improvements we can make for AST rewrite.

• Support conditional equi-joins with non-AST,
  • broadcast hash join rewrite [FEA] support split join condition for broadcast hash join into AST and non-AST #9759
• Bump up cudf version and update plugin accordingly to support latest AST features. This includes: [FEA] Add string comparison to AST expressions #8157 and [FEA] Add AST for safe upcasts #9645)
• Fallback message improvement: it's hardly to tell the actually fallback reason in GpuOverrides warning message.
  From spark.time(left.join(right, length(col("leftstr")) + length(col("rightstr")) >= left("id") + right("id"), joinType="leftouter").count()), it reports following message:

          !Expression <Cast> cast((length(leftstr#301) + length(rightstr#296)) as bigint) cannot run on GPU because AST is required and this expression does not support AST
              @Expression <Add> (length(leftstr#301) + length(rightstr#296)) could run on GPU
                !Expression <Length> length(leftstr#301) cannot run on GPU because AST is required and this expression does not support AST
                  !Expression <AttributeReference> leftstr#301 cannot run on GPU because AST is required and expression AttributeReference leftstr#301 produces an unsupported type StringType
                !Expression <Length> length(rightstr#296) cannot run on GPU because AST is required and this expression does not support AST
                  !Expression <AttributeReference> rightstr#296 cannot run on GPU because AST is required and expression AttributeReference rightstr#296 produces an unsupported type StringType

It's more straightforward to AST failure reason results from 1) AST not supported by cast; 2) Failed to rewrite AST condition since contains cross childern attributes. Other message can share the same reason (e.g., cannot run on GPU because cast in this case).

• Update AST join split condition code to avoid idempotent issues [BUG] Update AST join split condition code to avoid idempotent issues #9680
• Java AST String literal does not provide a desired encoding [BUG] java AST String literal does not provide a desired encoding rapidsai/cudf#14397

Describe the solution you'd like
Extract non AST-able condition and push down to child nodes for conditional equal equal.

Describe alternatives you've considered
More feature coverage in Cudf. But it needs more time and some cases are not likely to be implemented.

Additional context
This is a tracking issue for some follow-ups of #9635 which introduced the AST basic framework and related support on BroadcastNestedLoopJoin.

[winningsix]

Regards to "Support conditional equi-joins with non-AST rewrite" above, it seems not applicable to have non-ast-split there. It seems we should close it.

Given this left.join(broadcast(right), f.round(left.a).cast('integer') == f.round(f.log(right.r_a).cast('integer')), join_type) as an example, join condition will be split into left and right key expressions in GpuBroadcastHashJoinExec.scala

spark-rapids/sql-plugin/src/main/spark340/scala/org/apache/spark/sql/rapids/execution/GpuBroadcastHashJoinExec.scala

Lines 65 to 66 in 792b4c1

	leftKeys: Seq[Expression],
	rightKeys: Seq[Expression],

And it will be evaluated in separated project nodes for build and stream sides accordingly.

Build side:

spark-rapids/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/execution/GpuHashJoin.scala

Lines 324 to 327 in 792b4c1

	withResource(GpuProjectExec.project(built.getBatch, boundBuiltKeys)) { builtKeys =>
	// ensure that the build data can be spilled
	built.allowSpilling()
	joinGatherer(builtKeys, built, streamBatch)

Stream side:

spark-rapids/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/execution/GpuHashJoin.scala

Lines 394 to 398 in 792b4c1

	withResource(GpuProjectExec.project(streamCb.getBatch, boundStreamKeys)) { streamKeys =>
	// ensure we make the stream side spillable again
	streamCb.allowSpilling()
	joinGatherer(buildKeys, LazySpillableColumnarBatch.spillOnly(buildData), streamKeys, streamCb)
	}

The remain case I came up is that a join condition involves columns across both sides. In that case, it's also not a split-able case.

Explained Plan

GpuBroadcastHashJoin [gpuround(a#22, 0, IntegerType)], [gpuround(cast(LOG(cast(r_a#30 as double)) as int), 0, IntegerType)], Cross, GpuBuildRight
:- GpuRowToColumnar targetsize(104857600)
:  +- *(1) Scan ExistingRDD[a#22,b#23]
+- GpuBroadcastExchange HashedRelationBroadcastMode(List(cast(round(cast(ln(cast(input[0, int, true] as double)) as int), 0) as bigint)),false), [plan_id=107]
   +- GpuProject [a#26 AS r_a#30, b#27 AS r_b#33]
      +- GpuRowToColumnar targetsize(104857600)
         +- *(2) Scan ExistingRDD[a#26,b#27]

@revans2 @jlowe any thoughts here?