Preserving row group information during reads

[adriangb]

Hi folks,

I'm essentially trying to implement an external bloom filter for columns with randomly distributed values (where min/max stats don't help) and where reading bloom filter information from all files may be too expensive (100s of high latency object store requests), hence why I want to store this information externally.

The roadblock I'm running into is getting access to values per row group like bloom filters do.

Given that I have a DataFrame that I'm going to write out as I see it my options are:

• Write it out then using the metadata read back one row group at a time so I can get the values per row group. This I should be able to do as is but is gross and won't perform well even if I have the data in memory.
• Make changes to the writer APIs to allow hooking into the right spot where bloom filters are updated but instead call a callback, accumulate the values, etc. This would be the most efficient but requires API designs and changes to DataFusion.
• Add a feature to inject the row group a row came from when reading a file (still need to read it back, requires changes to DataFusion, but at least would be simple on my end)

Any suggestions on APIs or hook points I may be overlooking?

Thanks!

[alamb]

Add a feature to inject the row group a row came from when reading a file (still need to read it back, requires changes to DataFusion, but at least would be simple on my end)

One think you could do is to create a ParquetAccessPlan that specifies reading only the row group you are interested in.

You would have to run this for each row group of course, but if you have all the data and metadata in memory anyways, it probably isn't that bad (and you could scan them all in parallel) 🤔

[adriangb]

Makes sense! I did something similar but manually calling ParquetRecordBatchReaderBuilder.
Either way I end up having to go back to the bytes once for each row group, which seems like it would be less efficient than if I could do the read all together and then just group by row_group or similar.

[alamb]

I think to get per-batch row group information you would have to change the parquet-rs reader

Maybe it could optionally add "provenance" metadata or something to each RecordBatch that came out? Added as user defined Metadata on the schema of each RecordBatch

https://docs.rs/arrow/latest/arrow/datatypes/struct.Schema.html#structfield.metadata

🤔

[adriangb]

Yep that's basically what I'm asking for. Although I don't know if theres a fixed 1:m relationship between row groups and RecordBatch (i.e. if 1 RecordBatch can have more than 1 row group).

[alamb]

You know, something that is a bit of a hack but would probably work would be:

1. Read the data using a single Partition (or single ArrowReader)
2. Use the row_counts in the RowGroupMetaData to determine where the data in each row group ends

That would require reading the entire file with only a single core

Either way I end up having to go back to the bytes once for each row group, which seems like it would be less efficient than if I could do the read all together and then just group by row_group or similar.

I think it is a tradeoff -- it would be more work overall to read each RowGroup with a separate reader, but it could be done by multiple threads at once.