[tlfua]

Which issue does this PR close?

Closes #6815.

Rationale for this change

With AQE off, CometExecRule.rewriteInSubqueryPlan only converts a DPP SubqueryBroadcastExec to a CometSubqueryBroadcastExec when the broadcast's child is a CometNativeExec. The join's own broadcast converts to a CometBroadcastExchangeExec over more plans than that, because CometUnionExec, CometCoalesceExec, CometTakeOrderedAndProjectExec and CometShuffleExchangeExec are wrapped in a native placeholder while the join is converted. For those build sides the DPP subquery keeps a Spark BroadcastExchange, ReuseExchangeAndSubquery can't match it to the join's exchange, and the dimension is broadcast twice.

The AQE path already accepts any columnar CometPlan in CometPlanAdaptiveDynamicPruningFilters.isNativeBuildSide (#6786), so this only changes the non-AQE path.

What changes are included in this PR?

• CometExecRule.rewriteInSubqueryPlan now requires CometPlan && supportsColumnar instead of CometNativeExec, the same check as isNativeBuildSide. CometBroadcastExchangeExec only needs a child that supports executeColumnar(), so a child admitted by this check runs correctly. If the DPP side converts a plan the join side didn't, the only effect is that the two exchanges don't match and reuse is lost, as before.
• The same check now also requires spark.comet.exec.broadcastExchange.enabled. With it off, the join's broadcast stays a Spark BroadcastExchangeExec, so converting the DPP subquery would leave two exchanges that can't match and broadcast the dimension twice. The relaxed check above would have made that happen for more build sides, and it already happened for a CometNativeExec build side before this PR. The AQE path gets the same result by requiring a Comet join (isCometJoin).
• Updated the comment in CometPlanAdaptiveDynamicPruningFilters that pointed to the old non-AQE check, and a stale comment in an existing test that named CometNativeExec.

Build sides this change newly admits:

Top operator of the build side	Tested	Why not	How the reason was checked
CometUnionExec	yes
CometCoalesceExec	yes
CometShuffleExchangeExec	yes
CometTakeOrderedAndProjectExec	no	The join adds an isnotnull filter on a nullable key, and the optimizer can't push it below the limit, so a native CometFilter is the top operator	Seen in the plan of an ORDER BY ... LIMIT case in fork CI run https://github.com/tlfua/datafusion-comet/actions/runs/42359642929
CometCollectLimitExec	no	Spark plans CollectLimit only at the root of a query	Reasoned from the planner, not reproduced
CometLocalTableScanExec, CometEmptyRelation	no	The optimizer folds these away, and DPP needs a selective filter on the build side	Reasoned from the planner, not reproduced
CometSparkToColumnarExec, CometScanExec, CometBatchScanExec, CometInMemoryTableScanExec	no	DPP needs a selective filter on the build side, and Comet runs that filter natively above the scan, so these are not the top operator	Reasoned from the planner, not reproduced

How are these changes tested?

Added three tests to CometExecSuite, one for each tested shape in the table above. Each runs a non-AQE broadcast join with DPP in reuse-only mode. Each test asserts these things:

• The answer matches Spark.
• There is exactly one CometBroadcastExchangeExec.
• The broadcast's direct child is the expected operator, so the test can't pass on a build side that was already a CometNativeExec.
• The DPP subquery is a CometSubqueryBroadcastExec.
• The DPP subquery reuses the join's broadcast.

Added a fourth test that runs the UNION ALL case with spark.comet.exec.broadcastExchange.enabled=false. It asserts there is no CometBroadcastExchangeExec, the DPP subquery is a Spark SubqueryBroadcastExec, and it reuses the join's only BroadcastExchangeExec. The four tests share a helper that writes the tables.

No TPC-DS approved plan has a Spark BroadcastExchange under a SubqueryBroadcast, so no golden files change.

Local checks:

• ./mvnw compile test-compile scalafix:scalafix -Psemanticdb and ./mvnw spotless:check: passed.
• I did not run the new tests locally.

Fork CI: https://github.com/tlfua/datafusion-comet/actions/runs/42395827505 (passed; the [exec] job ran all four new tests). In an earlier run, an ORDER BY ... LIMIT case showed the CometFilter described in the table above, so I removed that case.

Follow-up

If the config is on but the join itself stays on Spark, for example with spark.comet.exec.broadcastHashJoin.enabled=false, the join's broadcast still converts to a Comet exchange and the non-AQE rule can't see the join to decide. That case predates this PR and isn't handled here.