RecordBatchTransformer: Handle schema migration and column re-ordering in table scans

[sdd]

Addresses parts 2 and 3 of #405.

Add support for type promotion and default values to the read pipeline.

• When the scan includes fields that have undergone type promotion since some of the underlying parquet files were written, any selected fields that have undergone type promotion will be promoted to the newer type before being returned to the user. For example, A table contained a field "a" that was a Float, and some rows were written. The table's schema was changed so that field "a" is now a Double. When a table scan is performed, record batches coming from files written prior to the type promotion will be dynamically converted so that field "a" is of type Double, matching any row batches returned from files written after the schema change.
• When the scan includes fields that have been added since some of the files were written, record batches will be dynamically converted as per the above to contain selected fields that were not present at the time the file that they were in was written. These will have a value of null if there is no default value present for the column but the column is not required. If the table schema specifies an initial-default-value for the field, then all rows will have that value for the new column instead.
• If any fields have been renamed, the record batch schemas for rows written before the rename occurred will be rewritten to contain the new field name.
• If projected_field_ids is provided, the columns in the response will be re-ordered to match the order in the projection.

[sdd]

I need to go back to the drawing board on this. The current implementation breaks when not all columns in the file are in the list of projected fields.

[sdd]

OK, I've addressed the problem with projections other than the equiv of SELECT *. All existing tests passing, and new tests extended to cover these cases.

[sdd]

@liurenjie1024 and @Xuanwo - ready for review when you get chance

[sdd]

@Xuanwo and @liurenjie1024: PTAL, I've rebased and refactored this to better handle the pass-through case. It would be great to get this merged - without it, I'm experiencing some annoying issues, as table scans against tables that have had new columns added where pre-existing data exists without the added column, and then had new data written in the new schema, results in record batches being streamed back that can have different schemas in the same stream, which is very difficult to deal with.

[Xuanwo]

Thanks a lot for working on this!

[Xuanwo]

Waiting for @liurenjie1024 to take another look.

[liurenjie1024]

Thanks @sdd for this high quality pr, looks great to me! I believe current approach works well without nested types, and we can add support for nested types later. Just one minor point about arrow dependency, others look great!

[sdd]

Hi @liurenjie1024 - I addressed your comment around not importing the whole of Arrow.

Additionally, I realised that there was an optimisation that could be made in the case that only the column names differed between the source and target schemas, so I've added code for that use case.

Also, the schemas can contain metadata which, where present, would cause the equality comparison to fail, so I changed the equality check to only compare on the aspects of the schema that we're interested in (data type, nullability, and column name).

This is now ready for re-review (FAO @Xuanwo for this also).

Thanks! :-)

[sdd]

@Xuanwo PTAL - you approved an earlier version but there are some small additional changes since then.

I added:

• a performance improvement for a particular scenario;
• a change to how schemas are compared for equality for this specific case.

(See here: 0ed5721)

[Xuanwo]

Thank you @sdd for the great work! Also thanks @liurenjie1024 for the review. Let's move.