sunchao opened a new pull request, #3745: URL: https://github.com/apache/celeborn/pull/3745
Supersedes #3687. ### What changes were proposed in this pull request? Add optional partition-range fields to `GetReducerFileGroup` requests and responses. For Spark reads, executors now request only `[startPartition, endPartition)` and cache the ranges they have loaded. The driver filters reducer file groups, successful partition IDs, and failed-batch metadata to that range. The first request still fetches mapper attempts; later range requests omit that shuffle-wide array. Scoped responses bypass the shuffle-wide RPC cache and Spark broadcast path. Legacy full-shuffle requests keep the existing behavior. The client also: - shares concurrent loads for the same range; - allows unrelated warm ranges to load independently; - invalidates in-flight loads safely during shuffle cleanup; - preserves interrupt behavior; and - treats responses from older drivers as shuffle-wide metadata. ### Why are the changes needed? `GetReducerFileGroup` currently returns metadata for every reducer in a shuffle. With very large reducer counts, each executor downloads and materializes the same shuffle-wide response even though a task reads only a small partition range. This can create large driver responses, executor memory pressure, and client-side load contention. Scoping the response to the task's partition range makes transfer and executor cache size proportional to the data that executor actually reads. JIRA: https://issues.apache.org/jira/browse/CELEBORN-2370 ### Does this PR introduce _any_ user-facing change? Yes. Spark clients request task-specific reducer metadata instead of shuffle-wide metadata when the driver supports the new optional fields. The protocol is backward compatible: - a new client with an old driver detects the legacy full response and caches it as shuffle-wide; - an old client with a new driver continues using the existing full-response cache and broadcast path. Flink continues to use the legacy full-shuffle API. ### How was this patch tested? - `ShuffleClientSuiteJ`: 29 tests - `UtilsSuite`: 28 tests, including V1 and V2 protocol round trips - `ReducerFileGroupFilterSuite`: 2 tests - `ConfigurationSuite`: 8 tests - Spark 4.0 / Scala 2.13 package compile - Flink 1.20 common client package compile - Spotless apply/check -- 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]
