Repository navigation
feat(scan): optionally retain column statistics on FileScanTask - #3424
anuragmantri wants to merge 3 commits into
Conversation
Add TableScanBuilder::with_column_stats, mirroring Java's Scan.includeColumnStats. The named columns resolve to field ids at build time with the scan's case sensitivity, and each planned FileScanTask carries a FileScanTaskColumnStats holding the data file's value, null and NaN counts and lower and upper bounds for only those field ids. Retention is opt-in: by default column_stats is None, nothing is copied from the manifest entry, and the field is omitted when a task is serialized, so serialized tasks are unchanged.
JeonDaehong
left a comment
There was a problem hiding this comment.
Thanks, this will be useful for #3343.
One thing I hit while round-tripping the new stats through serde: a float/double bound of ±Infinity (or NaN) serializes to "literal":null with serde_json, and deserializing it fails with None value at line 1 .... Every other primitive type round-trips (bool through decimal, including fixed, uuid and timestamp_ns).
±Infinity is a valid bound: the spec only excludes NaN, and Java's FloatFieldMetrics keeps infinite values as bounds. So a task for a file containing an infinite float would serialize but not deserialize. The root cause is Datum's serde rather than this PR, but this PR is what puts file bounds into serialized tasks. Could we add an infinite-bound case to the serde test and either fix it in Datum or document it?
Bounds went through Datum's JSON encoding, which writes non-finite floats as null and rejects many finite float values when reading them back, so a task with such a bound serialized but failed to deserialize. Serialize each bound as its type and the spec's single-value binary encoding, as the DataFile JSON serde already does, so every bound round-trips exactly.
| partition_spec: Option<PartitionSpecRef>, | ||
| unified_partition_type: Option<Arc<StructType>>, | ||
| sort_orders: Arc<HashMap<i64, SortOrderRef>>, | ||
| column_stats_field_ids: Option<Arc<Vec<i32>>>, |
There was a problem hiding this comment.
I kept i32 to match how field ids are typed everywhere else: the spec defines them as int, and NestedField::id, the DataFile stats maps this reads from, FileScanTask::project_field_ids and FileMetrics in #3347 all use i32. Switching here would mean converting at each of those boundaries.
|
|
||
| /// One column's statistics: field id, value count, null count, NaN count, lower bound | ||
| /// and upper bound. | ||
| type ColumnStat = (i32, u64, u64, Option<u64>, Datum, Datum); |
There was a problem hiding this comment.
I'd prob prefer a typed structure instead of tuple type.... unless tuple type is cheaper than struct
There was a problem hiding this comment.
Right, it's only a test helper, so each column's expected stats fit on one line. Happy to switch to a small struct if that reads better.
| } | ||
| } | ||
|
|
||
| /// Column statistics of a [`FileScanTask`]'s data file, from its manifest entry. |
There was a problem hiding this comment.
I feel the comments can be concised here
There was a problem hiding this comment.
Trimmed in 5345a45 to what a caller needs: what a missing entry means, that bounds stay inclusive when truncated and are typed with the manifest's schema, and that the statistics cover the whole file.
comphead
left a comment
There was a problem hiding this comment.
Overall the opt-in design looks sound and matches Java's includeColumnStats for the maps it copies. I found no correctness issues. A few small points below.
Question: Java's BaseFile(copy, requestedColumnIds) also keeps column_sizes for the requested ids. Is leaving it out here intentional, for example because FileMetrics in #3347 does not carry it? If so, a sentence in the FileScanTaskColumnStats docs would help.
|
|
||
| impl FileScanTaskColumnStats { | ||
| pub(crate) fn from_data_file(data_file: &DataFile, field_ids: &[i32]) -> Self { | ||
| fn retain<V: Clone>(map: &HashMap<i32, V>, field_ids: &[i32]) -> HashMap<i32, V> { |
There was a problem hiding this comment.
This runs once per manifest entry and builds five maps. collect() over filter_map has a lower size hint of 0, so each map regrows. Consider HashMap::with_capacity(field_ids.len().min(map.len())) and returning early when map is empty (metrics mode none is common), so the empty case does no work.
There was a problem hiding this comment.
Done in 5345a45. An empty map now returns before any lookups, and the result is sized with field_ids.len().min(map.len()).
| .collect() | ||
| } | ||
|
|
||
| fn collect_column_stats_field_ids( |
There was a problem hiding this comment.
The resolve plus Column {column_name} not found in table. Schema: {schema} construction duplicates collect_scan_field_ids. A small field_id_by_name(schema, name, case_sensitive) -> Result<i32> used by both would keep the two error messages from drifting. Related: column_names is an Option<Vec<String>> and this is a Vec<String> where empty means none. An Option would follow the existing field and drop the is_empty special case.
There was a problem hiding this comment.
Done in 5345a45. Both paths now share a require_field_id helper and its not-found error. I named it differently so it isn't confused with Schema::field_id_by_name. The requested names are now an Option<Vec<String>> like column_names, so an empty list means no columns and each task carries empty statistics, the same way select_empty works.
| partition_spec: Option<PartitionSpecRef>, | ||
| unified_partition_type: Option<Arc<StructType>>, | ||
| sort_orders: Arc<HashMap<i64, SortOrderRef>>, | ||
| column_stats_field_ids: Option<Arc<Vec<i32>>>, |
There was a problem hiding this comment.
Option<Arc<Vec<i32>>> is a double indirection. Option<Arc<[i32]>> reads the ids with one pointer hop and is built with Arc::from(vec). Same for the field on ManifestEntryContext and PlanContext.
There was a problem hiding this comment.
Switched to Option<Arc<[i32]>> in all three contexts in 5345a45.
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn test_plan_files_retains_requested_column_stats() { |
There was a problem hiding this comment.
The description calls out nested names as a deliberate difference from select, but no case here uses one. A nested leaf (for example a struct child, resolved by field_id_by_name) would pin that choice.
There was a problem hiding this comment.
Added test_collect_column_stats_field_ids_resolves_nested_field in 5345a45, which resolves s.f to the struct child's field id.
anuragmantri
left a comment
There was a problem hiding this comment.
@JeonDaehong thanks for catching this. I reproduced it, and it's a little wider than infinite bounds: Datum's JSON path also rejects many finite float values, such as Datum::float(100.1), because the float conversion compares against an absolute f32::EPSILON. That's the same root cause as #3409.
In e0f3919, bounds now serialize as their type plus the spec's single-value binary encoding, as the DataFile JSON serde already does, so every bound round-trips exactly. The serde test now covers a float of 100.1 and infinite float and double bounds. Partition values and predicate literals still go through the JSON path, so I've left those to #3409.
| partition_spec: Option<PartitionSpecRef>, | ||
| unified_partition_type: Option<Arc<StructType>>, | ||
| sort_orders: Arc<HashMap<i64, SortOrderRef>>, | ||
| column_stats_field_ids: Option<Arc<Vec<i32>>>, |
There was a problem hiding this comment.
I kept i32 to match how field ids are typed everywhere else: the spec defines them as int, and NestedField::id, the DataFile stats maps this reads from, FileScanTask::project_field_ids and FileMetrics in #3347 all use i32. Switching here would mean converting at each of those boundaries.
|
|
||
| /// One column's statistics: field id, value count, null count, NaN count, lower bound | ||
| /// and upper bound. | ||
| type ColumnStat = (i32, u64, u64, Option<u64>, Datum, Datum); |
There was a problem hiding this comment.
Right, it's only a test helper, so each column's expected stats fit on one line. Happy to switch to a small struct if that reads better.
Retain column_sizes for the requested ids too, matching Java's BaseFile(copy, requestedColumnIds). Share the column-name lookup and its error with column selection, keep the requested names as an Option like column_names so an empty list yields empty statistics, and pass the ids as Arc<[i32]>. Skip work for an empty map when filtering, trim the struct docs, and cover nested field names.
| } | ||
| } | ||
|
|
||
| /// Column statistics of a [`FileScanTask`]'s data file, from its manifest entry. |
There was a problem hiding this comment.
Trimmed in 5345a45 to what a caller needs: what a missing entry means, that bounds stay inclusive when truncated and are typed with the manifest's schema, and that the statistics cover the whole file.
|
|
||
| impl FileScanTaskColumnStats { | ||
| pub(crate) fn from_data_file(data_file: &DataFile, field_ids: &[i32]) -> Self { | ||
| fn retain<V: Clone>(map: &HashMap<i32, V>, field_ids: &[i32]) -> HashMap<i32, V> { |
There was a problem hiding this comment.
Done in 5345a45. An empty map now returns before any lookups, and the result is sized with field_ids.len().min(map.len()).
| .collect() | ||
| } | ||
|
|
||
| fn collect_column_stats_field_ids( |
There was a problem hiding this comment.
Done in 5345a45. Both paths now share a require_field_id helper and its not-found error. I named it differently so it isn't confused with Schema::field_id_by_name. The requested names are now an Option<Vec<String>> like column_names, so an empty list means no columns and each task carries empty statistics, the same way select_empty works.
| partition_spec: Option<PartitionSpecRef>, | ||
| unified_partition_type: Option<Arc<StructType>>, | ||
| sort_orders: Arc<HashMap<i64, SortOrderRef>>, | ||
| column_stats_field_ids: Option<Arc<Vec<i32>>>, |
There was a problem hiding this comment.
Switched to Option<Arc<[i32]>> in all three contexts in 5345a45.
| } | ||
|
|
||
| #[tokio::test] | ||
| async fn test_plan_files_retains_requested_column_stats() { |
There was a problem hiding this comment.
Added test_collect_column_stats_field_ids_resolves_nested_field in 5345a45, which resolves s.f to the struct child's field id.
|
Thanks @anuragmantri! Storing bounds as the spec's binary encoding is a nice fix, since it now matches the DataFile serde. Good catch on finite floats like 100.1 too. Leaving partition values and literals to #3409 sounds right to me. |
JeonDaehong
left a comment
There was a problem hiding this comment.
LGTM. Read through the full diff: opt-in default keeps existing serialization unchanged, stats are filtered to the requested ids, and the binary bound encoding round-trips. Thanks @anuragmantri!
Which issue does this PR close?
What changes are included in this PR?
Planning drops a data file's column statistics when it builds a
FileScanTask. Engines that want to order or merge files by key range, such as the sorted scan work in apache/datafusion-iceberg#55 and apache/datafusion-comet#5343, need per-file bounds on the sort columns. This PR lets a scan keep them, mirroring Java'sScan.includeColumnStats.TableScanBuilder::with_column_stats(columns)takes column names and resolves them to field ids inbuild()with the scan's case sensitivity. An unknown name fails the same wayselectdoes, as does a name that matches more than one field when the scan is case-insensitive. Each planned task then carries a newFileScanTaskColumnStatswith the file's column sizes, value, null and NaN counts, and lower and upper bounds for only those field ids. The maps have the same names and types as the crate-internalFileMetricsin #3347, so the two convert directly.The feature is opt-in, following the concerns raised in #607. By default
FileScanTask::column_stats()isNone, planning copies nothing from the manifest entry, and serialization leaves the field out, so tasks without stats serialize exactly as they do today. Bounds serialize as their type plus the spec's single-value binary encoding, as the DataFile JSON serde does, because the JSON encoding of single values does not round-trip every float (#3409). Nothing fromkey_metadataor otherDataFilefields is exposed.The docs spell out what the statistics mean. Bounds are inclusive and may be truncated. Any of them can be missing. They describe the whole file, with no adjustment for deletes, the filter or splits. They are typed with the schema the manifest was written with, so after type promotion a bound can be an
inton alongcolumn.Public API additions are
TableScanBuilder::with_column_stats,FileScanTask::column_statsand theFileScanTaskColumnStatsstruct with its getters and builder.Are these changes tested?
Yes. A new fixture writes manifest entries with statistics on several columns, where one file has no statistics for
x. The scan test checks that the default scan and an empty request return no stats, that requestingxanddblkeeps only those entries in every map whilexstays missing for the second file, that case-insensitive resolution works, and that unknown names fail in both case modes. The serde round trip with all optional fields now includes string and decimal bounds, and a new test checks that a task without stats serializes nocolumn_statskey and reads back asNone. I also mutated the filtering, the default, case handling, unknown-name handling and serialization, and each change made a test fail.Reviewers may want to look at two choices. Nested field names are accepted, unlike
select, because manifests record stats for nested leaves. An empty column list gives each task empty statistics, asselect_emptydoes.AI Disclosure
I used Claude Code (Claude Opus 5.5) to write this change and its tests, and I reviewed it manually.