gavin9402 commented on PR #10457: URL: https://github.com/apache/paimon/pull/10457#issuecomment-6092524047
> [P2] Avoid reducing existing downstream parallelism when adding the shuffle > > At [RepartitionLargePaimonScan.scala:75-81](https://github.com/apache/paimon/blob/2f0c354a75b0946ad01a5bdfbeaff40da0b4f9e4/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/optimizer/RepartitionLargePaimonScan.scala#L75-L81), an oversized input partition triggers a new shuffle even when `ceil(total bytes / targetSize)` is smaller than the current input-partition count. Since this option applies to the whole session, an otherwise well-parallelized scan can lose downstream concurrency when one large Blob partition is mixed with many small partitions. > > I reproduced this with a real Spark 3 Paimon data-evolution table partitioned by `id`, a Blob `payload`, 21 table partitions (20 one-byte payloads and one 256 KiB payload), `source.split.target-size=64kb`, default file-open cost, and AQE disabled. For `SELECT id, length(payload)`, the option disabled gives 21 RDD partitions; enabled gives 5. All row results are preserved, but all 21 source tasks remain and the added shuffle reduces the downstream stage to 5 tasks. For expensive, roughly uniform per-row UDF/map work on a cluster with more than five available cores, that caps concurrency below the existing plan. I have not measured a runtime slowdown; the observed regression is the reduction in available downstream parallelism. > > Please floor the computed count at the existing input-partition count, or skip insertion when it would not increase parallelism. The 14 existing rule tests pass, and a real 100-row Blob query also demonstrated the intended 2-to-15 downstream increase with identical results; adding the skewed-table case would cover this missing guard. > > Reviewed head: `2f0c354a75b0946ad01a5bdfbeaff40da0b4f9e4`. Thanks for the detailed reproduction. Updated the rule to skip repartitioning when the computed partition count is less than or equal to the existing input-partition count. In that case, the original scan is preserved without an additional shuffle. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
