(
&self,
state: &dyn Session,
conf: FileScanConfig,
)
| 419 | } |
| 420 | |
| 421 | async fn create_physical_plan( |
| 422 | &self, |
| 423 | state: &dyn Session, |
| 424 | conf: FileScanConfig, |
| 425 | ) -> Result<Arc<dyn ExecutionPlan>> { |
| 426 | // Consult configuration options for default values |
| 427 | let has_header = self |
| 428 | .options |
| 429 | .has_header |
| 430 | .unwrap_or_else(|| state.config_options().catalog.has_header); |
| 431 | let newlines_in_values = self |
| 432 | .options |
| 433 | .newlines_in_values |
| 434 | .unwrap_or_else(|| state.config_options().catalog.newlines_in_values); |
| 435 | |
| 436 | let mut csv_options = self.options.clone(); |
| 437 | csv_options.has_header = Some(has_header); |
| 438 | csv_options.newlines_in_values = Some(newlines_in_values); |
| 439 | |
| 440 | // Get the existing CsvSource and update its options |
| 441 | // We need to preserve the table_schema from the original source (which includes partition columns) |
| 442 | let csv_source = conf |
| 443 | .file_source |
| 444 | .downcast_ref::<CsvSource>() |
| 445 | .expect("file_source should be a CsvSource"); |
| 446 | let source = Arc::new(csv_source.clone().with_csv_options(csv_options)); |
| 447 | |
| 448 | let config = FileScanConfigBuilder::from(conf) |
| 449 | .with_file_compression_type(self.options.compression.into()) |
| 450 | .with_source(source) |
| 451 | .build(); |
| 452 | |
| 453 | Ok(DataSourceExec::from_data_source(config)) |
| 454 | } |
| 455 | |
| 456 | async fn create_writer_physical_plan( |
| 457 | &self, |
nothing calls this directly
no test coverage detected