alamb commented on code in PR #27163:
URL: https://github.com/apache/datafusion/pull/27163#discussion_r4538858431
##########
datafusion/core/tests/parquet/ordering.rs:
##########
@@ -89,6 +98,241 @@ async fn
test_create_table_with_order_writes_sorting_columns() -> Result<()> {
Ok(())
}
+#[tokio::test]
+async fn test_parquet_unordered_row_groups_require_sort() -> Result<()> {
Review Comment:
I found these tests very hard to read -- is there some way we can convert
them to mostly/all sqllogictests?
Or at least insta snapshot tests?
for example, a test that shows:
1. Input (with relevant sorting)
2. Query that calls for sorting
3. Verifificion of the correct results and plan
##########
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
Review Comment:
this code feels much more complicated than necessary to me.
Can you add some exampels of what it is actually doing?
##########
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());
Review Comment:
is this code to handle sorting of nested columns new? Is it needed for this
PR? If so, I think we should have end to end slt/sql examples that show what is
going on
##########
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)
+ != Some(0)
+ || column_order != Some(&ColumnOrder::IEEE_754_TOTAL_ORDER)
+ && !float_bounds_exclude_zero(&mins, &maxes, index)?
+ {
+ return Ok(false);
+ }
+ }
+ }
+ let null = ScalarValue::try_new_null(data_type)?;
+ let bound = |index: usize, lower: bool| -> Result<Option<ScalarValue>> {
+ if null_count(index).is_some_and(|count| count >
groups[index].num_rows() as u64)
+ {
+ return Ok(None);
+ }
+ // Unknown null counts include NULL in the possible range. A missing
+ // min/max value, unlike this typed NULL, is not a known NULL endpoint.
+ if all_null(index) || lower == options.nulls_first &&
null_count(index) != Some(0)
+ {
+ return Ok(Some(null.clone()));
+ }
+ let values = if lower != options.descending {
+ &mins
+ } else {
+ &maxes
+ };
+ if values.is_null(index) || !bounds_trusted(index) {
+ return Ok(None);
+ }
+ Ok(Some(ScalarValue::try_from_array(values.as_ref(), index)?))
+ };
+ for (index, already_separated) in separated.iter_mut().enumerate() {
+ if *already_separated {
+ continue;
+ }
+ let (Some(upper), Some(lower)) = (bound(index, false)?, bound(index +
1, true)?)
+ else {
+ return Ok(false);
+ };
+ match compare_rows(&[upper], &[lower], &[options])? {
+ Ordering::Less => *already_separated = true,
+ Ordering::Equal => {}
+ Ordering::Greater => return Ok(false),
+ }
+ }
+ Ok(true)
+}
- let expr = Arc::new(Column::new(name, index));
- let options = SortOptions {
- descending: sc.descending,
- nulls_first: sc.nulls_first,
- };
- Some(PhysicalSortExpr::new(expr, options))
- })
- .collect()
+fn float_bounds_exclude_zero(
Review Comment:
CHanging the sorting statistics for floating point columns seems like a
substantial new behavior as well
--
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]
