use TableSchema

Author: adriangb

datafusion-examples/examples/csv_json_opener.rs

@@ -31,7 +31,9 @@ use datafusion::{
  test_util::aggr_test_schema,
 };
 
-use datafusion::datasource::physical_plan::FileScanConfigBuilder;
+use datafusion::datasource::{
+ physical_plan::FileScanConfigBuilder, table_schema::TableSchema,
+};
 use futures::StreamExt;
 use object_store::{local::LocalFileSystem, memory::InMemory,ObjectStore};
 
@@ -67,7 +69,7 @@ async fn csv_opener() -> Result<()> {
 
 let config = CsvSource::new(true,b',',b'"')
 .with_comment(Some(b'#'))
-.with_schema(schema)
+.with_schema(TableSchema::new(schema,vec![]))
 .with_batch_size(8192)
 .with_projection(&scan_config);
 

datafusion/core/src/datasource/mod.rs

@@ -45,6 +45,7 @@ pub use datafusion_catalog::view;
 pubuse datafusion_datasource::schema_adapter;
 pubuse datafusion_datasource::sink;
 pubuse datafusion_datasource::source;
+pubuse datafusion_datasource::table_schema;
 pubuse datafusion_execution::object_store;
 pubuse datafusion_physical_expr::create_ordering;
 

datafusion/core/src/datasource/physical_plan/parquet.rs

@@ -54,7 +54,7 @@ mod tests {
 use datafusion_datasource::source::DataSourceExec;
 
 use datafusion_datasource::file::FileSource;
-use datafusion_datasource::{FileRange,PartitionedFile};
+use datafusion_datasource::{FileRange,PartitionedFile,TableSchema};
 use datafusion_datasource_parquet::source::ParquetSource;
 use datafusion_datasource_parquet::{
 DefaultParquetFileReaderFactory,ParquetFileReaderFactory,ParquetFormat,
@@ -186,7 +186,7 @@ mod tests {
  source = source.with_bloom_filter_on_read(false);
 }
 
- source.with_schema(Arc::clone(&table_schema))
+ source.with_schema(TableSchema::new(Arc::clone(&table_schema),vec![]))
 }
 
 fnbuild_parquet_exec(

datafusion/core/src/test_util/parquet.rs

@@ -186,7 +186,10 @@ impl TestParquetFile {
 ParquetSource::new(parquet_options)
 .with_predicate(Arc::clone(&physical_filter_expr)),
 )
-.with_schema(Arc::clone(&self.schema));
+.with_schema(datafusion_datasource::TableSchema::new(
+Arc::clone(&self.schema),
+vec![],
+));
 let config = scan_config_builder.with_source(source).build();
 let parquet_exec = DataSourceExec::from_data_source(config);
 

datafusion/core/tests/physical_optimizer/filter_pushdown/util.rs

@@ -24,6 +24,7 @@ use datafusion_datasource::{
  file_scan_config::FileScanConfigBuilder, file_stream::FileOpenFuture,
  file_stream::FileOpener, schema_adapter::DefaultSchemaAdapterFactory,
  schema_adapter::SchemaAdapterFactory, source::DataSourceExec,PartitionedFile,
+TableSchema,
 };
 use datafusion_physical_expr_common::physical_expr::fmt_sql;
 use datafusion_physical_optimizer::PhysicalOptimizerRule;
@@ -156,9 +157,9 @@ impl FileSource for TestSource {
 })
 }
 
-fnwith_schema(&self,schema:SchemaRef) -> Arc<dynFileSource>{
+fnwith_schema(&self,schema:TableSchema) -> Arc<dynFileSource>{
 Arc::new(TestSource{
-schema:Some(schema),
+schema:Some(schema.file_schema().clone()),
  ..self.clone()
 })
 }

datafusion/datasource-arrow/src/source.rs

@@ -20,9 +20,9 @@ use std::sync::Arc;
 
 use datafusion_datasource::as_file_source;
 use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
+use datafusion_datasource::TableSchema;
 
 use arrow::buffer::Buffer;
-use arrow::datatypes::SchemaRef;
 use arrow_ipc::reader::FileDecoder;
 use datafusion_common::error::Result;
 use datafusion_common::{exec_datafusion_err,Statistics};
@@ -73,7 +73,7 @@ impl FileSource for ArrowSource {
 Arc::new(Self{ ..self.clone()})
 }
 
-fnwith_schema(&self,_schema:SchemaRef) -> Arc<dynFileSource>{
+fnwith_schema(&self,_schema:TableSchema) -> Arc<dynFileSource>{
 Arc::new(Self{ ..self.clone()})
 }
 fnwith_statistics(&self,statistics:Statistics) -> Arc<dynFileSource>{

datafusion/datasource-avro/src/source.rs

@@ -29,6 +29,7 @@ use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_scan_config::FileScanConfig;
 use datafusion_datasource::file_stream::FileOpener;
 use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
+use datafusion_datasource::TableSchema;
 use datafusion_physical_expr_common::sort_expr::LexOrdering;
 use datafusion_physical_plan::metrics::ExecutionPlanMetricsSet;
 
@@ -84,9 +85,9 @@ impl FileSource for AvroSource {
 Arc::new(conf)
 }
 
-fnwith_schema(&self,schema:SchemaRef) -> Arc<dynFileSource>{
+fnwith_schema(&self,schema:TableSchema) -> Arc<dynFileSource>{
 letmut conf = self.clone();
- conf.schema = Some(schema);
+ conf.schema = Some(schema.file_schema().clone());
 Arc::new(conf)
 }
 fnwith_statistics(&self,statistics:Statistics) -> Arc<dynFileSource>{

datafusion/datasource-csv/src/source.rs

@@ -29,7 +29,7 @@ use datafusion_datasource::file_compression_type::FileCompressionType;
 use datafusion_datasource::file_stream::{FileOpenFuture,FileOpener};
 use datafusion_datasource::{
  as_file_source, calculate_range,FileRange,ListingTableUrl,PartitionedFile,
-RangeCalculation,
+RangeCalculation,TableSchema,
 };
 
 use arrow::csv;
@@ -258,9 +258,9 @@ impl FileSource for CsvSource {
 Arc::new(conf)
 }
 
-fnwith_schema(&self,schema:SchemaRef) -> Arc<dynFileSource>{
+fnwith_schema(&self,schema:TableSchema) -> Arc<dynFileSource>{
 letmut conf = self.clone();
- conf.file_schema = Some(schema);
+ conf.file_schema = Some(schema.file_schema().clone());
 Arc::new(conf)
 }
 

datafusion/datasource-json/src/source.rs

@@ -32,6 +32,7 @@ use datafusion_datasource::file_stream::{FileOpenFuture, FileOpener};
 use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
 use datafusion_datasource::{
  as_file_source, calculate_range,ListingTableUrl,PartitionedFile,RangeCalculation,
+TableSchema,
 };
 use datafusion_physical_plan::{ExecutionPlan,ExecutionPlanProperties};
 
@@ -122,7 +123,7 @@ impl FileSource for JsonSource {
 Arc::new(conf)
 }
 
-fnwith_schema(&self,_schema:SchemaRef) -> Arc<dynFileSource>{
+fnwith_schema(&self,_schema:TableSchema) -> Arc<dynFileSource>{
 Arc::new(Self{ ..self.clone()})
 }
 fnwith_statistics(&self,statistics:Statistics) -> Arc<dynFileSource>{

datafusion/datasource-parquet/src/file_format.rs

@@ -463,8 +463,7 @@ impl FileFormat for ParquetFormat {
  metadata_size_hint = Some(metadata);
 }
 
-letmut source = ParquetSource::new(self.options.clone())
-.with_table_partition_cols(conf.table_partition_cols().clone());
+letmut source = ParquetSource::new(self.options.clone());
 
 // Use the CachedParquetFileReaderFactory
 let metadata_cache = state.runtime_env().cache_manager.get_file_metadata_cache();