Remove SchemaAdapter

Author: adriangb

datafusion-examples/examples/custom_data_source/default_column_values.rs

@@ -63,9 +63,8 @@ const DEFAULT_VALUE_METADATA_KEY: &str = "example.default_value";
 /// 4. Use the DefaultPhysicalExprAdapter as a fallback for standard schema adaptation
 /// 5. Convert string default values to proper types using `ScalarValue::cast_to()` at planning time
 ///
-/// Important: PhysicalExprAdapter is specifically designed for rewriting filter predicates
-/// that get pushed down to file scans. For handling missing columns in projections,
-/// other mechanisms in DataFusion are used (like SchemaAdapter).
+/// Important: PhysicalExprAdapter handles rewriting both filter predicates and projection
+/// expressions for file scans, including handling missing columns.
 ///
 /// The metadata-based approach provides a flexible way to store default values as strings
 /// and cast them to the appropriate types at planning time, avoiding runtime overhead.

datafusion-examples/examples/data_io/parquet_index.rs

@@ -511,7 +511,7 @@ impl ParquetMetadataIndexBuilder {
 
         // Get the schema of the file. A real system might have to handle the
         // case where the schema of the file is not the same as the schema of
-        // the other files e.g. using SchemaAdapter.
+        // the other files e.g. using PhysicalExprAdapterFactory.
         if self.file_schema.is_none() {
             self.file_schema = Some(reader.schema().clone());
         }

datafusion/catalog-listing/src/config.rs

@@ -21,7 +21,6 @@ use datafusion_catalog::Session;
 use datafusion_common::{config_err, internal_err};
 use datafusion_datasource::ListingTableUrl;
 use datafusion_datasource::file_compression_type::FileCompressionType;
-use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
 use datafusion_physical_expr_adapter::PhysicalExprAdapterFactory;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -44,15 +43,12 @@ pub enum SchemaSource {
 /// # Schema Evolution Support
 ///
 /// This configuration supports schema evolution through the optional
-/// [`SchemaAdapterFactory`]. You might want to override the default factory when you need:
+/// [`PhysicalExprAdapterFactory`]. You might want to override the default factory when you need:
 ///
 /// - **Type coercion requirements**: When you need custom logic for converting between
 ///   different Arrow data types (e.g., Int32 ↔ Int64, Utf8 ↔ LargeUtf8)
 /// - **Column mapping**: You need to map columns with a legacy name to a new name
 /// - **Custom handling of missing columns**: By default they are filled in with nulls, but you may e.g. want to fill them in with `0` or `""`.
-///
-/// If not specified, a [`datafusion_datasource::schema_adapter::DefaultSchemaAdapterFactory`]
-/// will be used, which handles basic schema compatibility cases.
 #[derive(Debug, Clone, Default)]
 pub struct ListingTableConfig {
     /// Paths on the `ObjectStore` for creating [`crate::ListingTable`].
@@ -68,8 +64,6 @@ pub struct ListingTableConfig {
     pub options: Option<ListingOptions>,
     /// Tracks the source of the schema information
     pub(crate) schema_source: SchemaSource,
-    /// Optional [`SchemaAdapterFactory`] for creating schema adapters
-    pub(crate) schema_adapter_factory: Option<Arc<dyn SchemaAdapterFactory>>,
     /// Optional [`PhysicalExprAdapterFactory`] for creating physical expression adapters
     pub(crate) expr_adapter_factory: Option<Arc<dyn PhysicalExprAdapterFactory>>,
 }
@@ -218,8 +212,7 @@ impl ListingTableConfig {
                     file_schema,
                     options: _,
                     schema_source,
-                    schema_adapter_factory,
-                    expr_adapter_factory: physical_expr_adapter_factory,
+                    expr_adapter_factory,
                 } = self;
 
                 let (schema, new_schema_source) = match file_schema {
@@ -241,8 +234,7 @@ impl ListingTableConfig {
                     file_schema: Some(schema),
                     options: Some(options),
                     schema_source: new_schema_source,
-                    schema_adapter_factory,
-                    expr_adapter_factory: physical_expr_adapter_factory,
+                    expr_adapter_factory,
                 })
             }
             None => internal_err!("No `ListingOptions` set for inferring schema"),
@@ -282,71 +274,18 @@ impl ListingTableConfig {
                     file_schema: self.file_schema,
                     options: Some(options),
                     schema_source: self.schema_source,
-                    schema_adapter_factory: self.schema_adapter_factory,
                     expr_adapter_factory: self.expr_adapter_factory,
                 })
             }
             None => config_err!("No `ListingOptions` set for inferring schema"),
         }
     }
 
-    /// Set the [`SchemaAdapterFactory`] for the [`crate::ListingTable`]
-    ///
-    /// The schema adapter factory is used to create schema adapters that can
-    /// handle schema evolution and type conversions when reading files with
-    /// different schemas than the table schema.
-    ///
-    /// If not provided, a default schema adapter factory will be used.
-    ///
-    /// # Example: Custom Schema Adapter for Type Coercion
-    /// ```rust
-    /// # use std::sync::Arc;
-    /// # use datafusion_catalog_listing::{ListingTableConfig, ListingOptions};
-    /// # use datafusion_datasource::schema_adapter::{SchemaAdapterFactory, SchemaAdapter};
-    /// # use datafusion_datasource::ListingTableUrl;
-    /// # use datafusion_datasource_parquet::file_format::ParquetFormat;
-    /// # use arrow::datatypes::{SchemaRef, Schema, Field, DataType};
-    /// #
-    /// # #[derive(Debug)]
-    /// # struct MySchemaAdapterFactory;
-    /// # impl SchemaAdapterFactory for MySchemaAdapterFactory {
-    /// #     fn create(&self, _projected_table_schema: SchemaRef, _file_schema: SchemaRef) -> Box<dyn SchemaAdapter> {
-    /// #         unimplemented!()
-    /// #     }
-    /// # }
-    /// # let table_paths = ListingTableUrl::parse("file:///path/to/data").unwrap();
-    /// # let listing_options = ListingOptions::new(Arc::new(ParquetFormat::default()));
-    /// # let table_schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
-    /// let config = ListingTableConfig::new(table_paths)
-    ///     .with_listing_options(listing_options)
-    ///     .with_schema(table_schema)
-    ///     .with_schema_adapter_factory(Arc::new(MySchemaAdapterFactory));
-    /// ```
-    pub fn with_schema_adapter_factory(
-        self,
-        schema_adapter_factory: Arc<dyn SchemaAdapterFactory>,
-    ) -> Self {
-        Self {
-            schema_adapter_factory: Some(schema_adapter_factory),
-            ..self
-        }
-    }
-
-    /// Get the [`SchemaAdapterFactory`] for this configuration
-    pub fn schema_adapter_factory(&self) -> Option<&Arc<dyn SchemaAdapterFactory>> {
-        self.schema_adapter_factory.as_ref()
-    }
-
     /// Set the [`PhysicalExprAdapterFactory`] for the [`crate::ListingTable`]
     ///
     /// The expression adapter factory is used to create physical expression adapters that can
     /// handle schema evolution and type conversions when evaluating expressions
     /// with different schemas than the table schema.
-    ///
-    /// If not provided, a default physical expression adapter factory will be used unless a custom
-    /// `SchemaAdapterFactory` is set, in which case only the `SchemaAdapterFactory` will be used.
-    ///
-    /// See <https://github.com/apache/datafusion/issues/16800> for details on this transition.
     pub fn with_expr_adapter_factory(
         self,
         expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory>,

datafusion/catalog-listing/src/table.rs

@@ -29,7 +29,6 @@ use datafusion_datasource::file::FileSource;
 use datafusion_datasource::file_groups::FileGroup;
 use datafusion_datasource::file_scan_config::{FileScanConfig, FileScanConfigBuilder};
 use datafusion_datasource::file_sink_config::FileSinkConfig;
-use datafusion_datasource::schema_adapter::SchemaAdapterFactory;
 use datafusion_datasource::{
     ListingTableUrl, PartitionedFile, TableSchema, compute_all_files_statistics,
 };
@@ -191,8 +190,6 @@ pub struct ListingTable {
     constraints: Constraints,
     /// Column default expressions for columns that are not physically present in the data files
     column_defaults: HashMap<String, Expr>,
-    /// Optional [`SchemaAdapterFactory`] for creating schema adapters
-    schema_adapter_factory: Option<Arc<dyn SchemaAdapterFactory>>,
     /// Optional [`PhysicalExprAdapterFactory`] for creating physical expression adapters
     expr_adapter_factory: Option<Arc<dyn PhysicalExprAdapterFactory>>,
 }
@@ -235,7 +232,6 @@ impl ListingTable {
             collected_statistics: Arc::new(DefaultFileStatisticsCache::default()),
             constraints: Constraints::default(),
             column_defaults: HashMap::new(),
-            schema_adapter_factory: config.schema_adapter_factory,
             expr_adapter_factory: config.expr_adapter_factory,
         };
 
@@ -290,48 +286,8 @@ impl ListingTable {
         self.schema_source
     }
 
-    /// Set the [`SchemaAdapterFactory`] for this [`ListingTable`]
-    ///
-    /// The schema adapter factory is used to create schema adapters that can
-    /// handle schema evolution and type conversions when reading files with
-    /// different schemas than the table schema.
-    ///
-    /// # Example: Adding Schema Evolution Support
-    /// ```rust
-    /// # use std::sync::Arc;
-    /// # use datafusion_catalog_listing::{ListingTable, ListingTableConfig, ListingOptions};
-    /// # use datafusion_datasource::ListingTableUrl;
-    /// # use datafusion_datasource::schema_adapter::{DefaultSchemaAdapterFactory, SchemaAdapter};
-    /// # use datafusion_datasource_parquet::file_format::ParquetFormat;
-    /// # use arrow::datatypes::{SchemaRef, Schema, Field, DataType};
-    /// # let table_path = ListingTableUrl::parse("file:///path/to/data").unwrap();
-    /// # let options = ListingOptions::new(Arc::new(ParquetFormat::default()));
-    /// # let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
-    /// # let config = ListingTableConfig::new(table_path).with_listing_options(options).with_schema(schema);
-    /// # let table = ListingTable::try_new(config).unwrap();
-    /// let table_with_evolution = table
-    ///     .with_schema_adapter_factory(Arc::new(DefaultSchemaAdapterFactory));
-    /// ```
-    /// See [`ListingTableConfig::with_schema_adapter_factory`] for an example of custom SchemaAdapterFactory.
-    pub fn with_schema_adapter_factory(
-        self,
-        schema_adapter_factory: Arc<dyn SchemaAdapterFactory>,
-    ) -> Self {
-        Self {
-            schema_adapter_factory: Some(schema_adapter_factory),
-            ..self
-        }
-    }
-
-    /// Get the [`SchemaAdapterFactory`] for this table
-    pub fn schema_adapter_factory(&self) -> Option<&Arc<dyn SchemaAdapterFactory>> {
-        self.schema_adapter_factory.as_ref()
-    }
-
-    /// Creates a file source and applies schema adapter factory if available
-    fn create_file_source_with_schema_adapter(
-        &self,
-    ) -> datafusion_common::Result<Arc<dyn FileSource>> {
+    /// Creates a file source for this table
+    fn create_file_source(&self) -> Arc<dyn FileSource> {
         let table_schema = TableSchema::new(
             Arc::clone(&self.file_schema),
             self.options
@@ -341,13 +297,7 @@ impl ListingTable {
                 .collect(),
         );
 
-        let mut source = self.options.format.file_source(table_schema);
-        // Apply schema adapter to source if available.
-        // The source will use this SchemaAdapter to adapt data batches as they flow up the plan.
-        if let Some(factory) = &self.schema_adapter_factory {
-            source = source.with_schema_adapter_factory(Arc::clone(factory))?;
-        }
-        Ok(source)
+        self.options.format.file_source(table_schema)
     }
 
     /// If file_sort_order is specified, creates the appropriate physical expressions
@@ -490,7 +440,7 @@ impl TableProvider for ListingTable {
             )))));
         };
 
-        let file_source = self.create_file_source_with_schema_adapter()?;
+        let file_source = self.create_file_source();
 
         // create the execution plan
         let plan = self

datafusion/core/src/datasource/listing/table.rs

@@ -1443,11 +1443,10 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_statistics_mapping_with_default_factory() -> Result<()> {
+    async fn test_basic_table_scan() -> Result<()> {
         let ctx = SessionContext::new();
 
-        // Create a table without providing a custom schema adapter factory
-        // This should fall back to using DefaultSchemaAdapterFactory
+        // Test basic table creation and scanning
         let path = "table/file.json";
         register_test_store(&ctx, &[(path, 10)]);
 
@@ -1459,25 +1458,18 @@ mod tests {
         let config = ListingTableConfig::new(table_path)
             .with_listing_options(opt)
             .with_schema(Arc::new(schema));
-        // Note: NOT calling .with_schema_adapter_factory() to test default behavior
 
         let table = ListingTable::try_new(config)?;
 
-        // Verify that no custom schema adapter factory is set
-        assert!(table.schema_adapter_factory().is_none());
-
-        // The scan should work correctly with the default schema adapter
+        // The scan should work correctly
         let scan_result = table.scan(&ctx.state(), None, &[], None).await;
-        assert!(
-            scan_result.is_ok(),
-            "Scan should succeed with default schema adapter"
-        );
+        assert!(scan_result.is_ok(), "Scan should succeed");
 
-        // Verify that the default adapter handles basic schema compatibility
+        // Verify file listing works
         let result = table.list_files_for_scan(&ctx.state(), &[], None).await?;
         assert!(
             !result.file_groups.is_empty(),
-            "Should list files successfully with default adapter"
+            "Should list files successfully"
         );
 
         Ok(())

datafusion/core/src/datasource/mod.rs

@@ -42,7 +42,6 @@ pub use datafusion_catalog::default_table_source;
 pub use datafusion_catalog::memory;
 pub use datafusion_catalog::stream;
 pub use datafusion_catalog::view;
-pub use datafusion_datasource::schema_adapter;
 pub use datafusion_datasource::sink;
 pub use datafusion_datasource::source;
 pub use datafusion_datasource::table_schema;
@@ -60,15 +59,12 @@ mod tests {
         record_batch::RecordBatch,
     };
     use datafusion_common::{
-        record_batch,
         test_util::batches_to_sort_string,
         tree_node::{Transformed, TransformedResult, TreeNode},
         Result, ScalarValue,
     };
     use datafusion_datasource::{
-        file_scan_config::FileScanConfigBuilder,
-        schema_adapter::DefaultSchemaAdapterFactory, source::DataSourceExec,
-        PartitionedFile,
+        file_scan_config::FileScanConfigBuilder, source::DataSourceExec, PartitionedFile,
     };
     use datafusion_datasource_parquet::source::ParquetSource;
     use datafusion_physical_expr::expressions::{Column, Literal};
@@ -151,58 +147,6 @@ mod tests {
         "###);
     }
 
-    #[test]
-    fn default_schema_adapter() {
-        let table_schema = Schema::new(vec![
-            Field::new("a", DataType::Int32, true),
-            Field::new("b", DataType::Utf8, true),
-        ]);
-
-        // file has a subset of the table schema fields and different type
-        let file_schema = Schema::new(vec![
-            Field::new("c", DataType::Float64, true), // not in table schema
-            Field::new("b", DataType::Float64, true),
-        ]);
-
-        let adapter = DefaultSchemaAdapterFactory::from_schema(Arc::new(table_schema));
-        let (mapper, indices) = adapter.map_schema(&file_schema).unwrap();
-        assert_eq!(indices, vec![1]);
-
-        let file_batch = record_batch!(("b", Float64, vec![1.0, 2.0])).unwrap();
-
-        let mapped_batch = mapper.map_batch(file_batch).unwrap();
-
-        // the mapped batch has the correct schema and the "b" column has been cast to Utf8
-        let expected_batch = record_batch!(
-            ("a", Int32, vec![None, None]), // missing column filled with nulls
-            ("b", Utf8, vec!["1.0", "2.0"])  // b was cast to string and order was changed
-        )
-        .unwrap();
-        assert_eq!(mapped_batch, expected_batch);
-    }
-
-    #[test]
-    fn default_schema_adapter_non_nullable_columns() {
-        let table_schema = Schema::new(vec![
-            Field::new("a", DataType::Int32, false), // "a"" is declared non nullable
-            Field::new("b", DataType::Utf8, true),
-        ]);
-        let file_schema = Schema::new(vec![
-            // since file doesn't have "a" it will be filled with nulls
-            Field::new("b", DataType::Float64, true),
-        ]);
-
-        let adapter = DefaultSchemaAdapterFactory::from_schema(Arc::new(table_schema));
-        let (mapper, indices) = adapter.map_schema(&file_schema).unwrap();
-        assert_eq!(indices, vec![0]);
-
-        let file_batch = record_batch!(("b", Float64, vec![1.0, 2.0])).unwrap();
-
-        // Mapping fails because it tries to fill in a non-nullable column with nulls
-        let err = mapper.map_batch(file_batch).unwrap_err().to_string();
-        assert!(err.contains("Invalid argument error: Column 'a' is declared as non-nullable but contains null values"), "{err}");
-    }
-
     #[derive(Debug)]
     struct TestPhysicalExprAdapterFactory;