You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "alamb (via GitHub)" <gi...@apache.org> on 2023/02/12 11:50:10 UTC
[GitHub] [arrow-datafusion] alamb opened a new pull request, #5260: Minor: refactor ParquetExec roundtrip tests
alamb opened a new pull request, #5260:
URL: https://github.com/apache/arrow-datafusion/pull/5260
# Which issue does this PR close?
Part of https://github.com/apache/arrow-datafusion/issues/5104
# Rationale for this change
I am trying to write a test for https://github.com/apache/arrow-datafusion/issues/5104 and getting lost in the maze of parameters passed to `round_trip` and `round_trip_to_parquet`
It is also hard for me to evaluate which combinations of tests are covered
# What changes are included in this PR?
Move the parameters to run the roundtrip test into a structure, with better documentation
# Are these changes tested?
Yes
# Are there any user-facing changes?
no
--
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-datafusion] alamb commented on pull request #5260: Minor: refactor ParquetExec roundtrip tests
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #5260:
URL: https://github.com/apache/arrow-datafusion/pull/5260#issuecomment-1427797574
Oops I hit merge with a fmt issue. Will fix
--
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-datafusion] ursabot commented on pull request #5260: Minor: refactor ParquetExec roundtrip tests
Posted by "ursabot (via GitHub)" <gi...@apache.org>.
ursabot commented on PR #5260:
URL: https://github.com/apache/arrow-datafusion/pull/5260#issuecomment-1427799875
Benchmark runs are scheduled for baseline = 00e60f1678c0896c5007f08c7516e3e2a9aa8e2b and contender = 9710d7cdafb81e3c07e2cb7e7e242d55fc7fd88b. 9710d7cdafb81e3c07e2cb7e7e242d55fc7fd88b 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-datafusion-commits is not supported on ec2-t3-xlarge-us-east-2] [ec2-t3-xlarge-us-east-2](https://conbench.ursa.dev/compare/runs/e21febbb8ef642e688548789564b94ce...cccd7534f1d44989addde6fa2e36cd44/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on test-mac-arm] [test-mac-arm](https://conbench.ursa.dev/compare/runs/fa1fc7468a194665913559b4e1430d3b...92671a77c0274e5cab984efa27cf9384/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-i9-9960x] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/9f07baac7c654da4b8b7785b70e9b639...fbffb47f0ce649c49d9419a29f8313e1/)
[Skipped :warning: Benchmarking of arrow-datafusion-commits is not supported on ursa-thinkcentre-m75q] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/36ba6525faa04e038e92c7bf3b4e5a9c...1cb645cc16244405abc7a29dbc83b387/)
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-datafusion] viirya commented on a diff in pull request #5260: Minor: refactor ParquetExec roundtrip tests
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #5260:
URL: https://github.com/apache/arrow-datafusion/pull/5260#discussion_r1104068340
##########
datafusion/core/src/datasource/file_format/parquet.rs:
##########
@@ -550,6 +550,11 @@ pub(crate) mod test_util {
use parquet::file::properties::WriterProperties;
use tempfile::NamedTempFile;
+ /// Writes `batches` to a temporary parquet file
+ ///
+ /// If multi_page is set to `true`, the parquet file is written
+ /// with 2 rows per data page (used to test page filtering and
+ /// boundaries)
Review Comment:
```suggestion
/// If multi_page is set to `true`, all batches are written into
/// one temporary parquet file and the parquet file is written
/// with 2 rows per data page (used to test page filtering and
/// boundaries).
```
--
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-datafusion] alamb commented on pull request #5260: Minor: refactor ParquetExec roundtrip tests
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #5260:
URL: https://github.com/apache/arrow-datafusion/pull/5260#issuecomment-1427800206
fmt fix in https://github.com/apache/arrow-datafusion/pull/5263
--
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-datafusion] alamb merged pull request #5260: Minor: refactor ParquetExec roundtrip tests
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb merged PR #5260:
URL: https://github.com/apache/arrow-datafusion/pull/5260
--
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-datafusion] viirya commented on a diff in pull request #5260: Minor: refactor ParquetExec roundtrip tests
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #5260:
URL: https://github.com/apache/arrow-datafusion/pull/5260#discussion_r1104069998
##########
datafusion/core/src/physical_plan/file_format/parquet.rs:
##########
@@ -834,82 +834,116 @@ mod tests {
parquet_exec: Arc<ParquetExec>,
}
- /// writes each RecordBatch as an individual parquet file and re-reads
- /// the data back. Returns the data as [RecordBatch]es
- async fn round_trip_to_parquet(
- batches: Vec<RecordBatch>,
+ /// round-trip record batches by writing each individual RecordBatch to
+ /// a parquet file and then reading that parquet file with the specified
+ /// options
Review Comment:
```suggestion
/// round-trip record batches by writing each individual RecordBatch to
/// a parquet file and then reading that parquet file with the specified
/// options. If page_index_predicate is set to `true`, all RecordBatches
/// are written into a parquet file instead.
```
--
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-datafusion] alamb commented on a diff in pull request #5260: Minor: refactor ParquetExec roundtrip tests
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #5260:
URL: https://github.com/apache/arrow-datafusion/pull/5260#discussion_r1103787835
##########
datafusion/core/src/physical_plan/file_format/parquet.rs:
##########
@@ -1121,8 +1158,12 @@ mod tests {
let filter = col("c2").eq(lit(2_i64)).or(col("c2").eq(lit(1_i64)));
// read/write them files:
- let rt =
- round_trip(vec![batch1, batch2], None, None, Some(filter), true, false).await;
+ let rt = RoundTrip::new()
Review Comment:
I think this construction is clearer about what combination of options are being tested
##########
datafusion/core/src/physical_plan/file_format/parquet.rs:
##########
@@ -834,82 +834,116 @@ mod tests {
parquet_exec: Arc<ParquetExec>,
}
- /// writes each RecordBatch as an individual parquet file and re-reads
- /// the data back. Returns the data as [RecordBatch]es
- async fn round_trip_to_parquet(
- batches: Vec<RecordBatch>,
+ /// round-trip record batches by writing each individual RecordBatch to
+ /// a parquet file and then reading that parquet file with the specified
+ /// options
+ #[derive(Debug, Default)]
+ struct RoundTrip {
projection: Option<Vec<usize>>,
schema: Option<SchemaRef>,
predicate: Option<Expr>,
pushdown_predicate: bool,
- ) -> Result<Vec<RecordBatch>> {
- round_trip(
- batches,
- projection,
- schema,
- predicate,
- pushdown_predicate,
- false,
- )
- .await
- .batches
+ page_index_predicate: bool,
}
- /// Writes each RecordBatch as an individual parquet file and then
- /// reads them back. Returns the parquet exec as well as the data
- /// as [RecordBatch]es
- async fn round_trip(
- batches: Vec<RecordBatch>,
- projection: Option<Vec<usize>>,
- schema: Option<SchemaRef>,
- predicate: Option<Expr>,
- pushdown_predicate: bool,
- page_index_predicate: bool,
- ) -> RoundTripResult {
- let file_schema = match schema {
- Some(schema) => schema,
- None => Arc::new(
- Schema::try_merge(batches.iter().map(|b| b.schema().as_ref().clone()))
- .unwrap(),
- ),
- };
+ impl RoundTrip {
+ fn new() -> Self {
+ Default::default()
+ }
- let (meta, _files) = store_parquet(batches, page_index_predicate).await.unwrap();
- let file_groups = meta.into_iter().map(Into::into).collect();
+ fn with_projection(mut self, projection: Vec<usize>) -> Self {
+ self.projection = Some(projection);
+ self
+ }
- // prepare the scan
- let mut parquet_exec = ParquetExec::new(
- FileScanConfig {
- object_store_url: ObjectStoreUrl::local_filesystem(),
- file_groups: vec![file_groups],
- file_schema,
- statistics: Statistics::default(),
- projection,
- limit: None,
- table_partition_cols: vec![],
- output_ordering: None,
- infinite_source: false,
- },
- predicate,
- None,
- );
+ fn with_schema(mut self, schema: SchemaRef) -> Self {
+ self.schema = Some(schema);
+ self
+ }
- if pushdown_predicate {
- parquet_exec = parquet_exec
- .with_pushdown_filters(true)
- .with_reorder_filters(true);
+ fn with_predicate(mut self, predicate: Expr) -> Self {
+ self.predicate = Some(predicate);
+ self
}
- if page_index_predicate {
- parquet_exec = parquet_exec.with_enable_page_index(true);
+ fn with_pushdown_predicate(mut self) -> Self {
+ self.pushdown_predicate = true;
+ self
}
- let session_ctx = SessionContext::new();
- let task_ctx = session_ctx.task_ctx();
- let parquet_exec = Arc::new(parquet_exec);
- RoundTripResult {
- batches: collect(parquet_exec.clone(), task_ctx).await,
- parquet_exec,
+ fn with_page_index_predicate(mut self) -> Self {
+ self.page_index_predicate = true;
+ self
+ }
+
+ /// run the test, returning only the resulting RecordBatches
+ async fn round_trip_to_batches(
+ self,
+ batches: Vec<RecordBatch>,
+ ) -> Result<Vec<RecordBatch>> {
+ self.round_trip(batches).await.batches
+ }
+
+ /// run the test, returning the `RoundTripResult`
+ async fn round_trip(self, batches: Vec<RecordBatch>) -> RoundTripResult {
Review Comment:
I think whitespace blind diff shows the changes more clearly: https://github.com/apache/arrow-datafusion/pull/5260/files?w=1
##########
datafusion/core/src/physical_plan/file_format/parquet.rs:
##########
@@ -834,82 +834,116 @@ mod tests {
parquet_exec: Arc<ParquetExec>,
}
- /// writes each RecordBatch as an individual parquet file and re-reads
- /// the data back. Returns the data as [RecordBatch]es
- async fn round_trip_to_parquet(
- batches: Vec<RecordBatch>,
+ /// round-trip record batches by writing each individual RecordBatch to
+ /// a parquet file and then reading that parquet file with the specified
+ /// options
+ #[derive(Debug, Default)]
+ struct RoundTrip {
projection: Option<Vec<usize>>,
schema: Option<SchemaRef>,
predicate: Option<Expr>,
pushdown_predicate: bool,
- ) -> Result<Vec<RecordBatch>> {
- round_trip(
- batches,
- projection,
- schema,
- predicate,
- pushdown_predicate,
- false,
- )
- .await
- .batches
+ page_index_predicate: bool,
}
- /// Writes each RecordBatch as an individual parquet file and then
- /// reads them back. Returns the parquet exec as well as the data
- /// as [RecordBatch]es
- async fn round_trip(
- batches: Vec<RecordBatch>,
- projection: Option<Vec<usize>>,
- schema: Option<SchemaRef>,
- predicate: Option<Expr>,
- pushdown_predicate: bool,
- page_index_predicate: bool,
- ) -> RoundTripResult {
- let file_schema = match schema {
- Some(schema) => schema,
- None => Arc::new(
- Schema::try_merge(batches.iter().map(|b| b.schema().as_ref().clone()))
- .unwrap(),
- ),
- };
+ impl RoundTrip {
+ fn new() -> Self {
+ Default::default()
+ }
- let (meta, _files) = store_parquet(batches, page_index_predicate).await.unwrap();
- let file_groups = meta.into_iter().map(Into::into).collect();
+ fn with_projection(mut self, projection: Vec<usize>) -> Self {
+ self.projection = Some(projection);
+ self
+ }
- // prepare the scan
- let mut parquet_exec = ParquetExec::new(
- FileScanConfig {
- object_store_url: ObjectStoreUrl::local_filesystem(),
- file_groups: vec![file_groups],
- file_schema,
- statistics: Statistics::default(),
- projection,
- limit: None,
- table_partition_cols: vec![],
- output_ordering: None,
- infinite_source: false,
- },
- predicate,
- None,
- );
+ fn with_schema(mut self, schema: SchemaRef) -> Self {
+ self.schema = Some(schema);
+ self
+ }
- if pushdown_predicate {
- parquet_exec = parquet_exec
- .with_pushdown_filters(true)
- .with_reorder_filters(true);
+ fn with_predicate(mut self, predicate: Expr) -> Self {
+ self.predicate = Some(predicate);
+ self
}
- if page_index_predicate {
- parquet_exec = parquet_exec.with_enable_page_index(true);
+ fn with_pushdown_predicate(mut self) -> Self {
+ self.pushdown_predicate = true;
+ self
}
- let session_ctx = SessionContext::new();
- let task_ctx = session_ctx.task_ctx();
- let parquet_exec = Arc::new(parquet_exec);
- RoundTripResult {
- batches: collect(parquet_exec.clone(), task_ctx).await,
- parquet_exec,
+ fn with_page_index_predicate(mut self) -> Self {
+ self.page_index_predicate = true;
+ self
+ }
+
+ /// run the test, returning only the resulting RecordBatches
+ async fn round_trip_to_batches(
+ self,
+ batches: Vec<RecordBatch>,
+ ) -> Result<Vec<RecordBatch>> {
+ self.round_trip(batches).await.batches
+ }
+
+ /// run the test, returning the `RoundTripResult`
+ async fn round_trip(self, batches: Vec<RecordBatch>) -> RoundTripResult {
+ let Self {
+ projection,
+ schema,
+ predicate,
+ pushdown_predicate,
+ page_index_predicate,
+ } = self;
+
+ let file_schema = match schema {
+ Some(schema) => schema,
+ None => Arc::new(
+ Schema::try_merge(
+ batches.iter().map(|b| b.schema().as_ref().clone()),
+ )
+ .unwrap(),
+ ),
+ };
+
+ // If testing with page_index_predicate, write parquet
+ // files with multiple pages
+ let multi_page = page_index_predicate;
Review Comment:
I found this logic slightly confusing so I added some comments and tried to make it clearer -- no functional change is intended
--
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