diff --git a/datafusion/catalog-listing/src/table.rs b/datafusion/catalog-listing/src/table.rs index 816e81661fc37..dd3675bd2b39d 100644 --- a/datafusion/catalog-listing/src/table.rs +++ b/datafusion/catalog-listing/src/table.rs @@ -32,7 +32,7 @@ use datafusion_datasource::file_sink_config::{FileOutputMode, FileSinkConfig}; #[expect(deprecated)] use datafusion_datasource::schema_adapter::SchemaAdapterFactory; use datafusion_datasource::{ - ListingTableUrl, PartitionedFile, TableSchema, compute_all_files_statistics, + ListingTableUrl, PartitionedFile, TableSchemaBuilder, compute_all_files_statistics, }; use datafusion_execution::cache::TableScopedPath; use datafusion_execution::cache::cache_manager::FileStatisticsCache; @@ -321,7 +321,7 @@ impl ListingTable { /// Creates a file source for this table fn create_file_source(&self) -> Arc { - let table_schema = TableSchema::builder(Arc::clone(&self.file_schema)) + let table_schema = TableSchemaBuilder::from(&self.file_schema) .with_table_partition_cols( self.options .table_partition_cols diff --git a/datafusion/core/src/datasource/file_format/mod.rs b/datafusion/core/src/datasource/file_format/mod.rs index abf3400b2f383..c46b472bd6404 100644 --- a/datafusion/core/src/datasource/file_format/mod.rs +++ b/datafusion/core/src/datasource/file_format/mod.rs @@ -67,7 +67,7 @@ pub(crate) mod test_util { .await? }; - let table_schema = TableSchema::from(file_schema.clone()); + let table_schema = TableSchema::from(&file_schema); let statistics = format .infer_stats(state, &store, file_schema.clone(), &meta) diff --git a/datafusion/core/src/datasource/physical_plan/avro.rs b/datafusion/core/src/datasource/physical_plan/avro.rs index 3a63ca116e2db..c9ee2cc407783 100644 --- a/datafusion/core/src/datasource/physical_plan/avro.rs +++ b/datafusion/core/src/datasource/physical_plan/avro.rs @@ -34,7 +34,7 @@ mod tests { use datafusion_common::{Result, ScalarValue, test_util}; use datafusion_datasource::file_format::FileFormat; use datafusion_datasource::file_scan_config::FileScanConfigBuilder; - use datafusion_datasource::{PartitionedFile, TableSchema}; + use datafusion_datasource::{PartitionedFile, TableSchemaBuilder}; use datafusion_datasource_avro::AvroFormat; use datafusion_datasource_avro::source::AvroSource; use datafusion_execution::object_store::ObjectStoreUrl; @@ -223,7 +223,7 @@ mod tests { partitioned_file.partition_values = vec![ScalarValue::from("2021-10-26")]; let projection = Some(vec![0, 1, file_schema.fields().len(), 2]); - let table_schema = TableSchema::builder(file_schema.clone()) + let table_schema = TableSchemaBuilder::from(file_schema) .with_table_partition_cols(vec![Arc::new(Field::new( "date", DataType::Utf8, diff --git a/datafusion/core/src/datasource/physical_plan/csv.rs b/datafusion/core/src/datasource/physical_plan/csv.rs index 241bbbf7f7266..56642d583e414 100644 --- a/datafusion/core/src/datasource/physical_plan/csv.rs +++ b/datafusion/core/src/datasource/physical_plan/csv.rs @@ -122,7 +122,7 @@ mod tests { quote: b'"', ..Default::default() }; - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let source = Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options)); let config = @@ -194,7 +194,7 @@ mod tests { quote: b'"', ..Default::default() }; - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let source = Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options)); let config = @@ -265,7 +265,7 @@ mod tests { quote: b'"', ..Default::default() }; - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let source = Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options)); let config = @@ -335,7 +335,7 @@ mod tests { quote: b'"', ..Default::default() }; - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let source = Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options)); let config = @@ -371,7 +371,7 @@ mod tests { file_compression_type: FileCompressionType, ) -> Result<()> { use datafusion_common::ScalarValue; - use datafusion_datasource::TableSchema; + use datafusion_datasource::TableSchemaBuilder; let session_ctx = SessionContext::new(); let task_ctx = session_ctx.task_ctx(); @@ -400,7 +400,7 @@ mod tests { quote: b'"', ..Default::default() }; - let table_schema = TableSchema::builder(Arc::clone(&file_schema)) + let table_schema = TableSchemaBuilder::from(&file_schema) .with_table_partition_cols(vec![Arc::new(Field::new( "date", DataType::Utf8, @@ -511,7 +511,7 @@ mod tests { quote: b'"', ..Default::default() }; - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let source = Arc::new(CsvSource::new(table_schema.clone()).with_csv_options(options)); let config = diff --git a/datafusion/core/src/datasource/physical_plan/parquet.rs b/datafusion/core/src/datasource/physical_plan/parquet.rs index ef2eb0d93b312..87e7fb1af4dd5 100644 --- a/datafusion/core/src/datasource/physical_plan/parquet.rs +++ b/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::{PartitionedFile, TableSchema}; + use datafusion_datasource::{PartitionedFile, TableSchemaBuilder}; use datafusion_datasource_parquet::source::ParquetSource; use datafusion_datasource_parquet::{ DefaultParquetFileReaderFactory, ParquetFileReaderFactory, ParquetFormat, @@ -1642,7 +1642,7 @@ mod tests { ), ]); - let table_schema = TableSchema::builder(Arc::clone(&schema)) + let table_schema = TableSchemaBuilder::from(&schema) .with_table_partition_cols(vec![ Arc::new(Field::new("year", DataType::Utf8, false)), Arc::new(Field::new("month", DataType::UInt8, false)), diff --git a/datafusion/core/tests/physical_optimizer/projection_pushdown.rs b/datafusion/core/tests/physical_optimizer/projection_pushdown.rs index d8486c9daa2d6..9f83f070d0286 100644 --- a/datafusion/core/tests/physical_optimizer/projection_pushdown.rs +++ b/datafusion/core/tests/physical_optimizer/projection_pushdown.rs @@ -25,7 +25,7 @@ use datafusion::datasource::physical_plan::CsvSource; use datafusion::datasource::source::DataSourceExec; use datafusion_common::config::{ConfigOptions, CsvOptions}; use datafusion_common::{JoinSide, JoinType, NullEquality, Result, ScalarValue}; -use datafusion_datasource::TableSchema; +use datafusion_datasource::TableSchemaBuilder; use datafusion_datasource::file_scan_config::FileScanConfigBuilder; use datafusion_execution::object_store::ObjectStoreUrl; use datafusion_execution::{SendableRecordBatchStream, TaskContext}; @@ -1574,7 +1574,7 @@ fn partitioned_data_source() -> Arc { quote: b'"', ..Default::default() }; - let table_schema = TableSchema::builder(Arc::clone(&file_schema)) + let table_schema = TableSchemaBuilder::from(&file_schema) .with_table_partition_cols(vec![Arc::new(Field::new( "partition_col", DataType::Utf8, diff --git a/datafusion/datasource-arrow/src/file_format.rs b/datafusion/datasource-arrow/src/file_format.rs index e523b456ea1d1..1a3e0210145f8 100644 --- a/datafusion/datasource-arrow/src/file_format.rs +++ b/datafusion/datasource-arrow/src/file_format.rs @@ -37,7 +37,6 @@ use datafusion_common::{ internal_datafusion_err, not_impl_err, }; use datafusion_common_runtime::{JoinSet, SpawnedTask}; -use datafusion_datasource::TableSchema; use datafusion_datasource::display::FileGroupDisplay; use datafusion_datasource::file::FileSource; use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder}; @@ -45,6 +44,7 @@ use datafusion_datasource::sink::{DataSink, DataSinkExec}; use datafusion_datasource::write::{ ObjectWriterBuilder, SharedBuffer, get_writer_schema, }; +use datafusion_datasource::{TableSchema, TableSchemaBuilder}; use datafusion_execution::{SendableRecordBatchStream, TaskContext}; use datafusion_expr::dml::InsertOp; use datafusion_physical_expr_common::sort_expr::LexRequirement; @@ -197,7 +197,7 @@ impl FileFormat for ArrowFormat { .object_meta .location; - let table_schema = TableSchema::builder(Arc::clone(conf.file_schema())) + let table_schema = TableSchemaBuilder::from(conf.file_schema()) .with_table_partition_cols(conf.table_partition_cols().clone()) .build(); diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index 9eb5bf70ebbb1..09e77638776e5 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -1415,7 +1415,7 @@ mod test { stats::Precision, }; use datafusion_datasource::morsel::{Morsel, Morselizer}; - use datafusion_datasource::{PartitionedFile, TableSchema}; + use datafusion_datasource::{PartitionedFile, TableSchema, TableSchemaBuilder}; use datafusion_expr::{col, lit}; use datafusion_physical_expr::{ PhysicalExpr, @@ -1882,7 +1882,7 @@ mod test { Field::new("a", DataType::Int32, false), ])); - let table_schema_for_opener = TableSchema::builder(file_schema.clone()) + let table_schema_for_opener = TableSchemaBuilder::from(&file_schema) .with_table_partition_cols(vec![Arc::new(Field::new( "part", DataType::Int32, @@ -1954,7 +1954,7 @@ mod test { Field::new("a", DataType::Int32, false), Field::new("b", DataType::Float32, true), ])); - let table_schema_for_opener = TableSchema::builder(file_schema.clone()) + let table_schema_for_opener = TableSchemaBuilder::from(&file_schema) .with_table_partition_cols(vec![Arc::new(Field::new( "part", DataType::Int32, @@ -2029,7 +2029,7 @@ mod test { Field::new("a", DataType::Int32, false), ])); - let table_schema_for_opener = TableSchema::builder(file_schema.clone()) + let table_schema_for_opener = TableSchemaBuilder::from(&file_schema) .with_table_partition_cols(vec![Arc::new(Field::new( "part", DataType::Int32, @@ -2113,7 +2113,7 @@ mod test { Field::new("part", DataType::Int32, false), ])); - let table_schema_for_opener = TableSchema::builder(file_schema.clone()) + let table_schema_for_opener = TableSchemaBuilder::from(&file_schema) .with_table_partition_cols(vec![Arc::new(Field::new( "part", DataType::Int32, diff --git a/datafusion/datasource/src/file_scan_config/mod.rs b/datafusion/datasource/src/file_scan_config/mod.rs index 83ff1bd257307..3ebd588a0770f 100644 --- a/datafusion/datasource/src/file_scan_config/mod.rs +++ b/datafusion/datasource/src/file_scan_config/mod.rs @@ -1425,9 +1425,9 @@ mod tests { use std::collections::HashMap; use super::*; - use crate::TableSchema; use crate::source::DataSourceExec; use crate::test_util::col; + use crate::{TableSchema, TableSchemaBuilder}; use crate::{ generate_test_files, test_util::MockSource, tests::aggr_test_schema, verify_sort_integrity, @@ -1875,7 +1875,7 @@ mod tests { let file_schema = aggr_test_schema(); let object_store_url = ObjectStoreUrl::parse("test:///").unwrap(); - let table_schema = TableSchema::builder(Arc::clone(&file_schema)) + let table_schema = TableSchemaBuilder::from(&file_schema) .with_table_partition_cols(vec![Arc::new(Field::new( "date", wrap_partition_type_in_dict(DataType::Utf8), @@ -1943,7 +1943,7 @@ mod tests { let file_schema = aggr_test_schema(); let object_store_url = ObjectStoreUrl::parse("test:///").unwrap(); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); // Create a file source with a filter let file_source: Arc = Arc::new( @@ -1996,7 +1996,7 @@ mod tests { let file_schema = aggr_test_schema(); let object_store_url = ObjectStoreUrl::parse("test:///").unwrap(); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source: Arc = Arc::new(MockSource::new(table_schema.clone())); @@ -2058,7 +2058,7 @@ mod tests { )]; let file = PartitionedFile::new("test_file.parquet", 100); - let table_schema = TableSchema::builder(Arc::clone(&schema)) + let table_schema = TableSchemaBuilder::from(&schema) .with_table_partition_cols( partition_cols .iter() @@ -2314,7 +2314,7 @@ mod tests { let file_group = FileGroup::new(vec![PartitionedFile::new("test.parquet", 1024)]) .with_statistics(Arc::new(file_group_stats)); - let table_schema = TableSchema::from(Arc::clone(&schema)); + let table_schema = TableSchema::from(&schema); // Create a FileScanConfig with projection: only keep columns 0 and 2 let config = FileScanConfigBuilder::new( @@ -2545,7 +2545,7 @@ mod tests { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(InexactSortPushdownSource::new(table_schema)); let file_groups = vec![FileGroup::new(vec![ @@ -2660,7 +2660,7 @@ mod tests { fn sort_pushdown_unsupported_source_files_get_sorted() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(MockSource::new(table_schema)); let file_groups = vec![FileGroup::new(vec![ @@ -2694,7 +2694,7 @@ mod tests { fn sort_pushdown_unsupported_source_already_sorted() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(MockSource::new(table_schema)); let file_groups = vec![FileGroup::new(vec![ @@ -2718,7 +2718,7 @@ mod tests { fn sort_pushdown_unsupported_source_descending_sort() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(MockSource::new(table_schema)); let file_groups = vec![FileGroup::new(vec![ @@ -2757,7 +2757,7 @@ mod tests { fn sort_pushdown_exact_source_non_overlapping_returns_exact() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(ExactSortPushdownSource::new(table_schema)); let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); @@ -2791,7 +2791,7 @@ mod tests { fn sort_pushdown_exact_source_overlapping_downgraded_to_inexact() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(ExactSortPushdownSource::new(table_schema)); let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); @@ -2825,7 +2825,7 @@ mod tests { fn sort_pushdown_exact_source_out_of_order_returns_exact() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(ExactSortPushdownSource::new(table_schema)); let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); @@ -2863,7 +2863,7 @@ mod tests { fn sort_pushdown_unsupported_source_single_file_groups() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(MockSource::new(table_schema)); let file_groups = vec![ @@ -2889,7 +2889,7 @@ mod tests { fn sort_pushdown_unsupported_source_multiple_groups() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(MockSource::new(table_schema)); let file_groups = vec![ @@ -2929,7 +2929,7 @@ mod tests { fn sort_pushdown_unsupported_source_partial_statistics() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(MockSource::new(table_schema)); let file_groups = vec![ @@ -2969,7 +2969,7 @@ mod tests { fn sort_pushdown_inexact_source_with_statistics_sorting() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(InexactSortPushdownSource::new(table_schema)); let file_groups = vec![FileGroup::new(vec![ @@ -3006,7 +3006,7 @@ mod tests { // time (all values in group 0 < group 1), degrading to single-threaded I/O. let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(ExactSortPushdownSource::new(table_schema)); let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); @@ -3064,7 +3064,7 @@ mod tests { // sorting (which would undo the reversal). The result is Inexact. let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, false)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(InexactSortPushdownSource::new(table_schema)); let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); @@ -3131,7 +3131,7 @@ mod tests { // Should NOT upgrade to Exact — NULLs would appear in wrong position. let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, true)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(MockSource::new(table_schema)); let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); @@ -3164,7 +3164,7 @@ mod tests { // Files are non-overlapping, no NULLs → should upgrade to Exact let file_schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Float64, true)])); - let table_schema = TableSchema::from(Arc::clone(&file_schema)); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(MockSource::new(table_schema)); let sort_expr = PhysicalSortExpr::new_default(Arc::new(Column::new("a", 0))); diff --git a/datafusion/datasource/src/table_schema.rs b/datafusion/datasource/src/table_schema.rs index 76a3748b34657..8b6d18b0e5058 100644 --- a/datafusion/datasource/src/table_schema.rs +++ b/datafusion/datasource/src/table_schema.rs @@ -181,6 +181,12 @@ impl From for TableSchema { } } +impl From<&SchemaRef> for TableSchema { + fn from(schema: &SchemaRef) -> Self { + TableSchemaBuilder::new(Arc::clone(schema)).build() + } +} + /// Builder for [`TableSchema`]. /// /// The file schema is the only required input; partition columns are optional. @@ -237,6 +243,18 @@ impl TableSchemaBuilder { } } +impl From for TableSchemaBuilder { + fn from(schema: SchemaRef) -> Self { + TableSchemaBuilder::new(schema) + } +} + +impl From<&SchemaRef> for TableSchemaBuilder { + fn from(schema: &SchemaRef) -> Self { + TableSchemaBuilder::new(Arc::clone(schema)) + } +} + #[cfg(test)] mod tests { use super::{TableSchema, TableSchemaBuilder}; diff --git a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs index 71c3875b547a1..d88a360422b05 100644 --- a/datafusion/proto/tests/cases/roundtrip_physical_plan.rs +++ b/datafusion/proto/tests/cases/roundtrip_physical_plan.rs @@ -103,8 +103,8 @@ use datafusion_common::{ DataFusionError, NullEquality, Result, UnnestOptions, exec_datafusion_err, internal_datafusion_err, internal_err, not_impl_err, }; -use datafusion_datasource::TableSchema; use datafusion_datasource::file::FileSource; +use datafusion_datasource::{TableSchema, TableSchemaBuilder}; use datafusion_expr::async_udf::{AsyncScalarUDF, AsyncScalarUDFImpl}; use datafusion_expr::dml::InsertOp; use datafusion_expr::{ @@ -1008,7 +1008,7 @@ fn roundtrip_arrow_scan() -> Result<()> { let file_schema = Arc::new(Schema::new(vec![Field::new("col", DataType::Utf8, false)])); - let table_schema = TableSchema::from(file_schema.clone()); + let table_schema = TableSchema::from(&file_schema); let file_source = Arc::new(ArrowSource::new_file_source(table_schema)); let scan_config = @@ -1035,7 +1035,7 @@ async fn roundtrip_parquet_exec_with_table_partition_cols() -> Result<()> { vec![wrap_partition_value_in_dict(ScalarValue::Int64(Some(0)))]; let schema = Arc::new(Schema::new(vec![Field::new("col", DataType::Utf8, false)])); - let table_schema = TableSchema::builder(schema.clone()) + let table_schema = TableSchemaBuilder::from(&schema) .with_table_partition_cols(vec![Arc::new(Field::new( "part".to_string(), wrap_partition_type_in_dict(DataType::Int16),