rekbun commented on PR #59522: URL: https://github.com/apache/spark/pull/59522#issuecomment-6323703811
Thanks for the careful review, Peter, and for the context on Iceberg. That is helpful. The source where we hit this is an in-house DataSource V2 connector for Hive-format bucketed tables. The tables are partitioned by hour and bucketed by a key, and the reported partitioning contains the bucket key, not the hour. The connector currently emits one input per bucket file to retain its per-file sort order. For example, with one file per bucket in each hourly partition, reading a day produces 24 inputs with the same reported key, and a week produces 168. Before the refactor, Spark grouped those inputs by key, so these scans ran one task per bucket. After it, in the affected queries that still require a downstream shuffle, each input can execute as a separate scan task. That increased task setup, execution, and scheduling overhead in existing production jobs, with higher driver memory pressure and longer runtimes. I am not proposing a change to the default. Multiple splits can be an intentional sizing decision, and grouping a large or skewed key into one task can remove useful parallelism. I would like to keep the previous grouping policy reachable as an explicit, default-off compatibility option for users upgrading Spark whose connectors and workloads relied on it. If a legacy name would make that purpose clearer, I am happy to use one. COALESCE is a valid workaround when placed before an existing shuffle. For restoring the historical task count, the bucket count provides a natural target. The difficulty is applying that workaround across existing queries and choosing suitable parallelism as data volume and skew change. Generic coalescing is not key-aware, and placing it between a scan and a consumer that relies on keyed partitioning can introduce an exchange. Connector-side packing is also a reasonable direction, and we are investigating bounded packing. Our connector reports per-file ordering, so combining files requires either merging their sorted streams or withdrawing that ordering claim and letting Spark add a sort where needed. Packing can be implemented in the connector. The compatibility argument for Spark is that physical planning knows which partitioning and ordering properties the surrounding plan requires, while the connector does not generally see the final physical plan when choosing its inputs. Size-aware packing using HasPartitionStatistics.sizeInBytes() is a promising longer-term direction for addressing both small tasks and oversized tasks for large or skewed keys. I would be glad to work on that with you. The opt-in would provide a compatibility option while that is developed. I take the point about complexity. I’ll narrow the PR to the default-off scan-grouping option, preserve required ordering and existing SPJ behavior, and keep unrelated planning changes separate. The public reproducer shows 48 scan tasks becoming two, with the same results and the required shuffle retained. Would that narrower scope address your concern? -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
