Re: [PR] WIP: Test DataFusion with experimental parquet pushdown [datafusion]
Dandandan commented on code in PR #16208:
URL: https://github.com/apache/datafusion/pull/16208#discussion_r2115799584
##
datafusion/physical-plan/src/filter.rs:
##
@@ -689,6 +702,46 @@ fn filter_and_project(
})
}
+impl FilterExecStream {
+/// Evaluates the predicate filter on the given batch and appends and rows
that match
+/// to the in progress output batch builder.
+fn filter_batch(&mut self, batch: RecordBatch) -> Result<()> {
+self.predicate
+.evaluate(&batch)
+.and_then(|v| v.into_array(batch.num_rows()))
+.and_then(|filter| {
+let Some(filter) = filter.as_boolean_opt() else {
+return internal_err!(
+"Cannot create filter_array from non-boolean
predicates"
+);
+};
+
+let batch = match self.projection.as_ref() {
+Some(projection) => {
+let projected_columns = projection
+.iter()
+.map(|i| Arc::clone(batch.column(*i)))
+.collect();
+// Safety -- the input was a valid RecordBatch and
thus the projection is too
+unsafe {
+RecordBatch::new_unchecked(
+Arc::clone(&self.schema),
+projected_columns,
+batch.num_rows(),
+)
+}
+}
+None => batch,
+};
+let output_batch_builder = self
+.output_batch_builder
+.as_mut()
+.expect("output_batch_builder should be Some");
+Ok(output_batch_builder.append_filtered(batch, filter)?)
Review Comment:
> space each batch will take up front and can just straight up allocate
Hm yeah with primitive and views this might be okay - we might have to test
other data sources though (with normal binary types were it can't be
preallocated).
##
datafusion/physical-plan/src/filter.rs:
##
@@ -689,6 +702,46 @@ fn filter_and_project(
})
}
+impl FilterExecStream {
+/// Evaluates the predicate filter on the given batch and appends and rows
that match
+/// to the in progress output batch builder.
+fn filter_batch(&mut self, batch: RecordBatch) -> Result<()> {
+self.predicate
+.evaluate(&batch)
+.and_then(|v| v.into_array(batch.num_rows()))
+.and_then(|filter| {
+let Some(filter) = filter.as_boolean_opt() else {
+return internal_err!(
+"Cannot create filter_array from non-boolean
predicates"
+);
+};
+
+let batch = match self.projection.as_ref() {
+Some(projection) => {
+let projected_columns = projection
+.iter()
+.map(|i| Arc::clone(batch.column(*i)))
+.collect();
+// Safety -- the input was a valid RecordBatch and
thus the projection is too
+unsafe {
+RecordBatch::new_unchecked(
+Arc::clone(&self.schema),
+projected_columns,
+batch.num_rows(),
+)
+}
+}
+None => batch,
+};
+let output_batch_builder = self
+.output_batch_builder
+.as_mut()
+.expect("output_batch_builder should be Some");
+Ok(output_batch_builder.append_filtered(batch, filter)?)
Review Comment:
> space each batch will take up front and can just straight up allocate
Hm yeah with primitive and views this might be okay - we might have to test
other data sources though (with normal binary types were it can't be
preallocated).
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] WIP: Test DataFusion with experimental parquet pushdown [datafusion]
alamb commented on code in PR #16208:
URL: https://github.com/apache/datafusion/pull/16208#discussion_r2115668031
##
datafusion/physical-plan/src/filter.rs:
##
@@ -689,6 +702,46 @@ fn filter_and_project(
})
}
+impl FilterExecStream {
+/// Evaluates the predicate filter on the given batch and appends and rows
that match
+/// to the in progress output batch builder.
+fn filter_batch(&mut self, batch: RecordBatch) -> Result<()> {
+self.predicate
+.evaluate(&batch)
+.and_then(|v| v.into_array(batch.num_rows()))
+.and_then(|filter| {
+let Some(filter) = filter.as_boolean_opt() else {
+return internal_err!(
+"Cannot create filter_array from non-boolean
predicates"
+);
+};
+
+let batch = match self.projection.as_ref() {
+Some(projection) => {
+let projected_columns = projection
+.iter()
+.map(|i| Arc::clone(batch.column(*i)))
+.collect();
+// Safety -- the input was a valid RecordBatch and
thus the projection is too
+unsafe {
+RecordBatch::new_unchecked(
+Arc::clone(&self.schema),
+projected_columns,
+batch.num_rows(),
+)
+}
+}
+None => batch,
+};
+let output_batch_builder = self
+.output_batch_builder
+.as_mut()
+.expect("output_batch_builder should be Some");
+Ok(output_batch_builder.append_filtered(batch, filter)?)
Review Comment:
> This avoids reallocations / copying as the target capacity can be
calculated.
In order to avoid buffering too much batches probably have to limit this /
create a batch anyway after x batches or having x megabytes in memory.
I was trying to avoid having any reallocations in the
[`IncrementalRecordBatchBuilder`](https://github.com/apache/arrow-rs/pull/7513/files#diff-52e0762696101af1d088555b3d8bbe861f53dc4e9e6a6edd37eee22f950ec742R48)
-- since we know the target output batch size (`batch_size`) it knows how much
space each batch will take up front and can just straight up allocate it (
`instantiate_builder` function creates the builders with `with_capacity`)
However, now that I think about it, after a call to `finish()` the updated
builder doesn't have the right allocation 🤔
I'll look into that more later today
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] WIP: Test DataFusion with experimental parquet pushdown [datafusion]
Dandandan commented on code in PR #16208:
URL: https://github.com/apache/datafusion/pull/16208#discussion_r2115388263
##
datafusion/physical-plan/src/filter.rs:
##
@@ -689,6 +702,46 @@ fn filter_and_project(
})
}
+impl FilterExecStream {
+/// Evaluates the predicate filter on the given batch and appends and rows
that match
+/// to the in progress output batch builder.
+fn filter_batch(&mut self, batch: RecordBatch) -> Result<()> {
+self.predicate
+.evaluate(&batch)
+.and_then(|v| v.into_array(batch.num_rows()))
+.and_then(|filter| {
+let Some(filter) = filter.as_boolean_opt() else {
+return internal_err!(
+"Cannot create filter_array from non-boolean
predicates"
+);
+};
+
+let batch = match self.projection.as_ref() {
+Some(projection) => {
+let projected_columns = projection
+.iter()
+.map(|i| Arc::clone(batch.column(*i)))
+.collect();
+// Safety -- the input was a valid RecordBatch and
thus the projection is too
+unsafe {
+RecordBatch::new_unchecked(
+Arc::clone(&self.schema),
+projected_columns,
+batch.num_rows(),
+)
+}
+}
+None => batch,
+};
+let output_batch_builder = self
+.output_batch_builder
+.as_mut()
+.expect("output_batch_builder should be Some");
+Ok(output_batch_builder.append_filtered(batch, filter)?)
Review Comment:
I expect it should be faster rather than incrementally build a new batch
based on a number of arrays, to first evaluate a number of filters
(`self.predicate.evaluate(&batch)` until the number of true values reaches the
target batch size and then have a filter api to filter a list of batches. This
avoids reallocations / copying as the target capacity can be calculated
.
In order to avoid buffering too much batches probably have to limit this /
create a batch anyway after x batches or having x megabytes in memory.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] WIP: Test DataFusion with experimental parquet pushdown [datafusion]
Dandandan commented on code in PR #16208:
URL: https://github.com/apache/datafusion/pull/16208#discussion_r2115388263
##
datafusion/physical-plan/src/filter.rs:
##
@@ -689,6 +702,46 @@ fn filter_and_project(
})
}
+impl FilterExecStream {
+/// Evaluates the predicate filter on the given batch and appends and rows
that match
+/// to the in progress output batch builder.
+fn filter_batch(&mut self, batch: RecordBatch) -> Result<()> {
+self.predicate
+.evaluate(&batch)
+.and_then(|v| v.into_array(batch.num_rows()))
+.and_then(|filter| {
+let Some(filter) = filter.as_boolean_opt() else {
+return internal_err!(
+"Cannot create filter_array from non-boolean
predicates"
+);
+};
+
+let batch = match self.projection.as_ref() {
+Some(projection) => {
+let projected_columns = projection
+.iter()
+.map(|i| Arc::clone(batch.column(*i)))
+.collect();
+// Safety -- the input was a valid RecordBatch and
thus the projection is too
+unsafe {
+RecordBatch::new_unchecked(
+Arc::clone(&self.schema),
+projected_columns,
+batch.num_rows(),
+)
+}
+}
+None => batch,
+};
+let output_batch_builder = self
+.output_batch_builder
+.as_mut()
+.expect("output_batch_builder should be Some");
+Ok(output_batch_builder.append_filtered(batch, filter)?)
Review Comment:
I expect it should be faster rather than incrementally build a new batch
based on a number of arrays, to first evaluate a number of filters
(`self.predicate.evaluate(&batch)` until the number of true values reaches the
target batch size and then have a filter api to filter a list of batches. This
avoids reallocations / copying as the target capacity can be calculated.
In order to avoid buffering too much batches probably have to limit this /
create a batch anyway after x batches or having x megabytes in memory.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] WIP: Test DataFusion with experimental parquet pushdown [datafusion]
Dandandan commented on code in PR #16208:
URL: https://github.com/apache/datafusion/pull/16208#discussion_r2115400750
##
datafusion/physical-plan/src/filter.rs:
##
@@ -689,6 +702,46 @@ fn filter_and_project(
})
}
+impl FilterExecStream {
+/// Evaluates the predicate filter on the given batch and appends and rows
that match
+/// to the in progress output batch builder.
+fn filter_batch(&mut self, batch: RecordBatch) -> Result<()> {
+self.predicate
+.evaluate(&batch)
+.and_then(|v| v.into_array(batch.num_rows()))
+.and_then(|filter| {
+let Some(filter) = filter.as_boolean_opt() else {
+return internal_err!(
+"Cannot create filter_array from non-boolean
predicates"
+);
+};
+
+let batch = match self.projection.as_ref() {
+Some(projection) => {
+let projected_columns = projection
+.iter()
+.map(|i| Arc::clone(batch.column(*i)))
+.collect();
+// Safety -- the input was a valid RecordBatch and
thus the projection is too
+unsafe {
+RecordBatch::new_unchecked(
+Arc::clone(&self.schema),
+projected_columns,
+batch.num_rows(),
+)
+}
+}
+None => batch,
+};
+let output_batch_builder = self
+.output_batch_builder
+.as_mut()
+.expect("output_batch_builder should be Some");
+Ok(output_batch_builder.append_filtered(batch, filter)?)
Review Comment:
My feeling is this should to do this "multi batch filter" and then `concat`
anyway if smaller batches are generated by this approach, rather than using a
builder approach.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
-
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
Re: [PR] WIP: Test DataFusion with experimental parquet pushdown [datafusion]
zhuqi-lucas commented on PR #16208: URL: https://github.com/apache/datafusion/pull/16208#issuecomment-2921147300 Interesting, i am confused why it's different from arrow-rs clickbench result. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] WIP: Test DataFusion with experimental parquet pushdown [datafusion]
zhuqi-lucas commented on PR #16208: URL: https://github.com/apache/datafusion/pull/16208#issuecomment-2921150637 Ok, i misunderstood this testing is not for parquet filter pushdown enabled... -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] WIP: Test DataFusion with experimental parquet pushdown [datafusion]
alamb commented on PR #16208: URL: https://github.com/apache/datafusion/pull/16208#issuecomment-2920064834 > 🤖: Benchmark completed Hmm that is somewhat depressing. I will investigate tomorrow -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
Re: [PR] WIP: Test DataFusion with experimental parquet pushdown [datafusion]
alamb commented on PR #16208: URL: https://github.com/apache/datafusion/pull/16208#issuecomment-2920053295 🤖: Benchmark completed Details ``` Comparing HEAD and alamb_test_filter_pushdown Benchmark clickbench_extended.json ┏━━┳┳┳━━┓ ┃ Query┃ HEAD ┃ alamb_test_filter_pushdown ┃ Change ┃ ┡━━╇╇╇━━┩ │ QQuery 0 │ 1822.08ms │ 1950.22ms │ 1.07x slower │ │ QQuery 1 │ 686.15ms │ 725.22ms │ 1.06x slower │ │ QQuery 2 │ 1421.43ms │ 1437.38ms │no change │ │ QQuery 3 │ 684.69ms │ 705.23ms │no change │ │ QQuery 4 │ 1456.23ms │ 1479.06ms │no change │ │ QQuery 5 │ 15371.83ms │ 15471.55ms │no change │ │ QQuery 6 │ 2033.56ms │ 2057.13ms │no change │ │ QQuery 7 │ 2091.87ms │ 2101.95ms │no change │ │ QQuery 8 │ 849.28ms │ 854.59ms │no change │ └──┴┴┴──┘ ┏━━━┳┓ ┃ Benchmark Summary ┃┃ ┡━━━╇┩ │ Total Time (HEAD) │ 26417.11ms │ │ Total Time (alamb_test_filter_pushdown) │ 26782.33ms │ │ Average Time (HEAD) │ 2935.23ms │ │ Average Time (alamb_test_filter_pushdown) │ 2975.81ms │ │ Queries Faster│ 0 │ │ Queries Slower│ 2 │ │ Queries with No Change│ 7 │ └───┴┘ Benchmark clickbench_partitioned.json ┏━━┳┳┳━━━┓ ┃ Query┃ HEAD ┃ alamb_test_filter_pushdown ┃Change ┃ ┡━━╇╇╇━━━┩ │ QQuery 0 │15.73ms │14.77ms │ +1.06x faster │ │ QQuery 1 │31.94ms │32.66ms │ no change │ │ QQuery 2 │80.09ms │79.35ms │ no change │ │ QQuery 3 │98.77ms │95.13ms │ no change │ │ QQuery 4 │ 605.02ms │ 602.99ms │ no change │ │ QQuery 5 │ 868.32ms │ 856.53ms │ no change │ │ QQuery 6 │23.87ms │23.91ms │ no change │ │ QQuery 7 │37.19ms │38.29ms │ no change │ │ QQuery 8 │ 933.70ms │ 914.39ms │ no change │ │ QQuery 9 │ 1279.67ms │ 1209.80ms │ +1.06x faster │ │ QQuery 10│ 272.21ms │ 273.11ms │ no change │ │ QQuery 11│ 308.41ms │ 298.73ms │ no change │ │ QQuery 12│ 930.67ms │ 910.59ms │ no change │ │ QQuery 13│ 1348.50ms │ 1369.46ms │ no change │ │ QQuery 14│ 860.32ms │ 844.04ms │ no change │ │ QQuery 15│ 843.03ms │ 821.28ms │ no change │ │ QQuery 16│ 1746.35ms │ 1729.29ms │ no change │ │ QQuery 17│ 1634.90ms │ 1603.86ms │ no change │ │ QQuery 18│ 3126.70ms │ 3105.76ms │ no change │ │ QQuery 19│84.55ms │81.44ms │ no change │ │ QQuery 20│ 1152.02ms │ 1247.44ms │ 1.08x slower │ │ QQuery 21│ 1340.97ms │ 1376.02ms │ no change │ │ QQuery 22│ 2221.77ms │ 2373.59ms │ 1.07x slower │ │ QQuery 23│ 8067.48ms │ 8453.86ms │ no change │ │ QQuery 24│ 472.64ms │ 465.06ms │ no change │ │ QQuery 25│ 396.53ms │ 398.05ms │ no change │ │ QQuery 26│ 535.55ms │ 536.82ms │ no change │ │ QQuery 27│ 1576.77ms │ 1702.88ms │ 1.08x slower │ │ QQuery 28│ 12485.75ms │ 13253.81ms │ 1.06x slower │ │ QQuery 29│ 530.72ms │ 528.47ms │ no change │ │ QQuery 30│ 802.59ms │ 834.59ms │ no change │ │ QQuery 31│ 841.57ms │ 875.02ms │ no change │ │ QQuery 32│ 2642.67ms │ 2824.64ms │ 1.07x slower │ │ QQuery 33│ 3355.69ms │ 3530.76ms │ 1.05x slower │ │ QQuery 34│ 3376.55ms │
Re: [PR] WIP: Test DataFusion with experimental parquet pushdown [datafusion]
alamb commented on PR #16208: URL: https://github.com/apache/datafusion/pull/16208#issuecomment-2919955178 🤖 `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running Linux aal-dev 6.11.0-1013-gcp #13~24.04.1-Ubuntu SMP Wed Apr 2 16:34:16 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux Comparing alamb/test_filter_pushdown (f9dbcc077dbe1f1301f1f03142fe14075146a4f5) to 7002a0027876a17e5bdf275e63d2a25373331943 [diff](https://github.com/apache/datafusion/compare/7002a0027876a17e5bdf275e63d2a25373331943..f9dbcc077dbe1f1301f1f03142fe14075146a4f5) Benchmarks: tpch_mem clickbench_partitioned clickbench_extended Results will be posted here when complete -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] - To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
