(
&self,
input: Arc<dyn ExecutionPlan>,
state: &dyn Session,
conf: FileSinkConfig,
order_requirements: Option<LexRequirement>,
)
| 454 | } |
| 455 | |
| 456 | async fn create_writer_physical_plan( |
| 457 | &self, |
| 458 | input: Arc<dyn ExecutionPlan>, |
| 459 | state: &dyn Session, |
| 460 | conf: FileSinkConfig, |
| 461 | order_requirements: Option<LexRequirement>, |
| 462 | ) -> Result<Arc<dyn ExecutionPlan>> { |
| 463 | if conf.insert_op != InsertOp::Append { |
| 464 | return not_impl_err!("Overwrites are not implemented yet for CSV"); |
| 465 | } |
| 466 | |
| 467 | // `has_header` and `newlines_in_values` fields of CsvOptions may inherit |
| 468 | // their values from session from configuration settings. To support |
| 469 | // this logic, writer options are built from the copy of `self.options` |
| 470 | // with updated values of these special fields. |
| 471 | let has_header = self |
| 472 | .options() |
| 473 | .has_header |
| 474 | .unwrap_or_else(|| state.config_options().catalog.has_header); |
| 475 | let newlines_in_values = self |
| 476 | .options() |
| 477 | .newlines_in_values |
| 478 | .unwrap_or_else(|| state.config_options().catalog.newlines_in_values); |
| 479 | |
| 480 | let options = self |
| 481 | .options() |
| 482 | .clone() |
| 483 | .with_has_header(has_header) |
| 484 | .with_newlines_in_values(newlines_in_values); |
| 485 | |
| 486 | let writer_options = CsvWriterOptions::try_from(&options)?; |
| 487 | |
| 488 | let sink = Arc::new(CsvSink::new(conf, writer_options)); |
| 489 | |
| 490 | Ok(Arc::new(DataSinkExec::new(input, sink, order_requirements)) as _) |
| 491 | } |
| 492 | |
| 493 | fn file_source(&self, table_schema: TableSchema) -> Arc<dyn FileSource> { |
| 494 | let mut csv_options = self.options.clone(); |
no test coverage detected