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/06/05 04:39:04 UTC
[GitHub] [arrow-rs] Ted-Jiang opened a new pull request, #1791: feat:Implement page filtering with Row Alignment
Ted-Jiang opened a new pull request, #1791:
URL: https://github.com/apache/arrow-rs/pull/1791
# Which issue does this PR close?
Closes #1790.
# Rationale for this change
For now row group filter in datafusion pass a closure to arrow-rs
```
fn build_row_group_predicate(
pruning_predicate: &PruningPredicate,
metrics: ParquetFileMetrics,
) -> Box<dyn FnMut(&RowGroupMetaData, usize) -> bool> {
```
https://github.com/apache/arrow-datafusion/blob/585bc3a629b92ea7a86ebfe8bf762dbef4155710/datafusion/core/src/physical_plan/file_format/parquet.rs#L559-L562
So for page filter i try to pass
```
Box<dyn FnMut(&[pageIndex], &[pageLocation], usize) -> &[ranges]>
```
then use ranges to construct `RowRanges ` : row ranges in a row-group (one col) if col is sorted vec size will be 1.
For multi filter combine:
if there are two filters use `and` connect，use `RowRanges::intersection` to get the final rowRange; two filters use `or` connect，use `RowRanges::union` to get the final rowRange.
After this PR: i will working on `column_page_reader`, enable read specific row ranges record.
# 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.
-->
# Are there any user-facing changes?
<!---
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 a diff in pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889974862
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
+ }
+}
+
+/// Return the union of the two ranges,
Review Comment:
Is there a reason to omit these from the trait? Either as an associated function or a member function?
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
Review Comment:
```suggestion
self.end() + 1 - self.start()
```
I think otherwise this will potentially underflow - possibly worth a test
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
Review Comment:
```suggestion
fn is_before(&self, other: &Self) -> bool;
```
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
Review Comment:
```suggestion
fn is_after(&self, other: &Self) -> bool;
```
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
+ }
+}
+
+/// Return the union of the two ranges,
+/// Return `None` if there are hole between them.
+pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() + 1 >= *right.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() + 1 >= *left.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Returns the intersection of the two ranges,
+/// return null if they are not overlapped.
+pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() >= right.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() >= left.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, mut range: Range) {
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match union(last, &range) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ range = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(range);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the `RowRanges` of all the selected pages
+/// which represents a sequence of ranges of all the selected pages
+pub fn compute_row_ranges(
+ mask: &[bool],
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ let row_ranges = page_locations_to_row_ranges(locations, total_rows)?;
+ row_ranges.filter_with_mask(mask)
+}
+
+fn page_locations_to_row_ranges(
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ if locations.is_empty() {
Review Comment:
```suggestion
if locations.is_empty() || total_rows == 0 {
```
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
+ }
+}
+
+/// Return the union of the two ranges,
+/// Return `None` if there are hole between them.
+pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() + 1 >= *right.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() + 1 >= *left.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Returns the intersection of the two ranges,
+/// return null if they are not overlapped.
+pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() >= right.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() >= left.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
Review Comment:
```suggestion
/// Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
/// trying to union the specified range to the last ranges in the list. The specified range shall be larger than
/// the last one or might be overlapped with some of the last ones.
/// [a, b] < [c, d] if b < c
```
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
+ }
+}
+
+/// Return the union of the two ranges,
+/// Return `None` if there are hole between them.
+pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() + 1 >= *right.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() + 1 >= *left.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Returns the intersection of the two ranges,
+/// return null if they are not overlapped.
+pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() >= right.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() >= left.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, mut range: Range) {
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match union(last, &range) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ range = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(range);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the `RowRanges` of all the selected pages
+/// which represents a sequence of ranges of all the selected pages
+pub fn compute_row_ranges(
+ mask: &[bool],
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ let row_ranges = page_locations_to_row_ranges(locations, total_rows)?;
Review Comment:
`assert_eq(mask.len(), locations.len())` ?
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
+ }
+}
+
+/// Return the union of the two ranges,
+/// Return `None` if there are hole between them.
+pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() + 1 >= *right.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() + 1 >= *left.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Returns the intersection of the two ranges,
+/// return null if they are not overlapped.
+pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() >= right.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() >= left.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, mut range: Range) {
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match union(last, &range) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ range = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(range);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the `RowRanges` of all the selected pages
+/// which represents a sequence of ranges of all the selected pages
Review Comment:
```suggestion
/// Takes an array of [`PageLocation`], and a total number of rows, and based on the provided `page_mask`
/// returns the corresponding [`RowRanges`] to scan
```
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
+ }
+}
+
+/// Return the union of the two ranges,
+/// Return `None` if there are hole between them.
+pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() + 1 >= *right.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() + 1 >= *left.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Returns the intersection of the two ranges,
+/// return null if they are not overlapped.
+pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() >= right.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() >= left.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, mut range: Range) {
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match union(last, &range) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ range = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(range);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the `RowRanges` of all the selected pages
+/// which represents a sequence of ranges of all the selected pages
+pub fn compute_row_ranges(
+ mask: &[bool],
Review Comment:
```suggestion
page_mask: &[bool],
```
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r890094830
##########
parquet/src/file/metadata.rs:
##########
@@ -223,6 +223,7 @@ pub struct RowGroupMetaData {
num_rows: i64,
total_byte_size: i64,
schema_descr: SchemaDescPtr,
+ // Todo add filter result -> row range
Review Comment:
> It isn't even file-meta level, it isn't part of the footer but stored as separate pages 😅
yes separately from RowGroup, before the footer !😂
##########
parquet/src/file/metadata.rs:
##########
@@ -223,6 +223,7 @@ pub struct RowGroupMetaData {
num_rows: i64,
total_byte_size: i64,
schema_descr: SchemaDescPtr,
+ // Todo add filter result -> row range
Review Comment:
> It isn't even file-meta level, it isn't part of the footer but stored as separate pages 😅
yes separately from RowGroup, before the footer !😂
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r890005249
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
+ }
+}
+
+/// Return the union of the two ranges,
+/// Return `None` if there are hole between them.
+pub fn union(left: &Range, right: &Range) -> Option<Range> {
Review Comment:
Neither this method nor intersection appear to be used anywhere?
--
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] viirya commented on a diff in pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889657473
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, range: Range) {
+ let mut to_add = range;
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match Range::union(last, &to_add) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ to_add = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(to_add);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = Range::intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the row ranges `Vec(start, len)` of all the selected pages
+pub fn compute_row_ranges(
+ mask: &[bool],
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ let row_ranges = page_locations_to_row_ranges(locations, total_rows)?;
+ row_ranges.filter_with_mask(mask)
+}
+
+fn page_locations_to_row_ranges(
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ if locations.is_empty() {
+ return Ok(RowRanges::new_empty());
+ }
+
+ let mut vec_range: VecDeque<Range> = locations
+ .windows(2)
+ .map(|x| {
+ let start = x[0].first_row_index as usize;
+ let end = (x[1].first_row_index - 1) as usize;
+ Range {
+ from: start,
+ to: end,
+ }
+ })
+ .collect();
+
+ let last = Range {
+ from: locations.last().unwrap().first_row_index as usize,
+ to: total_rows,
+ };
+ vec_range.push_back(last);
+
+ Ok(RowRanges { ranges: vec_range })
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::basic::Type::INT32;
+ use crate::file::page_index::index::{NativeIndex, PageIndex};
+ use crate::file::page_index::range::{compute_row_ranges, Range, RowRanges};
+ use parquet_format::{BoundaryOrder, PageLocation};
+
+ #[test]
+ fn test_binary_search_overlap() {
+ let mut ranges = RowRanges::new_empty();
+ ranges.add(Range { from: 1, to: 3 });
+ ranges.add(Range { from: 6, to: 7 });
+
+ assert!(ranges.is_overlapping(&Range { from: 1, to: 2 }));
+ // include both [start, end]
+ assert!(ranges.is_overlapping(&Range { from: 0, to: 1 }));
+ assert!(ranges.is_overlapping(&Range { from: 0, to: 3 }));
+
+ assert!(ranges.is_overlapping(&Range { from: 0, to: 7 }));
+ assert!(ranges.is_overlapping(&Range { from: 2, to: 7 }));
+
+ assert!(!ranges.is_overlapping(&Range { from: 4, to: 5 }));
+ }
+
+ #[test]
+ fn test_add_func_ascending_disjunctive() {
+ let mut ranges_1 = RowRanges::new_empty();
+ ranges_1.add(Range { from: 1, to: 3 });
+ ranges_1.add(Range { from: 5, to: 6 });
+ ranges_1.add(Range { from: 8, to: 9 });
+ assert_eq!(ranges_1.count(), 3);
+ }
+
+ #[test]
+ fn test_add_func_ascending_merge() {
+ let mut ranges_1 = RowRanges::new_empty();
+ ranges_1.add(Range { from: 1, to: 3 });
+ ranges_1.add(Range { from: 4, to: 5 });
+ ranges_1.add(Range { from: 6, to: 7 });
+ assert_eq!(ranges_1.count(), 1);
+ }
+
+ #[test]
+ #[should_panic(expected = "Must add range in ascending!")]
+ fn test_add_func_not_ascending() {
+ let mut ranges_1 = RowRanges::new_empty();
+ ranges_1.add(Range { from: 6, to: 7 });
+ ranges_1.add(Range { from: 1, to: 3 });
+ ranges_1.add(Range { from: 4, to: 5 });
+ assert_eq!(ranges_1.count(), 1);
+ }
+
+ #[test]
+ fn test_union_func() {
+ let mut ranges_1 = RowRanges::new_empty();
+ ranges_1.add(Range { from: 1, to: 2 });
+ ranges_1.add(Range { from: 3, to: 4 });
+ ranges_1.add(Range { from: 5, to: 6 });
+
+ let mut ranges_2 = RowRanges::new_empty();
+ ranges_2.add(Range { from: 2, to: 3 });
+ ranges_2.add(Range { from: 4, to: 5 });
+ ranges_2.add(Range { from: 6, to: 7 });
+
+ let ranges = RowRanges::union(ranges_1, ranges_2);
+ assert_eq!(ranges.count(), 1);
+ let range = ranges.ranges.get(0).unwrap();
+ assert_eq!(range.from, 1);
+ assert_eq!(range.to, 7);
+
+ let mut ranges_a = RowRanges::new_empty();
+ ranges_a.add(Range { from: 1, to: 3 });
+ ranges_a.add(Range { from: 5, to: 8 });
+ ranges_a.add(Range { from: 11, to: 12 });
+
+ let mut ranges_b = RowRanges::new_empty();
+ ranges_b.add(Range { from: 0, to: 2 });
+ ranges_b.add(Range { from: 6, to: 7 });
+ ranges_b.add(Range { from: 10, to: 11 });
+
+ let ranges = RowRanges::union(ranges_a, ranges_b);
+ assert_eq!(ranges.count(), 3);
+
+ let range_1 = ranges.ranges.get(0).unwrap();
+ assert_eq!(range_1.from, 0);
+ assert_eq!(range_1.to, 3);
+ let range_2 = ranges.ranges.get(1).unwrap();
+ assert_eq!(range_2.from, 5);
+ assert_eq!(range_2.to, 8);
+ let range_3 = ranges.ranges.get(2).unwrap();
+ assert_eq!(range_3.from, 10);
+ assert_eq!(range_3.to, 12);
+ }
+
+ #[test]
+ fn test_intersection_func() {
+ let mut ranges_1 = RowRanges::new_empty();
+ ranges_1.add(Range { from: 1, to: 2 });
+ ranges_1.add(Range { from: 3, to: 4 });
+ ranges_1.add(Range { from: 5, to: 6 });
+
+ let mut ranges_2 = RowRanges::new_empty();
+ ranges_2.add(Range { from: 2, to: 3 });
+ ranges_2.add(Range { from: 4, to: 5 });
+ ranges_2.add(Range { from: 6, to: 7 });
+
+ let ranges = RowRanges::intersection(ranges_1, ranges_2);
+ assert_eq!(ranges.count(), 1);
+ let range = ranges.ranges.get(0).unwrap();
+ assert_eq!(range.from, 2);
+ assert_eq!(range.to, 6);
+
+ let mut ranges_a = RowRanges::new_empty();
+ ranges_a.add(Range { from: 1, to: 3 });
+ ranges_a.add(Range { from: 5, to: 8 });
+ ranges_a.add(Range { from: 11, to: 12 });
+
+ let mut ranges_b = RowRanges::new_empty();
+ ranges_b.add(Range { from: 0, to: 2 });
+ ranges_b.add(Range { from: 6, to: 7 });
+ ranges_b.add(Range { from: 10, to: 11 });
+
+ let ranges = RowRanges::intersection(ranges_a, ranges_b);
+ assert_eq!(ranges.count(), 3);
+
+ let range_1 = ranges.ranges.get(0).unwrap();
+ assert_eq!(range_1.from, 1);
+ assert_eq!(range_1.to, 2);
+ let range_2 = ranges.ranges.get(1).unwrap();
+ assert_eq!(range_2.from, 6);
+ assert_eq!(range_2.to, 7);
+ let range_3 = ranges.ranges.get(2).unwrap();
+ assert_eq!(range_3.from, 11);
+ assert_eq!(range_3.to, 11);
+ }
+
+ #[test]
+ fn test_compute_one() {
+ let locations = &[PageLocation {
+ offset: 50,
+ compressed_page_size: 10,
+ first_row_index: 0,
+ }];
+ let total_rows = 10;
+
+ let row_ranges = compute_row_ranges(&[true], locations, total_rows).unwrap();
+ assert_eq!(row_ranges.count(), 1);
+ assert_eq!(
+ row_ranges.ranges.get(0).unwrap(),
+ &Range { from: 0, to: 10 }
+ );
Review Comment:
`from` and `to` are inclusive, why `total_rows` is 10?
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r890005249
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
+ }
+}
+
+/// Return the union of the two ranges,
+/// Return `None` if there are hole between them.
+pub fn union(left: &Range, right: &Range) -> Option<Range> {
Review Comment:
Neither this method nor intersection appear to be used anywhere? And the same for the accompanying methods on RowRanges?
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889668210
##########
parquet/src/file/metadata.rs:
##########
@@ -223,6 +223,7 @@ pub struct RowGroupMetaData {
num_rows: i64,
total_byte_size: i64,
schema_descr: SchemaDescPtr,
+ // Todo add filter result -> row range
Review Comment:
The more I think about this the more I wonder whether the metadata structs are the right place to put the index information. They're parsed and interpreted separately from the main metadata, and so I think it makes sense for them to be stored separately?
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,472 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
Review Comment:
I'm curious why you opted for inclusive ranges, when exclusive ranges better match the source data and are more common in Rust
##########
parquet/src/file/page_index/mod.rs:
##########
@@ -17,3 +17,4 @@
pub mod index;
pub mod index_reader;
+pub mod range;
Review Comment:
At least whilst we're still iterating on these APIs perhaps `pub(crate)`
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,472 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
Review Comment:
I think this data structure is typically called am IntervalSet. There might even be an upstream crate implementation... 🤔
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,472 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
Review Comment:
Is there a particular reason you opted to not use std::ops:RangeInclusive. Potentially with free functions or am extension trait for the intersection, union, etc... ?
--
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] viirya commented on a diff in pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889658426
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, range: Range) {
+ let mut to_add = range;
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match Range::union(last, &to_add) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ to_add = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(to_add);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = Range::intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the row ranges `Vec(start, len)` of all the selected pages
+pub fn compute_row_ranges(
Review Comment:
How will this be used?
Do you need to make something like `with_predicate` in `ReadOptionsBuilder` to take a closure for filtering pages based on the ranges?
--
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] viirya commented on a diff in pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889650972
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+///Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+///the column index on the filtering.
Review Comment:
```suggestion
/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
/// the column index on the filtering.
```
There are some style issues like that.
--
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 #1791: feat:Add function for row alignment with page mask
Posted by GitBox <gi...@apache.org>.
tustvold merged PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r890103109
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
+ }
+}
+
+/// Return the union of the two ranges,
Review Comment:
yes moving in trait is better.
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r890148967
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
+ }
+}
+
+/// Return the union of the two ranges,
+/// Return `None` if there are hole between them.
+pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() + 1 >= *right.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() + 1 >= *left.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::max(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Returns the intersection of the two ranges,
+/// return null if they are not overlapped.
+pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.start() <= right.start() {
+ if left.end() >= right.start() {
+ return Some(Range::new(
+ *right.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ } else if right.end() >= left.start() {
+ return Some(Range::new(
+ *left.start(),
+ std::cmp::min(*left.end(), *right.end()),
+ ));
+ }
+ None
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
Review Comment:
This change appears to have been lost??
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889827232
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,472 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
Review Comment:
No particular reason, i think range is `from` row number `to` row number both should be inclusive.After use `std::ops::RangeInclusive`, we can call `into_slice_range` get slice like `[x..y)`
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889658890
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, range: Range) {
+ let mut to_add = range;
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match Range::union(last, &to_add) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ to_add = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(to_add);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = Range::intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the row ranges `Vec(start, len)` of all the selected pages
+pub fn compute_row_ranges(
+ mask: &[bool],
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ let row_ranges = page_locations_to_row_ranges(locations, total_rows)?;
+ row_ranges.filter_with_mask(mask)
+}
+
+fn page_locations_to_row_ranges(
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ if locations.is_empty() {
+ return Ok(RowRanges::new_empty());
+ }
+
+ let mut vec_range: VecDeque<Range> = locations
+ .windows(2)
+ .map(|x| {
+ let start = x[0].first_row_index as usize;
+ let end = (x[1].first_row_index - 1) as usize;
+ Range {
+ from: start,
+ to: end,
+ }
+ })
+ .collect();
+
+ let last = Range {
+ from: locations.last().unwrap().first_row_index as usize,
+ to: total_rows,
+ };
+ vec_range.push_back(last);
+
+ Ok(RowRanges { ranges: vec_range })
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::basic::Type::INT32;
+ use crate::file::page_index::index::{NativeIndex, PageIndex};
+ use crate::file::page_index::range::{compute_row_ranges, Range, RowRanges};
+ use parquet_format::{BoundaryOrder, PageLocation};
+
+ #[test]
+ fn test_binary_search_overlap() {
+ let mut ranges = RowRanges::new_empty();
+ ranges.add(Range { from: 1, to: 3 });
+ ranges.add(Range { from: 6, to: 7 });
+
+ assert!(ranges.is_overlapping(&Range { from: 1, to: 2 }));
+ // include both [start, end]
+ assert!(ranges.is_overlapping(&Range { from: 0, to: 1 }));
+ assert!(ranges.is_overlapping(&Range { from: 0, to: 3 }));
+
+ assert!(ranges.is_overlapping(&Range { from: 0, to: 7 }));
+ assert!(ranges.is_overlapping(&Range { from: 2, to: 7 }));
+
+ assert!(!ranges.is_overlapping(&Range { from: 4, to: 5 }));
+ }
+
+ #[test]
+ fn test_add_func_ascending_disjunctive() {
+ let mut ranges_1 = RowRanges::new_empty();
+ ranges_1.add(Range { from: 1, to: 3 });
+ ranges_1.add(Range { from: 5, to: 6 });
+ ranges_1.add(Range { from: 8, to: 9 });
+ assert_eq!(ranges_1.count(), 3);
+ }
+
+ #[test]
+ fn test_add_func_ascending_merge() {
+ let mut ranges_1 = RowRanges::new_empty();
+ ranges_1.add(Range { from: 1, to: 3 });
+ ranges_1.add(Range { from: 4, to: 5 });
+ ranges_1.add(Range { from: 6, to: 7 });
+ assert_eq!(ranges_1.count(), 1);
+ }
+
+ #[test]
+ #[should_panic(expected = "Must add range in ascending!")]
+ fn test_add_func_not_ascending() {
+ let mut ranges_1 = RowRanges::new_empty();
+ ranges_1.add(Range { from: 6, to: 7 });
+ ranges_1.add(Range { from: 1, to: 3 });
+ ranges_1.add(Range { from: 4, to: 5 });
+ assert_eq!(ranges_1.count(), 1);
+ }
+
+ #[test]
+ fn test_union_func() {
+ let mut ranges_1 = RowRanges::new_empty();
+ ranges_1.add(Range { from: 1, to: 2 });
+ ranges_1.add(Range { from: 3, to: 4 });
+ ranges_1.add(Range { from: 5, to: 6 });
+
+ let mut ranges_2 = RowRanges::new_empty();
+ ranges_2.add(Range { from: 2, to: 3 });
+ ranges_2.add(Range { from: 4, to: 5 });
+ ranges_2.add(Range { from: 6, to: 7 });
+
+ let ranges = RowRanges::union(ranges_1, ranges_2);
+ assert_eq!(ranges.count(), 1);
+ let range = ranges.ranges.get(0).unwrap();
+ assert_eq!(range.from, 1);
+ assert_eq!(range.to, 7);
+
+ let mut ranges_a = RowRanges::new_empty();
+ ranges_a.add(Range { from: 1, to: 3 });
+ ranges_a.add(Range { from: 5, to: 8 });
+ ranges_a.add(Range { from: 11, to: 12 });
+
+ let mut ranges_b = RowRanges::new_empty();
+ ranges_b.add(Range { from: 0, to: 2 });
+ ranges_b.add(Range { from: 6, to: 7 });
+ ranges_b.add(Range { from: 10, to: 11 });
+
+ let ranges = RowRanges::union(ranges_a, ranges_b);
+ assert_eq!(ranges.count(), 3);
+
+ let range_1 = ranges.ranges.get(0).unwrap();
+ assert_eq!(range_1.from, 0);
+ assert_eq!(range_1.to, 3);
+ let range_2 = ranges.ranges.get(1).unwrap();
+ assert_eq!(range_2.from, 5);
+ assert_eq!(range_2.to, 8);
+ let range_3 = ranges.ranges.get(2).unwrap();
+ assert_eq!(range_3.from, 10);
+ assert_eq!(range_3.to, 12);
+ }
+
+ #[test]
+ fn test_intersection_func() {
+ let mut ranges_1 = RowRanges::new_empty();
+ ranges_1.add(Range { from: 1, to: 2 });
+ ranges_1.add(Range { from: 3, to: 4 });
+ ranges_1.add(Range { from: 5, to: 6 });
+
+ let mut ranges_2 = RowRanges::new_empty();
+ ranges_2.add(Range { from: 2, to: 3 });
+ ranges_2.add(Range { from: 4, to: 5 });
+ ranges_2.add(Range { from: 6, to: 7 });
+
+ let ranges = RowRanges::intersection(ranges_1, ranges_2);
+ assert_eq!(ranges.count(), 1);
+ let range = ranges.ranges.get(0).unwrap();
+ assert_eq!(range.from, 2);
+ assert_eq!(range.to, 6);
+
+ let mut ranges_a = RowRanges::new_empty();
+ ranges_a.add(Range { from: 1, to: 3 });
+ ranges_a.add(Range { from: 5, to: 8 });
+ ranges_a.add(Range { from: 11, to: 12 });
+
+ let mut ranges_b = RowRanges::new_empty();
+ ranges_b.add(Range { from: 0, to: 2 });
+ ranges_b.add(Range { from: 6, to: 7 });
+ ranges_b.add(Range { from: 10, to: 11 });
+
+ let ranges = RowRanges::intersection(ranges_a, ranges_b);
+ assert_eq!(ranges.count(), 3);
+
+ let range_1 = ranges.ranges.get(0).unwrap();
+ assert_eq!(range_1.from, 1);
+ assert_eq!(range_1.to, 2);
+ let range_2 = ranges.ranges.get(1).unwrap();
+ assert_eq!(range_2.from, 6);
+ assert_eq!(range_2.to, 7);
+ let range_3 = ranges.ranges.get(2).unwrap();
+ assert_eq!(range_3.from, 11);
+ assert_eq!(range_3.to, 11);
+ }
+
+ #[test]
+ fn test_compute_one() {
+ let locations = &[PageLocation {
+ offset: 50,
+ compressed_page_size: 10,
+ first_row_index: 0,
+ }];
+ let total_rows = 10;
+
+ let row_ranges = compute_row_ranges(&[true], locations, total_rows).unwrap();
+ assert_eq!(row_ranges.count(), 1);
+ assert_eq!(
+ row_ranges.ranges.get(0).unwrap(),
+ &Range { from: 0, to: 10 }
+ );
Review Comment:
Opps! nice find! 👍
row_ranges should start with 0 end with total_rows - 1
fix in [8329f21](https://github.com/apache/arrow-rs/pull/1791/commits/8329f2135bc158f5aef74d92d6a8ced712d03094)
--
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] viirya commented on a diff in pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889658333
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, range: Range) {
+ let mut to_add = range;
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match Range::union(last, &to_add) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ to_add = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(to_add);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = Range::intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the row ranges `Vec(start, len)` of all the selected pages
Review Comment:
This doesn't return `Vec(start, len)` actually. It reads to confusing. You may document it returning `RowRanges` which represents a sequence of ranges of all the selected pages.
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
tustvold commented on PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#issuecomment-1146751327
Thank you, I will review this tomorrow (GMT)
--
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] viirya commented on a diff in pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889651705
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+///Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+///the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size{} is not equal to page size {}",
Review Comment:
```suggestion
"Mask size{} is not equal to number of pages {}",
```
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889821082
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,472 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
Review Comment:
it basic struct align with `parquet-mr` in 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] Ted-Jiang commented on a diff in pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889820667
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,472 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
Review Comment:
Thanks, before your notice, I didn't know there was a `ops:RangeInclusive` in std😂. I look through it, will extend `intersection, union` on 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] Ted-Jiang commented on pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#issuecomment-1147417751
> Edit: is the intention to use this for the first step?
Basically, yes !
First step contains:
1. Use `SerializedFileReader` read fileMeta get file_metadata
2. Use min_max on row group filter filter useless group (already exist)
3. Read specific row group page_index construct `Index`(#1761)
4. Use both `index` and `filter` generate `pagemask` (this one)
....
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889659815
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, range: Range) {
+ let mut to_add = range;
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match Range::union(last, &to_add) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ to_add = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(to_add);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = Range::intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the row ranges `Vec(start, len)` of all the selected pages
+pub fn compute_row_ranges(
Review Comment:
Yes, basically add a closure called `page_predicates` in `ReadOptionsBuilder ` like
```
Box<dyn FnMut(&[pageIndex], &[pageLocation], usize) -> &[bool]>
```
to generate mask then call this function `compute_row_ranges `.
I'm thinking about doing this at the end for testing with datafusion.
--
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] viirya commented on a diff in pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889657624
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, range: Range) {
+ let mut to_add = range;
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match Range::union(last, &to_add) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ to_add = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(to_add);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = Range::intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the row ranges `Vec(start, len)` of all the selected pages
+pub fn compute_row_ranges(
+ mask: &[bool],
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ let row_ranges = page_locations_to_row_ranges(locations, total_rows)?;
+ row_ranges.filter_with_mask(mask)
+}
+
+fn page_locations_to_row_ranges(
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ if locations.is_empty() {
+ return Ok(RowRanges::new_empty());
+ }
+
+ let mut vec_range: VecDeque<Range> = locations
+ .windows(2)
+ .map(|x| {
+ let start = x[0].first_row_index as usize;
+ let end = (x[1].first_row_index - 1) as usize;
+ Range {
+ from: start,
+ to: end,
+ }
+ })
+ .collect();
+
+ let last = Range {
+ from: locations.last().unwrap().first_row_index as usize,
+ to: total_rows,
+ };
Review Comment:
`total_rows` could be given an incorrect value like I just found in the test. Should we do a sanity check?
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889867688
##########
parquet/src/file/metadata.rs:
##########
@@ -223,6 +223,7 @@ pub struct RowGroupMetaData {
num_rows: i64,
total_byte_size: i64,
schema_descr: SchemaDescPtr,
+ // Todo add filter result -> row range
Review Comment:
Yes, you are right. page index stored in file-meta level.
My thought is read less pageIndex after rowgroup filter
https://github.com/apache/arrow-rs/blob/be388291036512b925796fdff50b39afdbceee20/parquet/src/file/serialized_reader.rs#L211-L224
https://github.com/apache/arrow-rs/blob/be388291036512b925796fdff50b39afdbceee20/parquet/src/file/serialized_reader.rs#L246-L249
So i want to read index here and insert it into RowGroupMetaData.
It was just a simple idea at first, maybe we can find a better way in the process of implementation
--
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 pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#issuecomment-1146749216
@tustvold @alamb @viirya PTAL 😊
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889653713
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+///Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+///the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size{} is not equal to page size {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, range: Range) {
+ let mut to_add = range;
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match Range::union(last, &to_add) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ to_add = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(to_add);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = Range::intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the row ranges `Vec(start, len)` of all the selected pages
+pub fn compute_row_ranges(
+ mask: &[bool],
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ let row_ranges = page_locations_to_row_ranges(locations, total_rows)?;
+ row_ranges.filter_with_mask(mask)
+}
+
+fn page_locations_to_row_ranges(
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ if locations.is_empty() {
+ return Ok(RowRanges::new_empty());
+ }
+
+ let mut vec_range: VecDeque<Range> = locations
+ .windows(2)
+ .map(|x| {
+ let start = x[0].first_row_index as usize;
+ let end = (x[1].first_row_index - 1) as usize;
Review Comment:
Yes, if we read directly from parquet `pageIndex`, maybe i will add it in annotation
--
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] viirya commented on a diff in pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889651850
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+///Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+///the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size{} is not equal to page size {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, range: Range) {
+ let mut to_add = range;
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match Range::union(last, &to_add) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ to_add = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(to_add);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = Range::intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the row ranges `Vec(start, len)` of all the selected pages
+pub fn compute_row_ranges(
+ mask: &[bool],
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ let row_ranges = page_locations_to_row_ranges(locations, total_rows)?;
+ row_ranges.filter_with_mask(mask)
+}
+
+fn page_locations_to_row_ranges(
+ locations: &[PageLocation],
+ total_rows: usize,
+) -> Result<RowRanges, ParquetError> {
+ if locations.is_empty() {
+ return Ok(RowRanges::new_empty());
+ }
+
+ let mut vec_range: VecDeque<Range> = locations
+ .windows(2)
+ .map(|x| {
+ let start = x[0].first_row_index as usize;
+ let end = (x[1].first_row_index - 1) as usize;
Review Comment:
So the selected pages must be continuous?
--
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] viirya commented on a diff in pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
viirya commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889651705
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+///Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+///the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size{} is not equal to page size {}",
Review Comment:
```suggestion
"Mask size {} is not equal to number of pages {}",
```
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889861992
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,472 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
Review Comment:
@tustvold fix in [adc48cf](https://github.com/apache/arrow-rs/pull/1791/commits/adc48cfb1dc46f3747a8e7364839e397a63bac33)
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
tustvold commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889966156
##########
parquet/src/file/metadata.rs:
##########
@@ -223,6 +223,7 @@ pub struct RowGroupMetaData {
num_rows: i64,
total_byte_size: i64,
schema_descr: SchemaDescPtr,
+ // Todo add filter result -> row range
Review Comment:
> page index stored in file-meta level.
It isn't even file-meta level, it isn't part of the footer but stored as separate pages :sweat_smile:
> It was just a simple idea at first, maybe we can find a better way in the process of implementation
Provided we take care to ensure we keep things pub(crate) so we don't break APIs, this seems like a good strategy :+1:
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889654323
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+///Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+///the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size{} is not equal to page size {}",
Review Comment:
Thanks ❤️
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r890096497
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,461 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+use std::ops::RangeInclusive;
+
+type Range = RangeInclusive<usize>;
+
+pub trait RangeOps {
+ fn is_before(&self, other: &Range) -> bool;
+
+ fn is_after(&self, other: &Range) -> bool;
+
+ fn count(&self) -> usize;
+}
+
+impl RangeOps for Range {
+ fn is_before(&self, other: &Range) -> bool {
+ self.end() < other.start()
+ }
+
+ fn is_after(&self, other: &Range) -> bool {
+ self.start() > other.end()
+ }
+
+ fn count(&self) -> usize {
+ self.end() - self.start() + 1
Review Comment:
👍 learned
--
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] codecov-commenter commented on pull request #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#issuecomment-1146743725
# [Codecov](https://codecov.io/gh/apache/arrow-rs/pull/1791?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#1791](https://codecov.io/gh/apache/arrow-rs/pull/1791?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (a4c9e60) into [master](https://codecov.io/gh/apache/arrow-rs/commit/c1a91dc6d2ea7b2ee7ed8d0f468fb6b77e82d39d?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (c1a91dc) will **increase** coverage by `0.02%`.
> The diff coverage is `90.70%`.
```diff
@@ Coverage Diff @@
## master #1791 +/- ##
==========================================
+ Coverage 83.39% 83.42% +0.02%
==========================================
Files 198 199 +1
Lines 56142 56397 +255
==========================================
+ Hits 46821 47049 +228
- Misses 9321 9348 +27
```
| [Impacted Files](https://codecov.io/gh/apache/arrow-rs/pull/1791?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [parquet/src/file/metadata.rs](https://codecov.io/gh/apache/arrow-rs/pull/1791/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGFycXVldC9zcmMvZmlsZS9tZXRhZGF0YS5ycw==) | `95.12% <ø> (ø)` | |
| [parquet/src/file/serialized\_reader.rs](https://codecov.io/gh/apache/arrow-rs/pull/1791/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGFycXVldC9zcmMvZmlsZS9zZXJpYWxpemVkX3JlYWRlci5ycw==) | `94.46% <ø> (-1.17%)` | :arrow_down: |
| [parquet/src/file/page\_index/range.rs](https://codecov.io/gh/apache/arrow-rs/pull/1791/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGFycXVldC9zcmMvZmlsZS9wYWdlX2luZGV4L3JhbmdlLnJz) | `90.70% <90.70%> (ø)` | |
| [parquet/src/util/cursor.rs](https://codecov.io/gh/apache/arrow-rs/pull/1791/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGFycXVldC9zcmMvdXRpbC9jdXJzb3IucnM=) | `62.18% <0.00%> (-1.69%)` | :arrow_down: |
| [parquet\_derive/src/parquet\_field.rs](https://codecov.io/gh/apache/arrow-rs/pull/1791/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGFycXVldF9kZXJpdmUvc3JjL3BhcnF1ZXRfZmllbGQucnM=) | `65.75% <0.00%> (-0.46%)` | :arrow_down: |
| [parquet/src/file/writer.rs](https://codecov.io/gh/apache/arrow-rs/pull/1791/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGFycXVldC9zcmMvZmlsZS93cml0ZXIucnM=) | `92.84% <0.00%> (-0.02%)` | :arrow_down: |
| [parquet/src/arrow/arrow\_writer.rs](https://codecov.io/gh/apache/arrow-rs/pull/1791/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGFycXVldC9zcmMvYXJyb3cvYXJyb3dfd3JpdGVyLnJz) | `97.76% <0.00%> (-0.02%)` | :arrow_down: |
| [parquet/src/arrow/schema.rs](https://codecov.io/gh/apache/arrow-rs/pull/1791/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cGFycXVldC9zcmMvYXJyb3cvc2NoZW1hLnJz) | `96.81% <0.00%> (-0.01%)` | :arrow_down: |
| [arrow/src/ffi.rs](https://codecov.io/gh/apache/arrow-rs/pull/1791/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-YXJyb3cvc3JjL2ZmaS5ycw==) | `86.97% <0.00%> (ø)` | |
| ... and [11 more](https://codecov.io/gh/apache/arrow-rs/pull/1791/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/arrow-rs/pull/1791?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/arrow-rs/pull/1791?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [c1a91dc...a4c9e60](https://codecov.io/gh/apache/arrow-rs/pull/1791?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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 #1791: feat:Implement page filtering with Row Alignment
Posted by GitBox <gi...@apache.org>.
Ted-Jiang commented on code in PR #1791:
URL: https://github.com/apache/arrow-rs/pull/1791#discussion_r889658973
##########
parquet/src/file/page_index/range.rs:
##########
@@ -0,0 +1,474 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::errors::ParquetError;
+use parquet_format::PageLocation;
+use std::cmp::Ordering;
+use std::collections::VecDeque;
+
+/// A row range
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Range {
+ /// Its start
+ pub from: usize,
+ /// Its end
+ pub to: usize,
+}
+
+impl Range {
+ // Creates a range of [from, to] (from and to are both inclusive)
+ pub fn new(from: usize, to: usize) -> Self {
+ assert!(from <= to);
+ Self { from, to }
+ }
+
+ pub fn count(&self) -> usize {
+ self.to - self.from + 1
+ }
+
+ pub fn is_before(&self, other: &Range) -> bool {
+ self.to < other.from
+ }
+
+ pub fn is_after(&self, other: &Range) -> bool {
+ self.from > other.to
+ }
+
+ /// Return the union of the two ranges,
+ /// Return `None` if there are hole between them.
+ pub fn union(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to + 1 >= right.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ } else if right.to + 1 >= left.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::max(left.to, right.to),
+ });
+ }
+ None
+ }
+
+ /// Returns the intersection of the two ranges,
+ /// return null if they are not overlapped.
+ pub fn intersection(left: &Range, right: &Range) -> Option<Range> {
+ if left.from <= right.from {
+ if left.to >= right.from {
+ return Some(Range {
+ from: right.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ } else if right.to >= left.from {
+ return Some(Range {
+ from: left.from,
+ to: std::cmp::min(left.to, right.to),
+ });
+ }
+ None
+ }
+}
+
+/// Struct representing row ranges in a row-group. These row ranges are calculated as a result of using
+/// the column index on the filtering.
+#[derive(Debug, Clone)]
+pub struct RowRanges {
+ pub ranges: VecDeque<Range>,
+}
+
+impl RowRanges {
+ //create an empty RowRanges
+ pub fn new_empty() -> Self {
+ RowRanges {
+ ranges: VecDeque::new(),
+ }
+ }
+
+ pub fn count(&self) -> usize {
+ self.ranges.len()
+ }
+
+ pub fn filter_with_mask(&self, mask: &[bool]) -> Result<RowRanges, ParquetError> {
+ if self.ranges.len() != mask.len() {
+ return Err(ParquetError::General(format!(
+ "Mask size {} is not equal to number of pages {}",
+ mask.len(),
+ self.count()
+ )));
+ }
+ let vec_range = mask
+ .iter()
+ .zip(self.ranges.clone())
+ .filter_map(|(&f, r)| if f { Some(r) } else { None })
+ .collect();
+ Ok(RowRanges { ranges: vec_range })
+ }
+
+ //Add a range to the end of the list of ranges. It maintains the disjunctive ascending order of the ranges by
+ //trying to union the specified range to the last ranges in the list. The specified range shall be larger than
+ //the last one or might be overlapped with some of the last ones.
+ // [a, b] < [c, d] if b < c
+ pub fn add(&mut self, range: Range) {
+ let mut to_add = range;
+ let count = self.count();
+ if count > 0 {
+ for i in 1..(count + 1) {
+ let index = count - i;
+ let last = self.ranges.get(index).unwrap();
+ assert!(!last.is_after(&range), "Must add range in ascending!");
+ // try to merge range
+ match Range::union(last, &to_add) {
+ None => {
+ break;
+ }
+ Some(r) => {
+ to_add = r;
+ self.ranges.remove(index);
+ }
+ }
+ }
+ }
+ self.ranges.push_back(to_add);
+ }
+
+ /// Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are no
+ /// elements between them. Otherwise, the two disjunctive ranges are stored separately.
+ /// For example:
+ /// [113, 241] ∪ [221, 340] = [113, 330]
+ /// [113, 230] ∪ [231, 340] = [113, 340]
+ /// while
+ /// [113, 230] ∪ [232, 340] = [113, 230], [232, 340]
+ ///
+ /// The result RowRanges object will contain all the row indexes that were contained in one of the specified objects.
+ pub fn union(mut left: RowRanges, mut right: RowRanges) -> RowRanges {
+ let v1 = &mut left.ranges;
+ let v2 = &mut right.ranges;
+ let mut result = RowRanges::new_empty();
+ if v2.is_empty() {
+ left.clone()
+ } else {
+ let mut range2 = v2.pop_front().unwrap();
+ while !v1.is_empty() {
+ let range1 = v1.pop_front().unwrap();
+ if range1.is_after(&range2) {
+ result.add(range2);
+ range2 = range1;
+ std::mem::swap(v1, v2);
+ } else {
+ result.add(range1);
+ }
+ }
+
+ result.add(range2);
+ while !v2.is_empty() {
+ result.add(v2.pop_front().unwrap())
+ }
+
+ result
+ }
+ }
+
+ /// Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common
+ /// elements otherwise the result is empty.
+ /// For example:
+ /// [113, 241] ∩ [221, 340] = [221, 241]
+ /// while
+ /// [113, 230] ∩ [231, 340] = <EMPTY>
+ ///
+ /// The result RowRanges object will contain all the row indexes there were contained in both of the specified objects
+ #[allow(clippy::mut_range_bound)]
+ pub fn intersection(left: RowRanges, right: RowRanges) -> RowRanges {
+ let mut result = RowRanges::new_empty();
+ let mut right_index = 0;
+ for l in left.ranges.iter() {
+ for i in right_index..right.ranges.len() {
+ let r = right.ranges.get(i).unwrap();
+ if l.is_before(r) {
+ break;
+ } else if l.is_after(r) {
+ right_index = i + 1;
+ continue;
+ }
+ if let Some(ra) = Range::intersection(l, r) {
+ result.add(ra);
+ }
+ }
+ }
+ result
+ }
+
+ pub fn row_count(&self) -> usize {
+ self.ranges.iter().map(|x| x.count()).sum()
+ }
+
+ pub fn is_overlapping(&self, x: &Range) -> bool {
+ self.ranges
+ .binary_search_by(|y| -> Ordering {
+ if y.is_before(x) {
+ Ordering::Less
+ } else if y.is_after(x) {
+ Ordering::Greater
+ } else {
+ Ordering::Equal
+ }
+ })
+ .is_ok()
+ }
+}
+
+/// Return the row ranges `Vec(start, len)` of all the selected pages
Review Comment:
Yes, i will make it clearer.
--
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