You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2022/08/12 15:27:31 UTC
[GitHub] [arrow-rs] tustvold opened a new pull request, #2435: Add ParquetRecordBatchReaderBuilder (#2427)
tustvold opened a new pull request, #2435:
URL: https://github.com/apache/arrow-rs/pull/2435
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
Closes #2427
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
This standardises the configuration of the async and sync arrow parquet readers, helping to avoid inconsistency and reducing duplication.
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
Adds a new `ParquetRecordBatchReaderBuilder` and deprecates the old APIs
# Are there any user-facing changes?
This deprecates old APIs, however, it doesn't remove any of them
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!---
If there are any breaking changes to public APIs, please add the `breaking change` label.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold commented on PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#issuecomment-1213233894
@Ted-Jiang could you perhaps take a look and make sure this makes sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r944580751
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -427,46 +606,36 @@ mod tests {
};
use crate::errors::Result;
use crate::file::properties::{EnabledStatistics, WriterProperties, WriterVersion};
- use crate::file::reader::{FileReader, SerializedFileReader};
use crate::file::writer::SerializedFileWriter;
use crate::schema::parser::parse_message_type;
use crate::schema::types::{Type, TypePtr};
use crate::util::test_common::rand_gen::RandGen;
#[test]
fn test_arrow_reader_all_columns() {
- let parquet_file_reader =
- get_test_reader("parquet/generated_simple_numerics/blogs.parquet");
+ let file = get_test_file("parquet/generated_simple_numerics/blogs.parquet");
- let mut arrow_reader = ParquetFileArrowReader::new(parquet_file_reader);
-
- let record_batch_reader = arrow_reader
- .get_record_reader(60)
- .expect("Failed to read into array!");
+ let builder = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
+ let original_schema = Arc::clone(&builder.schema());
+ let reader = builder.build().unwrap();
// Verify that the schema was correctly parsed
- let original_schema = arrow_reader.get_schema().unwrap().fields().clone();
- assert_eq!(original_schema, *record_batch_reader.schema().fields());
+ assert_eq!(original_schema.fields(), reader.schema().fields());
Review Comment:
I'm not really sure this test makes sense anymore, but I kept the spirit of it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r944581270
##########
parquet/src/arrow/async_reader.rs:
##########
@@ -194,112 +194,23 @@ impl<T: AsyncRead + AsyncSeek + Unpin + Send> AsyncFileReader for T {
}
}
+#[doc(hidden)]
+/// A newtype used within [`ReaderOptionsBuilder`] to distinguish sync readers from async
+pub struct AsyncReader<T>(T);
+
/// A builder used to construct a [`ParquetRecordBatchStream`] for a parquet file
///
/// In particular, this handles reading the parquet file metadata, allowing consumers
/// to use this information to select what specific columns, row groups, etc...
/// they wish to be read by the resulting stream
///
-pub struct ParquetRecordBatchStreamBuilder<T> {
- input: T,
-
- metadata: Arc<ParquetMetaData>,
-
- schema: SchemaRef,
-
- batch_size: usize,
-
- row_groups: Option<Vec<usize>>,
-
- projection: ProjectionMask,
-
- filter: Option<RowFilter>,
-
- selection: Option<RowSelection>,
-}
+pub type ParquetRecordBatchStreamBuilder<T> = ArrowReaderBuilder<AsyncReader<T>>;
-impl<T: AsyncFileReader + Send + 'static> ParquetRecordBatchStreamBuilder<T> {
+impl<T: AsyncFileReader + Send + 'static> ArrowReaderBuilder<AsyncReader<T>> {
/// Create a new [`ParquetRecordBatchStreamBuilder`] with the provided parquet file
pub async fn new(mut input: T) -> Result<Self> {
let metadata = input.get_metadata().await?;
-
- let schema = Arc::new(parquet_to_arrow_schema(
- metadata.file_metadata().schema_descr(),
- metadata.file_metadata().key_value_metadata(),
- )?);
-
- Ok(Self {
- input,
- metadata,
- schema,
- batch_size: 1024,
- row_groups: None,
- projection: ProjectionMask::all(),
- filter: None,
- selection: None,
- })
- }
-
- /// Returns a reference to the [`ParquetMetaData`] for this parquet file
- pub fn metadata(&self) -> &Arc<ParquetMetaData> {
Review Comment:
This logic is moved into ArrowReaderBuilder
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] thinkharderdev commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
thinkharderdev commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r945747604
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -48,9 +50,127 @@ pub(crate) use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
#[allow(unused_imports)]
pub(crate) use selection::{RowSelection, RowSelector};
+/// A generic builder for constructing sync or async arrow parquet readers. This is not intended
+/// to be used directly, instead you should use the specialization for the type of reader
+/// you wish to use
+///
+/// * For a synchronous API - [`ParquetRecordBatchReaderBuilder`]
+/// * For an asynchronous API - [`ParquetRecordBatchStreamBuilder`]
+///
+/// [`ParquetRecordBatchStreamBuilder`]: [crate::arrow::async_reader::ParquetRecordBatchStreamBuilder]
+pub struct ArrowReaderBuilder<T> {
+ pub(crate) input: T,
+
+ pub(crate) metadata: Arc<ParquetMetaData>,
+
+ pub(crate) schema: SchemaRef,
+
+ pub(crate) batch_size: usize,
+
+ pub(crate) row_groups: Option<Vec<usize>>,
+
+ pub(crate) projection: ProjectionMask,
+
+ pub(crate) filter: Option<RowFilter>,
+
+ pub(crate) selection: Option<RowSelection>,
+}
+
+impl<T> ArrowReaderBuilder<T> {
+ pub(crate) fn new_builder(
+ input: T,
+ metadata: Arc<ParquetMetaData>,
+ options: ArrowReaderOptions,
+ ) -> Result<Self> {
+ let kv_metadata = match options.skip_arrow_metadata {
+ true => None,
+ false => metadata.file_metadata().key_value_metadata(),
+ };
+
+ let schema = Arc::new(parquet_to_arrow_schema(
+ metadata.file_metadata().schema_descr(),
+ kv_metadata,
+ )?);
+
+ Ok(Self {
+ input,
+ metadata,
+ schema,
+ batch_size: 1024,
+ row_groups: None,
+ projection: ProjectionMask::all(),
+ filter: None,
+ selection: None,
+ })
+ }
+
+ /// Returns a reference to the [`ParquetMetaData`] for this parquet file
+ pub fn metadata(&self) -> &Arc<ParquetMetaData> {
+ &self.metadata
+ }
+
+ /// Returns the parquet [`SchemaDescriptor`] for this parquet file
+ pub fn parquet_schema(&self) -> &SchemaDescriptor {
+ self.metadata.file_metadata().schema_descr()
+ }
+
+ /// Returns the arrow [`SchemaRef`] for this parquet file
+ pub fn schema(&self) -> &SchemaRef {
+ &self.schema
+ }
+
+ /// Set the size of [`RecordBatch`] to produce
+ pub fn with_batch_size(self, batch_size: usize) -> Self {
+ Self { batch_size, ..self }
+ }
+
+ /// Only read data from the provided row group indexes
+ pub fn with_row_groups(self, row_groups: Vec<usize>) -> Self {
+ Self {
+ row_groups: Some(row_groups),
+ ..self
+ }
+ }
+
+ /// Only read data from the provided column indexes
+ pub fn with_projection(self, mask: ProjectionMask) -> Self {
+ Self {
+ projection: mask,
+ ..self
+ }
+ }
+
+ /// Provide a [`RowSelection] to filter out rows, and avoid fetching their
+ /// data into memory
+ ///
+ /// Row group filtering is applied prior to this, and rows from skipped
+ /// row groups should not be included in the [`RowSelection`]
+ ///
+ /// TODO: Make public once stable (#1792)
+ #[allow(unused)]
+ pub(crate) fn with_row_selection(self, selection: RowSelection) -> Self {
+ Self {
+ selection: Some(selection),
+ ..self
+ }
+ }
Review Comment:
Hmm, would it make sense to collapse `with_row_selection` and `with_row_filter`. The API is a bit confusing with both. And you could always just define a `RowSelection` as an `ArrowPredicate`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] ursabot commented on pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
ursabot commented on PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#issuecomment-1215081694
Benchmark runs are scheduled for baseline = 3f0e12d8d362752181c75836d25d424862acc424 and contender = 76cfe8306e316c2bf068ed38fe9612dd0e5bdc11. 76cfe8306e316c2bf068ed38fe9612dd0e5bdc11 is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Skipped :warning: Benchmarking of arrow-rs-commits is not supported on ec2-t3-xlarge-us-east-2] [ec2-t3-xlarge-us-east-2](https://conbench.ursa.dev/compare/runs/da116332cb16435b81b9c257e153c559...ecc83824649241628a1392d8fcc369f5/)
[Skipped :warning: Benchmarking of arrow-rs-commits is not supported on test-mac-arm] [test-mac-arm](https://conbench.ursa.dev/compare/runs/bac0ec89194646b59933b5d260aac5fd...610b7719475e498e99b567abb2516ba2/)
[Skipped :warning: Benchmarking of arrow-rs-commits is not supported on ursa-i9-9960x] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/364ea01195a84a57b90b690e189e9aca...e79af702f6704cde9a5900c9a6e99f88/)
[Skipped :warning: Benchmarking of arrow-rs-commits is not supported on ursa-thinkcentre-m75q] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/74cb2097d4c14b0898ee9ce961b40839...fd3e237ca99b4a36b16f94f64959a423/)
Buildkite builds:
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold commented on PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#issuecomment-1213939390
I filed https://github.com/apache/arrow-rs/issues/2430 to track adding page index support to the async reader. There is a slight additional complication as it needs to perform IO to read the corresponding bytes, but nothing intractable. Thinking a bit more, I wonder if this should be handled by AsyncFileReader? π€
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] Ted-Jiang commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r945093466
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -84,10 +204,14 @@ pub trait ArrowReader {
) -> Result<Self::RecordReader>;
}
+/// Options that control how metadata is read for a parquet file
+///
+/// See [`ArrowReaderBuilder`] for how to configure how the column data
+/// is then read from the file, including projection and filter pushdown
#[derive(Debug, Clone, Default)]
pub struct ArrowReaderOptions {
skip_arrow_metadata: bool,
- selection: Option<RowSelection>,
+ page_index: bool,
Review Comment:
Sounds reasonable.
I think read page_index should belong to the open file, have you find out how many times read page_index costπ€
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r944581777
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -48,9 +50,127 @@ pub(crate) use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
#[allow(unused_imports)]
pub(crate) use selection::{RowSelection, RowSelector};
+/// A generic builder for constructing sync or async arrow parquet readers. This is not intended
+/// to be used directly, instead you should use the specialization for the type of reader
+/// you wish to use
+///
+/// * For a synchronous API - [`ParquetRecordBatchReaderBuilder`]
+/// * For an asynchronous API - [`ParquetRecordBatchStreamBuilder`]
+///
+/// [`ParquetRecordBatchStreamBuilder`]: [crate::arrow::async_reader::ParquetRecordBatchStreamBuilder]
+pub struct ArrowReaderBuilder<T> {
Review Comment:
This is largely moved from `ParquetRecordBatchStreamBuilder`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] alamb commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
alamb commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r945699776
##########
parquet/src/arrow/array_reader/mod.rs:
##########
@@ -124,6 +124,49 @@ impl RowGroupCollection for Arc<dyn FileReader> {
}
}
+pub(crate) struct FileReaderRowGroupCollection {
+ reader: Arc<dyn FileReader>,
+ row_groups: Option<Vec<usize>>,
Review Comment:
I think it would help to document what `usize` means here -- I assume it is the index of the row group within the parquet file? And that if this is None, all row groups will be read?
##########
parquet/src/arrow/async_reader.rs:
##########
@@ -194,112 +194,23 @@ impl<T: AsyncRead + AsyncSeek + Unpin + Send> AsyncFileReader for T {
}
}
+#[doc(hidden)]
+/// A newtype used within [`ReaderOptionsBuilder`] to distinguish sync readers from async
Review Comment:
```suggestion
/// A newtype used within [`ReaderOptionsBuilder`] to distinguish sync readers from async
/// Allows sharing the same builder for both the sync and async versions, whilst also not
/// breaking the existing ParquetRecordBatchStreamBuilder API
```
##########
parquet/src/arrow/schema.rs:
##########
@@ -1662,14 +1662,9 @@ mod tests {
writer.close()?;
// read file back
- let mut arrow_reader = ParquetFileArrowReader::try_new(file).unwrap();
- let read_schema = arrow_reader.get_schema()?;
- assert_eq!(schema, read_schema);
-
- // read all fields by columns
Review Comment:
Isn't the usecase (and test) of reading a partial schema still valid?
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -48,9 +50,127 @@ pub(crate) use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
#[allow(unused_imports)]
pub(crate) use selection::{RowSelection, RowSelector};
+/// A generic builder for constructing sync or async arrow parquet readers. This is not intended
+/// to be used directly, instead you should use the specialization for the type of reader
+/// you wish to use
+///
+/// * For a synchronous API - [`ParquetRecordBatchReaderBuilder`]
+/// * For an asynchronous API - [`ParquetRecordBatchStreamBuilder`]
+///
+/// [`ParquetRecordBatchStreamBuilder`]: [crate::arrow::async_reader::ParquetRecordBatchStreamBuilder]
+pub struct ArrowReaderBuilder<T> {
+ pub(crate) input: T,
+
+ pub(crate) metadata: Arc<ParquetMetaData>,
+
+ pub(crate) schema: SchemaRef,
+
+ pub(crate) batch_size: usize,
+
+ pub(crate) row_groups: Option<Vec<usize>>,
+
+ pub(crate) projection: ProjectionMask,
+
+ pub(crate) filter: Option<RowFilter>,
+
+ pub(crate) selection: Option<RowSelection>,
+}
+
+impl<T> ArrowReaderBuilder<T> {
Review Comment:
This is looking like a very nice api π π¨βπ³
π© tip to you @tustvold @Ted-Jiang and @thinkharderdev for this. Very cool
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -48,9 +50,127 @@ pub(crate) use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
#[allow(unused_imports)]
pub(crate) use selection::{RowSelection, RowSelector};
+/// A generic builder for constructing sync or async arrow parquet readers. This is not intended
+/// to be used directly, instead you should use the specialization for the type of reader
+/// you wish to use
+///
+/// * For a synchronous API - [`ParquetRecordBatchReaderBuilder`]
+/// * For an asynchronous API - [`ParquetRecordBatchStreamBuilder`]
+///
+/// [`ParquetRecordBatchStreamBuilder`]: [crate::arrow::async_reader::ParquetRecordBatchStreamBuilder]
+pub struct ArrowReaderBuilder<T> {
+ pub(crate) input: T,
+
+ pub(crate) metadata: Arc<ParquetMetaData>,
+
+ pub(crate) schema: SchemaRef,
+
+ pub(crate) batch_size: usize,
+
+ pub(crate) row_groups: Option<Vec<usize>>,
+
+ pub(crate) projection: ProjectionMask,
+
+ pub(crate) filter: Option<RowFilter>,
+
+ pub(crate) selection: Option<RowSelection>,
+}
+
+impl<T> ArrowReaderBuilder<T> {
+ pub(crate) fn new_builder(
+ input: T,
+ metadata: Arc<ParquetMetaData>,
+ options: ArrowReaderOptions,
+ ) -> Result<Self> {
+ let kv_metadata = match options.skip_arrow_metadata {
+ true => None,
+ false => metadata.file_metadata().key_value_metadata(),
+ };
+
+ let schema = Arc::new(parquet_to_arrow_schema(
+ metadata.file_metadata().schema_descr(),
+ kv_metadata,
+ )?);
+
+ Ok(Self {
+ input,
+ metadata,
+ schema,
+ batch_size: 1024,
+ row_groups: None,
+ projection: ProjectionMask::all(),
+ filter: None,
+ selection: None,
+ })
+ }
+
+ /// Returns a reference to the [`ParquetMetaData`] for this parquet file
+ pub fn metadata(&self) -> &Arc<ParquetMetaData> {
+ &self.metadata
+ }
+
+ /// Returns the parquet [`SchemaDescriptor`] for this parquet file
+ pub fn parquet_schema(&self) -> &SchemaDescriptor {
+ self.metadata.file_metadata().schema_descr()
+ }
+
+ /// Returns the arrow [`SchemaRef`] for this parquet file
+ pub fn schema(&self) -> &SchemaRef {
+ &self.schema
+ }
+
+ /// Set the size of [`RecordBatch`] to produce
+ pub fn with_batch_size(self, batch_size: usize) -> Self {
+ Self { batch_size, ..self }
+ }
+
+ /// Only read data from the provided row group indexes
+ pub fn with_row_groups(self, row_groups: Vec<usize>) -> Self {
+ Self {
+ row_groups: Some(row_groups),
+ ..self
+ }
+ }
+
+ /// Only read data from the provided column indexes
+ pub fn with_projection(self, mask: ProjectionMask) -> Self {
+ Self {
+ projection: mask,
+ ..self
+ }
+ }
+
+ /// Provide a [`RowSelection] to filter out rows, and avoid fetching their
+ /// data into memory
+ ///
+ /// Row group filtering is applied prior to this, and rows from skipped
+ /// row groups should not be included in the [`RowSelection`]
+ ///
+ /// TODO: Make public once stable (#1792)
+ #[allow(unused)]
+ pub(crate) fn with_row_selection(self, selection: RowSelection) -> Self {
+ Self {
+ selection: Some(selection),
+ ..self
+ }
+ }
+
+ /// Provide a [`RowFilter`] to skip decoding rows
Review Comment:
```suggestion
/// Provide a [`RowFilter`] to skip decoding rows. Row filters are applied
/// after row group selection and row selection
```
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -84,10 +204,14 @@ pub trait ArrowReader {
) -> Result<Self::RecordReader>;
}
+/// Options that control how metadata is read for a parquet file
+///
+/// See [`ArrowReaderBuilder`] for how to configure how the column data
+/// is then read from the file, including projection and filter pushdown
#[derive(Debug, Clone, Default)]
pub struct ArrowReaderOptions {
skip_arrow_metadata: bool,
- selection: Option<RowSelection>,
+ page_index: bool,
Review Comment:
```suggestion
/// if true, forces decoding of the page index for all row groups
/// as the group selection isn't known at the point the metadata is read
page_index: bool,
```
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -48,9 +50,127 @@ pub(crate) use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
#[allow(unused_imports)]
pub(crate) use selection::{RowSelection, RowSelector};
+/// A generic builder for constructing sync or async arrow parquet readers. This is not intended
+/// to be used directly, instead you should use the specialization for the type of reader
+/// you wish to use
+///
+/// * For a synchronous API - [`ParquetRecordBatchReaderBuilder`]
+/// * For an asynchronous API - [`ParquetRecordBatchStreamBuilder`]
+///
+/// [`ParquetRecordBatchStreamBuilder`]: [crate::arrow::async_reader::ParquetRecordBatchStreamBuilder]
Review Comment:
Eventually it would be great it update the examples to use this (much nicer) API as well: https://docs.rs/parquet/20.0.0/parquet/arrow/index.html
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -48,9 +50,127 @@ pub(crate) use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
#[allow(unused_imports)]
pub(crate) use selection::{RowSelection, RowSelector};
+/// A generic builder for constructing sync or async arrow parquet readers. This is not intended
+/// to be used directly, instead you should use the specialization for the type of reader
+/// you wish to use
+///
+/// * For a synchronous API - [`ParquetRecordBatchReaderBuilder`]
+/// * For an asynchronous API - [`ParquetRecordBatchStreamBuilder`]
+///
+/// [`ParquetRecordBatchStreamBuilder`]: [crate::arrow::async_reader::ParquetRecordBatchStreamBuilder]
+pub struct ArrowReaderBuilder<T> {
+ pub(crate) input: T,
+
+ pub(crate) metadata: Arc<ParquetMetaData>,
+
+ pub(crate) schema: SchemaRef,
+
+ pub(crate) batch_size: usize,
+
+ pub(crate) row_groups: Option<Vec<usize>>,
+
+ pub(crate) projection: ProjectionMask,
+
+ pub(crate) filter: Option<RowFilter>,
+
+ pub(crate) selection: Option<RowSelection>,
+}
+
+impl<T> ArrowReaderBuilder<T> {
+ pub(crate) fn new_builder(
+ input: T,
+ metadata: Arc<ParquetMetaData>,
+ options: ArrowReaderOptions,
+ ) -> Result<Self> {
+ let kv_metadata = match options.skip_arrow_metadata {
+ true => None,
+ false => metadata.file_metadata().key_value_metadata(),
+ };
+
+ let schema = Arc::new(parquet_to_arrow_schema(
+ metadata.file_metadata().schema_descr(),
+ kv_metadata,
+ )?);
+
+ Ok(Self {
+ input,
+ metadata,
+ schema,
+ batch_size: 1024,
+ row_groups: None,
+ projection: ProjectionMask::all(),
+ filter: None,
+ selection: None,
+ })
+ }
+
+ /// Returns a reference to the [`ParquetMetaData`] for this parquet file
+ pub fn metadata(&self) -> &Arc<ParquetMetaData> {
+ &self.metadata
+ }
+
+ /// Returns the parquet [`SchemaDescriptor`] for this parquet file
+ pub fn parquet_schema(&self) -> &SchemaDescriptor {
+ self.metadata.file_metadata().schema_descr()
+ }
+
+ /// Returns the arrow [`SchemaRef`] for this parquet file
+ pub fn schema(&self) -> &SchemaRef {
+ &self.schema
+ }
+
+ /// Set the size of [`RecordBatch`] to produce
Review Comment:
```suggestion
/// Set the size of [`RecordBatch`] to produce. Defaults to 1024
```
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -109,27 +233,29 @@ impl ArrowReaderOptions {
}
}
- /// Scan rows from the parquet file according to the provided `selection`
+ /// Set this true to enable decoding of the [PageIndex] if present. This can be used
+ /// to push down predicates to the parquet scan, potentially eliminating unnecessary IO
///
- /// TODO: Revisit this API, as [`Self`] is provided before the file metadata is available
- #[allow(unused)]
- pub(crate) fn with_row_selection(self, selection: impl Into<RowSelection>) -> Self {
Review Comment:
Isn't it moved to `ArrowReaderBuilder::with_row_selection`?
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -1309,20 +1464,21 @@ mod tests {
let testdata = arrow::util::test_util::parquet_test_data();
let path = format!("{}/nested_structs.rust.parquet", testdata);
let file = File::open(&path).unwrap();
- let mut arrow_reader = ParquetFileArrowReader::try_new(file).unwrap();
- let record_batch_reader = arrow_reader
- .get_record_reader(60)
- .expect("Failed to read into array!");
+ let record_batch_reader = ParquetRecordBatchReader::try_new(file, 60).unwrap();
for batch in record_batch_reader {
batch.unwrap();
}
- let mask = ProjectionMask::leaves(arrow_reader.parquet_schema(), [3, 8, 10]);
- let projected_reader = arrow_reader
- .get_record_reader_by_columns(mask.clone(), 60)
+ let file = File::open(&path).unwrap();
+ let builder = ParquetRecordBatchReaderBuilder::try_new(file).unwrap();
+
+ let mask = ProjectionMask::leaves(builder.parquet_schema(), [3, 8, 10]);
Review Comment:
π very nice
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r944582529
##########
parquet/src/arrow/async_reader.rs:
##########
@@ -194,112 +194,23 @@ impl<T: AsyncRead + AsyncSeek + Unpin + Send> AsyncFileReader for T {
}
}
+#[doc(hidden)]
+/// A newtype used within [`ReaderOptionsBuilder`] to distinguish sync readers from async
+pub struct AsyncReader<T>(T);
Review Comment:
This is the type trickery that allows sharing the same builder for both the sync and async versions, whilst also not breaking the existing `ParquetRecordBatchStreamBuilder` API
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] Ted-Jiang commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r945093466
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -84,10 +204,14 @@ pub trait ArrowReader {
) -> Result<Self::RecordReader>;
}
+/// Options that control how metadata is read for a parquet file
+///
+/// See [`ArrowReaderBuilder`] for how to configure how the column data
+/// is then read from the file, including projection and filter pushdown
#[derive(Debug, Clone, Default)]
pub struct ArrowReaderOptions {
skip_arrow_metadata: bool,
- selection: Option<RowSelection>,
+ page_index: bool,
Review Comment:
Sounds reasonable.
I think read page_index should belong to the open file, have you find out how long read `page_index` costπ€
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold commented on PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#issuecomment-1215005969
> as I think everything (except RowFilter) is applicable to other uses as well.
I would rather leave this for when I eventually get to cleaning up the lower level APIs, this PR can be viewed as decoupling the arrow implementation from the other APIs, so that a subsequent PR can revisit them
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r944579710
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -84,10 +204,14 @@ pub trait ArrowReader {
) -> Result<Self::RecordReader>;
}
+/// Options that control how metadata is read for a parquet file
+///
+/// See [`ArrowReaderBuilder`] for how to configure how the column data
+/// is then read from the file, including projection and filter pushdown
#[derive(Debug, Clone, Default)]
pub struct ArrowReaderOptions {
skip_arrow_metadata: bool,
- selection: Option<RowSelection>,
+ page_index: bool,
Review Comment:
There is a detail worth highlighting here, this forces decoding of the page index for all row groups, as the row group selection isn't known at the point the metadata is read. I experimented with APIs to allow for this, but they were very clunky, and ultimately the index information should be relatively small and cheap to decode so I didn't think it was worth it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] thinkharderdev commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
thinkharderdev commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r945747604
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -48,9 +50,127 @@ pub(crate) use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
#[allow(unused_imports)]
pub(crate) use selection::{RowSelection, RowSelector};
+/// A generic builder for constructing sync or async arrow parquet readers. This is not intended
+/// to be used directly, instead you should use the specialization for the type of reader
+/// you wish to use
+///
+/// * For a synchronous API - [`ParquetRecordBatchReaderBuilder`]
+/// * For an asynchronous API - [`ParquetRecordBatchStreamBuilder`]
+///
+/// [`ParquetRecordBatchStreamBuilder`]: [crate::arrow::async_reader::ParquetRecordBatchStreamBuilder]
+pub struct ArrowReaderBuilder<T> {
+ pub(crate) input: T,
+
+ pub(crate) metadata: Arc<ParquetMetaData>,
+
+ pub(crate) schema: SchemaRef,
+
+ pub(crate) batch_size: usize,
+
+ pub(crate) row_groups: Option<Vec<usize>>,
+
+ pub(crate) projection: ProjectionMask,
+
+ pub(crate) filter: Option<RowFilter>,
+
+ pub(crate) selection: Option<RowSelection>,
+}
+
+impl<T> ArrowReaderBuilder<T> {
+ pub(crate) fn new_builder(
+ input: T,
+ metadata: Arc<ParquetMetaData>,
+ options: ArrowReaderOptions,
+ ) -> Result<Self> {
+ let kv_metadata = match options.skip_arrow_metadata {
+ true => None,
+ false => metadata.file_metadata().key_value_metadata(),
+ };
+
+ let schema = Arc::new(parquet_to_arrow_schema(
+ metadata.file_metadata().schema_descr(),
+ kv_metadata,
+ )?);
+
+ Ok(Self {
+ input,
+ metadata,
+ schema,
+ batch_size: 1024,
+ row_groups: None,
+ projection: ProjectionMask::all(),
+ filter: None,
+ selection: None,
+ })
+ }
+
+ /// Returns a reference to the [`ParquetMetaData`] for this parquet file
+ pub fn metadata(&self) -> &Arc<ParquetMetaData> {
+ &self.metadata
+ }
+
+ /// Returns the parquet [`SchemaDescriptor`] for this parquet file
+ pub fn parquet_schema(&self) -> &SchemaDescriptor {
+ self.metadata.file_metadata().schema_descr()
+ }
+
+ /// Returns the arrow [`SchemaRef`] for this parquet file
+ pub fn schema(&self) -> &SchemaRef {
+ &self.schema
+ }
+
+ /// Set the size of [`RecordBatch`] to produce
+ pub fn with_batch_size(self, batch_size: usize) -> Self {
+ Self { batch_size, ..self }
+ }
+
+ /// Only read data from the provided row group indexes
+ pub fn with_row_groups(self, row_groups: Vec<usize>) -> Self {
+ Self {
+ row_groups: Some(row_groups),
+ ..self
+ }
+ }
+
+ /// Only read data from the provided column indexes
+ pub fn with_projection(self, mask: ProjectionMask) -> Self {
+ Self {
+ projection: mask,
+ ..self
+ }
+ }
+
+ /// Provide a [`RowSelection] to filter out rows, and avoid fetching their
+ /// data into memory
+ ///
+ /// Row group filtering is applied prior to this, and rows from skipped
+ /// row groups should not be included in the [`RowSelection`]
+ ///
+ /// TODO: Make public once stable (#1792)
+ #[allow(unused)]
+ pub(crate) fn with_row_selection(self, selection: RowSelection) -> Self {
+ Self {
+ selection: Some(selection),
+ ..self
+ }
+ }
Review Comment:
Hmm, would it make sense to collapse `with_row_selection` and `with_row_filter`. The API is a bit confusing with both. And you could always just define a `RowSelection` as an `ArrowPredicate`.
Edit: To clarify a bit, I'm not sure I understand the use case in which you would have the `RowSelection` when constructing the reader. Obviously defining the selection in terms of an `ArrowPredicate` is not ideal since it is only applied after decoding, which pretty much defeats the purpose.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] thinkharderdev commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
thinkharderdev commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r945758339
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -48,9 +50,127 @@ pub(crate) use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
#[allow(unused_imports)]
pub(crate) use selection::{RowSelection, RowSelector};
+/// A generic builder for constructing sync or async arrow parquet readers. This is not intended
+/// to be used directly, instead you should use the specialization for the type of reader
+/// you wish to use
+///
+/// * For a synchronous API - [`ParquetRecordBatchReaderBuilder`]
+/// * For an asynchronous API - [`ParquetRecordBatchStreamBuilder`]
+///
+/// [`ParquetRecordBatchStreamBuilder`]: [crate::arrow::async_reader::ParquetRecordBatchStreamBuilder]
+pub struct ArrowReaderBuilder<T> {
+ pub(crate) input: T,
+
+ pub(crate) metadata: Arc<ParquetMetaData>,
+
+ pub(crate) schema: SchemaRef,
+
+ pub(crate) batch_size: usize,
+
+ pub(crate) row_groups: Option<Vec<usize>>,
+
+ pub(crate) projection: ProjectionMask,
+
+ pub(crate) filter: Option<RowFilter>,
+
+ pub(crate) selection: Option<RowSelection>,
+}
+
+impl<T> ArrowReaderBuilder<T> {
+ pub(crate) fn new_builder(
+ input: T,
+ metadata: Arc<ParquetMetaData>,
+ options: ArrowReaderOptions,
+ ) -> Result<Self> {
+ let kv_metadata = match options.skip_arrow_metadata {
+ true => None,
+ false => metadata.file_metadata().key_value_metadata(),
+ };
+
+ let schema = Arc::new(parquet_to_arrow_schema(
+ metadata.file_metadata().schema_descr(),
+ kv_metadata,
+ )?);
+
+ Ok(Self {
+ input,
+ metadata,
+ schema,
+ batch_size: 1024,
+ row_groups: None,
+ projection: ProjectionMask::all(),
+ filter: None,
+ selection: None,
+ })
+ }
+
+ /// Returns a reference to the [`ParquetMetaData`] for this parquet file
+ pub fn metadata(&self) -> &Arc<ParquetMetaData> {
+ &self.metadata
+ }
+
+ /// Returns the parquet [`SchemaDescriptor`] for this parquet file
+ pub fn parquet_schema(&self) -> &SchemaDescriptor {
+ self.metadata.file_metadata().schema_descr()
+ }
+
+ /// Returns the arrow [`SchemaRef`] for this parquet file
+ pub fn schema(&self) -> &SchemaRef {
+ &self.schema
+ }
+
+ /// Set the size of [`RecordBatch`] to produce
+ pub fn with_batch_size(self, batch_size: usize) -> Self {
+ Self { batch_size, ..self }
+ }
+
+ /// Only read data from the provided row group indexes
+ pub fn with_row_groups(self, row_groups: Vec<usize>) -> Self {
+ Self {
+ row_groups: Some(row_groups),
+ ..self
+ }
+ }
+
+ /// Only read data from the provided column indexes
+ pub fn with_projection(self, mask: ProjectionMask) -> Self {
+ Self {
+ projection: mask,
+ ..self
+ }
+ }
+
+ /// Provide a [`RowSelection] to filter out rows, and avoid fetching their
+ /// data into memory
+ ///
+ /// Row group filtering is applied prior to this, and rows from skipped
+ /// row groups should not be included in the [`RowSelection`]
+ ///
+ /// TODO: Make public once stable (#1792)
+ #[allow(unused)]
+ pub(crate) fn with_row_selection(self, selection: RowSelection) -> Self {
+ Self {
+ selection: Some(selection),
+ ..self
+ }
+ }
Review Comment:
Ah, yeah that makes sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] alamb commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
alamb commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r945955110
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -48,9 +50,127 @@ pub(crate) use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
#[allow(unused_imports)]
pub(crate) use selection::{RowSelection, RowSelector};
+/// A generic builder for constructing sync or async arrow parquet readers. This is not intended
+/// to be used directly, instead you should use the specialization for the type of reader
+/// you wish to use
+///
+/// * For a synchronous API - [`ParquetRecordBatchReaderBuilder`]
+/// * For an asynchronous API - [`ParquetRecordBatchStreamBuilder`]
+///
+/// [`ParquetRecordBatchStreamBuilder`]: [crate::arrow::async_reader::ParquetRecordBatchStreamBuilder]
+pub struct ArrowReaderBuilder<T> {
+ pub(crate) input: T,
+
+ pub(crate) metadata: Arc<ParquetMetaData>,
+
+ pub(crate) schema: SchemaRef,
+
+ pub(crate) batch_size: usize,
+
+ pub(crate) row_groups: Option<Vec<usize>>,
+
+ pub(crate) projection: ProjectionMask,
+
+ pub(crate) filter: Option<RowFilter>,
+
+ pub(crate) selection: Option<RowSelection>,
+}
+
+impl<T> ArrowReaderBuilder<T> {
+ pub(crate) fn new_builder(
+ input: T,
+ metadata: Arc<ParquetMetaData>,
+ options: ArrowReaderOptions,
+ ) -> Result<Self> {
+ let kv_metadata = match options.skip_arrow_metadata {
+ true => None,
+ false => metadata.file_metadata().key_value_metadata(),
+ };
+
+ let schema = Arc::new(parquet_to_arrow_schema(
+ metadata.file_metadata().schema_descr(),
+ kv_metadata,
+ )?);
+
+ Ok(Self {
+ input,
+ metadata,
+ schema,
+ batch_size: 1024,
+ row_groups: None,
+ projection: ProjectionMask::all(),
+ filter: None,
+ selection: None,
+ })
+ }
+
+ /// Returns a reference to the [`ParquetMetaData`] for this parquet file
+ pub fn metadata(&self) -> &Arc<ParquetMetaData> {
+ &self.metadata
+ }
+
+ /// Returns the parquet [`SchemaDescriptor`] for this parquet file
+ pub fn parquet_schema(&self) -> &SchemaDescriptor {
+ self.metadata.file_metadata().schema_descr()
+ }
+
+ /// Returns the arrow [`SchemaRef`] for this parquet file
+ pub fn schema(&self) -> &SchemaRef {
+ &self.schema
+ }
+
+ /// Set the size of [`RecordBatch`] to produce
+ pub fn with_batch_size(self, batch_size: usize) -> Self {
+ Self { batch_size, ..self }
+ }
+
+ /// Only read data from the provided row group indexes
+ pub fn with_row_groups(self, row_groups: Vec<usize>) -> Self {
+ Self {
+ row_groups: Some(row_groups),
+ ..self
+ }
+ }
+
+ /// Only read data from the provided column indexes
+ pub fn with_projection(self, mask: ProjectionMask) -> Self {
+ Self {
+ projection: mask,
+ ..self
+ }
+ }
+
+ /// Provide a [`RowSelection] to filter out rows, and avoid fetching their
+ /// data into memory
+ ///
+ /// Row group filtering is applied prior to this, and rows from skipped
+ /// row groups should not be included in the [`RowSelection`]
+ ///
+ /// TODO: Make public once stable (#1792)
+ #[allow(unused)]
+ pub(crate) fn with_row_selection(self, selection: RowSelection) -> Self {
+ Self {
+ selection: Some(selection),
+ ..self
+ }
+ }
Review Comment:
might be good to clarify this in the comments (as others will likely have the same question)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r945730452
##########
parquet/src/arrow/schema.rs:
##########
@@ -1662,14 +1662,9 @@ mod tests {
writer.close()?;
// read file back
- let mut arrow_reader = ParquetFileArrowReader::try_new(file).unwrap();
- let read_schema = arrow_reader.get_schema()?;
- assert_eq!(schema, read_schema);
-
- // read all fields by columns
Review Comment:
There is no separate get_schema_by_columns API anymore, and there is no difference between specifying ProjectionMask::all and not specifying a mask, so this additional bit of the test no longer makes sense. It was added from when get_schema_by_columns used completely different logic from the array reader.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r945751276
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -48,9 +50,127 @@ pub(crate) use filter::{ArrowPredicate, ArrowPredicateFn, RowFilter};
#[allow(unused_imports)]
pub(crate) use selection::{RowSelection, RowSelector};
+/// A generic builder for constructing sync or async arrow parquet readers. This is not intended
+/// to be used directly, instead you should use the specialization for the type of reader
+/// you wish to use
+///
+/// * For a synchronous API - [`ParquetRecordBatchReaderBuilder`]
+/// * For an asynchronous API - [`ParquetRecordBatchStreamBuilder`]
+///
+/// [`ParquetRecordBatchStreamBuilder`]: [crate::arrow::async_reader::ParquetRecordBatchStreamBuilder]
+pub struct ArrowReaderBuilder<T> {
+ pub(crate) input: T,
+
+ pub(crate) metadata: Arc<ParquetMetaData>,
+
+ pub(crate) schema: SchemaRef,
+
+ pub(crate) batch_size: usize,
+
+ pub(crate) row_groups: Option<Vec<usize>>,
+
+ pub(crate) projection: ProjectionMask,
+
+ pub(crate) filter: Option<RowFilter>,
+
+ pub(crate) selection: Option<RowSelection>,
+}
+
+impl<T> ArrowReaderBuilder<T> {
+ pub(crate) fn new_builder(
+ input: T,
+ metadata: Arc<ParquetMetaData>,
+ options: ArrowReaderOptions,
+ ) -> Result<Self> {
+ let kv_metadata = match options.skip_arrow_metadata {
+ true => None,
+ false => metadata.file_metadata().key_value_metadata(),
+ };
+
+ let schema = Arc::new(parquet_to_arrow_schema(
+ metadata.file_metadata().schema_descr(),
+ kv_metadata,
+ )?);
+
+ Ok(Self {
+ input,
+ metadata,
+ schema,
+ batch_size: 1024,
+ row_groups: None,
+ projection: ProjectionMask::all(),
+ filter: None,
+ selection: None,
+ })
+ }
+
+ /// Returns a reference to the [`ParquetMetaData`] for this parquet file
+ pub fn metadata(&self) -> &Arc<ParquetMetaData> {
+ &self.metadata
+ }
+
+ /// Returns the parquet [`SchemaDescriptor`] for this parquet file
+ pub fn parquet_schema(&self) -> &SchemaDescriptor {
+ self.metadata.file_metadata().schema_descr()
+ }
+
+ /// Returns the arrow [`SchemaRef`] for this parquet file
+ pub fn schema(&self) -> &SchemaRef {
+ &self.schema
+ }
+
+ /// Set the size of [`RecordBatch`] to produce
+ pub fn with_batch_size(self, batch_size: usize) -> Self {
+ Self { batch_size, ..self }
+ }
+
+ /// Only read data from the provided row group indexes
+ pub fn with_row_groups(self, row_groups: Vec<usize>) -> Self {
+ Self {
+ row_groups: Some(row_groups),
+ ..self
+ }
+ }
+
+ /// Only read data from the provided column indexes
+ pub fn with_projection(self, mask: ProjectionMask) -> Self {
+ Self {
+ projection: mask,
+ ..self
+ }
+ }
+
+ /// Provide a [`RowSelection] to filter out rows, and avoid fetching their
+ /// data into memory
+ ///
+ /// Row group filtering is applied prior to this, and rows from skipped
+ /// row groups should not be included in the [`RowSelection`]
+ ///
+ /// TODO: Make public once stable (#1792)
+ #[allow(unused)]
+ pub(crate) fn with_row_selection(self, selection: RowSelection) -> Self {
+ Self {
+ selection: Some(selection),
+ ..self
+ }
+ }
Review Comment:
They are related but different, in particular `with_row_selection` exists to allow you to specify a row selection before reading any data, e.g. based on information in the PageIndex
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold merged pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold merged PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-rs] tustvold commented on a diff in pull request #2435: Add ParquetRecordBatchReaderBuilder (#2427)
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #2435:
URL: https://github.com/apache/arrow-rs/pull/2435#discussion_r944579956
##########
parquet/src/arrow/arrow_reader/mod.rs:
##########
@@ -109,27 +233,29 @@ impl ArrowReaderOptions {
}
}
- /// Scan rows from the parquet file according to the provided `selection`
+ /// Set this true to enable decoding of the [PageIndex] if present. This can be used
+ /// to push down predicates to the parquet scan, potentially eliminating unnecessary IO
///
- /// TODO: Revisit this API, as [`Self`] is provided before the file metadata is available
- #[allow(unused)]
- pub(crate) fn with_row_selection(self, selection: impl Into<RowSelection>) -> Self {
Review Comment:
This API is removed, as it was impossible to use
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org