| 372 | } |
| 373 | |
| 374 | fn execute(&self, _ctx: Arc<TaskContext>) -> SendableRecordBatchStream { |
| 375 | let config = Arc::clone(&self.0); |
| 376 | let schema = Arc::clone(self.0.source.schema()); |
| 377 | let mut builder = RecordBatchReceiverStreamBuilder::new(schema, 2); |
| 378 | let tx = builder.tx(); |
| 379 | builder.spawn_blocking(move || { |
| 380 | let reader = config.reader()?; |
| 381 | for b in reader { |
| 382 | if tx.blocking_send(b.map_err(Into::into)).is_err() { |
| 383 | break; |
| 384 | } |
| 385 | } |
| 386 | Ok(()) |
| 387 | }); |
| 388 | builder.build() |
| 389 | } |
| 390 | } |
| 391 | |
| 392 | #[derive(Debug)] |