(
&self,
ctx: &SessionContext,
table: &str,
)
| 277 | } |
| 278 | |
| 279 | async fn get_table( |
| 280 | &self, |
| 281 | ctx: &SessionContext, |
| 282 | table: &str, |
| 283 | ) -> Result<Arc<dyn TableProvider>> { |
| 284 | let path = self.path.to_str().unwrap(); |
| 285 | let table_format = self.file_format.as_str(); |
| 286 | let target_partitions = self.partitions(); |
| 287 | |
| 288 | // Obtain a snapshot of the SessionState |
| 289 | let state = ctx.state(); |
| 290 | let (format, path, extension): (Arc<dyn FileFormat>, String, &'static str) = |
| 291 | match table_format { |
| 292 | // dbgen creates .tbl ('|' delimited) files without header |
| 293 | "tbl" => { |
| 294 | let path = format!("{path}/{table}.tbl"); |
| 295 | |
| 296 | let format = CsvFormat::default() |
| 297 | .with_delimiter(b'|') |
| 298 | .with_has_header(false); |
| 299 | |
| 300 | (Arc::new(format), path, ".tbl") |
| 301 | } |
| 302 | "csv" => { |
| 303 | let path = format!("{path}/csv/{table}"); |
| 304 | let format = CsvFormat::default() |
| 305 | .with_delimiter(b',') |
| 306 | .with_has_header(true); |
| 307 | |
| 308 | (Arc::new(format), path, DEFAULT_CSV_EXTENSION) |
| 309 | } |
| 310 | "parquet" => { |
| 311 | let path = format!("{path}/{table}"); |
| 312 | let format = ParquetFormat::default() |
| 313 | .with_options(ctx.state().table_options().parquet.clone()); |
| 314 | |
| 315 | (Arc::new(format), path, DEFAULT_PARQUET_EXTENSION) |
| 316 | } |
| 317 | other => { |
| 318 | unimplemented!("Invalid file format '{}'", other); |
| 319 | } |
| 320 | }; |
| 321 | |
| 322 | let table_path = ListingTableUrl::parse(path)?; |
| 323 | let options = ListingOptions::new(format) |
| 324 | .with_file_extension(extension) |
| 325 | .with_target_partitions(target_partitions) |
| 326 | .with_collect_stat(state.config().collect_statistics()); |
| 327 | let schema = match table_format { |
| 328 | "parquet" => options.infer_schema(&state, &table_path).await?, |
| 329 | "tbl" => Arc::new(get_tbl_tpch_table_schema(table)), |
| 330 | "csv" => Arc::new(get_tpch_table_schema(table)), |
| 331 | _ => unreachable!(), |
| 332 | }; |
| 333 | let options = if self.sorted { |
| 334 | let key_column_name = schema.fields()[0].name(); |
| 335 | options |
| 336 | .with_file_sort_order(vec![vec![col(key_column_name).sort(true, false)]]) |
no test coverage detected