feat(parquet): Support struct schema evolution matching by name

[rui-mo]

The default behavior of the schema evolution for row type is matching by index.
This PR supports matching by name for Parquet file format. Missing subfields
are identified by matching the file type and requested type on the names of
subfileds. When all the subfields in the requested type are missing and the
number of subfields is more than one, the struct is set as null. Otherwise,
'null' occupies the position of the missing subfields. Below table summarizes
the supported cases.

Parquet column schema	Requested output schema	Result
row({"a", "c"})	row({"a", "b", "c"})	row(a_val, null, c_val)
row({"a", "c"})	row({"b"})	row(null)
row({"a", "c"})	row({"b", "d"})	null
row({"a", "c"})	row({})	empty

[netlify]

✅ Deploy Preview for meta-velox canceled.

Name	Link
🔨 Latest commit	b8b9fbc
🔍 Latest deploy log	https://app.netlify.com/sites/meta-velox/deploys/679389a59fb0980008084717

[Yuhta]

Requested type is available as getSelector()->getSchemaWithId()->type. We may want to convert it to a type directly in the future, but for now let's not keep 2 copies of the same thing.

[Yuhta]

We need to decide what is the schema evolution strategy we want here. In our data warehouse, columns are not matched by name but by position, so any extra fields added need to be at the end of the children list. This allows column renaming. If we match by name here, we will lose the renaming functionality and this seems quite important in most data warehouse.

[rui-mo]

Thanks for your comment. Does that mean for a row(a, c) struct schema in parquet, the expected output can only be like row(a, c, xxx, ...)? In Spark, there is no such limitation to extra child fields.

[Yuhta]

Yes new subfields can only be appended. So in plain vanilla Spark, field renaming is not supported? There is also a third way to match by field ID (e.g. Iceberg), we need to start draft some design about this to cover all three cases.

[rui-mo]

How does field renaming is conducted in the data warehouse you mentioned? In Spark, for query like select a as b, it adds a projection node with Alias expression after scan.
And what do you suggest for the design, should I added some notes in this PR or something else?

[Yuhta]

With matching by name you need to know all the old field names (a in your query) in all old files, which is not practical in a normal data warehouse. I would suggest we pause this PR for a bit and design the right way to allow matching columns in different ways first.

[rui-mo]

Thanks. That looks good to me. Convert this PR to draft for now.

[stale]

This pull request has been automatically marked as stale because it has not had recent activity. If you'd still like this PR merged, please comment on the PR, make sure you've addressed reviewer comments, and rebase on the latest main. Thank you for your contributions!

[Yuhta]

@rui-mo This behavior needs to be configurable as the default behavior should be match by index, i.e.

Parquet column schema	User-specified output schema	Result
row({"a", "c"})	row({"a", "b", "c"})	row(a:a_val, b:c_cal, c:null)
row({"a", "c"})	row({"b"})	Should not be supported, no deletion of subfields
row({"a", "c"})	row({"b", "d"})	row(b:a_val, c:d_val)
row({"a", "c"})	row({})	Should not be supported

Also we should consider doing it in a format-agnostic way.

[rui-mo]

@rui-mo This behavior needs to be configurable as the default behavior should be match by index, i.e.
Also we should consider doing it in a format-agnostic way.

@Yuhta Thanks for your feedback. It makes sense to me and I will take further look.