replace custom ParquetSstReader to DataFusion's ParquetExec

[jiacai2050]

Describe This Problem

In current implementation

• ParquetSstReader use CacheableSerializedFileReader to read parquet file, and
• CacheableSerializedFileReader is modeled after https://github.com/apache/arrow-rs/blob/5.2.0/parquet/src/file/serialized_reader.rs, which contains lots of low-level details about how to parse parquet, and there are already some issues with it, such as When upgrade to parquet 23, some usages are deprecated. #271

Proposal

By leverage DataFusion's ParquetExec to implement this, we can remove all those low-level details and it's already feature rich, for example:

• Support predicate pushdown
• Support custom ParquetFileReaderFactory, which can be used to implement row-group level cache, just as what we do now

Additional Context

IOx already does this.

• https://github.com/influxdata/influxdb_iox/pull/5531/files

[ygf11]

Seems this issue is relative to:

• Support complex filter before merge procedure #256

In #256, we want to implement filter logics with datafusion's expr.

If we replace ParquetSstReader with ParquetExec, I think the filter logic is also there.

[jiacai2050]

@ygf11 yeah, those issue are related, but they work in different levels.

In current implementation, where condition is filtered multiple times,

• The underlying one is what this issue concerns, predicate works on sst file
• The next one is Support complex filter before merge procedure #256, predicate works before pass to MergeIterator, which is used to merge multiple source into one deduplicated.
• The last one is FilterExec, which utilize DataFusion to ensure all rows match where condition

[ygf11]

yeah, those issue are related, but they work in different levels.

Thanks, I almost got it. If I have mistake, please take it out.

First, We will read data from two places, Memtables and Sst files. And we will read sst with ParquetExec after this task.

Predicate works before pass to MergeIterator:

• For sst files, we can directly pass filter expr to ParquetExec.
• For Memtables, we still need implement filter logics manually, that's what Support complex filter before merge procedure #256 need do.

Further more, after #256 and this task, we can remove FilterExec, since we have pushed down all filters.

[ShiKaiWi]

There are some mistakes here:

For sst files, we can directly pass filter expr to ParquetExec.

This is true, but one thing worth mentioning that the data is not ensured to meet the requirements by pushed down predicates (actually the filter is implemented by utilizing the min-max index on the row-group level.)

For Memtables, we still need implement filter logics manually, that's what #256 need do.

So the further filter works for not only the memtables but also the data from sst files. In conclusion, both #256 and #291 are necessary.

[ygf11]

This is true, but one thing worth mentioning that the data is not ensured to meet the requirements by pushed down predicates (actually the filter is implemented by utilizing the min-max index on the row-group level.)

Thanks for figuring it out. It is clear now :D.

[jiacai2050]

Already fixed in main.