Add BatchAdapter to simplify using PhysicalExprAdapter / Projector to map RecordBatch between schemas (#19716)

I've now seen this pattern a couple of times, in our own codebase,
working on apache/datafusion-comet#3047.

I was going to add an example but I think adding an API to handle it for
users is a better experience.

This should also make it a bit easier to migrate from SchemaAdapter. In
fact, I think it's possible to implement a SchemaAdapter using this as
the foundation + some shim code. This won't be available in DF 51 to
ease migration but it's easy enough to backport (just copy the code in
this PR) for users that would find that helpful.

Author: adriangb

datafusion/datasource/src/schema_adapter.rs

@@ -115,10 +115,20 @@ pub trait SchemaMapper: Debug + Send + Sync {
 
 /// Deprecated: Default [`SchemaAdapterFactory`] for mapping schemas.
 ///
-/// This struct has been removed. Use [`PhysicalExprAdapterFactory`] instead.
+/// This struct has been removed.
+///
+/// Use [`PhysicalExprAdapterFactory`] instead to customize scans via
+/// [`FileScanConfigBuilder`], i.e. if you had implemented a custom [`SchemaAdapter`]
+/// and passed that into [`FileScanConfigBuilder`] / [`ParquetSource`].
+/// Use [`BatchAdapter`] if you want to map a stream of [`RecordBatch`]es
+/// between one schema and another, i.e. if you were calling [`SchemaMapper::map_batch`] manually.
+///
 /// See `upgrading.md` for more details.
 ///
 /// [`PhysicalExprAdapterFactory`]: datafusion_physical_expr_adapter::PhysicalExprAdapterFactory
+/// [`FileScanConfigBuilder`]: crate::file_scan_config::FileScanConfigBuilder
+/// [`ParquetSource`]: https://docs.rs/datafusion-datasource-parquet/latest/datafusion_datasource_parquet/source/struct.ParquetSource.html
+/// [`BatchAdapter`]: datafusion_physical_expr_adapter::BatchAdapter
 #[deprecated(
     since = "52.0.0",
     note = "DefaultSchemaAdapterFactory has been removed. Use PhysicalExprAdapterFactory instead. See upgrading.md for more details."
@@ -178,10 +188,20 @@ impl SchemaAdapter for DeprecatedSchemaAdapter {
 
 /// Deprecated: The SchemaMapping struct held a mapping from the file schema to the table schema.
 ///
-/// This struct has been removed. Use [`PhysicalExprAdapterFactory`] instead.
+/// This struct has been removed.
+///
+/// Use [`PhysicalExprAdapterFactory`] instead to customize scans via
+/// [`FileScanConfigBuilder`], i.e. if you had implemented a custom [`SchemaAdapter`]
+/// and passed that into [`FileScanConfigBuilder`] / [`ParquetSource`].
+/// Use [`BatchAdapter`] if you want to map a stream of [`RecordBatch`]es
+/// between one schema and another, i.e. if you were calling [`SchemaMapper::map_batch`] manually.
+///
 /// See `upgrading.md` for more details.
 ///
 /// [`PhysicalExprAdapterFactory`]: datafusion_physical_expr_adapter::PhysicalExprAdapterFactory
+/// [`FileScanConfigBuilder`]: crate::file_scan_config::FileScanConfigBuilder
+/// [`ParquetSource`]: https://docs.rs/datafusion-datasource-parquet/latest/datafusion_datasource_parquet/source/struct.ParquetSource.html
+/// [`BatchAdapter`]: datafusion_physical_expr_adapter::BatchAdapter
 #[deprecated(
     since = "52.0.0",
     note = "SchemaMapping has been removed. Use PhysicalExprAdapterFactory instead. See upgrading.md for more details."

datafusion/physical-expr-adapter/src/lib.rs

@@ -29,6 +29,7 @@
 pub mod schema_rewriter;
 
 pub use schema_rewriter::{
-    DefaultPhysicalExprAdapter, DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter,
-    PhysicalExprAdapterFactory, replace_columns_with_literals,
+    BatchAdapter, BatchAdapterFactory, DefaultPhysicalExprAdapter,
+    DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
+    replace_columns_with_literals,
 };