You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "westonpace (via GitHub)" <gi...@apache.org> on 2023/07/19 22:30:07 UTC
[GitHub] [arrow] westonpace opened a new pull request, #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
westonpace opened a new pull request, #36779:
URL: https://github.com/apache/arrow/pull/36779
### Rationale for this change
The rationale is described in #36778
### What changes are included in this PR?
New methods are added to `parquet::arrow::FileReader` which read the file asynchronously and respect the batch size property. In addition, these new methods are a bit simpler than `GetRecordBatchGenerator` as they are able to reuse a lot of the code in the synchronous methods.
### Are these changes tested?
Yes, I've added new unit tests.
### Are there any user-facing changes?
Yes, there are new methods available. There should be no breaking changes to any existing 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
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1275455867
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,123 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
Review Comment:
Yes.
> If it's in the end of a File, it would return a Finish?
This can also happen in the middle of a file if the user does not ask for all row groups.
The scan node always calls ReadRowGroupsAsync with a single row group so it is always a possibility (but this is ok because the scan node can handle it).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1269594472
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
Review Comment:
I could also handle this in the scan node. If I get more batches than expected then I can assume this happened and raise an error. Then I don't need the flag in the parquet reader. I wasn't sure if it was helpful for the user of the parquet reader to know this was a possibility or not. Let me know which you'd prefer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] wgtmac commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1299514336
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +250,63 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
Review Comment:
cpp/src/arrow/util/async_generator_fwd.h has defined AsyncGenerator below, should we reuse it?
```cpp
template <typename T>
using AsyncGenerator = std::function<Future<T>()>;
```
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1285,127 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::vector<std::shared_ptr<ChunkedArray>> current_cols;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ std::vector<Future<>> column_futs;
+ column_futs.reserve(state_->column_readers.size());
+ for (std::size_t column_index = 0; column_index < state_->column_readers.size();
+ column_index++) {
+ const auto& column_reader = state_->column_readers[column_index];
+ column_futs.push_back(
+ column_reader
+ ->NextBatchAsync(rows_in_batch, state_->io_executor, state_->cpu_executor)
+ .Then([state = state_,
+ column_index](const std::shared_ptr<ChunkedArray>& chunked_array) {
+ state->current_cols[column_index] = chunked_array;
+ }));
+ }
+ Future<> all_columns_finished_fut = ::arrow::AllFinished(std::move(column_futs));
+
+ // Grab the first batch of data and return it. If there is more than one batch then
+ // throw the reamining batches into overflow and they will be fetched on the next call
Review Comment:
```suggestion
// throw the remaining batches into overflow and they will be fetched on the next call
```
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -2514,6 +2516,137 @@ TEST(TestArrowReadWrite, GetRecordBatchGenerator) {
}
}
+TEST(TestArrowReadWrite, ReadRowGroupsAsync) {
+ constexpr int kNumRows = 1024;
+ constexpr int kRowGroupSize = 512;
+ constexpr int kNumColumns = 2;
+
+ std::shared_ptr<Table> table;
+ ASSERT_NO_FATAL_FAILURE(MakeDoubleTable(kNumColumns, kNumRows, 1, &table));
+
+ std::shared_ptr<Buffer> buffer;
+ ASSERT_NO_FATAL_FAILURE(WriteTableToBuffer(table, kRowGroupSize,
+ default_arrow_writer_properties(), &buffer));
+
+ for (std::vector<int> row_groups : std::vector<std::vector<int>>{{}, {0}, {0, 1}}) {
+ ARROW_SCOPED_TRACE("# row_groups = ", row_groups.size());
+ int32_t expected_total_rows = static_cast<int32_t>(row_groups.size()) * kRowGroupSize;
+
+ for (std::vector<int> columns : std::vector<std::vector<int>>{{}, {0}, {0, 1}}) {
+ ARROW_SCOPED_TRACE("# columns = ", columns.size());
+
+ for (int row_group_size : {128, 512, 1024, 2048}) {
Review Comment:
nit: row_group_size -> batch_size
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +250,63 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = true) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, decompression is also done on the I/O thread
+ /// which may not be ideal.
+ ///
+ /// The returned generator will respect the batch size set in the ArrowReaderProperties.
+ /// Batches will not be larger than the given batch size. However, batches may be
+ /// smaller. This can happen, for example, when there is not enough data or when a
+ /// string column is too large to fit into a single batch. The parameter
+ /// `allow_sliced_batches` can be set to false to disallow this later case. This can be
+ /// useful when you need to know exactly how many batches you will get from the
+ /// operation before you start.
+ ///
+ /// The returned generator is not async-reentrant. You must not fetch another future
+ /// from it until the previously fetched future has completed.
+ ///
+ /// Note: When reading multiple row groups there is no guarantee you will get one
+ /// record batch per row group. Data from multiple row groups could get combined into
+ /// a single batch.
+ ///
+ /// Note: If a row group has 0 rows it will effectively be ignored. If you are only
+ /// reading empty row groups then the returned generator will immediately finish. This
+ /// method should never generate an empty record batch.
+ ///
+ /// The I/O executor is obtained from the I/O context in the reader properties.
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param column_indices indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If false, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
Review Comment:
It does not return arrow::Result, so the docstring would be good to include expected return value, especially when `allow_sliced_batches` is false.
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -316,6 +374,14 @@ class PARQUET_EXPORT ColumnReader {
// the data available in the file.
virtual ::arrow::Status NextBatch(int64_t batch_size,
std::shared_ptr<::arrow::ChunkedArray>* out) = 0;
+
+ // Overload of NextBatch that returns a Result
+ virtual ::arrow::Result<std::shared_ptr<::arrow::ChunkedArray>> NextBatch(
Review Comment:
Is it required to expose this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] wgtmac commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1270811098
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
Review Comment:
It does not hurt to add this option. So I am fine with keeping it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1296154200
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
Review Comment:
I was attempting to maintain parity with the synchronous methods above. I only need one of these four methods and so if you'd prefer I'm happy to scope this down.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304712231
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +250,63 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = true) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, decompression is also done on the I/O thread
+ /// which may not be ideal.
+ ///
+ /// The returned generator will respect the batch size set in the ArrowReaderProperties.
+ /// Batches will not be larger than the given batch size. However, batches may be
+ /// smaller. This can happen, for example, when there is not enough data or when a
+ /// string column is too large to fit into a single batch. The parameter
+ /// `allow_sliced_batches` can be set to false to disallow this later case. This can be
+ /// useful when you need to know exactly how many batches you will get from the
+ /// operation before you start.
+ ///
+ /// The returned generator is not async-reentrant. You must not fetch another future
+ /// from it until the previously fetched future has completed.
+ ///
+ /// Note: When reading multiple row groups there is no guarantee you will get one
+ /// record batch per row group. Data from multiple row groups could get combined into
+ /// a single batch.
+ ///
+ /// Note: If a row group has 0 rows it will effectively be ignored. If you are only
+ /// reading empty row groups then the returned generator will immediately finish. This
+ /// method should never generate an empty record batch.
+ ///
+ /// The I/O executor is obtained from the I/O context in the reader properties.
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param column_indices indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If false, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
Review Comment:
I will add something here.
> Would Future in arrow actually include Status?
Yes. A Future in Arrow has an implicit Status. So it is generally invalid to see something like `Future<Result<T>>`. There are a few places where we do return `Result<Future<T>>`. This typically indicates that "something can fail quickly, in the synchronous part" (e.g. the thread pool is already shut down) and any failure in the asynchronous portion will be communicated in the `Future`. However, I generally still prefer to just return `Future<T>` in these cases (there is a utility `DeferResult` which will convert `Result<Future<T>>` to `Future<T>`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304748443
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1285,127 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::vector<std::shared_ptr<ChunkedArray>> current_cols;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ std::vector<Future<>> column_futs;
+ column_futs.reserve(state_->column_readers.size());
+ for (std::size_t column_index = 0; column_index < state_->column_readers.size();
+ column_index++) {
+ const auto& column_reader = state_->column_readers[column_index];
+ column_futs.push_back(
+ column_reader
+ ->NextBatchAsync(rows_in_batch, state_->io_executor, state_->cpu_executor)
+ .Then([state = state_,
+ column_index](const std::shared_ptr<ChunkedArray>& chunked_array) {
+ state->current_cols[column_index] = chunked_array;
+ }));
+ }
+ Future<> all_columns_finished_fut = ::arrow::AllFinished(std::move(column_futs));
+
+ // Grab the first batch of data and return it. If there is more than one batch then
+ // throw the remaining batches into overflow and they will be fetched on the next call
+ return all_columns_finished_fut.Then(
+ [state = state_, rows_in_batch]() -> Result<std::shared_ptr<RecordBatch>> {
+ std::shared_ptr<Table> table =
+ Table::Make(state->schema, state->current_cols, rows_in_batch);
+ ::arrow::TableBatchReader batch_reader(*table);
+ std::shared_ptr<RecordBatch> first;
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> next_batch,
+ batch_reader.Next());
+ if (!next_batch) {
+ break;
+ }
+ if (first) {
+ if (!state->allow_sliced_batches) {
+ return Status::Invalid(
+ "The setting allow_sliced_batches is set to false and data was "
+ "encountered that was too large to fit in a single batch.");
+ }
+ state->overflow.push(std::move(next_batch));
+ } else {
+ first = std::move(next_batch);
+ }
+ }
+ // If there were no rows of data we should have been exhausted. Otherwise we
+ // should have gotten at least one batch.
+ DCHECK(!!first);
+ return first;
+ });
+ }
+
+ private:
+ std::shared_ptr<AsyncBatchGeneratorState> state_;
+};
+
+Result<FileReaderImpl::AsyncBatchGenerator> FileReaderImpl::DoReadRowGroupsAsync(
+ const std::vector<int>& row_groups, const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor, bool allow_sliced_batches) {
+ RETURN_NOT_OK(BoundsCheck(row_groups, column_indices));
+
+ if (reader_properties_.pre_buffer()) {
+ BEGIN_PARQUET_CATCH_EXCEPTIONS
+ parquet_reader()->PreBuffer(row_groups, column_indices,
+ reader_properties_.io_context(),
+ reader_properties_.cache_options());
+ END_PARQUET_CATCH_EXCEPTIONS
+ }
+
+ auto generator_state = std::make_shared<AsyncBatchGeneratorState>();
+ generator_state->io_executor = reader_properties_.io_context().executor();
+ generator_state->cpu_executor = cpu_executor;
+ generator_state->use_threads = reader_properties_.use_threads();
Review Comment:
Final question, it's not related to correctness, but it's a bit confusing, I've seen the comment:
```
/// This method ignores the use_threads property of the ArrowReaderProperties. It will
/// always create a task for each column. To run without threads you should use a
/// serial executor as the CPU executor.
```
So why is `use_threads` introduced 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
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1276855247
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,123 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ Future<std::vector<std::shared_ptr<ChunkedArray>>> chunked_arrays_fut =
+ ::arrow::internal::OptionalParallelForAsync(
Review Comment:
Yes, thank you. I have now fixed this in https://github.com/apache/arrow/pull/36779/commits/5e38b00103e1fdd970de44b863acf8f45e416e3a
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] wgtmac commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1270814529
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -2514,6 +2517,137 @@ TEST(TestArrowReadWrite, GetRecordBatchGenerator) {
}
}
+TEST(TestArrowReadWrite, ReadRowGroupsAsync) {
+ constexpr int kNumRows = 1024;
+ constexpr int kRowGroupSize = 512;
+ constexpr int kNumColumns = 2;
+
+ std::shared_ptr<Table> table;
+ ASSERT_NO_FATAL_FAILURE(MakeDoubleTable(kNumColumns, kNumRows, 1, &table));
+
+ std::shared_ptr<Buffer> buffer;
+ ASSERT_NO_FATAL_FAILURE(WriteTableToBuffer(table, kRowGroupSize,
+ default_arrow_writer_properties(), &buffer));
+
+ for (std::vector<int> row_groups : std::vector<std::vector<int>>{{}, {0}, {0, 1}}) {
+ ARROW_SCOPED_TRACE("# row_groups = ", row_groups.size());
+ int32_t expected_total_rows = static_cast<int32_t>(row_groups.size()) * kRowGroupSize;
+
+ for (std::vector<int> columns : std::vector<std::vector<int>>{{}, {0}, {0, 1}}) {
+ ARROW_SCOPED_TRACE("# columns = ", columns.size());
+
+ for (int row_group_size : {128, 512, 1024, 2048}) {
+ ARROW_SCOPED_TRACE("row_group_size = ", row_group_size);
+
+ ArrowReaderProperties properties = default_arrow_reader_properties();
+ properties.set_batch_size(row_group_size);
+ std::unique_ptr<FileReader> unique_reader;
+ FileReaderBuilder builder;
+ ASSERT_OK(builder.Open(std::make_shared<BufferReader>(buffer)));
+ ASSERT_OK(builder.properties(properties)->Build(&unique_reader));
+ auto batch_generator = unique_reader->ReadRowGroupsAsync(
+ row_groups, columns, ::arrow::internal::GetCpuThreadPool());
+
+ int64_t num_expected_batches =
Review Comment:
My concern is that this assumption may no longer hold if filter pushdown of page level (or even row level) has been implemented. In that case, it is difficult to estimate how many rows will be read in advance, which results in unpredictable number of batches.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1296270804
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -3963,6 +4096,88 @@ TEST(TestArrowReaderAdHoc, LARGE_MEMORY_TEST(LargeStringColumn)) {
AssertTablesEqual(*table, *batched_table, /*same_chunk_layout=*/false);
}
+TEST(TestArrowReaderAdHoc, LARGE_MEMORY_TEST(LargeStringValue)) {
+ // ARROW-3762
+ ::arrow::StringBuilder builder;
+ // 16 rows of 256MiB bytes each is 4GiB. This will get put into
+ // 3 chunks of 7 rows, 7 rows, and 2 rows.
+ constexpr std::int32_t kValueSize = 1 << 28;
+ constexpr std::int32_t kNumRows = 4;
+ constexpr std::int32_t kNumChunks = 4;
+ std::vector<std::shared_ptr<Array>> chunks;
+ std::vector<uint8_t> value(kValueSize, '0');
+ for (int chunk_idx = 0; chunk_idx < kNumChunks; chunk_idx++) {
+ ASSERT_OK(builder.Resize(kNumRows));
+ ASSERT_OK(builder.ReserveData(kNumRows * kValueSize));
+ for (int64_t i = 0; i < kNumRows; ++i) {
+ builder.UnsafeAppend(value.data(), kValueSize);
+ }
+ std::shared_ptr<Array> array;
+ ASSERT_OK(builder.Finish(&array));
+ chunks.push_back(std::move(array));
+ }
+
+ // Eaglerly free up memory
+ value.clear();
Review Comment:
I put `value` in its own scope block.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1296270577
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -3963,6 +4096,88 @@ TEST(TestArrowReaderAdHoc, LARGE_MEMORY_TEST(LargeStringColumn)) {
AssertTablesEqual(*table, *batched_table, /*same_chunk_layout=*/false);
}
+TEST(TestArrowReaderAdHoc, LARGE_MEMORY_TEST(LargeStringValue)) {
+ // ARROW-3762
+ ::arrow::StringBuilder builder;
+ // 16 rows of 256MiB bytes each is 4GiB. This will get put into
+ // 3 chunks of 7 rows, 7 rows, and 2 rows.
+ constexpr std::int32_t kValueSize = 1 << 28;
+ constexpr std::int32_t kNumRows = 4;
+ constexpr std::int32_t kNumChunks = 4;
+ std::vector<std::shared_ptr<Array>> chunks;
+ std::vector<uint8_t> value(kValueSize, '0');
+ for (int chunk_idx = 0; chunk_idx < kNumChunks; chunk_idx++) {
+ ASSERT_OK(builder.Resize(kNumRows));
+ ASSERT_OK(builder.ReserveData(kNumRows * kValueSize));
+ for (int64_t i = 0; i < kNumRows; ++i) {
+ builder.UnsafeAppend(value.data(), kValueSize);
+ }
+ std::shared_ptr<Array> array;
+ ASSERT_OK(builder.Finish(&array));
Review Comment:
You are correct. I now build the array once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304748443
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1285,127 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::vector<std::shared_ptr<ChunkedArray>> current_cols;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ std::vector<Future<>> column_futs;
+ column_futs.reserve(state_->column_readers.size());
+ for (std::size_t column_index = 0; column_index < state_->column_readers.size();
+ column_index++) {
+ const auto& column_reader = state_->column_readers[column_index];
+ column_futs.push_back(
+ column_reader
+ ->NextBatchAsync(rows_in_batch, state_->io_executor, state_->cpu_executor)
+ .Then([state = state_,
+ column_index](const std::shared_ptr<ChunkedArray>& chunked_array) {
+ state->current_cols[column_index] = chunked_array;
+ }));
+ }
+ Future<> all_columns_finished_fut = ::arrow::AllFinished(std::move(column_futs));
+
+ // Grab the first batch of data and return it. If there is more than one batch then
+ // throw the remaining batches into overflow and they will be fetched on the next call
+ return all_columns_finished_fut.Then(
+ [state = state_, rows_in_batch]() -> Result<std::shared_ptr<RecordBatch>> {
+ std::shared_ptr<Table> table =
+ Table::Make(state->schema, state->current_cols, rows_in_batch);
+ ::arrow::TableBatchReader batch_reader(*table);
+ std::shared_ptr<RecordBatch> first;
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> next_batch,
+ batch_reader.Next());
+ if (!next_batch) {
+ break;
+ }
+ if (first) {
+ if (!state->allow_sliced_batches) {
+ return Status::Invalid(
+ "The setting allow_sliced_batches is set to false and data was "
+ "encountered that was too large to fit in a single batch.");
+ }
+ state->overflow.push(std::move(next_batch));
+ } else {
+ first = std::move(next_batch);
+ }
+ }
+ // If there were no rows of data we should have been exhausted. Otherwise we
+ // should have gotten at least one batch.
+ DCHECK(!!first);
+ return first;
+ });
+ }
+
+ private:
+ std::shared_ptr<AsyncBatchGeneratorState> state_;
+};
+
+Result<FileReaderImpl::AsyncBatchGenerator> FileReaderImpl::DoReadRowGroupsAsync(
+ const std::vector<int>& row_groups, const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor, bool allow_sliced_batches) {
+ RETURN_NOT_OK(BoundsCheck(row_groups, column_indices));
+
+ if (reader_properties_.pre_buffer()) {
+ BEGIN_PARQUET_CATCH_EXCEPTIONS
+ parquet_reader()->PreBuffer(row_groups, column_indices,
+ reader_properties_.io_context(),
+ reader_properties_.cache_options());
+ END_PARQUET_CATCH_EXCEPTIONS
+ }
+
+ auto generator_state = std::make_shared<AsyncBatchGeneratorState>();
+ generator_state->io_executor = reader_properties_.io_context().executor();
+ generator_state->cpu_executor = cpu_executor;
+ generator_state->use_threads = reader_properties_.use_threads();
Review Comment:
Final question, it's not related to correctness, but it's a bit confusing
```
/// This method ignores the use_threads property of the ArrowReaderProperties. It will
/// always create a task for each column. To run without threads you should use a
/// serial executor as the CPU executor.
```
So why is `use_threads` introduced 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
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1299608176
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +250,63 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = true) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, decompression is also done on the I/O thread
+ /// which may not be ideal.
+ ///
+ /// The returned generator will respect the batch size set in the ArrowReaderProperties.
+ /// Batches will not be larger than the given batch size. However, batches may be
+ /// smaller. This can happen, for example, when there is not enough data or when a
+ /// string column is too large to fit into a single batch. The parameter
+ /// `allow_sliced_batches` can be set to false to disallow this later case. This can be
+ /// useful when you need to know exactly how many batches you will get from the
+ /// operation before you start.
+ ///
+ /// The returned generator is not async-reentrant. You must not fetch another future
+ /// from it until the previously fetched future has completed.
+ ///
+ /// Note: When reading multiple row groups there is no guarantee you will get one
+ /// record batch per row group. Data from multiple row groups could get combined into
+ /// a single batch.
+ ///
+ /// Note: If a row group has 0 rows it will effectively be ignored. If you are only
+ /// reading empty row groups then the returned generator will immediately finish. This
+ /// method should never generate an empty record batch.
+ ///
+ /// The I/O executor is obtained from the I/O context in the reader properties.
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param column_indices indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If false, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
Review Comment:
Would `Future` in arrow actually include `Status`, so it would contain the `arrow::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
[GitHub] [arrow] wgtmac commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1275654597
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,123 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ Future<std::vector<std::shared_ptr<ChunkedArray>>> chunked_arrays_fut =
+ ::arrow::internal::OptionalParallelForAsync(
Review Comment:
@westonpace Do you want to fix this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304712231
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +250,63 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = true) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, decompression is also done on the I/O thread
+ /// which may not be ideal.
+ ///
+ /// The returned generator will respect the batch size set in the ArrowReaderProperties.
+ /// Batches will not be larger than the given batch size. However, batches may be
+ /// smaller. This can happen, for example, when there is not enough data or when a
+ /// string column is too large to fit into a single batch. The parameter
+ /// `allow_sliced_batches` can be set to false to disallow this later case. This can be
+ /// useful when you need to know exactly how many batches you will get from the
+ /// operation before you start.
+ ///
+ /// The returned generator is not async-reentrant. You must not fetch another future
+ /// from it until the previously fetched future has completed.
+ ///
+ /// Note: When reading multiple row groups there is no guarantee you will get one
+ /// record batch per row group. Data from multiple row groups could get combined into
+ /// a single batch.
+ ///
+ /// Note: If a row group has 0 rows it will effectively be ignored. If you are only
+ /// reading empty row groups then the returned generator will immediately finish. This
+ /// method should never generate an empty record batch.
+ ///
+ /// The I/O executor is obtained from the I/O context in the reader properties.
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param column_indices indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If false, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
Review Comment:
I will add something here.
> Would Future in arrow actually include Status?
Yes. A Future in Arrow has an implicit Status. So it is generally invalid to see something like `Future<Result<T>>`. There are a few places where we do return `Result<Future<T>>`. This typically indicates that "something can fail quickly, in the synchronous part" (e.g. the thread pool is already shut down) and any failure in the asynchronous portion will be communicated in the `Future`. However, I generally still prefer to just return `Future<T>` in these cases (there is a utility `DeferNotOk` which will convert `Result<Future<T>>` to `Future<T>`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304717495
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -114,6 +116,32 @@ class ColumnReaderImpl : public ColumnReader {
return Status::OK();
}
+ ::arrow::Result<std::shared_ptr<::arrow::ChunkedArray>> NextBatch(
+ int64_t batch_size) final {
+ std::shared_ptr<::arrow::ChunkedArray> out;
+ RETURN_NOT_OK(NextBatch(batch_size, &out));
+ return out;
+ }
+
+ Future<std::shared_ptr<ChunkedArray>> NextBatchAsync(
+ int64_t batch_size, ::arrow::internal::Executor* io_executor,
+ ::arrow::internal::Executor* cpu_executor) final {
+ Future<> load_fut = ::arrow::DeferNotOk(
+ io_executor->Submit([this, batch_size] { return LoadBatch(batch_size); }));
+ return load_fut.Then(
+ [this, batch_size, cpu_executor]() -> Future<std::shared_ptr<ChunkedArray>> {
+ return DeferNotOk(cpu_executor->Submit(
+ [this, batch_size]() -> Result<std::shared_ptr<ChunkedArray>> {
+ std::shared_ptr<ChunkedArray> out;
+ RETURN_NOT_OK(BuildArray(batch_size, &out));
+ for (int x = 0; x < out->num_chunks(); x++) {
+ RETURN_NOT_OK(out->chunk(x)->Validate());
Review Comment:
I'm only guessing as I didn't write the original implementation but my guess is that, for security reasons, it is almost always required to validate because a malicious user could otherwise craft a parquet file that triggers buffer overflow. For example, they could store a list array where one of the offsets is way out of range.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1296155567
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
Review Comment:
I don't have strong feelings on this default. I will switch it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1296268330
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
Review Comment:
Sounds good. I've removed the single-row variants.
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
Review Comment:
I have switched the default to true
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on PR #36779:
URL: https://github.com/apache/arrow/pull/36779#issuecomment-1654638082
@wgtmac / @mapleFU I believe I have addressed your concerns and this is ready for another round of review. The CI failures appear to be unrelated.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] wgtmac commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1268958773
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
Review Comment:
In which case sliced batches is a concern?
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, compression and column decoding is also done on
+ /// the I/O thread which may not be ideal. The stage after that (transferring the
+ /// decoded data into Arrow structures and fulfilling the future) should be done as a
+ /// new task on the cpu_executor.
+ ///
+ /// The returned generator will respect the batch size set in the reader properties.
+ /// Batches will not be larger than the given batch size. However, batches may be
+ /// smaller. This can happen, for example, when there is not enough data or when a
+ /// string column is too large to fit into a single batch. The parameter
+ /// `allow_sliced_batches` can be set to false to disallow this later case. This can be
+ /// useful when you need to know exactly how many batches you will get from the
+ /// operation before you start.
+ ///
+ /// Note: When reading multiple row groups there is no guarantee you will get one
+ /// record batch per row group. Data from multiple row groups could get combined into
+ /// a single batch.
+ ///
+ /// Note: If a row group has 0 rows it will effectively be ignored. If you are only
+ /// reading empty row groups then the returned generated will immediately finish. This
Review Comment:
```suggestion
/// reading empty row groups then the returned generator will immediately finish. This
```
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -2514,6 +2517,137 @@ TEST(TestArrowReadWrite, GetRecordBatchGenerator) {
}
}
+TEST(TestArrowReadWrite, ReadRowGroupsAsync) {
+ constexpr int kNumRows = 1024;
+ constexpr int kRowGroupSize = 512;
+ constexpr int kNumColumns = 2;
+
+ std::shared_ptr<Table> table;
+ ASSERT_NO_FATAL_FAILURE(MakeDoubleTable(kNumColumns, kNumRows, 1, &table));
+
+ std::shared_ptr<Buffer> buffer;
+ ASSERT_NO_FATAL_FAILURE(WriteTableToBuffer(table, kRowGroupSize,
+ default_arrow_writer_properties(), &buffer));
+
+ for (std::vector<int> row_groups : std::vector<std::vector<int>>{{}, {0}, {0, 1}}) {
+ ARROW_SCOPED_TRACE("# row_groups = ", row_groups.size());
+ int32_t expected_total_rows = static_cast<int32_t>(row_groups.size()) * kRowGroupSize;
+
+ for (std::vector<int> columns : std::vector<std::vector<int>>{{}, {0}, {0, 1}}) {
+ ARROW_SCOPED_TRACE("# columns = ", columns.size());
+
+ for (int row_group_size : {128, 512, 1024, 2048}) {
+ ARROW_SCOPED_TRACE("row_group_size = ", row_group_size);
+
+ ArrowReaderProperties properties = default_arrow_reader_properties();
+ properties.set_batch_size(row_group_size);
+ std::unique_ptr<FileReader> unique_reader;
+ FileReaderBuilder builder;
+ ASSERT_OK(builder.Open(std::make_shared<BufferReader>(buffer)));
+ ASSERT_OK(builder.properties(properties)->Build(&unique_reader));
+ auto batch_generator = unique_reader->ReadRowGroupsAsync(
+ row_groups, columns, ::arrow::internal::GetCpuThreadPool());
+
+ int64_t num_expected_batches =
Review Comment:
Just curious, is a predicable num_expected_batches useful in real case?
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,123 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ Future<std::vector<std::shared_ptr<ChunkedArray>>> chunked_arrays_fut =
+ ::arrow::internal::OptionalParallelForAsync(
+ state_->use_threads, state_->column_readers,
+ [rows_in_batch](std::size_t, std::shared_ptr<ColumnReaderImpl> column_reader)
+ -> Result<std::shared_ptr<ChunkedArray>> {
+ std::shared_ptr<ChunkedArray> chunked_array;
+ ARROW_RETURN_NOT_OK(
+ column_reader->NextBatch(rows_in_batch, &chunked_array));
+ return chunked_array;
+ });
+
+ // Grab the first batch of data and return it. If there is more than one batch then
+ // throw the reamining batches into overflow and they will be fetched on the next call
+ return chunked_arrays_fut.Then(
+ [state = state_,
+ rows_in_batch](const std::vector<std::shared_ptr<ChunkedArray>>& chunks)
+ -> Result<std::shared_ptr<RecordBatch>> {
+ std::shared_ptr<Table> table =
+ Table::Make(state->schema, chunks, rows_in_batch);
+ ::arrow::TableBatchReader batch_reader(*table);
Review Comment:
Is there a handy way to directly convert ChunkedArray to RecordBatch? I am not sure if going through a TableBatchReader is heavy or not.
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,123 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ Future<std::vector<std::shared_ptr<ChunkedArray>>> chunked_arrays_fut =
+ ::arrow::internal::OptionalParallelForAsync(
Review Comment:
The `cpu_executor` is not used 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
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304715707
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -114,6 +116,32 @@ class ColumnReaderImpl : public ColumnReader {
return Status::OK();
}
+ ::arrow::Result<std::shared_ptr<::arrow::ChunkedArray>> NextBatch(
+ int64_t batch_size) final {
+ std::shared_ptr<::arrow::ChunkedArray> out;
+ RETURN_NOT_OK(NextBatch(batch_size, &out));
+ return out;
+ }
+
+ Future<std::shared_ptr<ChunkedArray>> NextBatchAsync(
+ int64_t batch_size, ::arrow::internal::Executor* io_executor,
+ ::arrow::internal::Executor* cpu_executor) final {
+ Future<> load_fut = ::arrow::DeferNotOk(
+ io_executor->Submit([this, batch_size] { return LoadBatch(batch_size); }));
Review Comment:
Good observation. In a perfect world we would do all of that on a CPU thread. This would help to keep context switches to a minimum. The only work that would happen on the I/O thread would be the `RandomAccessFile` call to read the file. However, that requires pushing async further into the parquet code base which would be a lot of work. It's not clear that the benefit would be significant enough to require the work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on PR #36779:
URL: https://github.com/apache/arrow/pull/36779#issuecomment-1684363094
@wgtmac Would you like to review this again?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1296156006
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, compression and column decoding is also done on
+ /// the I/O thread which may not be ideal. The stage after that (transferring the
+ /// decoded data into Arrow structures and fulfilling the future) should be done as a
+ /// new task on the cpu_executor.
+ ///
+ /// The returned generator will respect the batch size set in the ArrowReaderProperties.
+ /// Batches will not be larger than the given batch size. However, batches may be
+ /// smaller. This can happen, for example, when there is not enough data or when a
+ /// string column is too large to fit into a single batch. The parameter
+ /// `allow_sliced_batches` can be set to false to disallow this later case. This can be
+ /// useful when you need to know exactly how many batches you will get from the
+ /// operation before you start.
+ ///
+ /// Note: When reading multiple row groups there is no guarantee you will get one
+ /// record batch per row group. Data from multiple row groups could get combined into
+ /// a single batch.
Review Comment:
No. This is the same way that the synchronous APIs behave.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304719263
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -114,6 +116,32 @@ class ColumnReaderImpl : public ColumnReader {
return Status::OK();
}
+ ::arrow::Result<std::shared_ptr<::arrow::ChunkedArray>> NextBatch(
+ int64_t batch_size) final {
+ std::shared_ptr<::arrow::ChunkedArray> out;
+ RETURN_NOT_OK(NextBatch(batch_size, &out));
+ return out;
+ }
+
+ Future<std::shared_ptr<ChunkedArray>> NextBatchAsync(
+ int64_t batch_size, ::arrow::internal::Executor* io_executor,
+ ::arrow::internal::Executor* cpu_executor) final {
+ Future<> load_fut = ::arrow::DeferNotOk(
+ io_executor->Submit([this, batch_size] { return LoadBatch(batch_size); }));
Review Comment:
Yes I see it in comment, just saying that it might harm. Perfect IO might be so tricky and need to wrap `RangeCache` and `RandomAccessFile`. This looks good to me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] pitrou commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1293707887
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
Review Comment:
Is it necessary to expose `ReadRowGroupAsync` in addition to `ReadRowGroupsAsync`? One is a trivial call to the other...
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
Review Comment:
I don't understand what "a batch has too much data for the given batch size" means exactly. Do you mean "a row group has too much data for the given batch size"?
Also, why is it false by default? It seems allowing it should be the default behaviour.
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, compression and column decoding is also done on
+ /// the I/O thread which may not be ideal. The stage after that (transferring the
+ /// decoded data into Arrow structures and fulfilling the future) should be done as a
+ /// new task on the cpu_executor.
+ ///
+ /// The returned generator will respect the batch size set in the ArrowReaderProperties.
+ /// Batches will not be larger than the given batch size. However, batches may be
+ /// smaller. This can happen, for example, when there is not enough data or when a
+ /// string column is too large to fit into a single batch. The parameter
+ /// `allow_sliced_batches` can be set to false to disallow this later case. This can be
+ /// useful when you need to know exactly how many batches you will get from the
+ /// operation before you start.
+ ///
+ /// Note: When reading multiple row groups there is no guarantee you will get one
+ /// record batch per row group. Data from multiple row groups could get combined into
+ /// a single batch.
Review Comment:
Interesting, and I agree it's probably desirable. Is it a deviation from other APIs?
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,124 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ Future<std::vector<std::shared_ptr<ChunkedArray>>> chunked_arrays_fut =
+ ::arrow::internal::OptionalParallelForAsync(
+ state_->use_threads, state_->column_readers,
+ [rows_in_batch](std::size_t, std::shared_ptr<ColumnReaderImpl> column_reader)
+ -> Result<std::shared_ptr<ChunkedArray>> {
+ std::shared_ptr<ChunkedArray> chunked_array;
+ ARROW_RETURN_NOT_OK(
+ column_reader->NextBatch(rows_in_batch, &chunked_array));
+ return chunked_array;
Review Comment:
Not necessary for this PR, but we'd probably like a `Result` returning variant of `NextBatch`.
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,124 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ Future<std::vector<std::shared_ptr<ChunkedArray>>> chunked_arrays_fut =
+ ::arrow::internal::OptionalParallelForAsync(
+ state_->use_threads, state_->column_readers,
+ [rows_in_batch](std::size_t, std::shared_ptr<ColumnReaderImpl> column_reader)
+ -> Result<std::shared_ptr<ChunkedArray>> {
+ std::shared_ptr<ChunkedArray> chunked_array;
+ ARROW_RETURN_NOT_OK(
+ column_reader->NextBatch(rows_in_batch, &chunked_array));
+ return chunked_array;
+ },
+ state_->cpu_executor);
+
+ // Grab the first batch of data and return it. If there is more than one batch then
+ // throw the reamining batches into overflow and they will be fetched on the next call
+ return chunked_arrays_fut.Then(
+ [state = state_,
+ rows_in_batch](const std::vector<std::shared_ptr<ChunkedArray>>& chunks)
+ -> Result<std::shared_ptr<RecordBatch>> {
+ std::shared_ptr<Table> table =
+ Table::Make(state->schema, chunks, rows_in_batch);
+ ::arrow::TableBatchReader batch_reader(*table);
+ std::shared_ptr<RecordBatch> first;
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> next_batch,
+ batch_reader.Next());
+ if (!next_batch) {
+ break;
+ }
+ if (first) {
+ if (!state->allow_sliced_batches) {
+ return Status::Invalid(
+ "The setting allow_sliced_batches is set to false and data was "
+ "encountered that was too large to fit in a single batch.");
+ }
+ state->overflow.push(std::move(next_batch));
Review Comment:
I suppose this makes the generator not async-reentrant, since `operator()` might be called from one thread while this callback runs on another thread?
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -3963,6 +4096,88 @@ TEST(TestArrowReaderAdHoc, LARGE_MEMORY_TEST(LargeStringColumn)) {
AssertTablesEqual(*table, *batched_table, /*same_chunk_layout=*/false);
}
+TEST(TestArrowReaderAdHoc, LARGE_MEMORY_TEST(LargeStringValue)) {
+ // ARROW-3762
+ ::arrow::StringBuilder builder;
+ // 16 rows of 256MiB bytes each is 4GiB. This will get put into
+ // 3 chunks of 7 rows, 7 rows, and 2 rows.
+ constexpr std::int32_t kValueSize = 1 << 28;
+ constexpr std::int32_t kNumRows = 4;
+ constexpr std::int32_t kNumChunks = 4;
+ std::vector<std::shared_ptr<Array>> chunks;
+ std::vector<uint8_t> value(kValueSize, '0');
+ for (int chunk_idx = 0; chunk_idx < kNumChunks; chunk_idx++) {
+ ASSERT_OK(builder.Resize(kNumRows));
+ ASSERT_OK(builder.ReserveData(kNumRows * kValueSize));
+ for (int64_t i = 0; i < kNumRows; ++i) {
+ builder.UnsafeAppend(value.data(), kValueSize);
+ }
+ std::shared_ptr<Array> array;
+ ASSERT_OK(builder.Finish(&array));
+ chunks.push_back(std::move(array));
+ }
+
+ // Eaglerly free up memory
+ value.clear();
Review Comment:
`clear` unfortunately leaves the vector capacity unchanged. Perhaps `value = {}` would work...
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -3963,6 +4096,88 @@ TEST(TestArrowReaderAdHoc, LARGE_MEMORY_TEST(LargeStringColumn)) {
AssertTablesEqual(*table, *batched_table, /*same_chunk_layout=*/false);
}
+TEST(TestArrowReaderAdHoc, LARGE_MEMORY_TEST(LargeStringValue)) {
+ // ARROW-3762
+ ::arrow::StringBuilder builder;
+ // 16 rows of 256MiB bytes each is 4GiB. This will get put into
+ // 3 chunks of 7 rows, 7 rows, and 2 rows.
+ constexpr std::int32_t kValueSize = 1 << 28;
+ constexpr std::int32_t kNumRows = 4;
+ constexpr std::int32_t kNumChunks = 4;
+ std::vector<std::shared_ptr<Array>> chunks;
+ std::vector<uint8_t> value(kValueSize, '0');
+ for (int chunk_idx = 0; chunk_idx < kNumChunks; chunk_idx++) {
+ ASSERT_OK(builder.Resize(kNumRows));
+ ASSERT_OK(builder.ReserveData(kNumRows * kValueSize));
+ for (int64_t i = 0; i < kNumRows; ++i) {
+ builder.UnsafeAppend(value.data(), kValueSize);
+ }
+ std::shared_ptr<Array> array;
+ ASSERT_OK(builder.Finish(&array));
Review Comment:
This seems to be building the same array for each chunk, am I mistaken? You could simply build the array once.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1269585568
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -2514,6 +2517,137 @@ TEST(TestArrowReadWrite, GetRecordBatchGenerator) {
}
}
+TEST(TestArrowReadWrite, ReadRowGroupsAsync) {
+ constexpr int kNumRows = 1024;
+ constexpr int kRowGroupSize = 512;
+ constexpr int kNumColumns = 2;
+
+ std::shared_ptr<Table> table;
+ ASSERT_NO_FATAL_FAILURE(MakeDoubleTable(kNumColumns, kNumRows, 1, &table));
+
+ std::shared_ptr<Buffer> buffer;
+ ASSERT_NO_FATAL_FAILURE(WriteTableToBuffer(table, kRowGroupSize,
+ default_arrow_writer_properties(), &buffer));
+
+ for (std::vector<int> row_groups : std::vector<std::vector<int>>{{}, {0}, {0, 1}}) {
+ ARROW_SCOPED_TRACE("# row_groups = ", row_groups.size());
+ int32_t expected_total_rows = static_cast<int32_t>(row_groups.size()) * kRowGroupSize;
+
+ for (std::vector<int> columns : std::vector<std::vector<int>>{{}, {0}, {0, 1}}) {
+ ARROW_SCOPED_TRACE("# columns = ", columns.size());
+
+ for (int row_group_size : {128, 512, 1024, 2048}) {
+ ARROW_SCOPED_TRACE("row_group_size = ", row_group_size);
+
+ ArrowReaderProperties properties = default_arrow_reader_properties();
+ properties.set_batch_size(row_group_size);
+ std::unique_ptr<FileReader> unique_reader;
+ FileReaderBuilder builder;
+ ASSERT_OK(builder.Open(std::make_shared<BufferReader>(buffer)));
+ ASSERT_OK(builder.properties(properties)->Build(&unique_reader));
+ auto batch_generator = unique_reader->ReadRowGroupsAsync(
+ row_groups, columns, ::arrow::internal::GetCpuThreadPool());
+
+ int64_t num_expected_batches =
Review Comment:
In Acero we currently assign each batch an index. If we read several files in parallel then it helps to know how many batches we will receive before we start reading the file.
```
parallel for file in files:
metadata[file.id] = inspect(file)
index_offset = 0
for file in files:
starting_index[file.id] = metadata[file.id].num_rows / batch_size
parallel for file in files:
index = starting_index[file.id]
for batch in scan_file(file):
batch.index = index++
yield batch
```
If we don't know how many batches are in a file (ahead of time) then we have to read the files serially which can be quite slow. In #35889 I introduce a user option `allow_jumbo_values`. If true, then we read serially. If false (the default), then we read in parallel and raise an error if we have to split a batch.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1269837001
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,123 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
Review Comment:
So the empty rowgroup is handled in two ways:
1. If it's in the middle of a File, the rowgroup could handled by `ColumnReaderImpl`, which skip that rowgroup
2. If it's in the end of a File, it would return a `Finish`?
Am I right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1269841725
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, compression and column decoding is also done on
+ /// the I/O thread which may not be ideal. The stage after that (transferring the
+ /// decoded data into Arrow structures and fulfilling the future) should be done as a
+ /// new task on the cpu_executor.
+ ///
+ /// The returned generator will respect the batch size set in the reader properties.
+ /// Batches will not be larger than the given batch size. However, batches may be
+ /// smaller. This can happen, for example, when there is not enough data or when a
Review Comment:
Oh I see, It cannot be "0", because `ColumnReaderImpl` would try it best to assemble the rows.
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,123 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
Review Comment:
So the empty rowgroup is handled in two ways:
1. If it's in the middle of a File, the rowgroup could handled by `ColumnReaderImpl`, which skip that rowgroup
2.If it's in the end of a File, it would return a `Finish`?
Am I right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1268951056
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, compression and column decoding is also done on
Review Comment:
```suggestion
/// thread, which is correct. However, compression and column decoding is also done on
```
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, compression and column decoding is also done on
+ /// the I/O thread which may not be ideal. The stage after that (transferring the
+ /// decoded data into Arrow structures and fulfilling the future) should be done as a
+ /// new task on the cpu_executor.
+ ///
+ /// The returned generator will respect the batch size set in the reader properties.
Review Comment:
```suggestion
/// The returned generator will respect the batch size set in the ArrowReaderProperties.
```
Should this be better?
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -316,6 +318,23 @@ class FileReaderImpl : public FileReader {
return ReadRowGroups(row_groups, Iota(reader_->metadata()->num_columns()), table);
}
+ Result<AsyncBatchGenerator> DoReadRowGroupsAsync(
+ const std::vector<int>& row_groups, const std::vector<int>& indices,
Review Comment:
```suggestion
const std::vector<int>& row_groups, const std::vector<int>& column_indices,
```
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+ /// \brief Read some columns from a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ /// \see ReadTable for details on how column indices are resolved
+ ///
+ /// \param i the index of the row group to read
+ /// \param column_indices leaf-indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
+ const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = false) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, compression and column decoding is also done on
+ /// the I/O thread which may not be ideal. The stage after that (transferring the
+ /// decoded data into Arrow structures and fulfilling the future) should be done as a
+ /// new task on the cpu_executor.
+ ///
+ /// The returned generator will respect the batch size set in the reader properties.
+ /// Batches will not be larger than the given batch size. However, batches may be
+ /// smaller. This can happen, for example, when there is not enough data or when a
Review Comment:
Could the batch size be "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
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1269591242
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,123 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ Future<std::vector<std::shared_ptr<ChunkedArray>>> chunked_arrays_fut =
+ ::arrow::internal::OptionalParallelForAsync(
+ state_->use_threads, state_->column_readers,
+ [rows_in_batch](std::size_t, std::shared_ptr<ColumnReaderImpl> column_reader)
+ -> Result<std::shared_ptr<ChunkedArray>> {
+ std::shared_ptr<ChunkedArray> chunked_array;
+ ARROW_RETURN_NOT_OK(
+ column_reader->NextBatch(rows_in_batch, &chunked_array));
+ return chunked_array;
+ });
+
+ // Grab the first batch of data and return it. If there is more than one batch then
+ // throw the reamining batches into overflow and they will be fetched on the next call
+ return chunked_arrays_fut.Then(
+ [state = state_,
+ rows_in_batch](const std::vector<std::shared_ptr<ChunkedArray>>& chunks)
+ -> Result<std::shared_ptr<RecordBatch>> {
+ std::shared_ptr<Table> table =
+ Table::Make(state->schema, chunks, rows_in_batch);
+ ::arrow::TableBatchReader batch_reader(*table);
Review Comment:
It should be pretty efficient though there may be a few shared_ptr copies. The `TableBatchReader` also does a little bit of extra work because it allows for different chunked arrays to have different chunk sizes. For example, if you have two chunked arrays with 2 chunks each `[10 rows, 30 rows]` and `[20 rows, 20 rows]` then the table batch reader will return three batches (`[10 rows, 10 rows, 20 rows]`).
I don't think this logic is needed here because I'm pretty sure the chunks will have the same sizes. So we could maybe write a slightly more efficient method if we can assert that all chunks are the same size.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1269013524
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,123 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ Future<std::vector<std::shared_ptr<ChunkedArray>>> chunked_arrays_fut =
+ ::arrow::internal::OptionalParallelForAsync(
Review Comment:
Nice catch, here it uses `internal::GetCpuThreadPool()`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1299610387
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -114,6 +116,32 @@ class ColumnReaderImpl : public ColumnReader {
return Status::OK();
}
+ ::arrow::Result<std::shared_ptr<::arrow::ChunkedArray>> NextBatch(
+ int64_t batch_size) final {
+ std::shared_ptr<::arrow::ChunkedArray> out;
+ RETURN_NOT_OK(NextBatch(batch_size, &out));
+ return out;
+ }
+
+ Future<std::shared_ptr<ChunkedArray>> NextBatchAsync(
+ int64_t batch_size, ::arrow::internal::Executor* io_executor,
+ ::arrow::internal::Executor* cpu_executor) final {
+ Future<> load_fut = ::arrow::DeferNotOk(
+ io_executor->Submit([this, batch_size] { return LoadBatch(batch_size); }));
+ return load_fut.Then(
+ [this, batch_size, cpu_executor]() -> Future<std::shared_ptr<ChunkedArray>> {
+ return DeferNotOk(cpu_executor->Submit(
+ [this, batch_size]() -> Result<std::shared_ptr<ChunkedArray>> {
+ std::shared_ptr<ChunkedArray> out;
+ RETURN_NOT_OK(BuildArray(batch_size, &out));
+ for (int x = 0; x < out->num_chunks(); x++) {
+ RETURN_NOT_OK(out->chunk(x)->Validate());
Review Comment:
(this is not related to this patch, but I want to ask, why it's neccessary to Validate it here?)
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +250,63 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read row groups from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
+ const std::vector<int>& row_groups, ::arrow::internal::Executor* cpu_executor,
+ bool allow_sliced_batches = true) = 0;
+
+ /// \brief Read some columns from the given rows groups from the file
+ ///
+ /// If pre-buffering is enabled then all of the data will be read using the pre-buffer
+ /// cache. See ParquetFileReader::PreBuffer for details on how this affects memory and
+ /// performance.
+ ///
+ /// This operation is not perfectly async. The read from disk will be done on an I/O
+ /// thread, which is correct. However, decompression is also done on the I/O thread
+ /// which may not be ideal.
+ ///
+ /// The returned generator will respect the batch size set in the ArrowReaderProperties.
+ /// Batches will not be larger than the given batch size. However, batches may be
+ /// smaller. This can happen, for example, when there is not enough data or when a
+ /// string column is too large to fit into a single batch. The parameter
+ /// `allow_sliced_batches` can be set to false to disallow this later case. This can be
+ /// useful when you need to know exactly how many batches you will get from the
+ /// operation before you start.
+ ///
+ /// The returned generator is not async-reentrant. You must not fetch another future
+ /// from it until the previously fetched future has completed.
+ ///
+ /// Note: When reading multiple row groups there is no guarantee you will get one
+ /// record batch per row group. Data from multiple row groups could get combined into
+ /// a single batch.
+ ///
+ /// Note: If a row group has 0 rows it will effectively be ignored. If you are only
+ /// reading empty row groups then the returned generator will immediately finish. This
+ /// method should never generate an empty record batch.
+ ///
+ /// The I/O executor is obtained from the I/O context in the reader properties.
+ ///
+ /// \param row_groups indices of the row groups to read
+ /// \param column_indices indices of the columns to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If false, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupsAsync(
Review Comment:
Would `Future` in arrow actually include `Status`?
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1285,127 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::vector<std::shared_ptr<ChunkedArray>> current_cols;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ std::vector<Future<>> column_futs;
+ column_futs.reserve(state_->column_readers.size());
+ for (std::size_t column_index = 0; column_index < state_->column_readers.size();
+ column_index++) {
+ const auto& column_reader = state_->column_readers[column_index];
+ column_futs.push_back(
+ column_reader
+ ->NextBatchAsync(rows_in_batch, state_->io_executor, state_->cpu_executor)
+ .Then([state = state_,
+ column_index](const std::shared_ptr<ChunkedArray>& chunked_array) {
+ state->current_cols[column_index] = chunked_array;
+ }));
+ }
+ Future<> all_columns_finished_fut = ::arrow::AllFinished(std::move(column_futs));
+
+ // Grab the first batch of data and return it. If there is more than one batch then
+ // throw the reamining batches into overflow and they will be fetched on the next call
+ return all_columns_finished_fut.Then(
+ [state = state_, rows_in_batch]() -> Result<std::shared_ptr<RecordBatch>> {
+ std::shared_ptr<Table> table =
+ Table::Make(state->schema, state->current_cols, rows_in_batch);
+ ::arrow::TableBatchReader batch_reader(*table);
+ std::shared_ptr<RecordBatch> first;
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> next_batch,
+ batch_reader.Next());
+ if (!next_batch) {
+ break;
+ }
+ if (first) {
+ if (!state->allow_sliced_batches) {
Review Comment:
Should we add this in document?
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -114,6 +116,32 @@ class ColumnReaderImpl : public ColumnReader {
return Status::OK();
}
+ ::arrow::Result<std::shared_ptr<::arrow::ChunkedArray>> NextBatch(
+ int64_t batch_size) final {
+ std::shared_ptr<::arrow::ChunkedArray> out;
+ RETURN_NOT_OK(NextBatch(batch_size, &out));
+ return out;
+ }
+
+ Future<std::shared_ptr<ChunkedArray>> NextBatchAsync(
+ int64_t batch_size, ::arrow::internal::Executor* io_executor,
+ ::arrow::internal::Executor* cpu_executor) final {
+ Future<> load_fut = ::arrow::DeferNotOk(
+ io_executor->Submit([this, batch_size] { return LoadBatch(batch_size); }));
Review Comment:
Would it matter that `io_executor` would consume some CPU?
`LeafReader::LoadBatch` might decompress page, decode records, parse def-rep levels. They're all cpu-intensive.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304743341
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -114,6 +116,32 @@ class ColumnReaderImpl : public ColumnReader {
return Status::OK();
}
+ ::arrow::Result<std::shared_ptr<::arrow::ChunkedArray>> NextBatch(
+ int64_t batch_size) final {
+ std::shared_ptr<::arrow::ChunkedArray> out;
+ RETURN_NOT_OK(NextBatch(batch_size, &out));
+ return out;
+ }
+
+ Future<std::shared_ptr<ChunkedArray>> NextBatchAsync(
+ int64_t batch_size, ::arrow::internal::Executor* io_executor,
+ ::arrow::internal::Executor* cpu_executor) final {
+ Future<> load_fut = ::arrow::DeferNotOk(
+ io_executor->Submit([this, batch_size] { return LoadBatch(batch_size); }));
Review Comment:
Another point is that seems that `LoadBatch` would only depends on sync-io api? So it would not cause deadlock when `LoadBatch` waiting for page io? (which could happen when `use_thread` with previous scan?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #36779:
URL: https://github.com/apache/arrow/pull/36779#issuecomment-1643391836
> I think the ColumnReader already supports this internally.
Oh I see, `ColumnReaderImpl::LoadBatch` would load multiple batches. It try it best to read expected rows. Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1275454158
##########
cpp/src/parquet/arrow/arrow_reader_writer_test.cc:
##########
@@ -2514,6 +2517,137 @@ TEST(TestArrowReadWrite, GetRecordBatchGenerator) {
}
}
+TEST(TestArrowReadWrite, ReadRowGroupsAsync) {
+ constexpr int kNumRows = 1024;
+ constexpr int kRowGroupSize = 512;
+ constexpr int kNumColumns = 2;
+
+ std::shared_ptr<Table> table;
+ ASSERT_NO_FATAL_FAILURE(MakeDoubleTable(kNumColumns, kNumRows, 1, &table));
+
+ std::shared_ptr<Buffer> buffer;
+ ASSERT_NO_FATAL_FAILURE(WriteTableToBuffer(table, kRowGroupSize,
+ default_arrow_writer_properties(), &buffer));
+
+ for (std::vector<int> row_groups : std::vector<std::vector<int>>{{}, {0}, {0, 1}}) {
+ ARROW_SCOPED_TRACE("# row_groups = ", row_groups.size());
+ int32_t expected_total_rows = static_cast<int32_t>(row_groups.size()) * kRowGroupSize;
+
+ for (std::vector<int> columns : std::vector<std::vector<int>>{{}, {0}, {0, 1}}) {
+ ARROW_SCOPED_TRACE("# columns = ", columns.size());
+
+ for (int row_group_size : {128, 512, 1024, 2048}) {
+ ARROW_SCOPED_TRACE("row_group_size = ", row_group_size);
+
+ ArrowReaderProperties properties = default_arrow_reader_properties();
+ properties.set_batch_size(row_group_size);
+ std::unique_ptr<FileReader> unique_reader;
+ FileReaderBuilder builder;
+ ASSERT_OK(builder.Open(std::make_shared<BufferReader>(buffer)));
+ ASSERT_OK(builder.properties(properties)->Build(&unique_reader));
+ auto batch_generator = unique_reader->ReadRowGroupsAsync(
+ row_groups, columns, ::arrow::internal::GetCpuThreadPool());
+
+ int64_t num_expected_batches =
Review Comment:
I am not worried about getting an exact number of rows. Empty batches and batches that have fewer rows than expected are ok. So if the file has 1 million rows and the batch size is 100k then I expect 10 or fewer batches. So I think this approach will still work with both page and row level filtering.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] wgtmac commented on pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "wgtmac (via GitHub)" <gi...@apache.org>.
wgtmac commented on PR #36779:
URL: https://github.com/apache/arrow/pull/36779#issuecomment-1643383887
> Is the logic for concat multiple batches from RowGroup implemented? Or it will be implement in the future?
I think the ColumnReader already supports this internally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304723273
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1285,127 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::vector<std::shared_ptr<ChunkedArray>> current_cols;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ std::vector<Future<>> column_futs;
+ column_futs.reserve(state_->column_readers.size());
+ for (std::size_t column_index = 0; column_index < state_->column_readers.size();
+ column_index++) {
+ const auto& column_reader = state_->column_readers[column_index];
+ column_futs.push_back(
+ column_reader
+ ->NextBatchAsync(rows_in_batch, state_->io_executor, state_->cpu_executor)
+ .Then([state = state_,
+ column_index](const std::shared_ptr<ChunkedArray>& chunked_array) {
+ state->current_cols[column_index] = chunked_array;
+ }));
+ }
+ Future<> all_columns_finished_fut = ::arrow::AllFinished(std::move(column_futs));
+
+ // Grab the first batch of data and return it. If there is more than one batch then
+ // throw the reamining batches into overflow and they will be fetched on the next call
+ return all_columns_finished_fut.Then(
+ [state = state_, rows_in_batch]() -> Result<std::shared_ptr<RecordBatch>> {
+ std::shared_ptr<Table> table =
+ Table::Make(state->schema, state->current_cols, rows_in_batch);
+ ::arrow::TableBatchReader batch_reader(*table);
+ std::shared_ptr<RecordBatch> first;
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> next_batch,
+ batch_reader.Next());
+ if (!next_batch) {
+ break;
+ }
+ if (first) {
+ if (!state->allow_sliced_batches) {
Review Comment:
Nope, that's my fault. The document looks good to me
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304748443
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1285,127 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::vector<std::shared_ptr<ChunkedArray>> current_cols;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ std::vector<Future<>> column_futs;
+ column_futs.reserve(state_->column_readers.size());
+ for (std::size_t column_index = 0; column_index < state_->column_readers.size();
+ column_index++) {
+ const auto& column_reader = state_->column_readers[column_index];
+ column_futs.push_back(
+ column_reader
+ ->NextBatchAsync(rows_in_batch, state_->io_executor, state_->cpu_executor)
+ .Then([state = state_,
+ column_index](const std::shared_ptr<ChunkedArray>& chunked_array) {
+ state->current_cols[column_index] = chunked_array;
+ }));
+ }
+ Future<> all_columns_finished_fut = ::arrow::AllFinished(std::move(column_futs));
+
+ // Grab the first batch of data and return it. If there is more than one batch then
+ // throw the remaining batches into overflow and they will be fetched on the next call
+ return all_columns_finished_fut.Then(
+ [state = state_, rows_in_batch]() -> Result<std::shared_ptr<RecordBatch>> {
+ std::shared_ptr<Table> table =
+ Table::Make(state->schema, state->current_cols, rows_in_batch);
+ ::arrow::TableBatchReader batch_reader(*table);
+ std::shared_ptr<RecordBatch> first;
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> next_batch,
+ batch_reader.Next());
+ if (!next_batch) {
+ break;
+ }
+ if (first) {
+ if (!state->allow_sliced_batches) {
+ return Status::Invalid(
+ "The setting allow_sliced_batches is set to false and data was "
+ "encountered that was too large to fit in a single batch.");
+ }
+ state->overflow.push(std::move(next_batch));
+ } else {
+ first = std::move(next_batch);
+ }
+ }
+ // If there were no rows of data we should have been exhausted. Otherwise we
+ // should have gotten at least one batch.
+ DCHECK(!!first);
+ return first;
+ });
+ }
+
+ private:
+ std::shared_ptr<AsyncBatchGeneratorState> state_;
+};
+
+Result<FileReaderImpl::AsyncBatchGenerator> FileReaderImpl::DoReadRowGroupsAsync(
+ const std::vector<int>& row_groups, const std::vector<int>& column_indices,
+ ::arrow::internal::Executor* cpu_executor, bool allow_sliced_batches) {
+ RETURN_NOT_OK(BoundsCheck(row_groups, column_indices));
+
+ if (reader_properties_.pre_buffer()) {
+ BEGIN_PARQUET_CATCH_EXCEPTIONS
+ parquet_reader()->PreBuffer(row_groups, column_indices,
+ reader_properties_.io_context(),
+ reader_properties_.cache_options());
+ END_PARQUET_CATCH_EXCEPTIONS
+ }
+
+ auto generator_state = std::make_shared<AsyncBatchGeneratorState>();
+ generator_state->io_executor = reader_properties_.io_context().executor();
+ generator_state->cpu_executor = cpu_executor;
+ generator_state->use_threads = reader_properties_.use_threads();
Review Comment:
```
/// This method ignores the use_threads property of the ArrowReaderProperties. It will
/// always create a task for each column. To run without threads you should use a
/// serial executor as the CPU executor.
```
So why is `use_threads` introduced 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
[GitHub] [arrow] pitrou commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "pitrou (via GitHub)" <gi...@apache.org>.
pitrou commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1296155508
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -249,6 +249,90 @@ class PARQUET_EXPORT FileReader {
virtual ::arrow::Status ReadRowGroups(const std::vector<int>& row_groups,
std::shared_ptr<::arrow::Table>* out) = 0;
+ using AsyncBatchGenerator =
+ std::function<::arrow::Future<std::shared_ptr<::arrow::RecordBatch>>()>;
+
+ /// \brief Read a single row group from the file
+ ///
+ /// \see ReadRowGroupsAsync for operation details
+ ///
+ /// \param i the index of the row group to read
+ /// \param cpu_executor an executor to use to run CPU tasks
+ /// \param allow_sliced_batches if false, an error is raised if a batch has too much
+ /// data for the given batch size. If true, smaller
+ /// batches will be returned instead.
+ virtual AsyncBatchGenerator ReadRowGroupAsync(int i,
Review Comment:
At least the single-row group `ReadRowGroupAsync` is a trivial redirect to the several-row groups variant, so removing those two would be fine.
--
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-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync [arrow]
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace closed pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
URL: https://github.com/apache/arrow/pull/36779
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1296270379
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,124 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ Future<std::vector<std::shared_ptr<ChunkedArray>>> chunked_arrays_fut =
+ ::arrow::internal::OptionalParallelForAsync(
+ state_->use_threads, state_->column_readers,
+ [rows_in_batch](std::size_t, std::shared_ptr<ColumnReaderImpl> column_reader)
+ -> Result<std::shared_ptr<ChunkedArray>> {
+ std::shared_ptr<ChunkedArray> chunked_array;
+ ARROW_RETURN_NOT_OK(
+ column_reader->NextBatch(rows_in_batch, &chunked_array));
+ return chunked_array;
+ },
+ state_->cpu_executor);
+
+ // Grab the first batch of data and return it. If there is more than one batch then
+ // throw the reamining batches into overflow and they will be fetched on the next call
+ return chunked_arrays_fut.Then(
+ [state = state_,
+ rows_in_batch](const std::vector<std::shared_ptr<ChunkedArray>>& chunks)
+ -> Result<std::shared_ptr<RecordBatch>> {
+ std::shared_ptr<Table> table =
+ Table::Make(state->schema, chunks, rows_in_batch);
+ ::arrow::TableBatchReader batch_reader(*table);
+ std::shared_ptr<RecordBatch> first;
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> next_batch,
+ batch_reader.Next());
+ if (!next_batch) {
+ break;
+ }
+ if (first) {
+ if (!state->allow_sliced_batches) {
+ return Status::Invalid(
+ "The setting allow_sliced_batches is set to false and data was "
+ "encountered that was too large to fit in a single batch.");
+ }
+ state->overflow.push(std::move(next_batch));
Review Comment:
Good point. I've added some lines to the method doc to make this explicit. There isn't much point in trying to make this async-reentrant since that would require parallel reading of a row group and we don't support that. It might, in theory, be possible, but I think most users get enough parallelism from multiple files / multiple row groups.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1296268694
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1274,124 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ Future<std::vector<std::shared_ptr<ChunkedArray>>> chunked_arrays_fut =
+ ::arrow::internal::OptionalParallelForAsync(
+ state_->use_threads, state_->column_readers,
+ [rows_in_batch](std::size_t, std::shared_ptr<ColumnReaderImpl> column_reader)
+ -> Result<std::shared_ptr<ChunkedArray>> {
+ std::shared_ptr<ChunkedArray> chunked_array;
+ ARROW_RETURN_NOT_OK(
+ column_reader->NextBatch(rows_in_batch, &chunked_array));
+ return chunked_array;
Review Comment:
I added a result-returning variant.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] github-actions[bot] commented on pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #36779:
URL: https://github.com/apache/arrow/pull/36779#issuecomment-1642850484
:warning: GitHub issue #36778 **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
[GitHub] [arrow] mapleFU commented on pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on PR #36779:
URL: https://github.com/apache/arrow/pull/36779#issuecomment-1643275407
Is the logic for concat multiple batches from RowGroup implemented? Or it will be implement in the future?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] mapleFU commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "mapleFU (via GitHub)" <gi...@apache.org>.
mapleFU commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304743341
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -114,6 +116,32 @@ class ColumnReaderImpl : public ColumnReader {
return Status::OK();
}
+ ::arrow::Result<std::shared_ptr<::arrow::ChunkedArray>> NextBatch(
+ int64_t batch_size) final {
+ std::shared_ptr<::arrow::ChunkedArray> out;
+ RETURN_NOT_OK(NextBatch(batch_size, &out));
+ return out;
+ }
+
+ Future<std::shared_ptr<ChunkedArray>> NextBatchAsync(
+ int64_t batch_size, ::arrow::internal::Executor* io_executor,
+ ::arrow::internal::Executor* cpu_executor) final {
+ Future<> load_fut = ::arrow::DeferNotOk(
+ io_executor->Submit([this, batch_size] { return LoadBatch(batch_size); }));
Review Comment:
Another point is that seems that `LoadBatch` would only depends on sync-io api? So it would not cause deadlock when `LoadBatch` waiting for page io? (which could happen when `use_thread` with previous scanner?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304719989
##########
cpp/src/parquet/arrow/reader.cc:
##########
@@ -1233,6 +1285,127 @@ Status FileReaderImpl::ReadRowGroups(const std::vector<int>& row_groups,
return Status::OK();
}
+struct AsyncBatchGeneratorState {
+ ::arrow::internal::Executor* io_executor;
+ ::arrow::internal::Executor* cpu_executor;
+ std::vector<std::shared_ptr<ColumnReaderImpl>> column_readers;
+ std::vector<std::shared_ptr<ChunkedArray>> current_cols;
+ std::queue<std::shared_ptr<RecordBatch>> overflow;
+ std::shared_ptr<::arrow::Schema> schema;
+ int64_t batch_size;
+ int64_t rows_remaining;
+ bool use_threads;
+ bool allow_sliced_batches;
+};
+
+class AsyncBatchGeneratorImpl {
+ public:
+ explicit AsyncBatchGeneratorImpl(std::shared_ptr<AsyncBatchGeneratorState> state)
+ : state_(std::move(state)) {}
+ Future<std::shared_ptr<RecordBatch>> operator()() {
+ if (!state_->overflow.empty()) {
+ std::shared_ptr<RecordBatch> next = std::move(state_->overflow.front());
+ state_->overflow.pop();
+ return next;
+ }
+
+ if (state_->rows_remaining == 0) {
+ // Exhausted
+ return Future<std::shared_ptr<RecordBatch>>::MakeFinished(
+ ::arrow::IterationEnd<std::shared_ptr<RecordBatch>>());
+ }
+
+ int64_t rows_in_batch = std::min(state_->rows_remaining, state_->batch_size);
+ state_->rows_remaining -= rows_in_batch;
+
+ // We read the columns in parallel. Each reader returns a chunked array. This is
+ // because we might need to chunk a column if that column is too large. We
+ // do provide a batch size but even for a small batch size it is possible that a
+ // column has extremely large strings which don't fit in a single batch.
+ std::vector<Future<>> column_futs;
+ column_futs.reserve(state_->column_readers.size());
+ for (std::size_t column_index = 0; column_index < state_->column_readers.size();
+ column_index++) {
+ const auto& column_reader = state_->column_readers[column_index];
+ column_futs.push_back(
+ column_reader
+ ->NextBatchAsync(rows_in_batch, state_->io_executor, state_->cpu_executor)
+ .Then([state = state_,
+ column_index](const std::shared_ptr<ChunkedArray>& chunked_array) {
+ state->current_cols[column_index] = chunked_array;
+ }));
+ }
+ Future<> all_columns_finished_fut = ::arrow::AllFinished(std::move(column_futs));
+
+ // Grab the first batch of data and return it. If there is more than one batch then
+ // throw the reamining batches into overflow and they will be fetched on the next call
+ return all_columns_finished_fut.Then(
+ [state = state_, rows_in_batch]() -> Result<std::shared_ptr<RecordBatch>> {
+ std::shared_ptr<Table> table =
+ Table::Make(state->schema, state->current_cols, rows_in_batch);
+ ::arrow::TableBatchReader batch_reader(*table);
+ std::shared_ptr<RecordBatch> first;
+ while (true) {
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<RecordBatch> next_batch,
+ batch_reader.Next());
+ if (!next_batch) {
+ break;
+ }
+ if (first) {
+ if (!state->allow_sliced_batches) {
Review Comment:
I think this parameter (`allow_sliced_batches`) is documented. Am I misunderstanding?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #36779: GH-36778: [C++][Parquet] Add ReadRowGroupAsync/ReadRowGroupsAsync
Posted by "westonpace (via GitHub)" <gi...@apache.org>.
westonpace commented on code in PR #36779:
URL: https://github.com/apache/arrow/pull/36779#discussion_r1304713384
##########
cpp/src/parquet/arrow/reader.h:
##########
@@ -316,6 +374,14 @@ class PARQUET_EXPORT ColumnReader {
// the data available in the file.
virtual ::arrow::Status NextBatch(int64_t batch_size,
std::shared_ptr<::arrow::ChunkedArray>* out) = 0;
+
+ // Overload of NextBatch that returns a Result
+ virtual ::arrow::Result<std::shared_ptr<::arrow::ChunkedArray>> NextBatch(
Review Comment:
I'll remove 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