(
&self,
state: &dyn Session,
args: ScanArgs<'a>,
)
| 472 | } |
| 473 | |
| 474 | async fn scan_with_args<'a>( |
| 475 | &self, |
| 476 | state: &dyn Session, |
| 477 | args: ScanArgs<'a>, |
| 478 | ) -> datafusion_common::Result<ScanResult> { |
| 479 | let projection = args.projection().map(|p| p.to_vec()); |
| 480 | let filters = args.filters().map(|f| f.to_vec()).unwrap_or_default(); |
| 481 | let limit = args.limit(); |
| 482 | |
| 483 | // extract types of partition columns |
| 484 | let table_partition_cols = self |
| 485 | .options |
| 486 | .table_partition_cols |
| 487 | .iter() |
| 488 | .map(|col| Ok(Arc::new(self.table_schema.field_with_name(&col.0)?.clone()))) |
| 489 | .collect::<datafusion_common::Result<Vec<_>>>()?; |
| 490 | |
| 491 | let table_partition_col_names = table_partition_cols |
| 492 | .iter() |
| 493 | .map(|field| field.name().as_str()) |
| 494 | .collect::<Vec<_>>(); |
| 495 | |
| 496 | // If the filters can be resolved using only partition cols, there is no need to |
| 497 | // pushdown it to TableScan, otherwise, `unhandled` pruning predicates will be generated |
| 498 | let (partition_filters, filters): (Vec<_>, Vec<_>) = |
| 499 | filters.iter().cloned().partition(|filter| { |
| 500 | can_be_evaluated_for_partition_pruning(&table_partition_col_names, filter) |
| 501 | }); |
| 502 | |
| 503 | // We should not limit the number of partitioned files to scan if there are filters and limit |
| 504 | // at the same time. This is because the limit should be applied after the filters are applied. |
| 505 | let statistic_file_limit = if filters.is_empty() { limit } else { None }; |
| 506 | |
| 507 | let ListFilesResult { |
| 508 | file_groups: mut partitioned_file_lists, |
| 509 | statistics, |
| 510 | grouped_by_partition: partitioned_by_file_group, |
| 511 | } = self |
| 512 | .list_files_for_scan(state, &partition_filters, statistic_file_limit) |
| 513 | .await?; |
| 514 | |
| 515 | // if no files need to be read, return an `EmptyExec` |
| 516 | if partitioned_file_lists.is_empty() { |
| 517 | let projected_schema = project_schema(&self.schema(), projection.as_ref())?; |
| 518 | return Ok(ScanResult::new(Arc::new(EmptyExec::new(projected_schema)))); |
| 519 | } |
| 520 | |
| 521 | let output_ordering = self.try_create_output_ordering( |
| 522 | state.execution_props(), |
| 523 | &partitioned_file_lists, |
| 524 | )?; |
| 525 | match state |
| 526 | .config_options() |
| 527 | .execution |
| 528 | .split_file_groups_by_statistics |
| 529 | .then(|| { |
| 530 | output_ordering.first().map(|output_ordering| { |
| 531 | FileScanConfig::split_groups_by_statistics_with_target_partitions( |
no test coverage detected