You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/23 04:17:48 UTC
(datafusion) 01/01: perf: balance file groups by size for scans with an output ordering (#26458)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26458-30f066d2275cbeb3e82ea72ff76323022e077302
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit a5ca29189d1fe5459ee700adf9620db6058b6abc
Author: Jay Zhan <ja...@gmail.com>
AuthorDate: Fri Oct 23 03:25:12 2026 +0000
perf: balance file groups by size for scans with an output ordering (#26458)
## Which issue does this PR close?
- Closes #26457.
## Rationale for this change
A scan with a declared ordering keeps the count-based file groups from
`split_files`, while an unordered scan is cut into size-balanced byte
ranges. With uneven files, one partition of an ordered scan gets most of
the data and every query over the table waits for it (see the issue: 233
ms ordered vs 93 ms unordered for the same full scan).
## What changes are included in this PR?
`FileScanConfig::repartitioned`: when the scan has an output ordering,
first ask the file source for the split it would use without an ordering
(the size-balanced split, which keeps files in group order and only
moves the group boundaries). Use it only if
`validated_output_ordering()` is unchanged, that is, statistics prove
every new group is still sorted. Otherwise fall back to the current
order-preserving split.
- Groups from `split_groups_by_statistics` interleave value ranges, so
the re-cut fails validation and they are kept as they are.
- Files without statistics cannot be proven sorted in a multi-file
group, so they also keep today's behavior.
- File sources that cannot split files (compressed files, Arrow IPC
stream) return `None` and keep today's behavior.
## What is the testing strategy for this PR?
- `repartition_balances_ordered_groups`: groups `[[8000 B, 8000 B],
[1000 B, 1000 B]]` become four 4500 B groups and keep the ordering.
- `repartition_keeps_ordered_groups_when_rebalancing_breaks_order`:
overlapping files across the old boundary keep the old groups. Fails if
the validation check is removed.
-
`repartition_falls_back_to_ordered_split_when_rebalancing_breaks_order`:
`[[f1, f2], [f3]]` with `f3` overlapping both; the balanced split is
rejected and the ordered split keeps `[f1, f2]` and splits `f3`.
- `repartition_keeps_ordered_groups_when_rebalancing_misplaces_nulls`:
the balanced split would read a file with NULLs after another file; it
is rejected.
- `repartition_scan.slt`: a new `ordered_files` table whose files are in
value order gets size-balanced groups, one of them spanning both files,
and returns the rows in order. The existing ordered plan is unchanged,
because its statistics cannot prove the files are sorted.
- `parquet_sorted_statistics.slt`: two files with NULLs for NULLS FIRST
and NULLS LAST, read in one partition. When the NULLs would be in the
wrong place the plan keeps a sort, otherwise the ordering is used; both
read the rows back.
- `parallelization_multiple_files` snapshot updated: at 8 partitions
each group is still one range of one file, only the group order changes;
the 3-partition case keeps its old groups because the balanced split
would put two files without statistics in one group.
Benchmarks (release-nonlto, median of 5 interleaved runs, ms). 16 sorted
files (4 x 8M rows, 12 x 0.5M), `WITH ORDER (a ASC)`, `target_partitions
= 8`:
| Query | pushdown_filters | main | this PR |
|---|---|---|---|
| `SELECT sum(a + b) FROM t WHERE b > 0.5` | off / on | 228 / 304 | 78 /
105 |
| `SELECT * FROM t WHERE b > 0.999999` | off / on | 31 / 29 | 12 / 11 |
| `SELECT a, count(*) FROM t WHERE b > 0.99 GROUP BY a ORDER BY a LIMIT
5` | off / on | 103 / 102 | 74 / 74 |
| `SELECT * FROM t WHERE b > 0.999999 LIMIT 10` | off / on | 11 / 3 | 8
/ 4 |
| `SELECT * FROM t WHERE b > 0.999 ORDER BY a LIMIT 10` | off / on | 4 /
4 | 4 / 5 |
| `SELECT * FROM t ORDER BY a LIMIT 10` | off / on | 3 / 3 | 3 / 3 |
TPC-H SF10 `lineitem` (12 even files) declared `WITH ORDER
(l_orderkey)`, 5 queries x pushdown on/off, two interleaved rounds: all
within noise (for example `ORDER BY l_orderkey LIMIT 10`: 19-20 vs 18
ms; `WHERE l_shipdate > '1998-11-25' ORDER BY l_orderkey LIMIT 10`:
35-37 vs 35-36 ms).
Worst case: when the old groups were already balanced, the new split
moves the boundaries and can leave a small byte range of a file in a
group (the new `repartition_scan.slt` plan has `1.parquet:516..521`), as
unordered scans already do.
## Are there any user-facing changes?
EXPLAIN output for ordered file scans shows byte-range file groups more
often. No API changes.
---
.../physical_optimizer/enforce_distribution.rs | 2 +-
datafusion/datasource/src/file_scan_config/mod.rs | 187 ++++++++++++++++++
.../test_files/parquet_sorted_statistics.slt | 215 +++++++++++++++++++++
.../sqllogictest/test_files/repartition_scan.slt | 47 +++++
4 files changed, 450 insertions(+), 1 deletion(-)
diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
index 283d22767f..295bace596 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
@@ -4760,7 +4760,7 @@ fn parallelization_multiple_files() -> Result<()> {
@r"
SortRequiredExec: [a@0 ASC]
FilterExec: c@2 = 0
- DataSourceExec: file_groups={8 groups: [[x:0..25], [y:0..25], [x:25..50], [y:25..50], [x:50..75], [y:50..75], [x:75..100], [y:75..100]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
+ DataSourceExec: file_groups={8 groups: [[x:0..25], [x:25..50], [x:50..75], [x:75..100], [y:0..25], [y:25..50], [y:50..75], [y:75..100]]}, projection=[a, b, c, d, e], output_ordering=[a@0 ASC], file_type=parquet
");
let plan_8_sort = test_config_concurrency_8.to_plan(plan, &SORT_DISTRIB_DISTRIB);
assert_plan!(plan_8_distrib, plan_8_sort);
diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs
index 64b12951ef..c94979a472 100644
--- a/datafusion/datasource/src/file_scan_config/mod.rs
+++ b/datafusion/datasource/src/file_scan_config/mod.rs
@@ -872,6 +872,23 @@ impl DataSource for FileScanConfig {
return Ok(None);
}
+ // An ordered scan can use the size-balanced split of an unordered
+ // scan when every new group is still sorted. That split keeps the
+ // files in group order and only moves the group boundaries, so groups
+ // like `[f1, f2], [f3, f4]` with a large `f1` become
+ // `[f1 head], [f1 tail, f2], ...`.
+ if output_ordering.is_some()
+ && let Some(balanced) = self.file_source.repartitioned(
+ target_partitions,
+ repartition_file_min_size,
+ None,
+ self,
+ )?
+ && balanced.validated_output_ordering() == self.validated_output_ordering()
+ {
+ return Ok(Some(Arc::new(balanced)));
+ }
+
let source = self.file_source.repartitioned(
target_partitions,
repartition_file_min_size,
@@ -3361,6 +3378,176 @@ mod tests {
))
}
+ fn ordered_scan_with_groups(file_groups: Vec<FileGroup>) -> FileScanConfig {
+ let schema =
+ Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, true)]));
+ let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0)));
+ FileScanConfigBuilder::new(
+ ObjectStoreUrl::local_filesystem(),
+ Arc::new(MockSource::new(schema)),
+ )
+ .with_file_groups(file_groups)
+ .with_output_ordering(vec![LexOrdering::new(vec![sort_expr]).unwrap()])
+ .build()
+ }
+
+ fn sized_file_with_stats(
+ name: &str,
+ size: u64,
+ min: f64,
+ max: f64,
+ ) -> PartitionedFile {
+ let mut file = make_file_with_stats(name, min, max);
+ file.object_meta.size = size;
+ file
+ }
+
+ /// The files of each group, with whether each file reads only a range
+ fn group_files(config: &FileScanConfig) -> Vec<Vec<(&str, bool)>> {
+ config
+ .file_groups
+ .iter()
+ .map(|group| {
+ group
+ .iter()
+ .map(|f| (f.object_meta.location.as_ref(), f.range.is_some()))
+ .collect()
+ })
+ .collect()
+ }
+
+ fn group_sizes(config: &FileScanConfig) -> Vec<u64> {
+ config
+ .file_groups
+ .iter()
+ .map(|group| group.iter().map(|f| f.effective_size()).sum())
+ .collect()
+ }
+
+ /// An ordered scan whose files are sorted across groups gets the same
+ /// size-balanced split as an unordered scan and keeps its ordering.
+ #[test]
+ fn repartition_balances_ordered_groups() -> Result<()> {
+ let config = ordered_scan_with_groups(vec![
+ FileGroup::new(vec![
+ sized_file_with_stats("f1", 8000, 0.0, 9.0),
+ sized_file_with_stats("f2", 8000, 10.0, 19.0),
+ ]),
+ FileGroup::new(vec![
+ sized_file_with_stats("f3", 1000, 20.0, 29.0),
+ sized_file_with_stats("f4", 1000, 30.0, 39.0),
+ ]),
+ ]);
+ let ordering = config.output_ordering.first().cloned();
+ assert_eq!(group_sizes(&config), vec![16000, 2000]);
+
+ let repartitioned = DataSource::repartitioned(&config, 4, 0, ordering)?
+ .expect("ordered groups should be rebalanced");
+ let repartitioned = repartitioned
+ .downcast_ref::<FileScanConfig>()
+ .expect("expected FileScanConfig");
+ assert_eq!(group_sizes(repartitioned), vec![4500, 4500, 4500, 4500]);
+ assert_eq!(
+ repartitioned.validated_output_ordering(),
+ config.validated_output_ordering()
+ );
+ Ok(())
+ }
+
+ /// If moving the group boundaries would put overlapping files in one
+ /// group, the scan keeps its groups and its ordering.
+ #[test]
+ fn repartition_keeps_ordered_groups_when_rebalancing_breaks_order() -> Result<()> {
+ let config = ordered_scan_with_groups(vec![
+ FileGroup::new(vec![
+ sized_file_with_stats("f1", 8000, 0.0, 9.0),
+ sized_file_with_stats("f2", 8000, 10.0, 19.0),
+ ]),
+ FileGroup::new(vec![
+ sized_file_with_stats("f3", 1000, 5.0, 15.0),
+ sized_file_with_stats("f4", 1000, 16.0, 25.0),
+ ]),
+ ]);
+ let ordering = config.output_ordering.first().cloned();
+ assert_eq!(config.validated_output_ordering().len(), 1);
+
+ assert!(DataSource::repartitioned(&config, 4, 0, ordering)?.is_none());
+ Ok(())
+ }
+
+ /// With `NULLS FIRST`, a file with NULLs can only be the first file of a
+ /// group, so a size-balanced split that puts it after another file is not
+ /// used.
+ #[test]
+ fn repartition_keeps_ordered_groups_when_rebalancing_misplaces_nulls() -> Result<()> {
+ let mut with_nulls = sized_file_with_stats("f2", 1000, 10.0, 19.0);
+ Arc::make_mut(with_nulls.statistics.as_mut().unwrap()).column_statistics[0]
+ .null_count = Precision::Exact(1);
+ let config = ordered_scan_with_groups(vec![
+ FileGroup::new(vec![sized_file_with_stats("f1", 8000, 0.0, 9.0)]),
+ FileGroup::new(vec![with_nulls]),
+ ]);
+ let ordering = config.output_ordering.first().cloned();
+ assert_eq!(config.validated_output_ordering().len(), 1);
+
+ let repartitioned = DataSource::repartitioned(&config, 4, 0, ordering)?
+ .expect("the large file should be split");
+ let repartitioned = repartitioned
+ .downcast_ref::<FileScanConfig>()
+ .expect("expected FileScanConfig");
+ assert_eq!(
+ group_files(repartitioned),
+ vec![
+ vec![("f1", true)],
+ vec![("f2", true)],
+ vec![("f1", true)],
+ vec![("f1", true)],
+ ]
+ );
+ assert_eq!(
+ repartitioned.validated_output_ordering(),
+ config.validated_output_ordering()
+ );
+ Ok(())
+ }
+
+ /// If moving the group boundaries would put overlapping files in one
+ /// group, the ordered split is used instead: it keeps multi-file groups
+ /// and only splits single-file groups.
+ #[test]
+ fn repartition_falls_back_to_ordered_split_when_rebalancing_breaks_order()
+ -> Result<()> {
+ let config = ordered_scan_with_groups(vec![
+ FileGroup::new(vec![
+ sized_file_with_stats("f1", 8000, 0.0, 9.0),
+ sized_file_with_stats("f2", 8000, 10.0, 19.0),
+ ]),
+ FileGroup::new(vec![sized_file_with_stats("f3", 1000, 5.0, 15.0)]),
+ ]);
+ let ordering = config.output_ordering.first().cloned();
+ assert_eq!(config.validated_output_ordering().len(), 1);
+
+ let repartitioned = DataSource::repartitioned(&config, 4, 0, ordering)?
+ .expect("the single-file group should be split");
+ let repartitioned = repartitioned
+ .downcast_ref::<FileScanConfig>()
+ .expect("expected FileScanConfig");
+ assert_eq!(
+ group_files(repartitioned),
+ vec![
+ vec![("f1", false), ("f2", false)],
+ vec![("f3", true)],
+ vec![("f3", true)],
+ vec![("f3", true)],
+ ]
+ );
+ assert_eq!(
+ repartitioned.validated_output_ordering(),
+ config.validated_output_ordering()
+ );
+ Ok(())
+ }
+
fn make_exact_empty_file(name: &str) -> PartitionedFile {
PartitionedFile::new(name.to_string(), 1024).with_statistics(Arc::new(
Statistics {
diff --git a/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt b/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt
index a877b88d5e..6da3689988 100644
--- a/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt
+++ b/datafusion/sqllogictest/test_files/parquet_sorted_statistics.slt
@@ -340,6 +340,221 @@ WHERE partition_col = 'D'
GROUP BY int_col;
----
+# Min/max statistics ignore NULLs. Files read one after another in a single
+# partition keep the declared ordering only if their NULLs end up at the start
+# (NULLS FIRST) or at the end (NULLS LAST) of the combined stream.
+statement ok
+set datafusion.execution.target_partitions = 1;
+
+# NULLS LAST with NULLs in the first file: needs a sort
+query I
+COPY (SELECT * FROM (VALUES (1), (NULL)) AS t(a))
+TO 'test_files/scratch/parquet_sorted_statistics/nulls_last_bad/1.parquet'
+STORED AS PARQUET;
+----
+2
+
+query I
+COPY (SELECT * FROM (VALUES (2), (3)) AS t(a))
+TO 'test_files/scratch/parquet_sorted_statistics/nulls_last_bad/2.parquet'
+STORED AS PARQUET;
+----
+2
+
+statement ok
+CREATE EXTERNAL TABLE nulls_last_bad (a BIGINT)
+STORED AS PARQUET
+WITH ORDER (a ASC NULLS LAST)
+LOCATION 'test_files/scratch/parquet_sorted_statistics/nulls_last_bad';
+
+query TT
+EXPLAIN SELECT a FROM nulls_last_bad ORDER BY a ASC NULLS LAST;
+----
+logical_plan
+01)Sort: nulls_last_bad.a ASC NULLS LAST
+02)--TableScan: nulls_last_bad projection=[a]
+physical_plan
+01)SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/nulls_last_bad/1.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/nulls_last_bad/2.parquet]]}, projection=[a], file_type=parquet, sort_order_for_reorder=[a@0 ASC NULLS LAST]
+
+query I
+SELECT a FROM nulls_last_bad ORDER BY a ASC NULLS LAST;
+----
+1
+2
+3
+NULL
+
+# NULLS LAST with NULLs only in the last file: the ordering is kept
+query I
+COPY (SELECT * FROM (VALUES (1), (2)) AS t(a))
+TO 'test_files/scratch/parquet_sorted_statistics/nulls_last_ok/1.parquet'
+STORED AS PARQUET;
+----
+2
+
+query I
+COPY (SELECT * FROM (VALUES (3), (NULL)) AS t(a))
+TO 'test_files/scratch/parquet_sorted_statistics/nulls_last_ok/2.parquet'
+STORED AS PARQUET;
+----
+2
+
+statement ok
+CREATE EXTERNAL TABLE nulls_last_ok (a BIGINT)
+STORED AS PARQUET
+WITH ORDER (a ASC NULLS LAST)
+LOCATION 'test_files/scratch/parquet_sorted_statistics/nulls_last_ok';
+
+query TT
+EXPLAIN SELECT a FROM nulls_last_ok ORDER BY a ASC NULLS LAST;
+----
+logical_plan
+01)Sort: nulls_last_ok.a ASC NULLS LAST
+02)--TableScan: nulls_last_ok projection=[a]
+physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/nulls_last_ok/1.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/nulls_last_ok/2.parquet]]}, projection=[a], output_ordering=[a@0 ASC NULLS LAST], file_type=parquet
+
+query I
+SELECT a FROM nulls_last_ok ORDER BY a ASC NULLS LAST;
+----
+1
+2
+3
+NULL
+
+# NULLS FIRST with NULLs in the last file: needs a sort
+query I
+COPY (SELECT * FROM (VALUES (1), (2)) AS t(a))
+TO 'test_files/scratch/parquet_sorted_statistics/nulls_first_bad/1.parquet'
+STORED AS PARQUET;
+----
+2
+
+query I
+COPY (SELECT * FROM (VALUES (NULL), (3)) AS t(a))
+TO 'test_files/scratch/parquet_sorted_statistics/nulls_first_bad/2.parquet'
+STORED AS PARQUET;
+----
+2
+
+statement ok
+CREATE EXTERNAL TABLE nulls_first_bad (a BIGINT)
+STORED AS PARQUET
+WITH ORDER (a ASC NULLS FIRST)
+LOCATION 'test_files/scratch/parquet_sorted_statistics/nulls_first_bad';
+
+query TT
+EXPLAIN SELECT a FROM nulls_first_bad ORDER BY a ASC NULLS FIRST;
+----
+logical_plan
+01)Sort: nulls_first_bad.a ASC NULLS FIRST
+02)--TableScan: nulls_first_bad projection=[a]
+physical_plan
+01)SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/nulls_first_bad/2.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/nulls_first_bad/1.parquet]]}, projection=[a], file_type=parquet, sort_order_for_reorder=[a@0 ASC]
+
+query I
+SELECT a FROM nulls_first_bad ORDER BY a ASC NULLS FIRST;
+----
+NULL
+1
+2
+3
+
+# NULLS FIRST with NULLs only in the first file: the ordering is kept
+query I
+COPY (SELECT * FROM (VALUES (NULL), (1)) AS t(a))
+TO 'test_files/scratch/parquet_sorted_statistics/nulls_first_ok/1.parquet'
+STORED AS PARQUET;
+----
+2
+
+query I
+COPY (SELECT * FROM (VALUES (2), (3)) AS t(a))
+TO 'test_files/scratch/parquet_sorted_statistics/nulls_first_ok/2.parquet'
+STORED AS PARQUET;
+----
+2
+
+statement ok
+CREATE EXTERNAL TABLE nulls_first_ok (a BIGINT)
+STORED AS PARQUET
+WITH ORDER (a ASC NULLS FIRST)
+LOCATION 'test_files/scratch/parquet_sorted_statistics/nulls_first_ok';
+
+query TT
+EXPLAIN SELECT a FROM nulls_first_ok ORDER BY a ASC NULLS FIRST;
+----
+logical_plan
+01)Sort: nulls_first_ok.a ASC NULLS FIRST
+02)--TableScan: nulls_first_ok projection=[a]
+physical_plan DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/nulls_first_ok/1.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/nulls_first_ok/2.parquet]]}, projection=[a], output_ordering=[a@0 ASC], file_type=parquet
+
+query I
+SELECT a FROM nulls_first_ok ORDER BY a ASC NULLS FIRST;
+----
+NULL
+1
+2
+3
+
+# Declaring INT over files that store BIGINT leaves the file statistics
+# without usable min/max values or null counts. The files cannot be proven
+# sorted.
+query I
+COPY (SELECT * FROM (VALUES (100), (200)) AS t(a))
+TO 'test_files/scratch/parquet_sorted_statistics/cast_bounds/1.parquet'
+STORED AS PARQUET;
+----
+2
+
+query I
+COPY (SELECT * FROM (VALUES (1), (2), (3)) AS t(a))
+TO 'test_files/scratch/parquet_sorted_statistics/cast_bounds/2.parquet'
+STORED AS PARQUET;
+----
+3
+
+statement ok
+CREATE EXTERNAL TABLE cast_bounds (a INT)
+STORED AS PARQUET
+WITH ORDER (a ASC)
+LOCATION 'test_files/scratch/parquet_sorted_statistics/cast_bounds';
+
+query TT
+EXPLAIN SELECT a FROM cast_bounds ORDER BY a ASC;
+----
+logical_plan
+01)Sort: cast_bounds.a ASC NULLS LAST
+02)--TableScan: cast_bounds projection=[a]
+physical_plan
+01)SortExec: expr=[a@0 ASC NULLS LAST], preserve_partitioning=[false]
+02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/cast_bounds/1.parquet, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/parquet_sorted_statistics/cast_bounds/2.parquet]]}, projection=[a], file_type=parquet, sort_order_for_reorder=[a@0 ASC NULLS LAST]
+
+query I
+SELECT a FROM cast_bounds ORDER BY a ASC;
+----
+1
+2
+3
+100
+200
+
+statement ok
+DROP TABLE cast_bounds;
+
+statement ok
+DROP TABLE nulls_last_bad;
+
+statement ok
+DROP TABLE nulls_last_ok;
+
+statement ok
+DROP TABLE nulls_first_bad;
+
+statement ok
+DROP TABLE nulls_first_ok;
+
# Config reset
# The SLT runner sets `target_partitions` to 4 instead of using the default, so
diff --git a/datafusion/sqllogictest/test_files/repartition_scan.slt b/datafusion/sqllogictest/test_files/repartition_scan.slt
index 150d5b1b43..edb0ce3404 100644
--- a/datafusion/sqllogictest/test_files/repartition_scan.slt
+++ b/datafusion/sqllogictest/test_files/repartition_scan.slt
@@ -147,6 +147,53 @@ DROP TABLE parquet_table;
statement ok
DROP TABLE parquet_table_with_order;
+## An ordered scan gets size-balanced groups when statistics prove that
+## every new group is still sorted
+
+statement ok
+COPY (VALUES (1), (2), (3), (4), (5))
+TO 'test_files/scratch/repartition_scan/ordered_files/1.parquet'
+STORED AS PARQUET;
+
+statement ok
+COPY (VALUES (100), (200))
+TO 'test_files/scratch/repartition_scan/ordered_files/2.parquet'
+STORED AS PARQUET;
+
+statement ok
+CREATE EXTERNAL TABLE ordered_files(column1 bigint)
+STORED AS PARQUET
+LOCATION 'test_files/scratch/repartition_scan/ordered_files'
+WITH ORDER (column1 ASC);
+
+query I
+SELECT column1 FROM ordered_files WHERE column1 <> 42 ORDER BY column1;
+----
+1
+2
+3
+4
+5
+100
+200
+
+# A group may span files when reading them in order keeps the declared
+# ordering: all of 1.parquet's values are below 2.parquet's
+query TT
+EXPLAIN SELECT column1 FROM ordered_files WHERE column1 <> 42 ORDER BY column1;
+----
+logical_plan
+01)Sort: ordered_files.column1 ASC NULLS LAST
+02)--Filter: ordered_files.column1 != Int64(42)
+03)----TableScan: ordered_files projection=[column1], partial_filters=[ordered_files.column1 != Int64(42)]
+physical_plan
+01)SortPreservingMergeExec: [column1@0 ASC NULLS LAST]
+02)--FilterExec: column1@0 != 42
+03)----DataSourceExec: file_groups={4 groups: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/ordered_files/1.parquet:0..258], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/ordered_files/1.parquet:258..516], [WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/ordered_files/1.parquet:516..521, WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/repartition_scan/ordered_files/2.parquet:0..253], [WORKSP [...]
+
+statement ok
+DROP TABLE ordered_files;
+
## Either row group range assignment reads each row of a split file once
statement ok
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org