This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new 8be1e8986 docs: correct scan partitioning description in tuning guide
(#2125)
8be1e8986 is described below
commit 8be1e89861bc702ac02658216e030561bffd4eeb
Author: Andy Grove <[email protected]>
AuthorDate: Tue Jul 21 14:49:46 2026 -0600
docs: correct scan partitioning description in tuning guide (#2125)
* docs: correct scan partitioning description in tuning guide
The tuning guide claimed Ballista treats each file as a single partition
and never decreases partition count. Both are wrong: file listing chunks
files into at most target_partitions groups, and the physical optimizer
splits large files by byte range.
Also document ballista.scheduler.max_partitions_per_task, which bounds
how many partitions the scheduler packs into one task and still defaults
to 1.
* docs: clarify that target_partitions defaults to 16 only via
new_with_ballista
---
docs/source/user-guide/tuning-guide.md | 61 +++++++++++++++++++++++++++-------
1 file changed, 49 insertions(+), 12 deletions(-)
diff --git a/docs/source/user-guide/tuning-guide.md
b/docs/source/user-guide/tuning-guide.md
index 5391c399d..4100e0b5a 100644
--- a/docs/source/user-guide/tuning-guide.md
+++ b/docs/source/user-guide/tuning-guide.md
@@ -25,18 +25,31 @@ The goal of any distributed compute engine is to
parallelize work as much as pos
by adding more compute resource.
The basic unit of concurrency and parallelism in Ballista is the concept of a
partition. The leaf nodes of a query
-are typically table scans that read from files on disk and Ballista currently
treats each file within a table as a
-single partition (in the future, Ballista will support splitting files into
partitions but this is not implemented yet).
-
-For example, if there is a table "customer" that consists of 200 Parquet
files, that table scan will naturally have
-200 partitions and the table scan and certain subsequent operations will also
have 200 partitions. Conversely, if the
-table only has a single Parquet file then there will be a single partition and
the work will not be able to scale even
-if the cluster has resource available. Ballista supports repartitioning within
a query to improve parallelism.
-The configuration setting `datafusion.execution.target_partitions`can be set
to the desired number of partitions. This is
-currently a global setting for the entire context. The default value for this
setting is 16.
-
-Note that Ballista will never decrease the number of partitions based on this
setting and will only repartition if
-the source operation has fewer partitions than this setting.
+are typically table scans that read files from object storage, and the number
of partitions such a scan produces is
+driven by `datafusion.execution.target_partitions`. This is currently a global
setting for the entire context.
+
+DataFusion's own default for this setting is the number of CPU cores available
to the process. Ballista overrides it
+to 16 in `SessionConfig::new_with_ballista()`, so a context created with
`SessionContext::remote()` or
+`SessionContext::standalone()` starts at 16. If you instead build your own
`SessionConfig` and pass it to
+`remote_with_state()` or `standalone_with_state()`, Ballista leaves the
setting alone and you get whatever that
+config carries, which is the client machine's core count for a plain
`SessionConfig::new()`. Set it explicitly if
+you care about the value.
+
+A scan is partitioned in two steps:
+
+1. When the table is listed, its files are sorted by path and distributed into
**at most** `target_partitions`
+ file groups. A "customer" table of 200 Parquet files read with
`target_partitions = 16` therefore produces
+ 16 partitions of roughly 13 files each, not 200 partitions.
+2. The physical optimizer then splits individual files by byte range, so a
table with fewer files than
+ `target_partitions` can still be read in parallel. A table consisting of a
single large Parquet file is split
+ into `target_partitions` byte ranges rather than being read by one task.
This step is controlled by
+ `datafusion.optimizer.repartition_file_scans` (default `true`) and only
applies when the scan reads at least
+ `datafusion.optimizer.repartition_file_min_size` bytes (default 10 MB).
Small tables below that threshold are
+ left as a single partition, since splitting them costs more than it saves.
+
+Ballista disables DataFusion's round-robin repartitioning, so this file-group
splitting is the only source of
+parallelism for a scan stage. Raising `target_partitions` is the way to
increase it, and lowering it is the way to
+reduce the number of partitions a large table is read with.
Example: Setting the desired number of shuffle partitions when creating a
context.
@@ -54,6 +67,30 @@ let state = SessionStateBuilder::new()
let ctx: SessionContext = SessionContext::remote_with_state(&url,state).await?;
```
+### Partitions and Tasks
+
+A stage's partition count is not the same as its task count. The scheduler
keeps a cursor over the stage's
+partitions and, each time an executor is assigned work, hands out a slice of
those partitions to run as a single
+task. The size of that slice is bounded by both the executor's free vcores and
+`ballista.scheduler.max_partitions_per_task`.
+
+That setting still defaults to `1`, so out of the box a 16-partition scan
stage runs as 16 single-partition tasks.
+Raising it lets the scheduler pack several partitions into one task, which
reduces task count and scheduling
+overhead and allows operators such as sort and hash join to work across
partitions within a task. Setting it to `0`
+removes the cap entirely, so each task is filled up to the assigned executor's
free vcore count: a 16-partition
+stage then runs as one task on an idle 16-vcore executor, or as four
4-partition tasks across four 4-vcore
+executors.
+
+| key | type | default | description
|
+| ------------------------------------------ | ------ | ------- |
-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
+| ballista.scheduler.max_partitions_per_task | UInt64 | 1 | Upper bound
on the number of input partitions packed into a single task. `1` means one task
per partition. `0` means unbounded, filling each task up to the executor's free
vcores. |
+
+Stages whose plan collapses all input into a single partition (for example
under a `CoalescePartitionsExec` or
+`SortPreservingMergeExec`) ignore this cap and always pack their full pending
queue into one task, since splitting
+them would produce partial results that downstream stages cannot merge.
+
+Each task's plan is rewritten before dispatch so that its scan sees only the
file groups belonging to its own slice.
+
## Configuring Executor Concurrency Levels
Each executor instance advertises a fixed number of virtual cores (vcores) to
the scheduler. This is specified by
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
