andygrove commented on code in PR #2438: URL: https://github.com/apache/datafusion-ballista/pull/2438#discussion_r3993825448
########## ci/scripts/rust_clippy.sh: ########## @@ -18,11 +18,13 @@ # under the License. set -ex -cargo clippy --all-targets --package ballista-core --all-features -- -D warnings -cargo clippy --all-targets --package ballista-scheduler --all-features -- -D warnings -cargo clippy --all-targets --package ballista-executor --all-features -- -D warnings -cargo clippy --all-targets --package ballista --all-features -- -D warnings +# Lint the whole workspace rather than an explicit crate list, so a new member +# is covered the day it is added. Naming crates one by one had already left +# ballista-api-types, ballista-history and ballista-chaos unlinted. +# +# ballista-cli is excluded here and linted separately below: its `tui` and `web` +# features target different platforms and are not meant to be enabled together, +# so --all-features is not meaningful for it. +cargo clippy --all-targets --workspace --exclude ballista-cli --all-features -- -D warnings Review Comment: Agreed on moving to `--workspace`, the explicit list had clearly gone stale. One side effect worth being aware of: `--workspace --all-features` now turns on `ballista-core`'s `force_hash_collisions` while linting every other crate, which the old per-package invocations did not do. It is lint-only so no test behaviour changes, just noting it in case a confusing cfg-dependent warning shows up later. ########## docs/source/contributors-guide/architecture.md: ########## @@ -204,22 +220,64 @@ compares to the pipelined shuffle used by engines such as DataFusion Distributed [shufflewriterexec]: https://github.com/apache/datafusion-ballista/blob/main/ballista/core/src/execution_plans/shuffle_writer.rs [shufflereaderexec]: https://github.com/apache/datafusion-ballista/blob/main/ballista/core/src/execution_plans/shuffle_reader.rs +### Multi-partition tasks + +Ballista dispatches at the _slice_ level, not the partition level. Each executor advertises a fixed number of +virtual cores (`vcores`), and the scheduler packs up to that many of a stage's output partitions into a single +task. All partitions in the slice execute concurrently under one DataFusion plan invocation: scans and shuffle +readers are rewritten to see only the assigned partition ids, and DataFusion's per-partition `execute(N)` +contract fans the work across the executor's threads. Slice size is bounded by the executor's free vcore count +and by `ballista.scheduler.max_partitions_per_task` (`0` = unbounded — the default; fills each task up to the +executor's free vcore count; `1` = one task per partition, the pre-multi-partition-tasks model). + +Compared to Apache Spark, whose unit of dispatch is one task per partition, Ballista's unit is one task per +slice of partitions bound to a single executor. Spark achieves cluster-scale parallelism the same way — many +tasks running concurrently across cores — but each task is single-threaded and doesn't share state with its +neighbours. Ballista's slice model preserves cluster-scale parallelism _and_ adds intra-task shared-memory +parallelism: partitions inside one slice share DataFusion's per-task memory pool budget, share the collect-left +build side of a broadcast hash join (one hash table probed by every partition, instead of one materialization +per task), share segment-tree indices needed for degenerate window aggregates (non-invertible aggregates like +MIN/MAX, or wide/data-dependent frames where a sliding accumulator degrades to O(n × frame)), and can cooperate +on shared-memory algorithms like PSRS parallel sort that a shuffle-based system can't express within a stage. +It also unlocks pipelines whose intra-task state must be global-per-slice — e.g. +`SELECT sum(v2) OVER (ORDER BY v2 RANGE 3 PRECEDING) FROM large`, which today collapses onto a single-partition +sort+window and OOMs at h2o 10 GB scale; with the KLL-adaptive range-repartition rewrite that builds on this +model, one slice-task per executor holds the sketch, buffered input, and per-partition halo state inside a +single plan. + +#### Composition with in-flight DataFusion AQE Review Comment: This subsection points at apache/datafusion#23026, which is still an open draft, and apache/datafusion#23167, which is closed and was never merged. Given that the whole thesis of this PR is that docs drift away from reality, I would rather not anchor a contributors-guide section to a closed PoC. The multi-partition tasks part above it is fine, that describes code we actually have. Could this subsection move to an issue instead? ########## docs/source/user-guide/configs.md: ########## @@ -190,8 +190,8 @@ _Example: Specifying configuration options when starting the scheduler_ | key | type | default | description | | -------------------------------------------- | ------ | ----------- | -------------------------------------------------------------------------------------------------------------------------- | -| scheduler-policy | Utf8 | pull-staged | Sets the task scheduling policy for the scheduler, possible values: pull-staged, push-staged. | -| event-loop-buffer-size | UInt32 | 10000 | Sets the event loop buffer size. for a system of high throughput, a larger value like 1000000 is recommended. | +| scheduler-policy | Utf8 | push-staged | Sets the task scheduling policy for the scheduler, possible values: pull-staged, push-staged. | +| event-loop-buffer-size | UInt32 | 1000 | Sets the event loop buffer size. for a system of high throughput, a larger value like 1000000 is recommended. | Review Comment: 1000 is correct for this table since it is the CLI default at `ballista/scheduler/src/config.rs:135`, so no change needed here. Worth knowing though that `SchedulerConfig::default()` at `ballista/scheduler/src/config.rs:429` is still 10000, so anyone constructing the config in Rust rather than via the CLI gets a different number than this now claims. That is a pre-existing inconsistency in the code, not something you introduced, but it would be good to open a follow-up issue for it. ########## .github/workflows/build.yml: ########## @@ -94,9 +94,13 @@ jobs: # Update output format to enable automatic inline annotations. - name: Run Ruff run: | + # Linted from the repository root, not just python/: dev/, benchmarks/ + # and docs/source/conf.py are Python too. Ruff resolves the nearest + # config per file, so python/ still uses python/pyproject.toml and + # everything else uses the root ruff.toml. cd python - uv run --no-project ruff check --output-format=github . - uv run --no-project ruff format --check . + uv run --no-project ruff check --output-format=github .. Review Comment: Widening the scope is right, but this loses the inline annotations for everything it adds. Ruff emits paths relative to the working directory, so running from `python/` with `..` reports a finding in `dev/foo.py` as `../dev/foo.py`, and GitHub cannot map that back to a file. The annotations quietly stop appearing for every file outside `python/`. Running from the repository root instead should fix it and keep the per-file config resolution you describe in the comment. ########## docs/source/user-guide/extending-components.md: ########## @@ -32,21 +32,24 @@ new configuration extensions, object stores, logical and physical codecs ... Ballista executor can be configured using `ExecutorProcessConfig` which supports overriding `function registry`,`runtime producer`, `config producer`, `logical codec`, `physical codec`. -Ballista scheduler can be tunned using `SchedulerConfig` which supports overriding `config producer`, `session builder`, `logical codec`, `physical codec` +Ballista scheduler can be tuned using `SchedulerConfig` which supports overriding `config producer`, `session builder`, `logical codec`, `physical codec` ## Example: Custom Object Store Integration Extending basic building blocks will be demonstrated by integrating S3 object store. For this, new `ObjectStoreRegistry` and `S3Options` will be provided. `ObjectStoreRegistry` creates new `ObjectStore` instances configured using `S3Options`. For this specific task `config producer`, `runtime producer` and `session builder` have to be provided, and client, scheduler and executor need to be configured. +These three functions ship in `ballista_core::object_store`, so the snippets below are the +shipped implementations rather than something you have to write from scratch. Review Comment: Small accuracy thing, given what this PR is for. The snippet below is not quite the shipped implementation. The real `session_state_with_s3_support` in `ballista/core/src/object_store.rs:89` also chains `.with_scalar_functions(ballista_scalar_functions())`, `.with_aggregate_functions(...)` and `.with_window_functions(...)`, which the snippet drops. Either match it or soften this sentence to say the snippet is abridged, otherwise it starts drifting again on day one. ########## docs/source/contributors-guide/user-personas.md: ########## @@ -82,17 +82,19 @@ to make sure a change that delights one audience does not quietly break another. paradigm. - **Coming from**: Apache Spark. - **Why Ballista**: Keep the Spark mental model — plans split into **stages** at - shuffle boundaries, one **task** per partition, shuffle files, executors with - task slots, and adaptive query execution (**AQE**) for runtime adaptivity — on - top of DataFusion and Arrow. + shuffle boundaries, **tasks** over partitions, shuffle files, executors with + vcores, and adaptive query execution (**AQE**) for runtime adaptivity — on + top of DataFusion and Arrow. Ballista packs several partitions into one task Review Comment: No objection to the content, I checked it and it is accurate. `ballista.scheduler.max_partitions_per_task` does default to 0 meaning unbounded, setting it to 1 does give the Spark model, and `--vcores` is the real flag with `--concurrent-tasks` kept as a deprecated alias. One process note though. This page is the contract we review PRs against, and it is append-only by its own terms, so rewording a persona's guarantees is a bigger deal than its diff size suggests. Could you call it out explicitly in the PR description? I would rather it be a visible decision than something folded into a 67-file docs change. ########## docs/source/user-guide/metrics.md: ########## @@ -23,20 +23,19 @@ > This is optional scheduler feature which should be enabled with > `prometheus-metrics` feature -Built with default features, the ballista scheduler will automatically collect and expose a standard set of prometheus metrics. -The metrics currently collected automatically include: +Built with the `prometheus-metrics` feature, the ballista scheduler collects and exposes a standard set of +prometheus metrics. The metrics collected are: - _job_exec_time_seconds_ - Histogram of successful job execution time in seconds - _planning_time_ms_ - Histogram of job planning time in milliseconds -- _failed_ - Counter of failed jobs Review Comment: Good catch, there is no exported metric called `failed`. It is registered as `job_failed_total` at `ballista/scheduler/src/metrics/prometheus.rs:76`, the struct field is just named `failed`. The same stale name is still in the rustdoc at `prometheus.rs:36` if you want to get it in the same pass. You are already touching two comment-only lines in `.rs` files, so it fits. ########## docs/source/user-guide/extending-components.md: ########## @@ -193,7 +198,6 @@ ctx.sql("SET s3.endpoint = 'http://localhost:9000'") .await? .show() .await?; -ctx.sql("SET s3.allow_http = true").await?.show().await?; Review Comment: I do not think this line should go. `examples/examples/custom-client.rs` still has it, and `ballista/core/src/object_store.rs:205` rejects an `http://` endpoint unless `s3.allow_http` is true. The line just above this sets `s3.endpoint = 'http://localhost:9000'`, so with this removed the documented sequence fails against minio as written. Can you put it back? -- 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]
