zhuqi-lucas commented on code in PR #27163:
URL: https://github.com/apache/datafusion/pull/27163#discussion_r4501936828
##########
datafusion/datasource-parquet/src/metadata.rs:
##########
@@ -1181,71 +1183,276 @@ pub(crate) fn lex_ordering_to_sorting_columns(
.collect()
}
-/// Extracts ordering information from Parquet metadata.
+/// Extracts the common leading sort keys whose order holds across every row
group.
///
-/// This function reads the sorting_columns from the first row group's metadata
-/// and converts them into a [`LexOrdering`] that can be used by the query
engine.
-///
-/// # Arguments
-/// * `metadata` - The Parquet metadata containing sorting_columns information
-/// * `schema` - The Arrow schema to use for column lookup
-///
-/// # Returns
-/// * `Ok(Some(ordering))` if valid ordering information was found
-/// * `Ok(None)` if no sorting columns were specified or they couldn't be
resolved
+/// Parquet declares ordering within each row group. File ordering also
requires
+/// ordered boundaries between row groups. Unknown or incompatible metadata
+/// leaves only the leading sort keys that can be proved.
pub fn ordering_from_parquet_metadata(
metadata: &ParquetMetaData,
schema: &SchemaRef,
) -> Result<Option<LexOrdering>> {
- // Get the sorting columns from the first row group metadata.
- // If no row groups exist or no sorting columns are specified, return None.
- let sorting_columns = metadata
+ let groups = metadata
.row_groups()
- .first()
- .and_then(|rg| rg.sorting_columns())
- .filter(|cols| !cols.is_empty());
-
- let Some(sorting_columns) = sorting_columns else {
+ .iter()
+ .filter(|group| group.num_rows() > 0)
+ .collect::<Vec<_>>();
+ let Some(first) = groups.first() else {
return Ok(None);
};
-
- let parquet_schema = metadata.file_metadata().schema_descr();
-
- let sort_exprs =
- sorting_columns_to_physical_exprs(sorting_columns, parquet_schema,
schema);
-
- if sort_exprs.is_empty() {
+ let mut sorting_columns = first
+ .sorting_columns()
+ .map(Vec::as_slice)
+ .unwrap_or_default();
+ for group in &groups[1..] {
+ let columns = group
+ .sorting_columns()
+ .map(Vec::as_slice)
+ .unwrap_or_default();
+ let prefix = sorting_columns
+ .iter()
+ .zip(columns)
+ .take_while(|(left, right)| left == right)
+ .count();
+ sorting_columns = &sorting_columns[..prefix];
+ }
+ if sorting_columns.is_empty() {
return Ok(None);
}
- Ok(LexOrdering::new(sort_exprs))
+ let file_metadata = metadata.file_metadata();
+ let parquet_schema = file_metadata.schema_descr();
+ let Ok(physical_schema) =
+ parquet_to_arrow_schema(parquet_schema,
file_metadata.key_value_metadata())
+ else {
+ return Ok(None);
+ };
+ let resolver = LeafResolver::new(parquet_schema, &physical_schema);
+ let mut ordering = Vec::with_capacity(sorting_columns.len());
+ let mut separated = vec![false; groups.len() - 1];
+ for column in sorting_columns {
+ let Some(column_index) = usize::try_from(column.column_idx).ok() else {
+ break;
+ };
+ let Some(parquet_column) = parquet_schema.columns().get(column_index)
else {
+ break;
+ };
+ if parquet_column.max_rep_level() != 0
+ || matches!(
+ parquet_column.sort_order(),
+ SortOrder::UNDEFINED | SortOrder::INT96_TIMESTAMP
+ )
+ {
+ break;
+ }
+ // A leaf name alone can identify a different top-level field. Nested
+ // keys need a field-access expression; never promote keys after one.
+ let [name] = parquet_column.path().parts() else {
+ break;
+ };
+ if resolver.leaf_index(name) != Some(column_index) {
+ break;
+ }
+ let Some((index, field)) = schema.column_with_name(name) else {
+ break;
+ };
+ let Ok(physical_field) = physical_schema.field_with_name(name) else {
+ break;
+ };
+ let data_type = ordering_value_type(field.data_type());
+ let physical_type = ordering_value_type(physical_field.data_type());
+ // A lossy cast can merge leading keys and invalidate later keys.
+ if data_type.is_nested() || !CastExpr::check_bigger_cast(data_type,
physical_type)
+ {
+ break;
+ }
+ let options = SortOptions {
+ descending: column.descending,
+ nulls_first: column.nulls_first,
+ };
+ if !row_group_sort_bounds(
+ metadata,
+ &groups,
+ column_index,
+ physical_field,
+ data_type,
+ options,
+ &mut separated,
+ )
+ .unwrap_or_else(|error| {
+ debug!("Cannot validate Parquet sort order: {error}");
+ false
+ }) {
+ break;
+ }
+ ordering.push(PhysicalSortExpr::new(
+ Arc::new(Column::new(name, index)),
+ options,
+ ));
+ }
+ Ok(LexOrdering::new(ordering))
}
-/// Converts Parquet sorting columns to physical sort expressions.
-fn sorting_columns_to_physical_exprs(
- sorting_columns: &[SortingColumn],
- parquet_schema: &SchemaDescriptor,
- arrow_schema: &SchemaRef,
-) -> Vec<PhysicalSortExpr> {
- use arrow::compute::SortOptions;
-
- sorting_columns
- .iter()
- .filter_map(|sc| {
- let parquet_column = parquet_schema.column(sc.column_idx as usize);
- let name = parquet_column.name();
+fn ordering_value_type(data_type: &DataType) -> &DataType {
+ match data_type {
+ DataType::Dictionary(_, value) => ordering_value_type(value),
+ _ => data_type,
+ }
+}
- // Find the column in the arrow schema
- let (index, _) = arrow_schema.column_with_name(name)?;
+/// Refine boundaries still tied on earlier keys. A strict earlier boundary
+/// remains ordered regardless of the later keys' statistics.
+fn row_group_sort_bounds<'a>(
+ metadata: &'a ParquetMetaData,
+ groups: &[&'a RowGroupMetaData],
+ column_index: usize,
+ physical_field: &'a Field,
+ data_type: &DataType,
+ options: SortOptions,
+ separated: &mut [bool],
+) -> Result<bool> {
+ let is_float = matches!(
+ ordering_value_type(physical_field.data_type()),
+ DataType::Float16 | DataType::Float32 | DataType::Float64
+ );
+ if separated.iter().all(|value| *value) && !is_float {
+ return Ok(true);
+ }
+ if groups.iter().any(|group| {
+ group
+ .columns()
+ .get(column_index)
+ .is_none_or(|column| column.num_values() != group.num_rows())
+ }) {
+ return Ok(false);
+ }
+ let file_metadata = metadata.file_metadata();
+ let parquet_schema = file_metadata.schema_descr();
+ let converter = StatisticsConverter::from_column_index(
+ column_index,
+ physical_field,
+ parquet_schema,
+ )?
+ .with_missing_null_counts_as_zero(false);
+ let mins = cast(
+ converter.row_group_mins(groups.iter().copied())?.as_ref(),
+ data_type,
+ )?;
+ let maxes = cast(
+ converter.row_group_maxes(groups.iter().copied())?.as_ref(),
+ data_type,
+ )?;
+ let null_counts = converter.row_group_null_counts(groups.iter().copied())?;
+ let null_count = |index: usize| {
+ if parquet_schema.column(column_index).max_def_level() == 0 {
+ Some(0)
+ } else {
+ null_counts
+ .is_valid(index)
+ .then(|| null_counts.value(index))
+ }
+ };
+ let all_null =
+ |index: usize| null_count(index) == Some(groups[index].num_rows() as
u64);
+ let column_order = file_metadata
+ .column_orders()
+ .and_then(|orders| orders.get(column_index));
+ let sort_order = parquet_schema.column(column_index).sort_order();
+ let modern_order_matches = column_order
+ == Some(&ColumnOrder::TYPE_DEFINED_ORDER(sort_order))
+ || is_float
+ && matches!(
+ column_order,
+ Some(
+ ColumnOrder::TYPE_DEFINED_ORDER(SortOrder::SIGNED)
+ | ColumnOrder::IEEE_754_TOTAL_ORDER
+ )
+ );
+ let bounds_trusted = |index: usize| {
+ groups[index]
+ .column(column_index)
+ .statistics()
+ .is_some_and(|stats| {
+ if stats.is_min_max_deprecated() {
+ sort_order == SortOrder::SIGNED
+ } else {
+ modern_order_matches
+ }
+ })
+ };
+ if is_float {
+ // TYPE_ORDER can equate signed zeros. NaNs can be absent from bounds
+ // even with IEEE_754_TOTAL_ORDER, including whole-file statistics.
+ for (index, group) in groups.iter().enumerate() {
+ if all_null(index) {
+ continue;
+ }
+ if !bounds_trusted(index)
+ || group
+ .column(column_index)
+ .statistics()
+ .and_then(ParquetStatistics::nan_count_opt)
Review Comment:
Files whose float statistics have no `nan_count` (e.g. written by pyarrow
23, which I checked locally) used to keep a float sort key and now lose it, so
the `SortExec` comes back for them. That seems right since NaNs are not in
min/max, but it may be worth a line in the upgrade guide.
--
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]
