comphead commented on code in PR #3424:
URL: https://github.com/apache/iceberg-rust/pull/3424#discussion_r4263759122
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -103,6 +103,26 @@ fn collect_scan_field_ids(
.collect()
}
+fn collect_column_stats_field_ids(
Review 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.
##########
crates/iceberg/src/scan/mod.rs:
##########
@@ -1106,6 +1150,112 @@ mod tests {
);
}
+ /// 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);
+
+ fn column_stats(columns: &[ColumnStat]) -> FileScanTaskColumnStats {
+ FileScanTaskColumnStats::builder()
+ .with_value_counts(columns.iter().map(|c| (c.0, c.1)).collect())
+ .with_null_value_counts(columns.iter().map(|c| (c.0,
c.2)).collect())
+ .with_nan_value_counts(columns.iter().filter_map(|c| Some((c.0,
c.3?))).collect())
+ .with_lower_bounds(columns.iter().map(|c| (c.0,
c.4.clone())).collect())
+ .with_upper_bounds(columns.iter().map(|c| (c.0,
c.5.clone())).collect())
+ .build()
+ }
+
+ async fn column_stats_by_file(scan: TableScan) ->
Vec<Option<FileScanTaskColumnStats>> {
+ let mut tasks: Vec<_> = scan
+ .plan_files()
+ .await
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ tasks.sort_by(|a, b| a.data_file_path().cmp(b.data_file_path()));
+ tasks
+ .into_iter()
+ .map(|task| task.column_stats().cloned())
+ .collect()
+ }
+
+ #[tokio::test]
+ async fn test_plan_files_retains_requested_column_stats() {
Review 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.
##########
crates/iceberg/src/scan/context.rs:
##########
@@ -52,6 +52,7 @@ pub(crate) struct ManifestFileContext {
partition_spec: Option<PartitionSpecRef>,
unified_partition_type: Option<Arc<StructType>>,
sort_orders: Arc<HashMap<i64, SortOrderRef>>,
+ column_stats_field_ids: Option<Arc<Vec<i32>>>,
Review 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`.
##########
crates/iceberg/src/scan/task.rs:
##########
@@ -760,6 +826,96 @@ impl From<FileScanTaskDeleteFile> for
Result<FileScanTaskDeleteFile> {
}
}
+/// Column statistics of a [`FileScanTask`]'s data file, from its manifest
entry.
+///
+/// Each map is keyed by field id and holds only the columns requested for the
scan. A
+/// requested column missing from a map has unknown statistics of that kind:
metrics mode
+/// `none` records nothing for the column, while metrics mode `counts` and an
all-null column
+/// record counts but no bounds.
+///
+/// Bounds are inclusive. A truncated lower bound is still less than or equal
to every value in
+/// the file, and a truncated upper bound is still greater than or equal to
every value. Bounds
+/// are typed with the schema the manifest was written with, so after type
promotion a bound
+/// can be an `int` while the scan schema says `long`.
+///
+/// The statistics describe the whole data file: they are not adjusted for
delete files, the
+/// scan's filter, or the byte range of a split task.
+#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize,
TypedBuilder)]
+#[builder(field_defaults(default, setter(prefix = "with_")))]
+pub struct FileScanTaskColumnStats {
+ /// Number of values per field id, including nulls and NaNs.
+ #[serde(default, skip_serializing_if = "HashMap::is_empty")]
+ value_counts: HashMap<i32, u64>,
+
+ /// Number of null values per field id.
+ #[serde(default, skip_serializing_if = "HashMap::is_empty")]
+ null_value_counts: HashMap<i32, u64>,
+
+ /// Number of NaN values per field id.
+ #[serde(default, skip_serializing_if = "HashMap::is_empty")]
+ nan_value_counts: HashMap<i32, u64>,
+
+ /// Inclusive lower bound per field id.
+ #[serde(
+ default,
+ skip_serializing_if = "HashMap::is_empty",
+ with = "crate::scan::task::_serde::bounds"
+ )]
+ lower_bounds: HashMap<i32, Datum>,
+
+ /// Inclusive upper bound per field id.
+ #[serde(
+ default,
+ skip_serializing_if = "HashMap::is_empty",
+ with = "crate::scan::task::_serde::bounds"
+ )]
+ upper_bounds: HashMap<i32, Datum>,
+}
+
+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> {
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
