| 130 | } |
| 131 | |
| 132 | fn execute( |
| 133 | &self, |
| 134 | partition: usize, |
| 135 | _context: Arc<TaskContext>, |
| 136 | ) -> DFResult<SendableRecordBatchStream> { |
| 137 | let splits = Arc::clone(self.planned_partitions.get(partition).ok_or_else(|| { |
| 138 | datafusion::error::DataFusionError::Internal(format!( |
| 139 | "PaimonTableScan: partition index {partition} out of range (total {})", |
| 140 | self.planned_partitions.len() |
| 141 | )) |
| 142 | })?); |
| 143 | |
| 144 | let table = self.table.clone(); |
| 145 | let schema = self.schema(); |
| 146 | let projected_columns = self.projected_columns.clone(); |
| 147 | let pushed_predicate = self.pushed_predicate.clone(); |
| 148 | |
| 149 | let fut = async move { |
| 150 | let mut read_builder = table.new_read_builder(); |
| 151 | |
| 152 | if let Some(ref columns) = projected_columns { |
| 153 | let col_refs: Vec<&str> = columns.iter().map(|s| s.as_str()).collect(); |
| 154 | read_builder.with_projection(&col_refs); |
| 155 | } |
| 156 | if let Some(filter) = pushed_predicate { |
| 157 | read_builder.with_filter(filter); |
| 158 | } |
| 159 | |
| 160 | let read = read_builder.new_read().map_err(to_datafusion_error)?; |
| 161 | let stream = read.to_arrow(&splits).map_err(to_datafusion_error)?; |
| 162 | let stream = stream.map(|r| r.map_err(to_datafusion_error)); |
| 163 | |
| 164 | Ok::<_, datafusion::error::DataFusionError>(RecordBatchStreamAdapter::new( |
| 165 | schema, |
| 166 | Box::pin(stream), |
| 167 | )) |
| 168 | }; |
| 169 | |
| 170 | Ok(Box::pin(RecordBatchStreamAdapter::new( |
| 171 | self.schema(), |
| 172 | futures::stream::once(fut).try_flatten(), |
| 173 | ))) |
| 174 | } |
| 175 | |
| 176 | fn partition_statistics(&self, partition: Option<usize>) -> DFResult<Statistics> { |
| 177 | let partitions: &[Arc<[DataSplit]>] = match partition { |