Bigtable Join Pushdown should be visible in Spark Explain Plan

[maheshgoyal15]

Feature Request:

Currently, the Bigtable Join Pushdown feature, intended to pass down row keys from the source DataFrame as filters, isn't explicitly shown in the physical plan. The plan displays a BroadcastHashJoin rather than indicating that the join optimization is happening directly within Bigtable. Even if the join is pushed down, the physical plan doesn't clearly reflect this optimization.

Code Snippet (problematic section):

The relevant code is within the JoinPushDown.scala file:

import com.google.cloud.spark.bigtable.join.BigtableJoinImplicit._

println("\n--- Performing and Explaining Joins ---")

// ResDf1
println("\n----- resDf1 (single join key string) -----")
val resDf1 = srcDf.joinWithBigtable(joinConfig, "word")
println("resDf1 explain:")
resDf1.explain(true) // Use explain(true) for extended plan
println("resDf1 output:")
resDf1.show(5, truncate = false) // Show output

Plan output :

The explain(true) output below shows BroadcastHashJoin instead of a Bigtable-specific join operator:

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [word#15, src_count#55, frequency_binary#72, count#73L]
   +- BroadcastHashJoin [word#15], [word#74], Inner, BuildLeft, false
     :- BroadcastExchange HashedRelationBroadcastMode(List(input[0, string, false]),false), [plan_id=167]
     :  +- Filter isnotnull(word#15)
     :    +- GlobalLimit 100, 0
     :      +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [plan_id=163]
     :        +- LocalLimit 100
     :          +- Project [_1#8 AS word#15, _2#9 AS src_count#55]
     :            +- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, knownnotnull(assertnotnull(input[0, scala.Tuple3, true]))._1, true, false, true) AS _1#8, knownnotnull(assertnotnull(input[0, scala.Tuple3, true]))._2 AS _2#9]
     :              +- Scan[obj#7]
     +- Filter isnotnull(word#74)
       +- Scan ExistingRDD[frequency_binary#72,count#73L,word#74]

Expected Behavior:

For the joinWithBigtable method to effectively demonstrate pushdown, the physical plan should indicate that the join operation is being performed at the Bigtable such as BigtableJoin, rather than by Spark itself. This would typically manifest as a specialized Bigtable scan or join operation within the plan, without the presence of BroadcastHashJoin or similar Spark-side join operators for the Bigtable portion of the join.