milenkovicm opened a new issue, #2843:
URL: https://github.com/apache/datafusion-ballista/issues/2843
## Is your feature request related to a problem or challenge?
Ballista can run a query in two ways:
- **Staged execution:** each stage writes its shuffle output in full before
the next stage starts. It recovers well from failures and suits long, heavy
jobs that write results somewhere, such as ETL, `COPY TO` and `INSERT INTO`.
- **Pipelined execution:** stages stream data to each other, so results come
back sooner. It suits interactive queries, where the user wants rows back fast
and the result goes to the client.
Today users must pick one mode for the whole session. Most sessions run a
mix of both kinds of query. Users either accept slower interactive queries or
lose fault tolerance on big write jobs. Users also must know about the two
modes and the trade-off between them in order to set the option at all.
## Describe the solution you'd like
### 1. Add an `auto` execution mode
```
ballista.execution.mode = staged | pipelined | auto
```
With `auto`, the scheduler picks the mode for each job when the job is
submitted.
### 2. Default heuristic: does the plan write output?
The built-in strategy treats a query that writes its result somewhere as
batch work, and any other query as interactive:
| Plan contains | Treated as | Mode |
|---|---|---|
| `CopyTo` node (`COPY ... TO ...`, `DataFrame::write_*`) | batch / ETL |
**staged** |
| No `CopyTo` node | interactive | **pipelined** |
This is cheap to check (one walk of the plan tree), easy to predict, and
easy to explain to users.
### 3. Make the strategy pluggable
Deployments differ: one might want to choose by estimated data size, number
of stages, tenant or job tags. So the decision should sit behind a trait, with
the default heuristic as one implementation of it:
```rust
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ExecutionMode {
Staged,
Pipelined,
}
/// Decides how a job runs when `ballista.execution.mode = auto`.
pub trait ExecutionModeStrategy: Send + Sync + std::fmt::Debug {
fn select_mode(
&self,
plan: &LogicalPlan, // or the physical plan (see open
questions)
config: &SessionConfig,
) -> datafusion::error::Result<ExecutionMode>;
}
/// Default: if the plan contains CopyTo, use staged; otherwise use
pipelined.
#[derive(Debug, Default)]
pub struct CopyToBasedStrategy;
```
Users register their own strategy the same way they register other Ballista
extensions, for example through the scheduler config or session state builder:
```rust
let config = SchedulerConfig::default()
.with_execution_mode_strategy(Arc::new(MyStrategy::new()));
```
Behaviour:
- `staged` or `pipelined` set explicitly: the strategy is not called. This
is the current behaviour.
- `auto`: the scheduler calls the registered strategy, or
`CopyToBasedStrategy` if none is registered.
- The chosen mode is logged and shown in job status, so users can see why a
job ran the way it did.
## Describe alternatives you've considered
- **Keep the mode as a session setting only:** this is simple, but it pushes
the decision onto users and handles mixed workloads poorly.
- **Hard-code the heuristic with no trait:** this is easier to build, but
many deployments will need different rules, so it would likely have to be
reworked later.
- **Choose using cost or statistics:** this could pick better, but it is
much harder to predict. It can be built later as a custom strategy on top of
this proposal.
## Open questions
1. **Logical or physical plan:** should the strategy look at the
`LogicalPlan`, where `CopyTo` is easy to spot, or the `ExecutionPlan`, where
`DataSinkExec` also covers `INSERT INTO`? Or should it get both?
2. **DML:** should `INSERT INTO` and other `DataSinkExec` writes count as
batch work in the default strategy, as `CopyTo` does?
3. **Where the decision is made:** on the client before submission, or on
the scheduler? Deciding on the scheduler keeps the choice in one place and lets
operators enforce rules for every client.
4. **Fallback:** if a pipelined job fails, should the scheduler retry it in
staged mode?
5. **Default value:** should `auto` become the default once it has proved
stable?
## Additional context
This change is additive. Existing configs that set `staged` or `pipelined`
keep working as before.
--
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]
