I found an issue when reading many parquet files from AWS. The logical to physical plan translator goes incredibly slow when running this code:

import daft
from daft import col
from <http://daft.io|daft.io> import IOConfig

io_config = IOConfig(
    s3=daft.io.S3Config.from_env()
)
dedup = daft.read_parquet("<s3://eventual-dev-benchmarking-fixtures/redpajama-parquet/dedupe1>", io_config=io_config)

cc_sizes = dedup.groupby("original_url", "original_date_download").agg(col("url").count().alias("component_size"))
cc_sizes.sort(by="component_size", desc=True).explain(True)

It seems to be doing a lot of sequential requests to AWS. (More info in thread)

won't this revert the perf benefits for local reading though?

That’s why I want a better solution. @Sammy Sidhu had some ideas

currently on main, we're fetching the parquet metadata for each file sequentially. In the case for conor's query that's like a few hundred small parquet files which ends up taking like a minute during the translation phase (since we exhaust the glob scan operator iterator). Since it's a bunch of small files, theres no benefit to fetch the parquet metadata upfront since it won't be split across scan tasks In the PR that conor has, the idea is to cache the parquet metadata if we are in the process of spitting a ScanTask. The rational is that: 1. We already have the metadata, so might as well cache it 2. If it's a large file, we can cache it across multiple ScanTasks 3. We only perform this step if the total number of files is less than some threshold, bounding the time in planning. Beyond that, we get plenty of parallelism anyways

I changed the way that the metadata is stored. Now if we split the file across scan tasks, each scan task only stores the metadata relevant to it. This makes reading in the adversarial parquet file really fast--suspiciously fast, even. It went from taking ~13s to read and

.collect

to only 1.9s. Hopefully this doesn’t mean I broke anything!

Amazing!!

It is