Skip to content

Commit 63ad991

Browse files
Add ListingOptions::output_partitioning and FileScanConfig::output_partitioning for pre-defined file partitioning (#22657)
## Which issue does this PR close? <!-- We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123. --> - Closes #22645. ## Rationale for this change <!-- Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed. Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes. --> This follows up on #22607 by replacing range-partitioning sqllogictest boilerplate with a general file/listing scan API for declared output partitioning. Related: #21992, #22607, #22607 (comment) ## What changes are included in this PR? <!-- There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR. --> - Add declared `output_partitioning` to file scan and listing table configuration. - Preserve declared partition counts during listing-table file grouping. - Serialize scan `output_partitioning` through physical plan proto. - Refactor `range_partitioning.slt` to use a CSV `ListingTable` instead of a custom test-only `TableProvider` / `DataSource`. Contract: - Declared partitioning expressions are written against the full table schema before scan projection. For example, `Range([range_key@0], [(10), (20)], 3)` remains valid if the scan projects `range_key` and falls back to `UnknownPartitioning(3)` if `range_key` is not projected. - Listing tables create one file group per declared output partition (which can exceed `target_partitions`). It is up to the user to plan their partitioning. For example, a 4-partition range declaration creates four scan file groups, adding empty trailing groups when fewer files are present. - File group index is part of the contract: file group `i` must contain rows for declared output partition `i`. DataFusion does not validate row placement, matching other user-declared properties such as sortedness. ## Are these changes tested? <!-- We typically require tests for all PRs in order to: 1. Prevent the code from being accidentally broken by subsequent changes 2. Serve as another way to document the expected behavior of the code If tests are not included in your PR, please explain why (for example, are they covered by existing tests)? --> Yes. ## Are there any user-facing changes? <!-- If there are user-facing changes then we may require documentation to be updated before approving the PR. --> Yes. This adds public API for declaring file/listing scan output partitioning. No breaking API changes. <!-- If there are any breaking changes to public APIs, please add the `api change` label. --> --------- Co-authored-by: Gabriel <45515538+gabotechs@users.noreply.github.com>
1 parent 91f30b9 commit 63ad991

18 files changed

Lines changed: 1020 additions & 395 deletions

File tree

Cargo.lock

Lines changed: 0 additions & 1 deletion
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

datafusion/catalog-listing/src/helpers.rs

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -325,7 +325,7 @@ pub fn evaluate_partition_prefix<'a>(
325325
}
326326
}
327327

328-
fn filter_partitions(
328+
pub fn filter_partitioned_file(
329329
pf: PartitionedFile,
330330
filters: &[Expr],
331331
df_schema: &DFSchema,
@@ -447,7 +447,7 @@ pub async fn pruned_partition_list<'a>(
447447
))
448448
})
449449
.try_filter_map(move |pf| {
450-
futures::future::ready(filter_partitions(pf, filters, &df_schema))
450+
futures::future::ready(filter_partitioned_file(pf, filters, &df_schema))
451451
})
452452
.boxed())
453453
}

datafusion/catalog-listing/src/options.rs

Lines changed: 53 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ use datafusion_catalog::Session;
2020
use datafusion_common::plan_err;
2121
use datafusion_datasource::ListingTableUrl;
2222
use datafusion_datasource::file_format::FileFormat;
23-
use datafusion_expr::SortExpr;
23+
use datafusion_expr::{Partitioning, SortExpr};
2424
use futures::StreamExt;
2525
use futures::TryStreamExt;
2626
use itertools::Itertools;
@@ -53,6 +53,46 @@ pub struct ListingOptions {
5353
/// multiple equivalent orderings, the outer `Vec` will have a
5454
/// single element.
5555
pub file_sort_order: Vec<Vec<SortExpr>>,
56+
/// Declared output partitioning for scans from this table.
57+
///
58+
/// Expressions are logical expressions over the full table schema. When set,
59+
/// [`ListingTable`](crate::ListingTable) creates one file group per
60+
/// declared output partition. When unset, file grouping uses the scan-time
61+
/// [`SessionConfig::target_partitions`](datafusion_execution::config::SessionConfig::target_partitions).
62+
///
63+
/// Files are listed in path order, split into whole-file groups across the
64+
/// declared partition count, and then padded with trailing empty groups when
65+
/// needed. DataFusion does not route files by partition values or validate
66+
/// row placement, so callers must ensure file group `i` contains rows for
67+
/// partition `i`. Layouts that require explicit file-to-partition assignment
68+
/// are not supported.
69+
///
70+
/// For example, range partitioning on column `a` with split points
71+
/// `[10, 20, 30]` declares four output partitions. With three path-ordered
72+
/// files, the trailing partition is preserved as empty:
73+
///
74+
/// ```text
75+
/// files in path order: f0, f1, f2
76+
///
77+
/// file groups:
78+
/// partition 0: [f0]
79+
/// partition 1: [f1]
80+
/// partition 2: [f2]
81+
/// partition 3: []
82+
/// ```
83+
///
84+
/// With five path-ordered files, a partition can contain multiple files:
85+
///
86+
/// ```text
87+
/// files in path order: f0, f1, f2, f3, f4
88+
///
89+
/// file groups:
90+
/// partition 0: [f0, f1]
91+
/// partition 1: [f2, f3]
92+
/// partition 2: [f4]
93+
/// partition 3: []
94+
/// ```
95+
pub output_partitioning: Option<Partitioning>,
5696
}
5797

5898
impl ListingOptions {
@@ -66,6 +106,7 @@ impl ListingOptions {
66106
format,
67107
table_partition_cols: vec![],
68108
file_sort_order: vec![],
109+
output_partitioning: None,
69110
}
70111
}
71112

@@ -113,6 +154,17 @@ impl ListingOptions {
113154
self
114155
}
115156

157+
/// Set declared output partitioning.
158+
///
159+
/// See [`Self::output_partitioning`] for the contract.
160+
pub fn with_output_partitioning(
161+
mut self,
162+
output_partitioning: Option<Partitioning>,
163+
) -> Self {
164+
self.output_partitioning = output_partitioning;
165+
self
166+
}
167+
116168
/// Set `table partition columns` on [`ListingOptions`] and returns self.
117169
///
118170
/// "partition columns," used to support [Hive Partitioning], are

0 commit comments

Comments
 (0)