[AURON #2390] Preserve key-grouped partitions in native Iceberg scans

[lyne7-sc]

Which issue does this PR close?

Closes #2390

Rationale for this change

Native Iceberg scans did not preserve Spark's key-grouped partition layout. The native input partitions could therefore differ from the declared KeyGroupedPartitioning, causing missing rows or task creation failures in storage-partitioned joins.

What changes are included in this PR?

• Preserve Spark's key-grouped partitions, including empty partition groups, when creating native Iceberg scan tasks.
• Add regression tests for storage-partitioned joins.

Are there any user-facing changes?

Bug fix only.

How was this patch tested?

Add test cases to AuronIcebergIntegrationSuite.

Was this patch authored or co-authored using generative AI tooling?

• Yes
• No

Generated-by: OpenAI Codex (gpt-5.6-sol)

If yes, include: Generated-by: <tool name and version>

ASF guidance: https://www.apache.org/legal/generative-tooling.html

[weiqingy]

runtimeFilters.nonEmpty and "runtime filtering actually happens" come apart in one case. With spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly on (the default) and no reusable broadcast exchange to match, PlanDynamicPruningFilters substitutes DynamicPruningExpression(Literal.TrueLiteral) (v3.5.8, PlanDynamicPruningFilters.scala:78-80) — a filter that filters nothing. The AQE path does the same at PlanAdaptiveDynamicPruningFilters.scala:66-67, so this isn't limited to non-adaptive execution. Spark treats that value as absent in two places: doCanonicalize strips it explicitly (BatchScanExec.scala:237-239), and filteredPartitions translates runtime filters through DataSourceV2Strategy.translateRuntimeFilterV2, which only matches InSubqueryExec (DataSourceV2Strategy.scala:644-655) — so TrueLiteral translates to nothing, filteredPartitions returns partitions unfiltered, and the groups can't change.

Where that lands: select ... from fact f join dim d on f.p = d.p where d.x = 1, both tables Iceberg-partitioned on p, gets SPJ precisely because it's a sort-merge join (dim too large to broadcast). No BHJ ⇒ no reusable exchange ⇒ the DPP filter degrades to TrueLiteral ⇒ runtimeFilters.nonEmpty is true ⇒ the whole scan drops back to Spark, with zero runtime filtering having been performed.

That costs native coverage rather than correctness — on master this query went native and returned wrong rows, so this is better either way. Is the wider guard deliberate, or would mirroring Spark's own doCanonicalize predicate be closer to the intent? One option, in case it's useful:

val hasEffectiveRuntimeFilters = exec.runtimeFilters
  .exists(_ != DynamicPruningExpression(Literal.TrueLiteral))
if (hasEffectiveRuntimeFilters) None else keyGroupedInputPartitions(exec, partitioning)…

(Literal is already imported at :30; only DynamicPruningExpression would need adding.)

Related: would a test pinning this branch help? Nothing in the suite reaches it today — spark.sql.sources.v2.bucketing.enabled appears only in the two new tests, and every DPP test carries an explicit /*+ BROADCAST(d) */ hint (:281-340, and also :357 / :675), which forces the reusable exchange and a real InSubqueryExec. An SPJ query carrying runtime filters, asserting a clean fallback (right answer, no NativeIcebergTableScanExec, no exception), would pin the intended behavior — and if it turns out awkward to write because TrueLiteral makes the fallback fire where you'd expect native execution, that's the answer to the question above.