apache / apache/arrow

[Python][Parquet] dataset filter doesn't apply correctly when parquet file requires a cast (dataset evolution)

Open
#36,302 6 comments 0 reactions 0 assignees View on GitHub
Component: Parquet Component: Python Type: bug
Dominant language
C++
Stars
17.1k
Forks
4.3k
Avg merge
3d 13h
Merged PRs (30d)
88

Description

### Describe the bug, including details regarding any error messages, version, and platform.

dataset.filter fails with below error when filtering on timestamp. when the schema provided doesn't match schema in parquet file

```
---------------------------------------------------------------------------
ArrowNotImplementedError Traceback (most recent call last)
Cell In[24], line 2
1 dataset = ds.dataset('./yellow_tripdata_2009-01.parquet', schema=schema)
----> 2 dataset.filter((pc.field("Trip_Pickup_DateTime") <= pc.strptime('2009-01-02', format='%Y-%m-%d', unit='s'))).head(10).to_pandas()

File ~/.local/lib/python3.10/site-packages/pyarrow/_dataset.pyx:702, in pyarrow._dataset.Dataset.head()

File ~/.local/lib/python3.10/site-packages/pyarrow/_dataset.pyx:3495, in pyarrow._dataset.Scanner.head()

File ~/.local/lib/python3.10/site-packages/pyarrow/error.pxi:144, in pyarrow.lib.pyarrow_internal_check_status()

File ~/.local/lib/python3.10/site-packages/pyarrow/error.pxi:121, in pyarrow.lib.check_status()

ArrowNotImplementedError: Function 'equal' has no kernel matching input types (timestamp[s], string)

```

## Pyarrow version

```
import pyarrow as pa
import pyarrow.compute as pc
import pyarrow.dataset as ds

pa.__version__

'12.0.0'
```

### Sample data

https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2009-01.parquet

### Failing code
```
schema = pa.schema(
pa.struct({
"vendor_name": pa.string(),
"Trip_Pickup_DateTime": pa.timestamp('s'),
"Trip_Dropoff_DateTime": pa.timestamp('s'),
"Passenger_Count": pa.int64(),
"Trip_Distance": pa.float64(),
"Start_Lon": pa.float64(),
"Start_Lat": pa.float64(),
"Rate_Code": pa.float64(),
"store_and_forward": pa.float64(),
"End_Lon": pa.float64(),
"End_Lat": pa.float64(),
"Payment_Type": pa.string(),
"Fare_Amt": pa.float64(),
"surcharge": pa.float64(),
"mta_tax": pa.float64(),
"Tip_Amt": pa.float64(),
"Tolls_Amt": pa.float64(),
"Total_Amt": pa.float64()
}))

dataset = ds.dataset('./yellow_tripdata_2009-01.parquet', schema=schema)

dataset.filter(
(
pc.field("Trip_Pickup_DateTime") <= pc.strptime('2009-01-02', format='%Y-%m-%d', unit='s'))

).head(10).to_pandas()

# check schema

dataset.schema

vendor_name: string
Trip_Pickup_DateTime: timestamp[s]
Trip_Dropoff_DateTime: timestamp[s]
Passenger_Count: int64
Trip_Distance: double
Start_Lon: double
Start_Lat: double
Rate_Code: double
store_and_forward: double
End_Lon: double
End_Lat: double
Payment_Type: string
Fare_Amt: double
surcharge: double
mta_tax: double
Tip_Amt: double
Tolls_Amt: double
Total_Amt: double

# check metadata
dataset.schema.metadata
' '
```
### Remove the schema and the filter works as string.
```
dataset = ds.dataset('./yellow_tripdata_2009-01.parquet')
dataset.filter((pc.field("Trip_Pickup_DateTime") <= '2009-01-02')).head(10).to_pandas()
```
```
dataset.schema

vendor_name: string
Trip_Pickup_DateTime: string
Trip_Dropoff_DateTime: string
Passenger_Count: int64
Trip_Distance: double
Start_Lon: double
Start_Lat: double
Rate_Code: double
store_and_forward: double
End_Lon: double
End_Lat: double
Payment_Type: string
Fare_Amt: double
surcharge: double
mta_tax: double
Tip_Amt: double
Tolls_Amt: double
Total_Amt: double
-- schema metadata --
pandas: '{"index_columns": [{"kind": "range", "name": null, "start": 0, "' + 2473

```
```
dataset.schema.metadata

{b'pandas': b'{"index_columns": [{"kind": "range", "name": null, "start": 0, "stop": 14092413, "step": 1}], "column_indexes": [{"name": null, "field_name": null, "pandas_type": "unicode", "numpy_type": "object", "metadata": {"encoding": "UTF-8"}}], "columns": [{"name": "vendor_name", "field_name": "vendor_name", "pandas_type": "unicode", "numpy_type": "object", "metadata": null}, {"name": "Trip_Pickup_DateTime", "field_name": "Trip_Pickup_DateTime", "pandas_type": "unicode", "numpy_type": "object", "metadata": null}, {"name": "Trip_Dropoff_DateTime", "field_name": "Trip_Dropoff_DateTime", "pandas_type": "unicode", "numpy_type": "object", "metadata": null}, {"name": "Passenger_Count", "field_name": "Passenger_Count", "pandas_type": "int64", "numpy_type": "int64", "metadata": null}, {"name": "Trip_Distance", "field_name": "Trip_Distance", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "Start_Lon", "field_name": "Start_Lon", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "Start_Lat", "field_name": "Start_Lat", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "Rate_Code", "field_name": "Rate_Code", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "store_and_forward", "field_name": "store_and_forward", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "End_Lon", "field_name": "End_Lon", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "End_Lat", "field_name": "End_Lat", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "Payment_Type", "field_name": "Payment_Type", "pandas_type": "unicode", "numpy_type": "object", "metadata": null}, {"name": "Fare_Amt", "field_name": "Fare_Amt", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "surcharge", "field_name": "surcharge", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "mta_tax", "field_name": "mta_tax", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "Tip_Amt", "field_name": "Tip_Amt", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "Tolls_Amt", "field_name": "Tolls_Amt", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}, {"name": "Total_Amt", "field_name": "Total_Amt", "pandas_type": "float64", "numpy_type": "float64", "metadata": null}], "creator": {"library": "pyarrow", "version": "8.0.0"}, "pandas_version": "1.2.3"}'}

```

### Filter works with explicit schema if applied on table.
```
dataset.head(10).filter(
(
pc.field("Trip_Pickup_DateTime") <= pc.strptime('2009-01-02', format='%Y-%m-%d', unit='s')
)
).to_pandas()
```

### Filter also works if I load the file with schema and save it to new parquet file and then load again with the same schema.

1. load file with custom schema and save to Parquet
````
dataset = ds.dataset('./yellow_tripdata_2009-01.parquet', schema=schema)

import pyarrow.parquet as pp
pp.write_to_dataset(dataset, root_path='./test')
````
2. Load new file with same schema and filter

````
dataset = ds.dataset('./test/8c0673b61cc34b4e8094dc1cb11534bd-0.parquet', schema=schema)

dataset.filter(
(
pc.field("Trip_Pickup_DateTime") <= pc.strptime('2009-01-02', format='%Y-%m-%d', unit='s')
)
).head(10).to_pandas()

````

Dataset.filter should work if custom schema is provided with data types different than parquet metadata

### Component(s)

Parquet, Python

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.