[Python][Parquet] dataset filter doesn't apply correctly when parquet file requires a cast (dataset evolution)
- 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
Assessment
This issue has not been assessed yet.