(
&self,
state: &dyn Session,
projection: Option<&Vec<usize>>,
_filters: &[Expr],
limit: Option<usize>,
)
| 89 | } |
| 90 | |
| 91 | async fn scan( |
| 92 | &self, |
| 93 | state: &dyn Session, |
| 94 | projection: Option<&Vec<usize>>, |
| 95 | _filters: &[Expr], |
| 96 | limit: Option<usize>, |
| 97 | ) -> Result<Arc<dyn ExecutionPlan>> { |
| 98 | let physical_sort = if !self.sort_order.is_empty() { |
| 99 | let df_schema = DFSchema::try_from(Arc::clone(&self.schema))?; |
| 100 | let eqp = state.execution_props(); |
| 101 | |
| 102 | let original_sort_exprs = |
| 103 | create_physical_sort_exprs(&self.sort_order, &df_schema, eqp)?; |
| 104 | |
| 105 | if let Some(p) = projection { |
| 106 | // When performing a projection, the output columns will not match |
| 107 | // the original physical sort expression indices. Also the sort columns |
| 108 | // may not be in the output projection. To correct for these issues |
| 109 | // we need to project the ordering based on the output schema. |
| 110 | let schema = Arc::new(self.schema.project(p)?); |
| 111 | LexOrdering::new(original_sort_exprs) |
| 112 | .and_then(|lex_ordering| project_ordering(&lex_ordering, &schema)) |
| 113 | .map(|lex_ordering| lex_ordering.to_vec()) |
| 114 | .unwrap_or_default() |
| 115 | } else { |
| 116 | original_sort_exprs |
| 117 | } |
| 118 | } else { |
| 119 | vec![] |
| 120 | }; |
| 121 | |
| 122 | Ok(Arc::new(StreamingTableExec::try_new( |
| 123 | Arc::clone(&self.schema), |
| 124 | self.partitions.clone(), |
| 125 | projection, |
| 126 | LexOrdering::new(physical_sort), |
| 127 | self.infinite, |
| 128 | limit, |
| 129 | )?)) |
| 130 | } |
| 131 | } |
nothing calls this directly
no test coverage detected