Support get_row_group in AsyncFileReader

[Ted-Jiang]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.
When i implementing apache/datafusion#4512
I found in AsyncFileReader(df used) can not get the specific RowGroupReader

If i got the RowGroupReader then call get_column_bloom_filter will return the bloomFilter

arrow-rs/parquet/src/file/reader.rs

Lines 77 to 94 in f78a9be

	pub trait FileReader: Send + Sync {
	/// Get metadata information about this file.
	fn metadata(&self) -> &ParquetMetaData;
	/// Get the total number of row groups for this file.
	fn num_row_groups(&self) -> usize;
	/// Get the `i`th row group reader. Note this doesn't do bound check.
	fn get_row_group(&self, i: usize) -> Result<Box<dyn RowGroupReader + '_>>;
	/// Get full iterator of `Row`s from a file (over all row groups).
	///
	/// Iterator will automatically load the next row group to advance.
	///
	/// Projected schema can be a subset of or equal to the file schema, when it is None,
	/// full file schema is assumed.
	fn get_row_iter(&self, projection: Option<SchemaType>) -> Result<RowIter>;
	}

async version:

arrow-rs/parquet/src/arrow/async_reader/mod.rs

Lines 128 to 148 in de9f826

	pub trait AsyncFileReader: Send {
	/// Retrieve the bytes in `range`
	fn get_bytes(&mut self, range: Range<usize>) -> BoxFuture<'_, Result<Bytes>>;
	/// Retrieve multiple byte ranges. The default implementation will call `get_bytes` sequentially
	fn get_byte_ranges(
	&mut self,
	ranges: Vec<Range<usize>>,
	) -> BoxFuture<'_, Result<Vec<Bytes>>> {
	async move {
	let mut result = Vec::with_capacity(ranges.len());
	for range in ranges.into_iter() {
	let data = self.get_bytes(range).await?;
	result.push(data);
	}
	Ok(result)
	}
	.boxed()
	}

I think they should be consistent, Is there any other reason not supported?
Describe the solution you'd like

So i try to create a new struct AsyncRowGroupReader

Describe alternatives you've considered

Additional context

[Ted-Jiang]

@tustvold @alamb Could you take a look, is this a reasonable change 🤔 ?

[tustvold]

I don't think it is that simple as the IO to fetch the bloom filter data needs to be done ahead of time. I'll have a think over the next couple of days and write up how to support this

[Ted-Jiang]

btw, i think should support read the special bloom filter by row_group_id and col_id unlike page Index.

[madgene]

@Ted-Jiang 如果用SerializedFileReader先把parquet文件中的所有bloom filter数据都读取出来，然后调用prune_row_groups_by_bloom_filter方法做prune，PruningPredicate::prune中首先创建一个RecordBatch再通过self.predicate_expr.evaluate(&statistics_batch)方法。请教一下怎么用bloom filter数据构建RecordBatch再evaluate，或者是通过什么别的方式能够evaluate？

[Ted-Jiang]

@Ted-Jiang 如果用SerializedFileReader先把parquet文件中的所有bloom filter数据都读取出来，然后调用prune_row_groups_by_bloom_filter方法做prune，PruningPredicate::prune中首先创建一个RecordBatch再通过self.predicate_expr.evaluate(&statistics_batch)方法。请教一下怎么用bloom filter数据构建RecordBatch再evaluate，或者是通过什么别的方式能够evaluate？

Sorry，plz communicate in English

[madgene]

@Ted-Jiang Perhaps we can prune rowgroups by invoking prune_row_groups_by_bloom_filter method with all bloom filter data which is read with SerializedFileReader. In PruningPredicate::prune, a RecordBatch is first created and then evaluated using the self.predicate_expr.evaluate(&statistics_batch) method. Could you please advise me how to construct a RecordBatch using bloom filter data and then evaluate it, or is there any other way to evaluate?

[hengfeiyang]

@tustvold Any suggestions for this? we want to read the bloom filter in Datafusion. apache/datafusion#4512

[tustvold]

The simplest option is likely to add an async function to ParquetRecordBatchStreamBuilder that fetches and decodes the bloom filters for a given column and row group, and returns this.

We likely don't want to eagerly fetch and decode bloom filters as the formulation parquet uses tends to be very large, and so we should only pay for this after other forms of pruning and if we have an equality predicate on that column.

[tustvold]

Closed by #4917