(datafusion-ballista) branch main updated: docs: correct scan partitioning description in tuning guide (#2125)

[ag...@apache.org]

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 <ag...@apache.org>
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: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org