villebro opened a new issue, #2484:
URL: https://github.com/apache/datafusion-ballista/issues/2484
## Motivation
When a job completes, the scheduler returns each output partition's
location, including the executor's Flight host and port. By default the client
fetches every partition directly from its executor. Where clients can't reach
executors, as is common on Kubernetes and in isolated networks, the scheduler
can proxy the fetches instead and relay the result bytes itself.
```
peer-to-peer: client ──fetch──▶ each executor
scheduler proxy: client ──fetch──▶ scheduler ──▶ executor (result bytes
relayed by the scheduler)
```
Neither mode fits these deployments well:
- **Result serving congests the scheduler.** Scheduling is latency-sensitive
coordination; result serving is bulk transfer that grows with result size and
client count. When the scheduler proxies, a large result or a slow client
degrades planning for every other query.
- **Result serving can't scale on its own.** The proxy scales only with the
scheduler (in practice, one process), and peer-to-peer capacity is tied to the
executor count. Trino spooling has workers write result segments in parallel
while the coordinator only hands out URLs.
- **There is nowhere to put tiered result storage.** A result lives only on
its producing executor's disk; if that executor is lost, the result goes with
it. Memory, cache, or object-storage tiers need a component that owns where a
result lives and how it is served.
- **Clients see internal topology.** Executor host and port appear in the
job-status response and in every fetch ticket, revealing node count, network
layout and reachable endpoints that serving doesn't need.
Each of these points to a dedicated result service, separate from the
scheduler and the executors.
## Proposal
We propose adding an optional, stateless **Result Service** (a new
`ballista-result-service` binary) between clients and executors. The
scheduler's embedded result proxy would be deprecated and later removed (see
below).
```
client ──(1) submit──▶ scheduler (control plane only)
▲ │
└──(2) partition locations + advertised result-service endpoint
(3) client ──fetch──▶ LB / ingress ──▶ Result Service (N stateless
replicas) ──▶ executor
```
Each fetch ticket already names the executor holding the partition, so a
replica only has to forward the fetch and stream the bytes back. It is
stateless and scales as a Kubernetes Deployment behind a Service. Clients reach
it through the scheduler's existing `advertise_flight_endpoint` setting. Flight
uses long-lived gRPC streams, so the load balancer or ingress in front of the
fleet must be gRPC-aware. Since the endpoint is a plain `host:port`, path-based
routing isn't supported.
Result serving moves into `ballista-core` behind a small backend trait,
`fetch(ticket) -> stream`, so that the executor and the Result Service share
one implementation. The Result Service adds a forwarding backend, and future
storage tiers become further backends.
Once the Result Service exists, the scheduler's embedded proxy is redundant.
Keeping it indefinitely would mean maintaining a second proxy and would leave a
supported configuration in which the scheduler remains on the data path.
However, removing it at the same time the Result Service is introduced would
leave no compatibility or rollback window for existing deployments.
The transition therefore spans two major releases:
1. In the first major release, add the Result Service and deprecate
`enable_embedded_flight_proxy`. The embedded proxy continues to work, but emits
a startup warning directing operators to deploy a Result Service and configure
`advertise_flight_endpoint`. Documentation and deployment examples recommend
the Result Service for Kubernetes and isolated networks. Both configurations
remain supported for the whole release cycle, so deployments can migrate,
compare behavior under production load, and roll back without downgrading
Ballista.
2. In the following major release, remove `enable_embedded_flight_proxy` and
the scheduler's Flight proxy implementation. `advertise_flight_endpoint`
remains the way to point clients at a Result Service. Deployments that still
use the embedded proxy must run at least one Result Service replica before
upgrading.
Clients, the wire format, executors and peer-to-peer mode stay unchanged
throughout the transition. The existing `local` variant in
`GetJobStatusResult.flight_proxy` should remain reserved or accepted for wire
compatibility even after schedulers stop emitting it.
The eventual removal of `enable_embedded_flight_proxy` is the breaking
change, announced one major release in advance. The Result Service is opt-in
during the deprecation release and becomes the supported proxy mode after
removal.
Authentication on the result fetch
path is an existing gap that applies equally to executors today, so we'd
discuss it separately.
## What this opens up
This first step fixes congestion and independent scaling. The other two
motivations, tiered storage and topology hiding, now have somewhere to land,
but they are separate discussions and are not proposed here.
The most visible gap is that clients still receive executor locations,
because the fleet just forwards whatever each ticket names. The end state is an
opaque handle that the service resolves to a location. It would be a new result
descriptor rather than a change to `PartitionLocation`, which is shared with
shuffle. Our preference is to resolve handles against the shared state that
scheduler HA will introduce, which puts this step after that work.
Storage tiers follow naturally. Forwarding is a relay, not a durability
layer: if the executor is lost before the fetch, the fetch fails just as it
does with peer-to-peer. Executor memory, a distributed cache, and object
storage with pre-signed URLs would each be a backend behind the same trait.
Pre-signed URLs would let clients pull large results straight from storage
without ever seeing an executor. These tiers bring lifecycle work, such as
TTLs, deletion after fetch, and bounded memory budgets, and could share a
storage layer with remote 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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
