zhuqi-lucas commented on code in PR #18817:
URL: https://github.com/apache/datafusion/pull/18817#discussion_r2569302094
##########
datafusion/datasource/src/file_scan_config.rs:
##########
@@ -718,6 +717,101 @@ impl DataSource for FileScanConfig {
}
}
}
+
+ fn try_pushdown_sort(
+ &self,
+ order: &[PhysicalSortExpr],
+ ) -> Result<Option<Arc<dyn DataSource>>> {
+ let current_ordering = match self.output_ordering.first() {
+ Some(ordering) => ordering.as_ref(),
+ None => return Ok(None),
+ };
+
+ // Only support reverse ordering pushdown
+ if !is_reverse_ordering(order, current_ordering) {
+ return Ok(None);
+ }
+
+ // Ask the file source if it can handle the sort pushdown
+ // (e.g., ParquetSource will enable reverse_scan)
+ let new_file_source = match self.file_source.try_pushdown_sort(order)?
{
+ Some(source) => source,
+ None => return Ok(None),
+ };
+
+ let mut new_config = self.clone();
+
+ // Reverse file groups: when scanning in reverse, we need to read files
+ // in reverse order to maintain the correct global ordering
+ new_config.file_groups = new_config
+ .file_groups
+ .into_iter()
+ .map(|group| {
+ let mut files = group.into_inner();
+ files.reverse();
+ files.into()
+ })
+ .collect();
+
+ // Build the new output ordering by reversing each sort expression's
direction
+ // E.g., [number DESC] becomes [number ASC]
+ let mut reversed_ordering = Vec::new();
+ for sort_expr in current_ordering {
+ reversed_ordering.push(PhysicalSortExpr {
+ expr: Arc::clone(&sort_expr.expr),
+ options: !sort_expr.options,
+ });
+ }
+
+ new_config.output_ordering = vec![LexOrdering::new(reversed_ordering)
+ .ok_or_else(|| {
+ DataFusionError::Plan(
+ "Failed to create ordering: invalid sort
expressions".to_string(),
+ )
+ })?];
+
+ new_config.file_source = new_file_source;
+
+ Ok(Some(Arc::new(new_config)))
+ }
+}
+
+/// Check if the requested ordering can be satisfied by reversing the current
ordering.
+///
+/// This function supports **prefix matching**: if the file has ordering [A
DESC, B ASC]
+/// and we need [A ASC], reversing the scan gives us [A ASC, B DESC], which
satisfies
+/// the requirement since [A ASC] is a prefix.
+///
+/// # Arguments
+/// * `requested` - The ordering required by the query
+/// * `current` - The natural ordering of the data source (e.g., from file
metadata)
+///
+/// # Returns
+/// `true` if reversing the current ordering would satisfy the requested
ordering
+///
+/// # Example
+/// ```text
+/// Current: [number DESC, letter ASC]
+/// Requested: [number ASC]
+/// Reversed: [number ASC, letter DESC] ✓ Prefix match!
+/// ```
+fn is_reverse_ordering(
+ requested: &[PhysicalSortExpr],
+ current: &[PhysicalSortExpr],
+) -> bool {
+ // Allow prefix matching - we can satisfy a prefix of the current ordering
+ // by reversing the scan
+ if requested.len() > current.len() {
+ return false;
+ }
+
+ requested.iter().zip(current.iter()).all(|(req, cur)| {
+ let exprs_match = req.expr.to_string() == cur.expr.to_string();
Review Comment:
Thanks @alamb , i add more comments, compare the sort exprs is not we want
here, we want the sort string without sort order. Correct me if i am wrong.
--
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]
