Support for "Schema evolution" / Schema Adapters

[alamb]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

Sometimes it is desired to convert RecordBatches from one schema so they match another. This is often done when data is stored in several different sources (like parquet files) that are "compatible" but not exactly the same (e.g. maybe newer files have new columns)

Common transformations are:

• Reorder columns by name (so a file with (a int, b char) and one with (b char, a int) could be read as a single stream
• Insert missing columns (so a file with (a int, b char) and a file with (a int) could be merged

It is also common to want to fill in missing columns with either Null or some constant (e.g. 0) so a controllable policy would be nice

Note that these these usecases are pretty similar to casting Structs (e.g. reordering fields with the same name but different position)

• Support StructArray in Cast Kernel #4908

Often computing the transformation may be non trivial (e.g. matching columns by name) so it would be nice to do the mapping calculation once per schema rather than once per batch / StructArrayschema. For example DF's SchemaAdapter computes the mapping once and can then apply that to multiple batches.

Describe the solution you'd like
Add some API in Arrow-rs to do this mapping

One alternative, suggested by @tustvold would be to add a first-party schema adapter into arrow-rs.

Describe alternatives you've considered

For anyone interested, here is the API that is in DataFusion (it now even has ASCII art and Examples, thanks to @itsjunetime and myself):

• https://docs.rs/datafusion/latest/datafusion/datasource/schema_adapter/struct.DefaultSchemaAdapterFactory.html

We can/should probably change the names and reduce the levels of indirection of we upstreamed this into arrow-rs
Additional context

[alamb]

Possibly a duplicate of

• Add a way to map RecordBatch schema from one to another #5996

[adriangb]

Often computing the transformation may be non trivial (e.g. matching columns by name) so it would be nice to do the mapping calculation once per schema rather than once per batch / StructArrayschema. For example DF's SchemaAdapter computes the mapping once and can then apply that to multiple batches.

I'm tot sure how this would happen in practice: there's no state in UDFs in DataFusion. So if we e.g. wanted to implement cast(...) in terms of a SchemaAdapter we have nowhere to store the pre-computed value. I think we'd have to introduce some sort of build step that goes around the expression tree and optimizes expressions for the given input / output schemas.

[alamb]

Often computing the transformation may be non trivial (e.g. matching columns by name) so it would be nice to do the mapping calculation once per schema rather than once per batch / StructArrayschema. For example DF's SchemaAdapter computes the mapping once and can then apply that to multiple batches.

I'm tot sure how this would happen in practice: there's no state in UDFs in DataFusion. So if we e.g. wanted to implement cast(...) in terms of a SchemaAdapter we have nowhere to store the pre-computed value. I think we'd have to introduce some sort of build step that goes around the expression tree and optimizes expressions for the given input / output schemas.

I wonder if we could use the new snapshot machinery 🤔

https://docs.rs/datafusion/latest/datafusion/physical_expr/trait.PhysicalExpr.html#method.snapshot

[adriangb]

Often computing the transformation may be non trivial (e.g. matching columns by name) so it would be nice to do the mapping calculation once per schema rather than once per batch / StructArrayschema. For example DF's SchemaAdapter computes the mapping once and can then apply that to multiple batches.

I'm tot sure how this would happen in practice: there's no state in UDFs in DataFusion. So if we e.g. wanted to implement cast(...) in terms of a SchemaAdapter we have nowhere to store the pre-computed value. I think we'd have to introduce some sort of build step that goes around the expression tree and optimizes expressions for the given input / output schemas.

I wonder if we could use the new snapshot machinery 🤔

https://docs.rs/datafusion/latest/datafusion/physical_expr/trait.PhysicalExpr.html#method.snapshot

snapshot doesn't know anything about it's input data types. I did propose something similar in apache/datafusion#15057 (comment) which is actually part of what lead to the idea of PhysicalExpr::snapshot.

I think it could be a win in performance to "specialize" the whole expression tree to the work it's actually going to do but it will likely depend on the balance of upfront cost vs. execution gains, which would require benchmarks, etc.

My intuition is that for now we could start by doing something naive: build a SchemaAdapter every time CastExpr::evaluate` gets called and then throw it away. I think that will be more expensive than the current cast kernel but maybe not by much. If that is too expensive then we can think of some system to specialize PhysicalExprs.

[alamb]

I think starting simple and adding optimization guided by profiling sounds like a great idea

[mapleFU]

I noticed that in parquet internal there're multiple kinds of evolution:

1. Without any real cast. This happens on type changes (1) Like Utf8 -> Utf8View, List -> ListView they're store same in parquet (2) Int8/Int16 -> Int32. Parquet stores at least int32 type.
2. Need some casting? Similiar to (1) but might need some explicit cast. ( Schema Adapters)
3. New default / Null column
4. New projection/Expression added. Which might need some expression added. In fact a reader can expose "NULL"

I think (3)(1) might able to push in a layer in parquet/arrow, but (2)(4) might need others ( Maybe iceberg-rust or datafusion can be a good example?)