| 255 | } |
| 256 | |
| 257 | fn execute( |
| 258 | &self, |
| 259 | partition: usize, |
| 260 | ctx: Arc<TaskContext>, |
| 261 | ) -> Result<SendableRecordBatchStream> { |
| 262 | let stream = self.partitions[partition].execute(Arc::clone(&ctx)); |
| 263 | let projected_stream = match self.projection.clone() { |
| 264 | Some(projection) => Box::pin(RecordBatchStreamAdapter::new( |
| 265 | Arc::clone(&self.projected_schema), |
| 266 | stream.map(move |x| { |
| 267 | x.and_then(|b| b.project(projection.as_ref()).map_err(Into::into)) |
| 268 | }), |
| 269 | )), |
| 270 | None => stream, |
| 271 | }; |
| 272 | let stream = make_cooperative(projected_stream); |
| 273 | |
| 274 | Ok(match self.limit { |
| 275 | None => stream, |
| 276 | Some(fetch) => { |
| 277 | let baseline_metrics = BaselineMetrics::new(&self.metrics, partition); |
| 278 | Box::pin(LimitStream::new(stream, 0, Some(fetch), baseline_metrics)) |
| 279 | } |
| 280 | }) |
| 281 | } |
| 282 | |
| 283 | /// Tries to embed `projection` to its input (`streaming table`). |
| 284 | /// If possible, returns [`StreamingTableExec`] as the top plan. Otherwise, |