You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "binmahone (via GitHub)" <gi...@apache.org> on 2023/11/23 13:22:39 UTC
[PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
binmahone opened a new pull request, #38867:
URL: https://github.com/apache/arrow/pull/38867
### this PR is based on arrow-13 for early review
### Rationale for this change
see https://github.com/apache/arrow/issues/38865
### What changes are included in this PR?
add a parameter in GetRecordBatchReader to accept row_ranges.
The row ranges can be used to:
1. skip decompressing and decoding unnecessary pages. This part is done by leveraging an existing hook called DataPageFilter2
2. skip unwanted rows in the necessary pages. This part is done by a new added class called RecordSkipper. In Parquet, row number is not aligned across different columns' pages, so each Column Reader does NOT share RecordSkipper
### Are these changes tested?
a new test file callled range_reader_test.cc is added
### Are there any user-facing changes?
a new GetRecordBatchReader API overload is added. NO existing API is broken
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1452324625
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -206,6 +219,7 @@ class PARQUET_EXPORT FileReader {
std::shared_ptr<::arrow::RecordBatchReader>* out);
::arrow::Status GetRecordBatchReader(std::shared_ptr<::arrow::RecordBatchReader>* out);
+
Review Comment:
done
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
Review Comment:
done
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1439052028
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) { // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.getRanges()) {
Review Comment:
it requires more code to proper handle the case in which page index is missing. Exception on missing seems to be a simple&easy solution for now
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1439052476
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
Review Comment:
added a bitmap interface in RowRanges. However it is not implemented yet.
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
Review Comment:
this is deleted
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1439052390
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) { // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.getRanges()) {
+ if (!row_ranges.isOverlapping(page)) {
+ skip_pages.add(page, false);
+ }
+ }
+
+ /// Since the skipped pages will be slienly skipped without updating
+ /// current_rg_processed_records or records_read_, we need to pre-process the row
+ /// ranges as if these skipped pages never existed
+ adjust_ranges(skip_pages, row_ranges);
+
+ total_rows_to_process = pages.rowCount() - skip_pages.rowCount();
+ }
+
+ /// \brief Return the number of records to read or to skip
+ /// if return values is positive, it means to read N records
+ /// if return values is negative, it means to skip N records
+ /// if return values is 0, it means end of RG
+ int64_t advise_next(const int64_t current_rg_procesed) {
Review Comment:
total rows that current RG has processed
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
Review Comment:
done
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405597070
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
Review Comment:
I think in general too much code is inlined in these classes.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1451694236
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -336,19 +342,66 @@ class FileReaderImpl : public FileReader {
return ReadRowGroup(i, Iota(reader_->metadata()->num_columns()), table);
}
+ // This is a internal API owned by FileReaderImpl, not exposed in FileReader
+ Status GetRecordBatchReaderWithRowRanges(const std::vector<int>& row_group_indices,
+ const std::vector<int>& column_indices,
+ const std::shared_ptr<std::vector<RowRanges>> & row_ranges_map,
+ std::unique_ptr<RecordBatchReader>* out);
+
+ Status GetRecordBatchReader(const RowRanges& rows_to_return,
+ const std::vector<int>& column_indices,
+ std::unique_ptr<RecordBatchReader>* out) override {
+ const auto metadata = reader_->metadata();
+ // check if the row ranges are valid
+ if (!rows_to_return.IsValid()) {
+ return Status::Invalid("The provided row range is invalid, keep it monotone and non-interleaving: " +
+ rows_to_return.ToString());
+ }
+ // check if the row ranges are within the row group boundaries
+ if (rows_to_return.RowCount() != 0 && rows_to_return.GetRanges().back().end >= metadata->num_rows()) {
+ return Status::Invalid("The provided row range " + rows_to_return.ToString() +
+ " exceeds the number of rows in the file: " +
+ std::to_string(metadata->num_rows()));
+ }
+
+ std::vector<int64_t> split_points;
+ int64_t rows_so_far = 0;
+ for (int i = 0 ; i < metadata->num_row_groups() - 1; i++) {
+ rows_so_far += metadata->RowGroup(i)->num_rows();
+ split_points.push_back(rows_so_far);
+ }
+ // We'll assign a RowRanges for each RG, even if it's not required to return any rows
+ const std::vector<RowRanges> splits = rows_to_return.SplitAt(split_points);
+ // Call row_ranges_map because array index is the row group index
+ const std::shared_ptr<std::vector<RowRanges>> row_ranges_map =
+ std::make_shared<std::vector<RowRanges>>();
+ rows_so_far = 0;
+ std::vector<int> row_group_indices;
+ for (int i = 0 ; i < metadata->num_row_groups(); i++) {
+ row_ranges_map->push_back(splits[i].shift(-rows_so_far));
+ rows_so_far += metadata->RowGroup(i)->num_rows();
+ if (row_ranges_map->at(i).RowCount() > 0)
+ row_group_indices.push_back(i);
+ }
+
+ return GetRecordBatchReaderWithRowRanges(row_group_indices, column_indices, row_ranges_map, out);
Review Comment:
I'm trying to reuse the existing GetRecordBatchReaderWithRowRanges function, so row_group_indices must be kept for other usage. row_ranges_map itself is indexed by row group number, in LeafReader::NextRowGroup , empty RG check is performed by `if (row_ranges.RowCount() != 0)`
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405605267
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) { // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.getRanges()) {
+ if (!row_ranges.isOverlapping(page)) {
+ skip_pages.add(page, false);
+ }
+ }
+
+ /// Since the skipped pages will be slienly skipped without updating
+ /// current_rg_processed_records or records_read_, we need to pre-process the row
+ /// ranges as if these skipped pages never existed
+ adjust_ranges(skip_pages, row_ranges);
+
+ total_rows_to_process = pages.rowCount() - skip_pages.rowCount();
+ }
+
+ /// \brief Return the number of records to read or to skip
+ /// if return values is positive, it means to read N records
+ /// if return values is negative, it means to skip N records
+ /// if return values is 0, it means end of RG
+ int64_t advise_next(const int64_t current_rg_procesed) {
+ if (row_ranges.getRanges().size() == row_range_idx) {
+ return 0;
+ }
+
+ if (row_ranges[row_range_idx].to < current_rg_procesed) {
+ row_range_idx++;
+ if (row_ranges.getRanges().size() == row_range_idx) {
+ // negative, skip the ramaining rows
+ return current_rg_procesed - total_rows_to_process;
+ }
+ }
+
+ if (row_ranges[row_range_idx].from > current_rg_procesed) {
+ // negative, skip
+ return current_rg_procesed - row_ranges[row_range_idx].from;
+ }
+
+ const auto ret = row_ranges[row_range_idx].to - current_rg_procesed + 1;
+ assert(ret > 0);
Review Comment:
assert generally aren't used in the code base. use either DHECK or raise an exception.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1439052028
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) { // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.getRanges()) {
Review Comment:
it requires more code to proper handle the case in which page index is missing. Exception on missing seems to be a simple&easy solution for now
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -512,8 +564,87 @@ class LeafReader : public ColumnReaderImpl {
private:
std::shared_ptr<ChunkedArray> out_;
+
+ void checkAndGetPageRanges(const std::shared_ptr<RowRanges>& row_ranges,
+ std::shared_ptr<RowRanges>& page_ranges) {
+ // check offset exists
+ auto rg_pg_index_reader =
+ ctx_->reader->GetPageIndexReader()->RowGroup(input_->current_row_group());
+
+ if (!rg_pg_index_reader) {
+ throw ParquetException(
+ "Attempting to read with Ranges but Page Index is not found for Row "
+ "Group: " +
+ std::to_string(input_->current_row_group()));
+ }
+ auto offset_index = rg_pg_index_reader->GetOffsetIndex(input_->column_index());
+
+ if (!offset_index) {
Review Comment:
it requires more code to proper handle the case in which page index is missing. Exception on missing seems to be a simple&easy solution for now
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405422564
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -512,8 +564,87 @@ class LeafReader : public ColumnReaderImpl {
private:
std::shared_ptr<ChunkedArray> out_;
+
+ void checkAndGetPageRanges(const std::shared_ptr<RowRanges>& row_ranges,
+ std::shared_ptr<RowRanges>& page_ranges) {
+ // check offset exists
+ auto rg_pg_index_reader =
+ ctx_->reader->GetPageIndexReader()->RowGroup(input_->current_row_group());
+
+ if (!rg_pg_index_reader) {
+ throw ParquetException(
+ "Attempting to read with Ranges but Page Index is not found for Row "
+ "Group: " +
+ std::to_string(input_->current_row_group()));
+ }
+ auto offset_index = rg_pg_index_reader->GetOffsetIndex(input_->column_index());
+
+ if (!offset_index) {
Review Comment:
Is this too strict? Sometimes not all columns have populated offset index.
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) { // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.getRanges()) {
Review Comment:
This requires records not to exceed page boundary. So we may need to check when RecordSkipper cannot be used.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "thisisnic (via GitHub)" <gi...@apache.org>.
thisisnic commented on PR #38867:
URL: https://github.com/apache/arrow/pull/38867#issuecomment-3547439455
Thank you for your contribution. Unfortunately, this
pull request has been marked as stale because it has had no activity in the past 365 days. Please remove the stale label
or comment below, or this PR will be closed in 14 days. Feel free to re-open this if it has been closed in error. If you
do not have repository permissions to reopen the PR, please tag a maintainer.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1406475066
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
Review Comment:
Indeed and it's also counter-productive to use a `shared_ptr` around such a tiny struct.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1451693871
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -512,8 +599,79 @@ class LeafReader : public ColumnReaderImpl {
private:
std::shared_ptr<ChunkedArray> out_;
+
+ void checkAndGetPageRanges(const RowRanges & row_ranges,
+ std::shared_ptr<RowRanges>& page_ranges) const {
Review Comment:
why std::shared_ptr<RowRanges>& -> std::shared_ptr<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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on PR #38867:
URL: https://github.com/apache/arrow/pull/38867#issuecomment-1873386218
> I think it might be worth having more discussion on the issue with representation of RowRanges and on the newly exposed API on the issue to make sure we come to a consensus
The latest version follows the design in https://docs.google.com/document/d/1SeVcYudu6uD9rb9zRAnlLGgdauutaNZlAaS0gVzjkgM/edit
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1451680864
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const IntervalRange& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<IntervalRange>& ranges) { this->ranges = ranges; }
+
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges Intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const IntervalRange& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const IntervalRange& r = right.ranges[i];
+ if (l.IsBefore(r)) {
+ break;
+ } else if (l.IsAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.Add(IntervalRange::Intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ void Add(const IntervalRange& range) {
+ const IntervalRange rangeToAdd = range;
+ if (ranges.size() > 1 && rangeToAdd.start <= ranges.back().end) {
+ throw ParquetException("Ranges must be added in order");
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t RowCount() const {
+ size_t cnt = 0;
+ for (const IntervalRange& range : ranges) {
+ cnt += range.Count();
+ }
+ return cnt;
+ }
+
+ bool IsValid() const {
+ if (ranges.size() == 0) return true;
+ if (ranges[0].start < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].start <= ranges[i - 1].end) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool IsOverlapping(int64_t start, int64_t end) const {
+ const IntervalRange searchRange(start, end);
+ return IsOverlapping(searchRange);
+ }
+
+ bool IsOverlapping(const IntervalRange& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const IntervalRange& r1, const IntervalRange& r2) { return r1.IsBefore(r2); });
+ return it != ranges.end() && !(*it).IsAfter(searchRange);
+ }
+
+ std::vector<IntervalRange>& GetRanges() { return ranges; }
+
+ const std::vector<IntervalRange>& GetRanges() const { return ranges; }
+
+ // Split the ranges into N+1 parts at the given split point, where N =
+ // split_points.size() The RowRows object itself is not modified
+ std::vector<RowRanges> SplitAt(const std::vector<int64_t>& split_points) const {
+ if (split_points.size() == 0) {
+ return {*this};
+ }
+
+ std::vector<RowRanges> result;
+ int64_t last_split_point = -1;
+ for (const int64_t split_point : split_points) {
+ if (split_point <= 0) {
+ throw ParquetException("Invalid split point " + std::to_string(split_point));
+ }
+ if (split_point <= last_split_point) {
+ throw ParquetException("Split points must be in ascending order");
+ }
+ last_split_point = split_point;
+ }
+
+ RowRanges spaces;
+ for (size_t i = 0; i < split_points.size(); ++i) {
+ auto start = i == 0 ? 0 : split_points[i - 1];
+ auto end = split_points[i] - 1;
+ spaces.Add({start, end});
+ }
+ spaces.Add(
+ {split_points[split_points.size() - 1], std::numeric_limits<int64_t>::max()});
+
+ for (IntervalRange space : spaces.GetRanges()) {
+ RowRanges intersection = RowRanges::Intersection(RowRanges(space), *this);
+ result.push_back(intersection);
+ }
+
+ return result;
+ }
+
+ const IntervalRange& operator[](size_t index) const {
+ // check index
+ if (index >= ranges.size() || index < 0) {
+ throw ParquetException("Index out of range");
+ }
+ return ranges[index];
+ }
+
+ RowRanges shift(const int64_t offset) const {
+ RowRanges result;
+ for (const IntervalRange& range : ranges) {
+ result.Add({range.start + offset, range.end + offset});
+ }
+ return result;
+ }
+
+ std::string ToString() const {
+ std::string result = "[";
+ for (const IntervalRange& range : ranges) {
+ result +=
+ "(" + std::to_string(range.start) + ", " + std::to_string(range.end) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+
+ /// The following APIs are to be implemented
+ /// Comment out for now to pass compile
+
+ // // Returns a vector of PageLocations that must be read all to get values for
+ // all included in this range virtual std::vector<PageLocation>
+ // PageIndexesToInclude(const std::vector<PageLocation>& all_pages) = 0; class
+ // Iterator {
+ // virtual std::variant<IntervalRange, BitmapRange, End> NextRange() = 0;
+ // };
+ // virtual std::unique_ptr<Iterator> NewIterator() = 0;
+
+ private:
+ std::vector<IntervalRange> ranges;
+};
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, const RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) {
+ // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.GetRanges()) {
+ if (!row_ranges.IsOverlapping(page)) {
+ skip_pages.Add(page);
+ }
+ }
+
+ /// Since the skipped pages will be silently skipped without updating
+ /// current_rg_processed_records or records_read_, we need to pre-process the row
+ /// ranges as if these skipped pages never existed
+ adjust_ranges(skip_pages, row_ranges);
+
+ total_rows_to_process = pages.RowCount() - skip_pages.RowCount();
+ }
+
+ /// \brief Return the number of records to read or to skip
+ /// if return values is positive, it means to read N records
+ /// if return values is negative, it means to skip N records
+ /// if return values is 0, it means end of RG
+ int64_t advise_next(const int64_t current_rg_processed) {
+ if (row_ranges.GetRanges().size() == row_range_idx) {
+ return 0;
+ }
+
+ if (row_ranges[row_range_idx].end < current_rg_processed) {
+ row_range_idx++;
+ if (row_ranges.GetRanges().size() == row_range_idx) {
+ // negative, skip the ramaining rows
+ return current_rg_processed - total_rows_to_process;
+ }
+ }
+
+ if (row_ranges[row_range_idx].start > current_rg_processed) {
+ // negative, skip
+ return current_rg_processed - row_ranges[row_range_idx].start;
+ }
+
+ const auto ret = row_ranges[row_range_idx].end - current_rg_processed + 1;
+ return ret;
+ }
+
+ private:
+ void adjust_ranges(RowRanges& skip_pages, RowRanges& to_adjust) {
+ size_t skipped_rows = 0;
+ auto iter = to_adjust.GetRanges().begin();
+ auto skip_iter = skip_pages.GetRanges().begin();
+ while (iter != to_adjust.GetRanges().end()) {
+ while (skip_iter != skip_pages.GetRanges().end() && skip_iter->IsBefore(*iter)) {
+ skipped_rows += skip_iter->Count();
+ ++skip_iter;
+ }
+ iter->start -= skipped_rows;
+ iter->end -= skipped_rows;
+ ++iter;
+ }
+ }
+
+ /// Keep copy of ranges, because adjust_ranges() will modify them
+ RowRanges row_ranges;
+
+ size_t row_range_idx = 0;
+ size_t total_rows_to_process = 0;
Review Comment:
Got. The convention seems not enforced everywhere, for example: ReaderContext::reader
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1403408586
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -187,6 +188,11 @@ class PARQUET_EXPORT FileReader {
const std::vector<int>& row_group_indices, const std::vector<int>& column_indices,
std::unique_ptr<::arrow::RecordBatchReader>* out) = 0;
+ virtual ::arrow::Status GetRecordBatchReader(
+ const std::vector<int>& row_group_indices, const std::vector<int>& column_indices,
+ const std::shared_ptr<std::map<int, RowRangesPtr>>& row_ranges_map,
Review Comment:
So, does `RowRanges` index means row-idx in this RowGroup or the row-idx in this file?
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on PR #38867:
URL: https://github.com/apache/arrow/pull/38867#issuecomment-1828959046
hi @emkornfield , thanks for your thorough review. Looks like I should have provided some background to reviewers:
1. We're working on native parquet reading in Clickhouse, which maintains a arrow-13 fork. That's why this PR draft is based on arrow-13.
2. It's my first time to contribute to Arrow, so I'm consulting @wgtmac and @mapleFU offline to check whether this PR is needed by Arrow community and how to revise the PR to meet community standard.
3. @wgtmac has quickly summarized a detailed design doc at https://docs.google.com/document/d/1SeVcYudu6uD9rb9zRAnlLGgdauutaNZlAaS0gVzjkgM. When the design doc has reached consensus, I'll come back to this PR and address all the review comments.
Looking forward to contributing.
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1439052143
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) { // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.getRanges()) {
Review Comment:
When will this case happen?
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
Review Comment:
done
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
Review Comment:
done
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1452325757
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const IntervalRange& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<IntervalRange>& ranges) { this->ranges = ranges; }
+
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
Review Comment:
done
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const IntervalRange& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<IntervalRange>& ranges) { this->ranges = ranges; }
+
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges Intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const IntervalRange& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const IntervalRange& r = right.ranges[i];
+ if (l.IsBefore(r)) {
+ break;
+ } else if (l.IsAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.Add(IntervalRange::Intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ void Add(const IntervalRange& range) {
+ const IntervalRange rangeToAdd = range;
+ if (ranges.size() > 1 && rangeToAdd.start <= ranges.back().end) {
+ throw ParquetException("Ranges must be added in order");
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t RowCount() const {
+ size_t cnt = 0;
+ for (const IntervalRange& range : ranges) {
+ cnt += range.Count();
+ }
+ return cnt;
+ }
+
+ bool IsValid() const {
+ if (ranges.size() == 0) return true;
+ if (ranges[0].start < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].start <= ranges[i - 1].end) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool IsOverlapping(int64_t start, int64_t end) const {
+ const IntervalRange searchRange(start, end);
+ return IsOverlapping(searchRange);
+ }
+
+ bool IsOverlapping(const IntervalRange& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const IntervalRange& r1, const IntervalRange& r2) { return r1.IsBefore(r2); });
+ return it != ranges.end() && !(*it).IsAfter(searchRange);
+ }
+
+ std::vector<IntervalRange>& GetRanges() { return ranges; }
+
+ const std::vector<IntervalRange>& GetRanges() const { return ranges; }
+
+ // Split the ranges into N+1 parts at the given split point, where N =
+ // split_points.size() The RowRows object itself is not modified
+ std::vector<RowRanges> SplitAt(const std::vector<int64_t>& split_points) const {
+ if (split_points.size() == 0) {
+ return {*this};
+ }
+
+ std::vector<RowRanges> result;
+ int64_t last_split_point = -1;
+ for (const int64_t split_point : split_points) {
+ if (split_point <= 0) {
+ throw ParquetException("Invalid split point " + std::to_string(split_point));
+ }
+ if (split_point <= last_split_point) {
+ throw ParquetException("Split points must be in ascending order");
+ }
+ last_split_point = split_point;
+ }
+
+ RowRanges spaces;
+ for (size_t i = 0; i < split_points.size(); ++i) {
+ auto start = i == 0 ? 0 : split_points[i - 1];
+ auto end = split_points[i] - 1;
+ spaces.Add({start, end});
+ }
+ spaces.Add(
+ {split_points[split_points.size() - 1], std::numeric_limits<int64_t>::max()});
+
+ for (IntervalRange space : spaces.GetRanges()) {
+ RowRanges intersection = RowRanges::Intersection(RowRanges(space), *this);
+ result.push_back(intersection);
+ }
+
+ return result;
+ }
+
+ const IntervalRange& operator[](size_t index) const {
+ // check index
+ if (index >= ranges.size() || index < 0) {
+ throw ParquetException("Index out of range");
+ }
+ return ranges[index];
+ }
+
+ RowRanges shift(const int64_t offset) const {
+ RowRanges result;
+ for (const IntervalRange& range : ranges) {
+ result.Add({range.start + offset, range.end + offset});
+ }
+ return result;
+ }
+
+ std::string ToString() const {
+ std::string result = "[";
+ for (const IntervalRange& range : ranges) {
+ result +=
+ "(" + std::to_string(range.start) + ", " + std::to_string(range.end) + "), ";
Review Comment:
done
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405599558
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
Review Comment:
please follow class layout conventions of putting private members at the end.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405596067
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
Review Comment:
please provide documentation on what this method is intended to do.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1439052533
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
Review Comment:
can you elaborate on this? I don't quite get it.
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) { // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.getRanges()) {
+ if (!row_ranges.isOverlapping(page)) {
+ skip_pages.add(page, false);
+ }
+ }
+
+ /// Since the skipped pages will be slienly skipped without updating
+ /// current_rg_processed_records or records_read_, we need to pre-process the row
+ /// ranges as if these skipped pages never existed
+ adjust_ranges(skip_pages, row_ranges);
+
+ total_rows_to_process = pages.rowCount() - skip_pages.rowCount();
+ }
+
+ /// \brief Return the number of records to read or to skip
+ /// if return values is positive, it means to read N records
+ /// if return values is negative, it means to skip N records
+ /// if return values is 0, it means end of RG
+ int64_t advise_next(const int64_t current_rg_procesed) {
+ if (row_ranges.getRanges().size() == row_range_idx) {
+ return 0;
+ }
+
+ if (row_ranges[row_range_idx].to < current_rg_procesed) {
+ row_range_idx++;
+ if (row_ranges.getRanges().size() == row_range_idx) {
+ // negative, skip the ramaining rows
+ return current_rg_procesed - total_rows_to_process;
+ }
+ }
+
+ if (row_ranges[row_range_idx].from > current_rg_procesed) {
+ // negative, skip
+ return current_rg_procesed - row_ranges[row_range_idx].from;
+ }
+
+ const auto ret = row_ranges[row_range_idx].to - current_rg_procesed + 1;
+ assert(ret > 0);
Review Comment:
done
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1439052350
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
Review Comment:
Will do this before merging PR
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1443997791
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
Review Comment:
It would be good to relocate the RowRanges related classes to a separate source file.
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const IntervalRange& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<IntervalRange>& ranges) { this->ranges = ranges; }
+
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges Intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const IntervalRange& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const IntervalRange& r = right.ranges[i];
+ if (l.IsBefore(r)) {
+ break;
+ } else if (l.IsAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.Add(IntervalRange::Intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ void Add(const IntervalRange& range) {
+ const IntervalRange rangeToAdd = range;
+ if (ranges.size() > 1 && rangeToAdd.start <= ranges.back().end) {
+ throw ParquetException("Ranges must be added in order");
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t RowCount() const {
+ size_t cnt = 0;
+ for (const IntervalRange& range : ranges) {
+ cnt += range.Count();
+ }
+ return cnt;
+ }
+
+ bool IsValid() const {
+ if (ranges.size() == 0) return true;
+ if (ranges[0].start < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].start <= ranges[i - 1].end) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool IsOverlapping(int64_t start, int64_t end) const {
+ const IntervalRange searchRange(start, end);
+ return IsOverlapping(searchRange);
+ }
+
+ bool IsOverlapping(const IntervalRange& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const IntervalRange& r1, const IntervalRange& r2) { return r1.IsBefore(r2); });
+ return it != ranges.end() && !(*it).IsAfter(searchRange);
+ }
+
+ std::vector<IntervalRange>& GetRanges() { return ranges; }
+
+ const std::vector<IntervalRange>& GetRanges() const { return ranges; }
+
+ // Split the ranges into N+1 parts at the given split point, where N =
+ // split_points.size() The RowRows object itself is not modified
+ std::vector<RowRanges> SplitAt(const std::vector<int64_t>& split_points) const {
+ if (split_points.size() == 0) {
+ return {*this};
+ }
+
+ std::vector<RowRanges> result;
+ int64_t last_split_point = -1;
+ for (const int64_t split_point : split_points) {
+ if (split_point <= 0) {
+ throw ParquetException("Invalid split point " + std::to_string(split_point));
+ }
+ if (split_point <= last_split_point) {
+ throw ParquetException("Split points must be in ascending order");
+ }
+ last_split_point = split_point;
+ }
+
+ RowRanges spaces;
+ for (size_t i = 0; i < split_points.size(); ++i) {
+ auto start = i == 0 ? 0 : split_points[i - 1];
+ auto end = split_points[i] - 1;
+ spaces.Add({start, end});
+ }
+ spaces.Add(
+ {split_points[split_points.size() - 1], std::numeric_limits<int64_t>::max()});
+
+ for (IntervalRange space : spaces.GetRanges()) {
+ RowRanges intersection = RowRanges::Intersection(RowRanges(space), *this);
+ result.push_back(intersection);
+ }
+
+ return result;
+ }
+
+ const IntervalRange& operator[](size_t index) const {
+ // check index
+ if (index >= ranges.size() || index < 0) {
+ throw ParquetException("Index out of range");
+ }
+ return ranges[index];
+ }
+
+ RowRanges shift(const int64_t offset) const {
+ RowRanges result;
+ for (const IntervalRange& range : ranges) {
+ result.Add({range.start + offset, range.end + offset});
+ }
+ return result;
+ }
+
+ std::string ToString() const {
+ std::string result = "[";
+ for (const IntervalRange& range : ranges) {
+ result +=
+ "(" + std::to_string(range.start) + ", " + std::to_string(range.end) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+
+ /// The following APIs are to be implemented
+ /// Comment out for now to pass compile
+
+ // // Returns a vector of PageLocations that must be read all to get values for
+ // all included in this range virtual std::vector<PageLocation>
+ // PageIndexesToInclude(const std::vector<PageLocation>& all_pages) = 0; class
+ // Iterator {
+ // virtual std::variant<IntervalRange, BitmapRange, End> NextRange() = 0;
+ // };
+ // virtual std::unique_ptr<Iterator> NewIterator() = 0;
+
+ private:
+ std::vector<IntervalRange> ranges;
+};
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, const RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) {
+ // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.GetRanges()) {
+ if (!row_ranges.IsOverlapping(page)) {
+ skip_pages.Add(page);
+ }
+ }
+
+ /// Since the skipped pages will be silently skipped without updating
+ /// current_rg_processed_records or records_read_, we need to pre-process the row
+ /// ranges as if these skipped pages never existed
+ adjust_ranges(skip_pages, row_ranges);
+
+ total_rows_to_process = pages.RowCount() - skip_pages.RowCount();
+ }
+
+ /// \brief Return the number of records to read or to skip
+ /// if return values is positive, it means to read N records
+ /// if return values is negative, it means to skip N records
+ /// if return values is 0, it means end of RG
+ int64_t advise_next(const int64_t current_rg_processed) {
+ if (row_ranges.GetRanges().size() == row_range_idx) {
+ return 0;
+ }
+
+ if (row_ranges[row_range_idx].end < current_rg_processed) {
+ row_range_idx++;
+ if (row_ranges.GetRanges().size() == row_range_idx) {
+ // negative, skip the ramaining rows
+ return current_rg_processed - total_rows_to_process;
+ }
+ }
+
+ if (row_ranges[row_range_idx].start > current_rg_processed) {
+ // negative, skip
+ return current_rg_processed - row_ranges[row_range_idx].start;
+ }
+
+ const auto ret = row_ranges[row_range_idx].end - current_rg_processed + 1;
+ return ret;
+ }
+
+ private:
+ void adjust_ranges(RowRanges& skip_pages, RowRanges& to_adjust) {
+ size_t skipped_rows = 0;
+ auto iter = to_adjust.GetRanges().begin();
+ auto skip_iter = skip_pages.GetRanges().begin();
+ while (iter != to_adjust.GetRanges().end()) {
+ while (skip_iter != skip_pages.GetRanges().end() && skip_iter->IsBefore(*iter)) {
+ skipped_rows += skip_iter->Count();
+ ++skip_iter;
+ }
+ iter->start -= skipped_rows;
+ iter->end -= skipped_rows;
+ ++iter;
+ }
+ }
+
+ /// Keep copy of ranges, because adjust_ranges() will modify them
+ RowRanges row_ranges;
+
+ size_t row_range_idx = 0;
+ size_t total_rows_to_process = 0;
Review Comment:
```suggestion
size_t row_range_idx_ = 0;
size_t total_rows_to_process_ = 0;
```
This is the style convection for class member variables.
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -512,8 +599,79 @@ class LeafReader : public ColumnReaderImpl {
private:
std::shared_ptr<ChunkedArray> out_;
+
+ void checkAndGetPageRanges(const RowRanges & row_ranges,
+ std::shared_ptr<RowRanges>& page_ranges) const {
Review Comment:
```suggestion
void CheckAndGetPageRanges(const RowRanges& row_ranges,
std::shared_ptr<RowRanges>* page_ranges) const {
```
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const IntervalRange& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<IntervalRange>& ranges) { this->ranges = ranges; }
+
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges Intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const IntervalRange& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const IntervalRange& r = right.ranges[i];
+ if (l.IsBefore(r)) {
+ break;
+ } else if (l.IsAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.Add(IntervalRange::Intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ void Add(const IntervalRange& range) {
+ const IntervalRange rangeToAdd = range;
+ if (ranges.size() > 1 && rangeToAdd.start <= ranges.back().end) {
+ throw ParquetException("Ranges must be added in order");
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t RowCount() const {
+ size_t cnt = 0;
+ for (const IntervalRange& range : ranges) {
+ cnt += range.Count();
+ }
+ return cnt;
+ }
+
+ bool IsValid() const {
+ if (ranges.size() == 0) return true;
+ if (ranges[0].start < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].start <= ranges[i - 1].end) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool IsOverlapping(int64_t start, int64_t end) const {
+ const IntervalRange searchRange(start, end);
+ return IsOverlapping(searchRange);
+ }
+
+ bool IsOverlapping(const IntervalRange& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const IntervalRange& r1, const IntervalRange& r2) { return r1.IsBefore(r2); });
+ return it != ranges.end() && !(*it).IsAfter(searchRange);
+ }
+
+ std::vector<IntervalRange>& GetRanges() { return ranges; }
+
+ const std::vector<IntervalRange>& GetRanges() const { return ranges; }
+
+ // Split the ranges into N+1 parts at the given split point, where N =
+ // split_points.size() The RowRows object itself is not modified
+ std::vector<RowRanges> SplitAt(const std::vector<int64_t>& split_points) const {
+ if (split_points.size() == 0) {
+ return {*this};
+ }
+
+ std::vector<RowRanges> result;
+ int64_t last_split_point = -1;
+ for (const int64_t split_point : split_points) {
+ if (split_point <= 0) {
+ throw ParquetException("Invalid split point " + std::to_string(split_point));
+ }
+ if (split_point <= last_split_point) {
+ throw ParquetException("Split points must be in ascending order");
+ }
+ last_split_point = split_point;
+ }
+
+ RowRanges spaces;
+ for (size_t i = 0; i < split_points.size(); ++i) {
+ auto start = i == 0 ? 0 : split_points[i - 1];
+ auto end = split_points[i] - 1;
+ spaces.Add({start, end});
+ }
+ spaces.Add(
+ {split_points[split_points.size() - 1], std::numeric_limits<int64_t>::max()});
+
+ for (IntervalRange space : spaces.GetRanges()) {
+ RowRanges intersection = RowRanges::Intersection(RowRanges(space), *this);
+ result.push_back(intersection);
+ }
+
+ return result;
+ }
+
+ const IntervalRange& operator[](size_t index) const {
+ // check index
+ if (index >= ranges.size() || index < 0) {
+ throw ParquetException("Index out of range");
+ }
+ return ranges[index];
+ }
+
+ RowRanges shift(const int64_t offset) const {
+ RowRanges result;
+ for (const IntervalRange& range : ranges) {
+ result.Add({range.start + offset, range.end + offset});
+ }
+ return result;
+ }
+
+ std::string ToString() const {
+ std::string result = "[";
+ for (const IntervalRange& range : ranges) {
+ result +=
+ "(" + std::to_string(range.start) + ", " + std::to_string(range.end) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+
+ /// The following APIs are to be implemented
+ /// Comment out for now to pass compile
+
+ // // Returns a vector of PageLocations that must be read all to get values for
+ // all included in this range virtual std::vector<PageLocation>
+ // PageIndexesToInclude(const std::vector<PageLocation>& all_pages) = 0; class
+ // Iterator {
+ // virtual std::variant<IntervalRange, BitmapRange, End> NextRange() = 0;
+ // };
+ // virtual std::unique_ptr<Iterator> NewIterator() = 0;
+
+ private:
+ std::vector<IntervalRange> ranges;
Review Comment:
IMHO, `RowRanges` class should hide the implemetation detail and the private member `std::vector<IntervalRange> ranges` should not appear here. If we don't want to pay the cost for virtual class, we can use PIMPL idiom by following example in `parquet/metadata.h`
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
Review Comment:
Should we simply define the IntervalRange class as below and move any functions to an individual utility class?
```
struct IntervalRange {
// inclusive
int64_t start;
// inclusive
int64_t end;
};
struct IntervalRangeUtils {
static IntervalRange Intersection(const IntervalRange& rhs,
const IntervalRange& rhs);
static bool IsBefore(const IntervalRange& lhs, const IntervalRange& rhs);
...
}
```
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -206,6 +219,7 @@ class PARQUET_EXPORT FileReader {
std::shared_ptr<::arrow::RecordBatchReader>* out);
::arrow::Status GetRecordBatchReader(std::shared_ptr<::arrow::RecordBatchReader>* out);
+
Review Comment:
remove the blank line.
##########
cpp/src/parquet/arrow/reader_internal.h:
##########
@@ -109,6 +113,7 @@ struct ReaderContext {
FileColumnIteratorFactory iterator_factory;
bool filter_leaves;
std::shared_ptr<std::unordered_set<int>> included_leaves;
+ std::shared_ptr<std::vector<RowRanges>> row_ranges_map;
Review Comment:
```suggestion
std::shared_ptr<std::vector<RowRanges>> row_ranges;
```
This does not look like a map.
##########
cpp/src/parquet/arrow/reader_internal.h:
##########
@@ -88,11 +89,14 @@ class FileColumnIterator {
int column_index() const { return column_index_; }
+ int current_row_group() const { return current_rg; }
+
protected:
int column_index_;
ParquetFileReader* reader_;
const SchemaDescriptor* schema_;
std::deque<int> row_groups_;
+ int current_rg = 0;
Review Comment:
```suggestion
int current_row_group_;
```
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -72,6 +74,7 @@ using arrow::internal::Iota;
// Help reduce verbosity
using ParquetReader = parquet::ParquetFileReader;
+using parquet::IntervalRange;
Review Comment:
This seems unnecessary
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const IntervalRange& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<IntervalRange>& ranges) { this->ranges = ranges; }
+
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges Intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const IntervalRange& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const IntervalRange& r = right.ranges[i];
+ if (l.IsBefore(r)) {
+ break;
+ } else if (l.IsAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.Add(IntervalRange::Intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ void Add(const IntervalRange& range) {
+ const IntervalRange rangeToAdd = range;
+ if (ranges.size() > 1 && rangeToAdd.start <= ranges.back().end) {
+ throw ParquetException("Ranges must be added in order");
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t RowCount() const {
+ size_t cnt = 0;
+ for (const IntervalRange& range : ranges) {
+ cnt += range.Count();
+ }
+ return cnt;
+ }
+
+ bool IsValid() const {
+ if (ranges.size() == 0) return true;
+ if (ranges[0].start < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].start <= ranges[i - 1].end) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool IsOverlapping(int64_t start, int64_t end) const {
+ const IntervalRange searchRange(start, end);
+ return IsOverlapping(searchRange);
+ }
+
+ bool IsOverlapping(const IntervalRange& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const IntervalRange& r1, const IntervalRange& r2) { return r1.IsBefore(r2); });
+ return it != ranges.end() && !(*it).IsAfter(searchRange);
+ }
+
+ std::vector<IntervalRange>& GetRanges() { return ranges; }
+
+ const std::vector<IntervalRange>& GetRanges() const { return ranges; }
+
+ // Split the ranges into N+1 parts at the given split point, where N =
+ // split_points.size() The RowRows object itself is not modified
+ std::vector<RowRanges> SplitAt(const std::vector<int64_t>& split_points) const {
+ if (split_points.size() == 0) {
+ return {*this};
+ }
+
+ std::vector<RowRanges> result;
+ int64_t last_split_point = -1;
+ for (const int64_t split_point : split_points) {
+ if (split_point <= 0) {
+ throw ParquetException("Invalid split point " + std::to_string(split_point));
+ }
+ if (split_point <= last_split_point) {
+ throw ParquetException("Split points must be in ascending order");
+ }
+ last_split_point = split_point;
+ }
+
+ RowRanges spaces;
+ for (size_t i = 0; i < split_points.size(); ++i) {
+ auto start = i == 0 ? 0 : split_points[i - 1];
+ auto end = split_points[i] - 1;
+ spaces.Add({start, end});
+ }
+ spaces.Add(
+ {split_points[split_points.size() - 1], std::numeric_limits<int64_t>::max()});
+
+ for (IntervalRange space : spaces.GetRanges()) {
+ RowRanges intersection = RowRanges::Intersection(RowRanges(space), *this);
+ result.push_back(intersection);
+ }
+
+ return result;
+ }
+
+ const IntervalRange& operator[](size_t index) const {
+ // check index
+ if (index >= ranges.size() || index < 0) {
+ throw ParquetException("Index out of range");
+ }
+ return ranges[index];
+ }
+
+ RowRanges shift(const int64_t offset) const {
+ RowRanges result;
+ for (const IntervalRange& range : ranges) {
+ result.Add({range.start + offset, range.end + offset});
+ }
+ return result;
+ }
+
+ std::string ToString() const {
+ std::string result = "[";
+ for (const IntervalRange& range : ranges) {
+ result +=
+ "(" + std::to_string(range.start) + ", " + std::to_string(range.end) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+
+ /// The following APIs are to be implemented
+ /// Comment out for now to pass compile
+
+ // // Returns a vector of PageLocations that must be read all to get values for
+ // all included in this range virtual std::vector<PageLocation>
+ // PageIndexesToInclude(const std::vector<PageLocation>& all_pages) = 0; class
+ // Iterator {
+ // virtual std::variant<IntervalRange, BitmapRange, End> NextRange() = 0;
+ // };
+ // virtual std::unique_ptr<Iterator> NewIterator() = 0;
+
+ private:
+ std::vector<IntervalRange> ranges;
+};
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, const RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) {
+ // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.GetRanges()) {
+ if (!row_ranges.IsOverlapping(page)) {
+ skip_pages.Add(page);
+ }
+ }
+
+ /// Since the skipped pages will be silently skipped without updating
+ /// current_rg_processed_records or records_read_, we need to pre-process the row
+ /// ranges as if these skipped pages never existed
+ adjust_ranges(skip_pages, row_ranges);
+
+ total_rows_to_process = pages.RowCount() - skip_pages.RowCount();
+ }
+
+ /// \brief Return the number of records to read or to skip
+ /// if return values is positive, it means to read N records
+ /// if return values is negative, it means to skip N records
+ /// if return values is 0, it means end of RG
+ int64_t advise_next(const int64_t current_rg_processed) {
Review Comment:
```suggestion
int64_t AdviseNext(const int64_t current_rg_processed) {
```
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
Review Comment:
IMO, the above separation would make the public API `RowRanges` looks much simpler to external user and those utility functions are what we need only in the implementation.
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const IntervalRange& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<IntervalRange>& ranges) { this->ranges = ranges; }
+
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges Intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const IntervalRange& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const IntervalRange& r = right.ranges[i];
+ if (l.IsBefore(r)) {
+ break;
+ } else if (l.IsAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.Add(IntervalRange::Intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ void Add(const IntervalRange& range) {
+ const IntervalRange rangeToAdd = range;
+ if (ranges.size() > 1 && rangeToAdd.start <= ranges.back().end) {
+ throw ParquetException("Ranges must be added in order");
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t RowCount() const {
+ size_t cnt = 0;
+ for (const IntervalRange& range : ranges) {
+ cnt += range.Count();
+ }
+ return cnt;
+ }
+
+ bool IsValid() const {
+ if (ranges.size() == 0) return true;
+ if (ranges[0].start < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].start <= ranges[i - 1].end) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool IsOverlapping(int64_t start, int64_t end) const {
+ const IntervalRange searchRange(start, end);
+ return IsOverlapping(searchRange);
+ }
+
+ bool IsOverlapping(const IntervalRange& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const IntervalRange& r1, const IntervalRange& r2) { return r1.IsBefore(r2); });
+ return it != ranges.end() && !(*it).IsAfter(searchRange);
+ }
+
+ std::vector<IntervalRange>& GetRanges() { return ranges; }
+
+ const std::vector<IntervalRange>& GetRanges() const { return ranges; }
+
+ // Split the ranges into N+1 parts at the given split point, where N =
+ // split_points.size() The RowRows object itself is not modified
+ std::vector<RowRanges> SplitAt(const std::vector<int64_t>& split_points) const {
+ if (split_points.size() == 0) {
+ return {*this};
+ }
+
+ std::vector<RowRanges> result;
+ int64_t last_split_point = -1;
+ for (const int64_t split_point : split_points) {
+ if (split_point <= 0) {
+ throw ParquetException("Invalid split point " + std::to_string(split_point));
+ }
+ if (split_point <= last_split_point) {
+ throw ParquetException("Split points must be in ascending order");
+ }
+ last_split_point = split_point;
+ }
+
+ RowRanges spaces;
+ for (size_t i = 0; i < split_points.size(); ++i) {
+ auto start = i == 0 ? 0 : split_points[i - 1];
+ auto end = split_points[i] - 1;
+ spaces.Add({start, end});
+ }
+ spaces.Add(
+ {split_points[split_points.size() - 1], std::numeric_limits<int64_t>::max()});
+
+ for (IntervalRange space : spaces.GetRanges()) {
+ RowRanges intersection = RowRanges::Intersection(RowRanges(space), *this);
+ result.push_back(intersection);
+ }
+
+ return result;
+ }
+
+ const IntervalRange& operator[](size_t index) const {
Review Comment:
ditto
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -512,8 +599,79 @@ class LeafReader : public ColumnReaderImpl {
private:
std::shared_ptr<ChunkedArray> out_;
+
+ void checkAndGetPageRanges(const RowRanges & row_ranges,
+ std::shared_ptr<RowRanges>& page_ranges) const {
+ // check offset exists
+ const auto rg_pg_index_reader =
+ ctx_->reader->GetPageIndexReader()->RowGroup(input_->current_row_group());
+
+ if (!rg_pg_index_reader) {
+ throw ParquetException(
+ "Attempting to read with Ranges but Page Index is not found for Row "
+ "Group: " +
+ std::to_string(input_->current_row_group()));
+ }
+ const auto offset_index = rg_pg_index_reader->GetOffsetIndex(input_->column_index());
+
+ if (!offset_index) {
+ throw ParquetException(
+ "Attempting to read with Ranges but Offset index is not found for "
+ "column: " +
+ field_->name());
+ }
+
+ const auto page_locations = offset_index->page_locations();
+ page_ranges = std::make_shared<RowRanges>();
+ for (size_t i = 0; i < page_locations.size() - 1; i++) {
+ page_ranges->Add(
+ {page_locations[i].first_row_index, page_locations[i + 1].first_row_index - 1});
+ }
+ if (page_locations.size() >= 1) {
+ page_ranges->Add(
+ {page_locations[page_locations.size() - 1].first_row_index,
+ ctx_->reader->metadata()->RowGroup(input_->current_row_group())->num_rows() -
Review Comment:
This cost (`RowGroup(input_->current_row_group())`) may not be negligible if there are many columns to do this repeatedly.
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -17,6 +17,8 @@
#include "parquet/arrow/reader.h"
+#include <parquet/page_index.h>
Review Comment:
```suggestion
#include "parquet/page_index.h"
```
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const IntervalRange& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<IntervalRange>& ranges) { this->ranges = ranges; }
+
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges Intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const IntervalRange& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const IntervalRange& r = right.ranges[i];
+ if (l.IsBefore(r)) {
+ break;
+ } else if (l.IsAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.Add(IntervalRange::Intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ void Add(const IntervalRange& range) {
+ const IntervalRange rangeToAdd = range;
+ if (ranges.size() > 1 && rangeToAdd.start <= ranges.back().end) {
+ throw ParquetException("Ranges must be added in order");
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t RowCount() const {
+ size_t cnt = 0;
+ for (const IntervalRange& range : ranges) {
+ cnt += range.Count();
+ }
+ return cnt;
+ }
+
+ bool IsValid() const {
+ if (ranges.size() == 0) return true;
+ if (ranges[0].start < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].start <= ranges[i - 1].end) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool IsOverlapping(int64_t start, int64_t end) const {
+ const IntervalRange searchRange(start, end);
+ return IsOverlapping(searchRange);
+ }
+
+ bool IsOverlapping(const IntervalRange& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const IntervalRange& r1, const IntervalRange& r2) { return r1.IsBefore(r2); });
+ return it != ranges.end() && !(*it).IsAfter(searchRange);
+ }
+
+ std::vector<IntervalRange>& GetRanges() { return ranges; }
Review Comment:
This one and below expose all ranges as InteveralRange which does not look right. The reason is that they are public APIs but we do not promise all ranges are IntervalRanges.
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -451,6 +504,40 @@ class RowGroupReaderImpl : public RowGroupReader {
// ----------------------------------------------------------------------
// Column reader implementations
+struct RowRangesPageFilter {
Review Comment:
Please add some comment to explain what it does.
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -336,19 +342,66 @@ class FileReaderImpl : public FileReader {
return ReadRowGroup(i, Iota(reader_->metadata()->num_columns()), table);
}
+ // This is a internal API owned by FileReaderImpl, not exposed in FileReader
+ Status GetRecordBatchReaderWithRowRanges(const std::vector<int>& row_group_indices,
+ const std::vector<int>& column_indices,
+ const std::shared_ptr<std::vector<RowRanges>> & row_ranges_map,
+ std::unique_ptr<RecordBatchReader>* out);
+
+ Status GetRecordBatchReader(const RowRanges& rows_to_return,
+ const std::vector<int>& column_indices,
+ std::unique_ptr<RecordBatchReader>* out) override {
+ const auto metadata = reader_->metadata();
+ // check if the row ranges are valid
+ if (!rows_to_return.IsValid()) {
+ return Status::Invalid("The provided row range is invalid, keep it monotone and non-interleaving: " +
+ rows_to_return.ToString());
+ }
+ // check if the row ranges are within the row group boundaries
+ if (rows_to_return.RowCount() != 0 && rows_to_return.GetRanges().back().end >= metadata->num_rows()) {
+ return Status::Invalid("The provided row range " + rows_to_return.ToString() +
+ " exceeds the number of rows in the file: " +
+ std::to_string(metadata->num_rows()));
+ }
+
+ std::vector<int64_t> split_points;
+ int64_t rows_so_far = 0;
+ for (int i = 0 ; i < metadata->num_row_groups() - 1; i++) {
+ rows_so_far += metadata->RowGroup(i)->num_rows();
+ split_points.push_back(rows_so_far);
+ }
+ // We'll assign a RowRanges for each RG, even if it's not required to return any rows
+ const std::vector<RowRanges> splits = rows_to_return.SplitAt(split_points);
+ // Call row_ranges_map because array index is the row group index
+ const std::shared_ptr<std::vector<RowRanges>> row_ranges_map =
+ std::make_shared<std::vector<RowRanges>>();
+ rows_so_far = 0;
+ std::vector<int> row_group_indices;
+ for (int i = 0 ; i < metadata->num_row_groups(); i++) {
+ row_ranges_map->push_back(splits[i].shift(-rows_so_far));
+ rows_so_far += metadata->RowGroup(i)->num_rows();
+ if (row_ranges_map->at(i).RowCount() > 0)
+ row_group_indices.push_back(i);
+ }
+
+ return GetRecordBatchReaderWithRowRanges(row_group_indices, column_indices, row_ranges_map, out);
Review Comment:
Why not merge row_group_indices and row_ranges_map into a map? Thus we can eliminate empty row range of some row groups.
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -451,6 +504,40 @@ class RowGroupReaderImpl : public RowGroupReader {
// ----------------------------------------------------------------------
// Column reader implementations
+struct RowRangesPageFilter {
+ explicit RowRangesPageFilter(const RowRanges& row_ranges_,
Review Comment:
Please remove underscore from input parameters.
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -451,6 +504,40 @@ class RowGroupReaderImpl : public RowGroupReader {
// ----------------------------------------------------------------------
// Column reader implementations
+struct RowRangesPageFilter {
+ explicit RowRangesPageFilter(const RowRanges& row_ranges_,
+ const std::shared_ptr<RowRanges>& page_ranges_)
+ : row_ranges(row_ranges_), page_ranges(page_ranges_) {
+
+ if (page_ranges == nullptr || page_ranges->GetRanges().size() == 0) {
+ throw ParquetException("Page ranges is empty");
+ }
+ }
+
+ bool operator()(const DataPageStats& stats) {
+ ++page_range_idx;
+
+ IntervalRange current_page_range = (*page_ranges)[page_range_idx];
Review Comment:
Could we find some way to not directly assume all ranges are IntervalRange? Probably we can use the iterator of RowRanges and throw when a BitmapRange is obtained for now. Otherwise it would be difficult for future refactoring.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1439051852
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -187,6 +188,11 @@ class PARQUET_EXPORT FileReader {
const std::vector<int>& row_group_indices, const std::vector<int>& column_indices,
std::unique_ptr<::arrow::RecordBatchReader>* out) = 0;
+ virtual ::arrow::Status GetRecordBatchReader(
+ const std::vector<int>& row_group_indices, const std::vector<int>& column_indices,
+ const std::shared_ptr<std::map<int, RowRangesPtr>>& row_ranges_map,
Review Comment:
changed to File based
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405602438
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) { // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.getRanges()) {
+ if (!row_ranges.isOverlapping(page)) {
+ skip_pages.add(page, false);
+ }
+ }
+
+ /// Since the skipped pages will be slienly skipped without updating
Review Comment:
```suggestion
/// Since the skipped pages will be silently skipped without updating
```
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405606499
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
Review Comment:
I don't think we've come to a consensure on these types of acronyms but they are typically discouraged by the C++ style guide. @pitrou does that match your understanding or is there more up-to-date advice on defining them.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405604799
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -187,6 +188,11 @@ class PARQUET_EXPORT FileReader {
const std::vector<int>& row_group_indices, const std::vector<int>& column_indices,
std::unique_ptr<::arrow::RecordBatchReader>* out) = 0;
+ virtual ::arrow::Status GetRecordBatchReader(
+ const std::vector<int>& row_group_indices, const std::vector<int>& column_indices,
+ const std::shared_ptr<std::map<int, RowRangesPtr>>& row_ranges_map,
Review Comment:
Its also not clear to me that we should include both RowGroupIndices and ranges within those indicies. It seems like a cleaner API to only include one or the other (which also makes passing in ranges easier?)
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405594885
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
Review Comment:
Please document the struct.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "thisisnic (via GitHub)" <gi...@apache.org>.
thisisnic closed pull request #38867: GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader
URL: https://github.com/apache/arrow/pull/38867
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1451681328
##########
cpp/src/parquet/arrow/reader_internal.h:
##########
@@ -109,6 +113,7 @@ struct ReaderContext {
FileColumnIteratorFactory iterator_factory;
bool filter_leaves;
std::shared_ptr<std::unordered_set<int>> included_leaves;
+ std::shared_ptr<std::vector<RowRanges>> row_ranges_map;
Review Comment:
named with map because it's indexed by RowGroup number. Anyway I'll changed to row_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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405603852
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) { // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.getRanges()) {
+ if (!row_ranges.isOverlapping(page)) {
+ skip_pages.add(page, false);
+ }
+ }
+
+ /// Since the skipped pages will be slienly skipped without updating
+ /// current_rg_processed_records or records_read_, we need to pre-process the row
+ /// ranges as if these skipped pages never existed
+ adjust_ranges(skip_pages, row_ranges);
+
+ total_rows_to_process = pages.rowCount() - skip_pages.rowCount();
+ }
+
+ /// \brief Return the number of records to read or to skip
+ /// if return values is positive, it means to read N records
Review Comment:
is sign reversal really necessary? Couldn't the reader we have a conention were the first element returned is always records to read and then alternate between inclusion and exclusion?
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405602282
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
Review Comment:
second variable looks off
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on PR #38867:
URL: https://github.com/apache/arrow/pull/38867#issuecomment-1827116945
> !!! this PR is based on [arrow-13](https://issues.apache.org/jira/browse/ARROW-13) for early review
Generally there isn't much benefit to targeting a PR at a maintenance branch, could you retarget to main/master
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on PR #38867:
URL: https://github.com/apache/arrow/pull/38867#issuecomment-1873387179
hi @emkornfield , I've just pushed my refinements to this PR and will go through a review with @wgtmac . It may save you your time if you wait until @wgtmac has approved 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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1403425465
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -187,6 +188,11 @@ class PARQUET_EXPORT FileReader {
const std::vector<int>& row_group_indices, const std::vector<int>& column_indices,
std::unique_ptr<::arrow::RecordBatchReader>* out) = 0;
+ virtual ::arrow::Status GetRecordBatchReader(
+ const std::vector<int>& row_group_indices, const std::vector<int>& column_indices,
+ const std::shared_ptr<std::map<int, RowRangesPtr>>& row_ranges_map,
Review Comment:
row-idx in this RowGroup
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405594492
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -187,6 +188,11 @@ class PARQUET_EXPORT FileReader {
const std::vector<int>& row_group_indices, const std::vector<int>& column_indices,
std::unique_ptr<::arrow::RecordBatchReader>* out) = 0;
+ virtual ::arrow::Status GetRecordBatchReader(
+ const std::vector<int>& row_group_indices, const std::vector<int>& column_indices,
+ const std::shared_ptr<std::map<int, RowRangesPtr>>& row_ranges_map,
Review Comment:
Please document this in the method contract. The way this is being passed (as a shared pointer and values being pointer seems a little strange to me) is there a reason how this contract was arrived at (I need to take a closer look at how Range was arrived at).
I'm not sure if keeping consistency for a brand new method is worth it or if this should return arrow::Result<std::unique_ptr<RecordBatchReader>> at some point we should convert all the methods to return result.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405595767
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
Review Comment:
Generally, the [Google style guide is used](https://arrow.apache.org/docs/developers/cpp/development.html#code-style-linting-and-ci) this means typically method names start with an upper-case character.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405601389
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
Review Comment:
This seems like a relatively heavy-weight represenation if most ranges aren't runs. did you consider other alternatives?
CC @fatemehp
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1439052286
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
Review Comment:
this method is deleted. and I have added some comments to critical classes & methods
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1451680957
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const IntervalRange& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<IntervalRange>& ranges) { this->ranges = ranges; }
+
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges Intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const IntervalRange& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const IntervalRange& r = right.ranges[i];
+ if (l.IsBefore(r)) {
+ break;
+ } else if (l.IsAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.Add(IntervalRange::Intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ void Add(const IntervalRange& range) {
+ const IntervalRange rangeToAdd = range;
+ if (ranges.size() > 1 && rangeToAdd.start <= ranges.back().end) {
+ throw ParquetException("Ranges must be added in order");
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t RowCount() const {
+ size_t cnt = 0;
+ for (const IntervalRange& range : ranges) {
+ cnt += range.Count();
+ }
+ return cnt;
+ }
+
+ bool IsValid() const {
+ if (ranges.size() == 0) return true;
+ if (ranges[0].start < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].start <= ranges[i - 1].end) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool IsOverlapping(int64_t start, int64_t end) const {
+ const IntervalRange searchRange(start, end);
+ return IsOverlapping(searchRange);
+ }
+
+ bool IsOverlapping(const IntervalRange& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const IntervalRange& r1, const IntervalRange& r2) { return r1.IsBefore(r2); });
+ return it != ranges.end() && !(*it).IsAfter(searchRange);
+ }
+
+ std::vector<IntervalRange>& GetRanges() { return ranges; }
+
+ const std::vector<IntervalRange>& GetRanges() const { return ranges; }
+
+ // Split the ranges into N+1 parts at the given split point, where N =
+ // split_points.size() The RowRows object itself is not modified
+ std::vector<RowRanges> SplitAt(const std::vector<int64_t>& split_points) const {
+ if (split_points.size() == 0) {
+ return {*this};
+ }
+
+ std::vector<RowRanges> result;
+ int64_t last_split_point = -1;
+ for (const int64_t split_point : split_points) {
+ if (split_point <= 0) {
+ throw ParquetException("Invalid split point " + std::to_string(split_point));
+ }
+ if (split_point <= last_split_point) {
+ throw ParquetException("Split points must be in ascending order");
+ }
+ last_split_point = split_point;
+ }
+
+ RowRanges spaces;
+ for (size_t i = 0; i < split_points.size(); ++i) {
+ auto start = i == 0 ? 0 : split_points[i - 1];
+ auto end = split_points[i] - 1;
+ spaces.Add({start, end});
+ }
+ spaces.Add(
+ {split_points[split_points.size() - 1], std::numeric_limits<int64_t>::max()});
+
+ for (IntervalRange space : spaces.GetRanges()) {
+ RowRanges intersection = RowRanges::Intersection(RowRanges(space), *this);
+ result.push_back(intersection);
+ }
+
+ return result;
+ }
+
+ const IntervalRange& operator[](size_t index) const {
+ // check index
+ if (index >= ranges.size() || index < 0) {
+ throw ParquetException("Index out of range");
+ }
+ return ranges[index];
+ }
+
+ RowRanges shift(const int64_t offset) const {
+ RowRanges result;
+ for (const IntervalRange& range : ranges) {
+ result.Add({range.start + offset, range.end + offset});
+ }
+ return result;
+ }
+
+ std::string ToString() const {
+ std::string result = "[";
+ for (const IntervalRange& range : ranges) {
+ result +=
+ "(" + std::to_string(range.start) + ", " + std::to_string(range.end) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+
+ /// The following APIs are to be implemented
+ /// Comment out for now to pass compile
+
+ // // Returns a vector of PageLocations that must be read all to get values for
+ // all included in this range virtual std::vector<PageLocation>
+ // PageIndexesToInclude(const std::vector<PageLocation>& all_pages) = 0; class
+ // Iterator {
+ // virtual std::variant<IntervalRange, BitmapRange, End> NextRange() = 0;
+ // };
+ // virtual std::unique_ptr<Iterator> NewIterator() = 0;
+
+ private:
+ std::vector<IntervalRange> ranges;
+};
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, const RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) {
+ // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.GetRanges()) {
+ if (!row_ranges.IsOverlapping(page)) {
+ skip_pages.Add(page);
+ }
+ }
+
+ /// Since the skipped pages will be silently skipped without updating
+ /// current_rg_processed_records or records_read_, we need to pre-process the row
+ /// ranges as if these skipped pages never existed
+ adjust_ranges(skip_pages, row_ranges);
+
+ total_rows_to_process = pages.RowCount() - skip_pages.RowCount();
+ }
+
+ /// \brief Return the number of records to read or to skip
+ /// if return values is positive, it means to read N records
+ /// if return values is negative, it means to skip N records
+ /// if return values is 0, it means end of RG
+ int64_t advise_next(const int64_t current_rg_processed) {
+ if (row_ranges.GetRanges().size() == row_range_idx) {
+ return 0;
+ }
+
+ if (row_ranges[row_range_idx].end < current_rg_processed) {
+ row_range_idx++;
+ if (row_ranges.GetRanges().size() == row_range_idx) {
+ // negative, skip the ramaining rows
+ return current_rg_processed - total_rows_to_process;
+ }
+ }
+
+ if (row_ranges[row_range_idx].start > current_rg_processed) {
+ // negative, skip
+ return current_rg_processed - row_ranges[row_range_idx].start;
+ }
+
+ const auto ret = row_ranges[row_range_idx].end - current_rg_processed + 1;
+ return ret;
+ }
+
+ private:
+ void adjust_ranges(RowRanges& skip_pages, RowRanges& to_adjust) {
+ size_t skipped_rows = 0;
+ auto iter = to_adjust.GetRanges().begin();
+ auto skip_iter = skip_pages.GetRanges().begin();
+ while (iter != to_adjust.GetRanges().end()) {
+ while (skip_iter != skip_pages.GetRanges().end() && skip_iter->IsBefore(*iter)) {
+ skipped_rows += skip_iter->Count();
+ ++skip_iter;
+ }
+ iter->start -= skipped_rows;
+ iter->end -= skipped_rows;
+ ++iter;
+ }
+ }
+
+ /// Keep copy of ranges, because adjust_ranges() will modify them
+ RowRanges row_ranges;
+
+ size_t row_range_idx = 0;
+ size_t total_rows_to_process = 0;
Review Comment:
I'll change all member variables in my PR
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1406978633
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
Review Comment:
I guess the reason of using a shared pointer here is that the row ranges are shared by all column readers.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405598724
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
+
namespace internal {
+class PARQUET_EXPORT RecordSkipper {
+ public:
+ RecordSkipper(RowRanges& pages, RowRanges& row_ranges_)
+ : row_ranges(row_ranges_) { // copy row_ranges
+ RowRanges will_process_pages, skip_pages;
+ for (auto& page : pages.getRanges()) {
+ if (!row_ranges.isOverlapping(page)) {
+ skip_pages.add(page, false);
+ }
+ }
+
+ /// Since the skipped pages will be slienly skipped without updating
+ /// current_rg_processed_records or records_read_, we need to pre-process the row
+ /// ranges as if these skipped pages never existed
+ adjust_ranges(skip_pages, row_ranges);
+
+ total_rows_to_process = pages.rowCount() - skip_pages.rowCount();
+ }
+
+ /// \brief Return the number of records to read or to skip
+ /// if return values is positive, it means to read N records
+ /// if return values is negative, it means to skip N records
+ /// if return values is 0, it means end of RG
+ int64_t advise_next(const int64_t current_rg_procesed) {
Review Comment:
its not claear to me what the input parameter is here?
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "emkornfield (via GitHub)" <gi...@apache.org>.
emkornfield commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1405601897
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
Review Comment:
please document intended semantics here. Is this method really necessary? why wouldn't it consider partial 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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #38867:
URL: https://github.com/apache/arrow/pull/38867#issuecomment-1824433736
:warning: GitHub issue #38865 **has been automatically assigned in GitHub** to PR creator.
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1444429627
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
Review Comment:
`IntervalRange({-1, -1}).Count() == 1` ? Would this be weird?
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const IntervalRange& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<IntervalRange>& ranges) { this->ranges = ranges; }
+
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges Intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const IntervalRange& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const IntervalRange& r = right.ranges[i];
+ if (l.IsBefore(r)) {
+ break;
+ } else if (l.IsAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.Add(IntervalRange::Intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ void Add(const IntervalRange& range) {
+ const IntervalRange rangeToAdd = range;
+ if (ranges.size() > 1 && rangeToAdd.start <= ranges.back().end) {
+ throw ParquetException("Ranges must be added in order");
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t RowCount() const {
+ size_t cnt = 0;
+ for (const IntervalRange& range : ranges) {
+ cnt += range.Count();
+ }
+ return cnt;
+ }
+
+ bool IsValid() const {
+ if (ranges.size() == 0) return true;
+ if (ranges[0].start < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].start <= ranges[i - 1].end) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool IsOverlapping(int64_t start, int64_t end) const {
+ const IntervalRange searchRange(start, end);
+ return IsOverlapping(searchRange);
+ }
+
+ bool IsOverlapping(const IntervalRange& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const IntervalRange& r1, const IntervalRange& r2) { return r1.IsBefore(r2); });
+ return it != ranges.end() && !(*it).IsAfter(searchRange);
+ }
+
+ std::vector<IntervalRange>& GetRanges() { return ranges; }
+
+ const std::vector<IntervalRange>& GetRanges() const { return ranges; }
+
+ // Split the ranges into N+1 parts at the given split point, where N =
+ // split_points.size() The RowRows object itself is not modified
+ std::vector<RowRanges> SplitAt(const std::vector<int64_t>& split_points) const {
+ if (split_points.size() == 0) {
+ return {*this};
+ }
+
+ std::vector<RowRanges> result;
+ int64_t last_split_point = -1;
+ for (const int64_t split_point : split_points) {
+ if (split_point <= 0) {
+ throw ParquetException("Invalid split point " + std::to_string(split_point));
+ }
+ if (split_point <= last_split_point) {
+ throw ParquetException("Split points must be in ascending order");
+ }
+ last_split_point = split_point;
+ }
+
+ RowRanges spaces;
+ for (size_t i = 0; i < split_points.size(); ++i) {
+ auto start = i == 0 ? 0 : split_points[i - 1];
+ auto end = split_points[i] - 1;
+ spaces.Add({start, end});
+ }
+ spaces.Add(
+ {split_points[split_points.size() - 1], std::numeric_limits<int64_t>::max()});
+
+ for (IntervalRange space : spaces.GetRanges()) {
+ RowRanges intersection = RowRanges::Intersection(RowRanges(space), *this);
+ result.push_back(intersection);
+ }
+
+ return result;
+ }
+
+ const IntervalRange& operator[](size_t index) const {
+ // check index
+ if (index >= ranges.size() || index < 0) {
+ throw ParquetException("Index out of range");
+ }
+ return ranges[index];
+ }
+
+ RowRanges shift(const int64_t offset) const {
+ RowRanges result;
+ for (const IntervalRange& range : ranges) {
+ result.Add({range.start + offset, range.end + offset});
+ }
+ return result;
+ }
+
+ std::string ToString() const {
+ std::string result = "[";
+ for (const IntervalRange& range : ranges) {
+ result +=
+ "(" + std::to_string(range.start) + ", " + std::to_string(range.end) + "), ";
Review Comment:
Uses `range.ToString()`?
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
+
+ bool IsBefore(const IntervalRange& other) const { return end < other.start; }
+
+ bool IsAfter(const IntervalRange& other) const { return start > other.end; }
+
+ bool IsOverlap(const IntervalRange& other) const {
+ return !IsBefore(other) && !IsAfter(other);
+ }
+
+ bool IsValid() const { return start >= 0 && end >= 0 && end >= start; }
+
+ std::string ToString() const {
+ return "[" + std::to_string(start) + ", " + std::to_string(end) + "]";
+ }
+
+ // inclusive
+ int64_t start;
+ // inclusive
+ int64_t end;
+};
+
+struct BitmapRange {
+ int64_t offset;
+ // zero added to, if there are less than 64 elements left in the column.
+ uint64_t bitmap;
+};
+
+struct End {};
+
+// Represent a set of ranges to read. The ranges are sorted and non-overlapping.
+class RowRanges {
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const IntervalRange& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<IntervalRange>& ranges) { this->ranges = ranges; }
+
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
Review Comment:
= default?
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1439052693
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,8 +303,274 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+struct Range {
+ static Range unionRange(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to + 1 >= right.from) {
+ return {left.from, std::max(left.to, right.to)};
+ }
+ } else if (right.to + 1 >= left.from) {
+ return {right.from, std::max(left.to, right.to)};
+ }
+ return {-1, -1};
+ }
+
+ static Range intersection(const Range& left, const Range& right) {
+ if (left.from <= right.from) {
+ if (left.to >= right.from) {
+ return {right.from, std::min(left.to, right.to)};
+ }
+ } else if (right.to >= left.from) {
+ return {left.from, std::min(left.to, right.to)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ int64_t from;
+ int64_t to;
+
+ Range(const int64_t from_, const int64_t to_) : from(from_), to(to_) {
+ assert(from <= to);
+ }
+
+ size_t count() const { return to - from + 1; }
+
+ bool isBefore(const Range& other) const { return to < other.from; }
+
+ bool isAfter(const Range& other) const { return from > other.to; }
+
+ bool isOverlap(const Range& other) const { return !isBefore(other) && !isAfter(other); }
+
+ std::string toString() const {
+ return "[" + std::to_string(from) + ", " + std::to_string(to) + "]";
+ }
+};
+
+class RowRanges {
+ std::vector<Range> ranges;
+
+ public:
+ RowRanges() = default;
+
+ explicit RowRanges(const Range& range) { ranges.push_back(range); }
+
+ RowRanges(const std::vector<Range>& ranges) { this->ranges = ranges; }
+
+ // copy cstr
+ RowRanges(const RowRanges& other) { ranges = other.ranges; }
+
+ RowRanges(RowRanges&& other) noexcept { ranges = std::move(other.ranges); }
+
+ static RowRanges unionRanges(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+ auto it1 = left.ranges.begin();
+ auto it2 = right.ranges.begin();
+ if (it2 != right.ranges.end()) {
+ Range range2 = *it2;
+ while (it1 != left.ranges.end()) {
+ Range range1 = *it1;
+ if (range1.isAfter(range2)) {
+ result.add(range2);
+ range2 = range1;
+ const auto tmp = it1;
+ it1 = it2;
+ it2 = tmp;
+ } else {
+ result.add(range1);
+ }
+ ++it1;
+ }
+ result.add(range2);
+ } else {
+ it2 = it1;
+ }
+ while (it2 != right.ranges.end()) {
+ result.add(*it2);
+ ++it2;
+ }
+
+ return result;
+ }
+
+ static RowRanges intersection(const RowRanges& left, const RowRanges& right) {
+ RowRanges result;
+
+ size_t rightIndex = 0;
+ for (const Range& l : left.ranges) {
+ for (size_t i = rightIndex, n = right.ranges.size(); i < n; ++i) {
+ const Range& r = right.ranges[i];
+ if (l.isBefore(r)) {
+ break;
+ } else if (l.isAfter(r)) {
+ rightIndex = i + 1;
+ continue;
+ }
+ result.add(Range::intersection(l, r));
+ }
+ }
+
+ return result;
+ }
+
+ RowRanges slice(const int64_t from, const int64_t to) const {
+ RowRanges result;
+ for (const Range& range : ranges) {
+ if (range.from >= from && range.to <= to) {
+ result.add(range);
+ }
+ }
+ return result;
+ }
+
+ void add(const Range& range, bool merge = true) {
+ Range rangeToAdd = range;
+ if (merge) {
+ for (int i = static_cast<int>(ranges.size()) - 1; i >= 0; --i) {
+ Range last = ranges[i];
+ if (last.isAfter(range)) {
+ throw ParquetException(range.toString() + " cannot be added to " +
+ this->toString());
+ }
+ const Range u = Range::unionRange(last, rangeToAdd);
+ if (u.from == -1 && u.to == -1) {
+ break;
+ }
+ rangeToAdd = u;
+ ranges.erase(ranges.begin() + i);
+ }
+ } else {
+ if (ranges.size() > 1) assert(rangeToAdd.from > ranges.back().to);
+ }
+ ranges.push_back(rangeToAdd);
+ }
+
+ size_t rowCount() const {
+ size_t cnt = 0;
+ for (const Range& range : ranges) {
+ cnt += range.count();
+ }
+ return cnt;
+ }
+
+ bool isValid() const {
+ if (ranges.size() == 0) {
+ return false;
+ }
+ if (ranges[0].from < 0) {
+ return false;
+ }
+ for (size_t i = 1; i < ranges.size(); i++) {
+ if (ranges[i].from <= ranges[i - 1].to) {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ bool isOverlapping(int64_t from, int64_t to) const {
+ const Range searchRange(from, to);
+ return isOverlapping(searchRange);
+ }
+
+ bool isOverlapping(const Range& searchRange) const {
+ auto it = std::lower_bound(
+ ranges.begin(), ranges.end(), searchRange,
+ [](const Range& r1, const Range& r2) { return r1.isBefore(r2); });
+ return it != ranges.end() && !(*it).isAfter(searchRange);
+ }
+
+ std::vector<Range>& getRanges() { return ranges; }
+
+ const Range& operator[](size_t index) const { return ranges[index]; }
+
+ std::string toString() const {
+ std::string result = "[";
+ for (const Range& range : ranges) {
+ result +=
+ "(" + std::to_string(range.from) + ", " + std::to_string(range.to) + "), ";
+ }
+ if (!ranges.empty()) {
+ result = result.substr(0, result.size() - 2);
+ }
+ result += "]";
+ return result;
+ }
+};
+
+using RowRangesPtr = std::shared_ptr<RowRanges>;
Review Comment:
removed such kind of type alias (but still, I don't know why it is not encouraged?)
--
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
Re: [PR] GH-38865 [C++][Parquet] support passing a RowRange to RecordBatchReader [arrow]
Posted by "binmahone (via GitHub)" <gi...@apache.org>.
binmahone commented on code in PR #38867:
URL: https://github.com/apache/arrow/pull/38867#discussion_r1452254496
##########
cpp/src/parquet/column_reader.h:
##########
@@ -302,7 +300,284 @@ class TypedColumnReader : public ColumnReader {
int32_t* dict_len) = 0;
};
+// Represent a range to read. The range is inclusive on both ends.
+struct IntervalRange {
+ static IntervalRange Intersection(const IntervalRange& left,
+ const IntervalRange& right) {
+ if (left.start <= right.start) {
+ if (left.end >= right.start) {
+ return {right.start, std::min(left.end, right.end)};
+ }
+ } else if (right.end >= left.start) {
+ return {left.start, std::min(left.end, right.end)};
+ }
+ return {-1, -1}; // Return a default Range object if no intersection range found
+ }
+
+ IntervalRange(const int64_t start_, const int64_t end_) : start(start_), end(end_) {
+ if (start > end) {
+ throw ParquetException("Invalid range with start: " + std::to_string(start) +
+ " and end: " + std::to_string(end));
+ }
+ }
+
+ size_t Count() const { return end - start + 1; }
Review Comment:
added a check of validness
--
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