don't fully enforce matching schema

[nicklan]

When we read from a checkpoint, there can be disagreement about the schema we think we've read, and what was actually in the parquet. This can cause issues when we try and interact with engine data via expressions. For example here, where we use the "correct" schema and do not mark dvs as nullable. Also, in our arrow_conversions we make assumptions about the names of the fields that mark map keys and values (see here), which also causes issues when the actual materialized names are different.

An example error trying to work with a checkpoint file:

Arrow(
    InvalidArgumentError(
        "Incorrect datatype for StructArray field \"deletionVector\", expected Struct([Field { name: \"storageType\", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"pathOrInlineDv\", data_type: Utf8, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"offset\", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"sizeInBytes\", data_type: Int32, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"cardinality\", data_type: Int64, nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }]) got Struct([Field { name: \"storageType\", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"pathOrInlineDv\", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"offset\", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"sizeInBytes\", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"cardinality\", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, Field { name: \"maxRowIndex\", data_type: Int64, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }])",
    ),
)

This arises for two reasons:

1. To avoid "parquet corruption", spark will write a nullable struct as having all fields nullable. Thus, for example, since DVs are nullable, all fields of the DV struct are marked nullable. This means the "nullability" of fields will not be the same as the specified read schema
2. We can't guarantee exactly what names the "meta" fields for maps/lists will have in the raw schema

This PR does as much validation as possible, but doesn't check things we can't control. So for each named field in the output schema it ensures the types are the same, and does so recursively into structs, maps, and lists.

Then it simply uses the schema that the parquet/json reader had already associated with the data.

This "works". Major issues:

1. The schema you ask for might not be exactly the schema you get. Although differences in nullability are the only "observable" difference from kernel perspective
2. A bigger issue is that this complicates the writing of expression evaluators for engines, as we'll have to carefully document and explain potential schema mismatches.

After some discussion, we will go with this for the time being.

[scovich]

FYI spark does document this behavior:

When reading Parquet files, all columns are automatically converted to be nullable for compatibility reasons.

[roeap]

FYI spark does document this behavior:

When reading Parquet files, all columns are automatically converted to be nullable for compatibility reasons.

After reading, I am not completely sure, how specifically this is meant. When it says columns, does that imply all leaf columns, or just top level columns, or everything recursively? i.e. just applying to all leafs might yield incorrect data (e.g. map keys). So my guess would be just roots?

[scovich]

FYI spark does document this behavior:

When reading Parquet files, all columns are automatically converted to be nullable for compatibility reasons.

After reading, I am not completely sure, how specifically this is meant. When it says columns, does that imply all leaf columns, or just top level columns, or everything recursively? i.e. just applying to all leafs might yield incorrect data (e.g. map keys). So my guess would be just roots?

My understanding is: Any time we have a struct containing non-null fields which is itself a nullable field (whether in a parent struct or top-level schema), then parquet cannot express that situation. Writers have to compensate by forcing all non-null-with-nullable-ancestor fields to be nullable. It is not necessary to make the entire schema nullable, but I'm guessing many writers do so because it's simpler?

[scovich]

This is avoiding the double-? issue we hit before?

(but if ensure_data_types is really just a validator rather than a mapper... a lot of this code can probably go away?)

[nicklan]

It's to fix an error like:

error[E0282]: type annotations needed
   --> kernel/src/engine/arrow_expression.rs:256:21
    |
256 |                     Ok(ArrowField::new(input_field.name(), array.data_type().clone(), array.is_nullable()))
    |                     ^^ cannot infer type of the type parameter `E` declared on the enum `Result`
    |
help: consider specifying the generic arguments
    |
256 |                     Ok::<arrow_schema::Field, E>(ArrowField::new(input_field.name(), array.data_type().clone(), array.is_nullable()))
    |                       ++++++++++++++++++++++++++

I've fixed it slightly differently to avoid having to use the turbofish.

[scovich]

nit: Seems like a good place for require! followed by the for-loop?

[scovich]

nit: I suspect this would also work:

.map(|(array, input_field)| -> DeltaResult<_> {
    ensure_data_types(...)?;
    Ok(ArrowFIeld::new(...));
}

[roeap]

👍