andygrove opened a new pull request, #2813: URL: https://github.com/apache/datafusion-ballista/pull/2813
# Which issue does this PR close? Closes #1851. # Rationale for this change Every `PartitionLocation` holds its own copy of the executor's full `ExecutorMetadata`, but readers only use its `id`, `host` and `port`. A stage that reads many partitions from a few executors repeats the same strings once per location, in scheduler and executor memory and in the task plans sent over gRPC, where they count against the message size limit. This follows the design @milenkovicm and @sandugood worked out in #1853, which went stale while waiting on review. # What changes are included in this PR? - **In memory:** a new `ExecutorEndpoint` holds just `id`, `host` and `port`, and `PartitionLocation` and `PipelinedPartitionLocation` hold it in an `Arc`. The scheduler creates one per status update rather than one per location, and a decoded reader has one per executor. - **In task plans:** the shuffle, range shuffle and pipelined shuffle reader nodes list each executor once, and each location refers to one by position. An out-of-range position is a decode error. This changes the wire format between schedulers and executors, so `BALLISTA_PROTOCOL_VERSION` goes from 3 to 4. - **In job status:** every location still carries its executor, now with just those three fields, so the Python, Flight SQL and Rust clients read it as before. #1853 discussed moving this to a table as well. It stays per location here because a job status has about one location per output partition, so a table would save little and every client would have to resolve positions. The protobuf `ExecutorMetadata` message is renamed `ExecutorEndpoint` and loses `grpc_port`, `specification` and `os_info`, which nothing reads from a location. The generated Python stubs are updated to match. The encoded size and decoded heap of one `ShuffleReaderExec`, using UUID executor ids and the default OS info, before and after this change: | Locations | Executors | Encoded before | Encoded after | Decoded heap before | Decoded heap after | | --- | --- | --- | --- | --- | --- | | 128 × 128 | 16 | 2.96 MB | 0.56 MB | 6.7 MB | 2.2 MB | | 512 × 512 | 8 | 47.6 MB | 9.2 MB | 107.5 MB | 35.4 MB | On a local cluster of one scheduler and two executors, TPC-H SF1 results matched single-process DataFusion (`tpch benchmark ballista --verify`) for all 22 queries under both push and pull scheduling, and for queries 1, 3, 5, 10, 12 and 18 in pipelined mode. # Are there any user-facing changes? Yes. This changes the public API and the wire format, and the 56.0.0 upgrade guide covers both: - `PartitionLocation::executor_meta` and `PipelinedPartitionLocation::executor_meta` are now `Arc<ExecutorEndpoint>`. - The protobuf message is renamed as described above, and the location messages gain `executor_index`. - Schedulers and executors must be upgraded together. -- 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]
