[Python][Parquet][C++] Null values in a single partition of Parquet dataset, results in invalid schema on read

[asfimport]

import pyarrow as pa
import pyarrow.parquet as pq
import pandas as pd

from datetime import datetime, timedelta


def generate_data(event_type, event_id, offset=0):
    """Generate data."""
    now = datetime.utcnow() + timedelta(seconds=offset)
    obj = {
        'event_type': event_type,
        'event_id': event_id,
        'event_date': now.date(),
        'foo': None,
        'bar': u'hello',
    }
    if event_type == 2:
        obj['foo'] = 1
        obj['bar'] = u'world'
    if event_type == 3:
        obj['different'] = u'data'
        obj['bar'] = u'event type 3'
    else:
        obj['different'] = None
    return obj


data = [
    generate_data(1, 1, 1),
    generate_data(1, 1, 3600 * 72),
    generate_data(2, 1, 1),
    generate_data(2, 1, 3600 * 72),
    generate_data(3, 1, 1),
    generate_data(3, 1, 3600 * 72),
]

df = pd.DataFrame.from_records(data, index='event_id')
table = pa.Table.from_pandas(df)

pq.write_to_dataset(table, root_path='/tmp/events', partition_cols=['event_type', 'event_date'])

dataset = pq.ParquetDataset('/tmp/events')
table = dataset.read()
print(table.num_rows)

Expected output:

6

Actual:

python example_failure.py
Traceback (most recent call last):
  File "example_failure.py", line 43, in <module>
    dataset = pq.ParquetDataset('/tmp/events')
  File "/Users/sam/.virtualenvs/test-parquet/lib/python2.7/site-packages/pyarrow/parquet.py", line 745, in __init__
    self.validate_schemas()
  File "/Users/sam/.virtualenvs/test-parquet/lib/python2.7/site-packages/pyarrow/parquet.py", line 775, in validate_schemas
    dataset_schema))
ValueError: Schema in partition[event_type=2, event_date=0] /tmp/events/event_type=3/event_date=2018-07-16 00:00:00/be001bf576674d09825539f20e99ebe5.parquet was different.
bar: string
different: string
foo: double
event_id: int64
metadata
--------
{'pandas': '{"pandas_version": "0.23.3", "index_columns": ["event_id"], "columns": [{"metadata": null, "field_name": "bar", "name": "bar", "numpy_type": "object", "pandas_type": "unicode"}, {"metadata": null, "field_name": "different", "name": "different", "numpy_type": "object", "pandas_type": "unicode"}, {"metadata": null, "field_name": "foo", "name": "foo", "numpy_type": "float64", "pandas_type": "float64"}, {"metadata": null, "field_name": "event_id", "name": "event_id", "numpy_type": "int64", "pandas_type": "int64"}], "column_indexes": [{"metadata": null, "field_name": null, "name": null, "numpy_type": "object", "pandas_type": "bytes"}]}'}

vs

bar: string
different: null
foo: double
event_id: int64
metadata
--------
{'pandas': '{"pandas_version": "0.23.3", "index_columns": ["event_id"], "columns": [{"metadata": null, "field_name": "bar", "name": "bar", "numpy_type": "object", "pandas_type": "unicode"}, {"metadata": null, "field_name": "different", "name": "different", "numpy_type": "object", "pandas_type": "empty"}, {"metadata": null, "field_name": "foo", "name": "foo", "numpy_type": "float64", "pandas_type": "float64"}, {"metadata": null, "field_name": "event_id", "name": "event_id", "numpy_type": "int64", "pandas_type": "int64"}], "column_indexes": [{"metadata": null, "field_name": null, "name": null, "numpy_type": "object", "pandas_type": "bytes"}]}'}

Apparently what is happening is that pyarrow is interpreting the schema from each of the partitions individually and the partitions for event_type=3 / event_date=\* both have values for the column different whereas the other columns do not. The discrepancy causes the None values of the other partitions to be labeled as pandas_type empty instead of unicode.

Reporter: Sam Oluwalana

Related issues:

• [Python] More graceful reading of empty String columns in ParquetDataset (relates to)
• [Python][Dataset] Support using dataset API in pyarrow.parquet with a minimal ParquetDataset shim (depends upon)
• [C++][Dataset] Support null -> other type promotion in Dataset scanning (depends upon)

_{Note: This issue was originally created as ARROW-2860. Please see the migration documentation for further details.}

[asfimport]

Wes McKinney / @wesm:
There's other issues reported about this. We should handle schema normalization across components of a dataset. I don't have time to work on this for 0.11 but I hope that someone else can take a look soon

[asfimport]

Tanya Schlusser / @tanyaschlusser:
I think this was resolved with https://issues.apache.org/jira/browse/ARROW-2891

pull request 2302
#2302

When I run example_failure.py it does not fail and returns the expected result.

[asfimport]

Wes McKinney / @wesm:
Thanks for checking that. There's a couple of related issues that may be the same thing

[asfimport]

Karl Koster:
I am running into the same problem. It looks like https://issues.apache.org/jira/browse/ARROW-2891 addresses this within a single write (i.e. multiple partitions on a single write where one or more have empty column sets). The problem I am running in to is across writes (e.g. incrementally adding partitions) when a write contains a column that has only NaN's but other partitions have values. Accessing the earliest or latest partition for the schema would likely solve this. Something like the read does when reading multiple partitions (it takes the earliest schema).

[asfimport]

Krisztian Szucs / @kszucs:
Allowing null typed columns to any nullable columns fix this particular issue. On long term We could check column type compatibility more robustly, however upcasting may not always be desirable.

[asfimport]

Wes McKinney / @wesm:
We need to address schema normalization in Parquet files more generally, otherwise we are going to continue to have other kinds of bug reports.

[asfimport]

Krisztian Szucs / @kszucs:
@wesm What do You mean under schema normalization a bit precisely? AFAICS it's about checking whether file_schema is castable to (compatible with) dataset_schema in validate_schemas and handle the conversions during read (if required).

If You can recall the related issues please link them!

[asfimport]

Wes McKinney / @wesm:
If a field is nullable in one file and non-nullable in another (which happens), it is rejected by ParquetDataset. We should probably also permit automatic integer type promotions (as well as null -> any promotions)

As soon as we can write partitioned datasets to Parquet without going through pandas for the groupby splitting (which is causing a lot of these problems), some issues will go away automatically

[asfimport]

Wes McKinney / @wesm:
See https://github.com/apache/spark/blob/4427a96bcea625bc51fc5e0e999f170ad537a2fc/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala#L555

See also schema resolution logic in Impala https://github.com/apache/impala/blob/07fd332089c262fa8813605f12b927c8602ac0d2/be/src/exec/parquet/parquet-metadata-utils.h#L124. This is a little bit different because the objective is to establish a mapping between the actual representation of data in the file and the logical schema that is fixed in the Hive metastore

[asfimport]

Krisztian Szucs / @kszucs:
So You think something like dt.castable() in ibis, but implemented in type.cc (and propagated to Schema)?

[asfimport]

Wes McKinney / @wesm:
I think we are a bit beyond the scope of the immediate issue. There are some other issues around schema normalization (if not, let's create an appropriate one); let's take the discussion there and make sure we understand the feature requirements and use cases. I have a bit too much on my plate to think very much about it right now

[asfimport]

Wes McKinney / @wesm:
Like ARROW-2659, I'm moving this to 0.13 so that we can take the time to address this properly

[asfimport]

Joris Van den Bossche / @jorisvandenbossche:
So several things here:

• The original root cause of this issue (write_to_dataset outputting parquet files with different schemas) has already been solved a while ago in ARROW-2891
• Of course, you can still end up with parquet files where one of the files has a null type by other means, and we should be able to read those:
• We have no plans to fix this in the current ParquetDataset python implementation, but
• It already works fine in the new Dataset API, and this can be used in ParquetDataset after ARROW-8039 is merged.
• So in the new Dataset API, basic schema normalization/evolution already works: nullable vs non-nullable, null -> any type promotion, addition/removal of a column, reordering the columns. For other types mentioned above, there are some open JIRAs (eg ARROW-8282 for ints, ARROW-8284 for timestamps)

[asfimport]

Joris Van den Bossche / @jorisvandenbossche:
So once we use the datasets API under the hood in pyarrow.parquet (ARROW-8039), this issue should be solved (we can still add a test for it to close this issue)

[asfimport]

Todd Farmer / @toddfarmer:
This issue was last updated over 90 days ago, which may be an indication it is no longer being actively worked. To better reflect the current state, the issue is being unassigned. Please feel free to re-take assignment of the issue if it is being actively worked, or if you plan to start that work soon.