feat: Convert predicate to arrow filter and push down to parquet reader #295

viirya · 2024-03-24T04:35:10Z

This implements the feature of row filtering when reading Parquet files in Iceberg scan. It is achieved by converting predicates into Parquet Arrow filter which is used to filter rows during reading in Parquet reader.

This implements AlwaysTrue, AlwaysFalse, And, Or, Not, Binary, partial Unary predicates. Unimplemented predicates (some Unary and Set predicates) are because no existing kernels to be used in arrow. I'll implement them in following works.

close #265

crates/iceberg/src/arrow.rs

liurenjie1024 · 2024-04-01T07:27:36Z

cc @viirya Is this ready for review or you still need to do more update?

viirya · 2024-04-01T17:10:34Z

@liurenjie1024 It is ready for review. I will fix the conflicts.

liurenjie1024

Hi, @viirya Thanks for this pr, the general idea of reusing arrow kernel looks great! But I found some small problems which could be improved.

crates/iceberg/src/arrow.rs

liurenjie1024 · 2024-04-02T10:35:00Z

crates/iceberg/src/arrow.rs

+            PredicateOperator::NotNull => Ok(Box::new(ArrowPredicateFn::new(
+                self.projection_mask.clone(),
+                move |batch| {
+                    let column = batch.column(term_index);


This maybe incorrect for nested column, I think maybe we should either return projection_mask for each leave column, or implement a general purpose flatten method for struct array.

The Parquet reader API will flatten required columns based on the projection_mask we provide. I.e., If the projection mask selects one nested column a.b, it will be the first column of the record batch when calling evaluate of ArrowPredicate API.

I did some test but it seems that it doesn't work like this, say the schem is like following:

message { struct a { int b } }

And we pass ProjectionMask::leaves([0]), it will return struct array, so batch.column[term_index] will return StructArray

Hmm, that's what I got from reading the Parquet code. Let me take further look and test.

Just wrote a test to test it in arrow-rs. Yea, for a nested column such as struct.a, the batch passed to evaluate contains a struct array with the column a. Looks like Parquet will project the requested column indices and construct the upper nested type (i.e., struct) before passing to evaluate,

Added the test to arrow-rs to clarify its usage: apache/arrow-rs#5600

~~I think to have a flatten method for struct array sounds more simple way. I'm looking in arrow-rs to see if there is existing one, if not, we need to implement it here.~~

This maybe incorrect for nested column, I think maybe we should either return projection_mask for each leave column, or implement a general purpose flatten method for struct array.

I tried to change to return projection_mask for each leave column, it is pretty straightforward to implement. Please let me know if it looks good to you. Thanks.

viirya · 2024-04-03T21:33:39Z

I've addressed some of above reviews. I will resolve other reviews soon. Thanks.

crates/iceberg/src/arrow/reader.rs

viirya · 2024-04-12T18:07:15Z

@liurenjie1024 I've addressed all comments. Thank you.

liurenjie1024

Thanks @viirya for this pr, it looks promising! Given #334 has been merged, do you mind to rewrite some of the visitors with new api? This helps to make code easier to maintain.

crates/iceberg/src/arrow/reader.rs

viirya · 2024-04-28T01:59:50Z

@liurenjie1024 Thanks for review. Sorry for late. I addressed the comments by rewriting the visitors using the new API. I replied with another questions.

liurenjie1024

Thanks @viirya for this great pr! It's really hard work, and you did it in a quite elegant way. I found some small problems to fix.

crates/iceberg/src/arrow/reader.rs

liurenjie1024 · 2024-04-28T14:12:20Z

crates/iceberg/src/arrow/reader.rs

+    fn always_true(&mut self) -> Result<Self::T> {
+        Ok(Box::new(ArrowPredicateFn::new(
+            self.projection_mask.clone(),
+            |batch| Ok(BooleanArray::from(vec![true; batch.num_rows()])),


This maybe not a blocker, but is it possible to build a const array in arrow?

As I know, there is no const array in arrow for now. It has dictionary array which I think is close to const array, but the ArrowPredicateFn API defines returned type to be BooleanArray.

I've created an issue to track this: apache/arrow-rs#5701

crates/iceberg/src/arrow/reader.rs

liurenjie1024 · 2024-04-28T14:27:59Z

crates/iceberg/src/arrow/reader.rs

+        reference: &BoundReference,
+        _predicate: &BoundPredicate,
+    ) -> Result<Self::T> {
+        let projected_mask = self.bound_reference(reference)?;


This seems incorrect to me. Let's say the predicate is a is null AND b >1, then the batch passed to this ArrowPredicateFn is constructed by projection mask of [a, b]. I think one possible solution is to use same project mask for all predicates, and pass the column_idx to get_leaf_column.

Hmm, as RowFilter API design accepts multiple ArrowPredicates. Each ArrowPredicate has its projection and the API doc said the API will be passed batches that contains the columns specified in projection. I think it should be projected for each ArrowPredicate and projection.

I will go to test it in arrow-rs to verify it.

Ah, no, although I used multiple ArrowPredicates before, it was changed to one ArrowPredicates after you suggested. So now we generate only one ArrowPredicates.

Hmm, let me think how do deal with it.

As we use one ArrowPredicates to represent the entire predicate, we should use one projection mask which contains all leaf columns in the predicate.

But it brings another question, how do we access the correct array from the RecordBatch. For top-level column, it should be straightforward, but for nested column, I don't find a way to get it quickly.

I created one issue at arrow-rs: apache/arrow-rs#5699

Based on what I searched and the kindly reply on the issue, I think there is no way to do nested projection on RecordBatch currently.

To implement the feature in arrow-rs might block this. I tend to finish top-level column only in this PR.

WDYT, @liurenjie1024 ?

I agree that we can support top-level column only first to move on.

crates/iceberg/src/arrow/reader.rs

liurenjie1024 · 2024-04-28T14:42:05Z

crates/iceberg/src/scan.rs

+    }
+
+    #[tokio::test]
+    async fn test_filter_on_arrow_is_not_null() {


Thanks for the tests, is it possible to add serveral test cases for more complex types such as AND, OR?

Yes. I added some more tests using AND and OR.

liurenjie1024 · 2024-04-29T12:26:10Z

crates/iceberg/src/arrow/reader.rs

+                },
+            )))
+        } else {
+            self.build_always_true()


When a column is missing, I think we should treat it as null, so this should be false?

Yes, fixed it.

liurenjie1024 · 2024-04-29T12:29:08Z

crates/iceberg/src/arrow/reader.rs

+                Ok(BooleanArray::from(vec![true; batch.num_rows()]))
+            })))
+        } else {
+            self.build_always_true()


Iceberg spec has no definition for NULL is_nan, but java defines it as false: https://github.com/apache/iceberg/blob/c07f2aabc0a1d02f068ecf1514d2479c0fbdd3b0/api/src/main/java/org/apache/iceberg/util/NaNUtil.java#L25

crates/iceberg/src/arrow/reader.rs

liurenjie1024 · 2024-04-29T12:34:03Z

crates/iceberg/src/arrow/reader.rs

+                },
+            )))
+        } else {
+            self.build_always_true()


Java impl comparator uses null first: https://github.com/apache/iceberg/blob/c07f2aabc0a1d02f068ecf1514d2479c0fbdd3b0/api/src/main/java/org/apache/iceberg/expressions/Literals.java#L616

I think we should return false here?

crates/iceberg/src/arrow/reader.rs

viirya · 2024-05-10T19:28:46Z

@liurenjie1024 I've addressed your comments. Please take a look when you can. Thanks.

liurenjie1024

Thanks @viirya for this effort!

liurenjie1024 · 2024-05-14T14:49:39Z

Oh, sorry, seems we need to resolve conflicts. Others LGTM, thanks!

viirya · 2024-05-14T19:16:34Z

Thanks @liurenjie1024. I just resolved the conflicts.

liurenjie1024 · 2024-05-15T01:46:39Z

Thanks @viirya for this great effort!

viirya · 2024-05-15T02:24:42Z

Thanks @liurenjie1024 for your review!

viirya force-pushed the filter_to_row_filter branch from c9e53f0 to 22af140 Compare March 24, 2024 05:45

viirya marked this pull request as draft March 24, 2024 17:26

viirya force-pushed the filter_to_row_filter branch 3 times, most recently from 5606bbe to 67c3f05 Compare March 25, 2024 00:43

viirya marked this pull request as ready for review March 25, 2024 00:43

feat: Convert predicate to arrow filter and push down to parquet reader

9262774

viirya force-pushed the filter_to_row_filter branch from 67c3f05 to 9262774 Compare March 25, 2024 00:50

viirya commented Mar 26, 2024

View reviewed changes

crates/iceberg/src/arrow.rs Outdated Show resolved Hide resolved

viirya commented Mar 26, 2024

View reviewed changes

crates/iceberg/src/arrow.rs Outdated Show resolved Hide resolved

liurenjie1024 reviewed Apr 2, 2024

View reviewed changes

Merge remote-tracking branch 'upstream/main' into HEAD

61221f3

viirya mentioned this pull request Apr 3, 2024

Add Send + Sync traits for Datum apache/arrow-rs#5587

Closed

For review

a537092

viirya added 4 commits April 3, 2024 16:07

Fix clippy

d612d71

Change from vector of BoundPredicate to BoundPredicate

86bde19

Merge remote-tracking branch 'upstream/main' into filter_to_row_filter

67dac6f

Add test for CollectFieldIdVisitor

f68a556

viirya mentioned this pull request Apr 7, 2024

test: Add a test for RowFilter with nested type apache/arrow-rs#5600

Merged

Return projection_mask for leaf column

733e0ca

viirya commented Apr 8, 2024

View reviewed changes

crates/iceberg/src/arrow/reader.rs Show resolved Hide resolved

liurenjie1024 reviewed Apr 16, 2024

View reviewed changes

Fokko mentioned this pull request Apr 24, 2024

Tracking issues of iceberg-rust v0.3.0 #348

Open

72 tasks

viirya added 2 commits April 27, 2024 17:56

Merge remote-tracking branch 'upstream/main' into filter_to_row_filter

bca1d57

Update

732d43f

For review

89e3aa6

liurenjie1024 reviewed Apr 28, 2024

View reviewed changes

viirya added 2 commits April 28, 2024 12:02

For review

e06a5b9

For review

78f35e6

viirya mentioned this pull request Apr 29, 2024

Retrieve array from RecordBatch for a leaf column apache/arrow-rs#5699

Open

liurenjie1024 mentioned this pull request Apr 29, 2024

Support constant array. apache/arrow-rs#5701

Open

liurenjie1024 reviewed Apr 29, 2024

View reviewed changes

viirya added 6 commits May 5, 2024 22:26

For review

8e1a352

More

9b9d401

fix

559321a

Fix clippy

5453c57

More

abe7afc

Fix clippy

c36560c

viirya force-pushed the filter_to_row_filter branch from 49721e8 to c36560c Compare May 8, 2024 03:56

liurenjie1024 approved these changes May 14, 2024

View reviewed changes

Merge branch 'main' into filter_to_row_filter

9093d32

fix clippy

185e185

liurenjie1024 merged commit 81df940 into apache:main May 15, 2024
7 checks passed

feat: Convert predicate to arrow filter and push down to parquet reader #295

feat: Convert predicate to arrow filter and push down to parquet reader #295

Conversation

viirya commented Mar 24, 2024 • edited

liurenjie1024 commented Apr 1, 2024

viirya commented Apr 1, 2024

liurenjie1024 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya Apr 8, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya commented Apr 3, 2024

viirya commented Apr 12, 2024

liurenjie1024 left a comment

Choose a reason for hiding this comment

viirya commented Apr 28, 2024

liurenjie1024 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

viirya commented May 10, 2024

liurenjie1024 left a comment

Choose a reason for hiding this comment

liurenjie1024 commented May 14, 2024

viirya commented May 14, 2024

liurenjie1024 commented May 15, 2024

viirya commented May 15, 2024

viirya commented Mar 24, 2024 •

edited

viirya Apr 8, 2024 •

edited