dask: read_parquet + getitem optimization no longer working

The optimization introduced in #1785 doesn’t work in master.

import dask
import dask.dataframe as dd
import pandas as pd

# Create test dataset
df = pd.DataFrame({'a': [1] * 10000, 'b': [2] * 10000})
df.to_parquet('test.parquet', engine='fastparquet')

# Open it, get a subset of items and try to optimize the graph
ddi = dd.read_parquet('test.parquet', engine='fastparquet')[['a']]
dsk = ddi.dask
dsk2 = dd.optimize(dsk, ddi.__dask_keys__())

dsk2[('read-parquet-getitem-e42c1003708b81c7d6bfcbf6eaa5ac3d', 0)]

returns

(subgraph_callable, (<function _read_parquet_row_group>,..., ['a', 'b'], ...), ['a'])

whereas in the past we were seeing:

(<function _read_parquet_row_group>,..., ['a'], ...)

The problem seems to be in dask.optimization.fuse_selections not handling subgraph_callable. I see 2 possible solutions:

• Convert read_parquet into a HighLevelGraph, so that the optimization can be performed between 2 layers, instead of among tasks
• Modify fuse_selections to handle subgraph_callable

What would you suggest? I could find some time to work on it. Thank you for your help!