Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you all sign our Contributor License Agreement before we can accept your contribution.
1 out of 2 committers have signed the CLA.

✅ yxheartipp
❌ liuneng1994

This is probably outside the scope of this PR, but I think there's a better way to do PREWHERE-like prefiltering and input_format_parquet_bloom_filter_push_down. Especially when reading from network (rather than local file), especially with high latency. It seems that the best read strategy would be to read in 4 stages, pipelined as 4 sliding windows of tasks running in parallel (with some dependencies between them):

1. Filter prefetch: read file byte ranges corresponding to bloom filters and/or the smallest/most-promising filter columns (similar to PREWHERE). Runs on IO threads with very high parallelism (important when reading from network), for lots of row groups at once, because these filters/columns are small and don't take a lot of memory per row group.
2. Decompress and decode those bloom filters and filter columns. Evaluate filter expressions and produce filter masks. On non-IO threads, with parallelism matching the number of cores. One task per row group is ok, no need to parallelize by individual columns. The resulting filter masks are small (especially if we bit-pack them), so we can afford these tasks to run far ahead of the main data reading, usually all the way to the end of the file (for good read size estimate for the progress bar).
3. Main data prefetch: read file byte ranges corresponding to the remaining columns. Use filter masks from previous step and page offset index to determine byte ranges to read. That's why this is a separate step from filtering - we want to know tight read ranges before we start reading. Runs on IO threads, with medium parallelism. Can't have very high parallelism because this takes a lot of memory. But higher than decompression+decoding because data is still compressed at this stage. Can read byte ranges within the same row group in parallel, can merge multiple small row groups into one byte range.
4. Decompress and decode the remaining columns. Runs on non-IO threads, with parallelism matching the number of cores, or limited by memory. Parallelized at the level of primitive columns, not row groups (this is a problem for the current reader that can use at most one thread per row group, but we often don't have enough memory for many row groups) (this is another reason to build filter masks separately - it allows reading columns independently from each other). Composite columns like tuples can be reassembled at the very end before returning the block from ParquetBlockInputFormat.

Each of these would be pipelined across row groups, with a low and high watermark. The dependencies are simple: next stage for a row group can begin when the previous stage completed for that row group (but we won't necessarily begin one row group at a time; for small row groups we'll wait for a few row groups to be ready and schedule them as one task, based a target byte size).

(As a slight generalization, there can be multiple filtering stages: first by the smallest and most promising filter columns, then by other filter columns, then read main data. This hopefully won't be needed.)

(Another consideration is that sometimes when reading filter byte ranges we'll also read some data along the way, if gaps between ranges are small enough that we decided to read instead of seeking. Would be nice to preserve and reuse that read data in later stages instead of reading again. But other times we would rather deallocate it to free up more memory for later stages. Not sure how exactly to handle that.)

Maybe I'll make a native reader v3 at some point :) . This is speculative though, please finish this PR too.

(I previously assumed that this PR would have close to optimal performance and would need at most need incremental optimizations/simplifications. But now that I wrote down the above idea, it sounds very different from this PR, and potentially significantly faster and simpler (though it's likely that I missed something and it won't work). Probably faster to prototype it from scratch than modifying reader v2.)

This is probably outside the scope of this PR, but I think there's a better way to do PREWHERE-like prefiltering and input_format_parquet_bloom_filter_push_down. Especially when reading from network (rather than local file), especially with high latency. It seems that the best read strategy would be to read in 4 stages, pipelined as 4 sliding windows of tasks running in parallel (with some dependencies between them):

1. Filter prefetch: read file byte ranges corresponding to bloom filters and/or the smallest/most-promising filter columns (similar to PREWHERE). Runs on IO threads with very high parallelism (important when reading from network), for lots of row groups at once, because these filters/columns are small and don't take a lot of memory per row group.
2. Decompress and decode those bloom filters and filter columns. Evaluate filter expressions and produce filter masks. On non-IO threads, with parallelism matching the number of cores. One task per row group is ok, no need to parallelize by individual columns. The resulting filter masks are small (especially if we bit-pack them), so we can afford these tasks to run far ahead of the main data reading, usually all the way to the end of the file (for good read size estimate for the progress bar).
3. Main data prefetch: read file byte ranges corresponding to the remaining columns. Use filter masks from previous step and page offset index to determine byte ranges to read. That's why this is a separate step from filtering - we want to know tight read ranges before we start reading. Runs on IO threads, with medium parallelism. Can't have very high parallelism because this takes a lot of memory. But higher than decompression+decoding because data is still compressed at this stage. Can read byte ranges within the same row group in parallel, can merge multiple small row groups into one byte range.
4. Decompress and decode the remaining columns. Runs on non-IO threads, with parallelism matching the number of cores, or limited by memory. Parallelized at the level of primitive columns, not row groups (this is a problem for the current reader that can use at most one thread per row group, but we often don't have enough memory for many row groups) (this is another reason to build filter masks separately - it allows reading columns independently from each other). Composite columns like tuples can be reassembled at the very end before returning the block from ParquetBlockInputFormat.

Each of these would be pipelined across row groups, with a low and high watermark. The dependencies are simple: next stage for a row group can begin when the previous stage completed for that row group (but we won't necessarily begin one row group at a time; for small row groups we'll wait for a few row groups to be ready and schedule them as one task, based a target byte size).

(As a slight generalization, there can be multiple filtering stages: first by the smallest and most promising filter columns, then by other filter columns, then read main data. This hopefully won't be needed.)

(Another consideration is that sometimes when reading filter byte ranges we'll also read some data along the way, if gaps between ranges are small enough that we decided to read instead of seeking. Would be nice to preserve and reuse that read data in later stages instead of reading again. But other times we would rather deallocate it to free up more memory for later stages. Not sure how exactly to handle that.)

Maybe I'll make a native reader v3 at some point :) . This is speculative though, please finish this PR too.

(I previously assumed that this PR would have close to optimal performance and would need at most need incremental optimizations/simplifications. But now that I wrote down the above idea, it sounds very different from this PR, and potentially significantly faster and simpler (though it's likely that I missed something and it won't work). Probably faster to prototype it from scratch than modifying reader v2.)

This is a very professional suggestion. I originally made this reader hoping to have a more efficient single-threaded reader. I did not consider the parallel issues too much and only did some parallel processing on IO.

Dear @al13n321, this PR hasn't been updated for a while. You will be unassigned. Will you continue working on it? If so, please feel free to reassign yourself.

I'm working on another parquet reader: #78380 (only the last 3 commits in that PR are relevant). Current plan is to merge both this PR ("v2") and that PR ("v3"), then see if one is pareto-better than the other and maybe remove one (and also remove v1).

The number of parquet readers in the codebase is getting kind of ridiculous, we should do something to separate them more cleanly and make them easy to remove. I'm thinking:

• Move the writer, native readers v1, v2 (yours), and v3 (mine) to 4 subdirectories of Formats/Impl/Parquet. Please move the v2 into a directory in this PR, then I'll send a separate PR to move the existing files (writer and v1).
• Don't add parquet-specific things to KeyCondition. Looks like the KeyCondition always comes from a call to SourceWithKeyCondition::setKeyCondition(const std::optional<ActionsDAG>, ContextPtr). So maybe we can just remove the overload setKeyCondition(const std::shared_ptr<const KeyCondition>) and make SourceWithKeyCondition store an ActionsDAG instead of KeyCondition? The format can create KeyCondition itself. Please try this. (Ideally we would also do something to reuse the KeyCondition/ColumnFilterHelper when reading many files in one query; I'm working on it, will try to unify it with Utilizing a Shared Parsing Pool for Multiple Parquet Streams #66253 as well.) Use Refactor how IInputFormat deals with reading many files at once #80931 instead.

(I was also hoping that some v3 things would be easy to reuse in v2, but that doesn't seem to be the case. V3 integrates with prewhere, and will probably use KeyCondition for all non-prewhere filtering instead of re-analyzing the ActionsDAG; but that doesn't seem to fit v2 because v2 wants to split all conditions by columns, so it would need to pointlessly reimplement the splitting for KeyCondition and prewhere separately. V3 has some complicated fine-grained task scheduling and prefetching to extract ~all available parallelism and to avoid prefetching data before filters are applied; it doesn't seem feasible to reuse at all.)

Please make these changes, then we can get this merged and start benchmarking (and hopefully v3 will be ready soon so we can benchmark everything at once). You don't have to address all the nitpicks from review comments, up to you.

Don't add parquet-specific things to KeyCondition.

Use #80931 to get access to ActionsDAG from ParquetBlockInputFormat.

@liuneng1994, thank you so much for the contribution, which provided the basis of the currently merged implementation!