You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by GitBox <gi...@apache.org> on 2022/08/02 18:22:35 UTC
[GitHub] [arrow] westonpace opened a new pull request, #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
westonpace opened a new pull request, #13782:
URL: https://github.com/apache/arrow/pull/13782
**Primary Goal:** Create a scanner that "cancels" properly. In other words, when the scan node is marked finished then all scan-related thread tasks will be finished. This is different than the current model where I/O tasks are allowed to keep parts of the scan alive via captures of shared_ptr state.
**Secondary Goal:** Remove our dependency on the merged generator and make the scanner more accessible. The merged generator is complicated and does not support cancellation, and it currently only understood by a very small set of people.
Performance isn't a goal for this rework but ideally this should not degrade performance.
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r980546461
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,11 +38,107 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath column_reference;
+ /// \brief The requested type of the column
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. In this case the scan node will cast the column
+ /// to the appropriate type later.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
Review Comment:
It could be. I think I'd like to leave it for now and see how it looks when we add CSV support for the new 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] lidavidm commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r937908586
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -59,6 +156,17 @@ class ARROW_DS_EXPORT Fragment : public std::enable_shared_from_this<Fragment> {
virtual Result<RecordBatchGenerator> ScanBatchesAsync(
const std::shared_ptr<ScanOptions>& options) = 0;
+ /// \brief Inspect a fragment to learn basic information
Review Comment:
I don't see the problem with splitting scanning into these steps, FWIW. I think this would also let us down the line support schemes where metadata (footers) are cached in an index separately from the actual data (which would let us skip a synchronous I/O step)
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r980598407
##########
cpp/src/arrow/compute/exec/expression.h:
##########
@@ -277,6 +279,53 @@ ARROW_EXPORT Expression or_(Expression lhs, Expression rhs);
ARROW_EXPORT Expression or_(const std::vector<Expression>&);
ARROW_EXPORT Expression not_(Expression operand);
+/// Modify an Expression with pre-order and post-order visitation.
+/// `pre` will be invoked on each Expression. `pre` will visit Calls before their
+/// arguments, `post_call` will visit Calls (and no other Expressions) after their
+/// arguments. Visitors should return the Identical expression to indicate no change; this
+/// will prevent unnecessary construction in the common case where a modification is not
+/// possible/necessary/...
+///
+/// If an argument was modified, `post_call` visits a reconstructed Call with the modified
+/// arguments but also receives a pointer to the unmodified Expression as a second
+/// argument. If no arguments were modified the unmodified Expression* will be nullptr.
+template <typename PreVisit, typename PostVisitCall>
+Result<Expression> Modify(Expression expr, const PreVisit& pre,
Review Comment:
I couldn't mark the structs as static. I moved `Modify` to `compute/exec/util.h` which is also a private header and I renamed it `ModifyExpression` so it is more clear.
--
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] lidavidm commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r975605874
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan a batch of data from the file
+ /// \param batch_number The index of the batch to read
+ virtual Future<std::shared_ptr<RecordBatch>> ScanBatch(int batch_number) = 0;
+ /// \brief Calculate an estimate of how many data bytes the given batch will represent
+ ///
+ /// "Data bytes" should be the total size of all the buffers once the data has been
+ /// decoded into the Arrow format.
+ virtual int64_t EstimatedDataBytes(int batch_number) = 0;
+ /// \brief The number of batches in the fragment to scan
+ virtual int NumBatches() = 0;
+};
+
+struct InspectedFragment {
+ explicit InspectedFragment(std::vector<std::string> column_names)
+ : column_names(std::move(column_names)) {}
+ std::vector<std::string> column_names;
Review Comment:
It may make sense to keep it around; encoding everything in a schema might be unwieldy. (Though I suppose it buys you serialization for free.)
--
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] lidavidm commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r974198545
##########
cpp/src/arrow/compute/exec/expression.h:
##########
@@ -277,6 +279,53 @@ ARROW_EXPORT Expression or_(Expression lhs, Expression rhs);
ARROW_EXPORT Expression or_(const std::vector<Expression>&);
ARROW_EXPORT Expression not_(Expression operand);
+/// Modify an Expression with pre-order and post-order visitation.
+/// `pre` will be invoked on each Expression. `pre` will visit Calls before their
+/// arguments, `post_call` will visit Calls (and no other Expressions) after their
+/// arguments. Visitors should return the Identical expression to indicate no change; this
+/// will prevent unnecessary construction in the common case where a modification is not
+/// possible/necessary/...
+///
+/// If an argument was modified, `post_call` visits a reconstructed Call with the modified
+/// arguments but also receives a pointer to the unmodified Expression as a second
+/// argument. If no arguments were modified the unmodified Expression* will be nullptr.
+template <typename PreVisit, typename PostVisitCall>
+Result<Expression> Modify(Expression expr, const PreVisit& pre,
Review Comment:
Probably because those weren't declared `static`, only `inline`
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r980546636
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,11 +38,107 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath column_reference;
+ /// \brief The requested type of the column
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. In this case the scan node will cast the column
+ /// to the appropriate type later.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+/// \brief A collection of reads issued by a FragmentScanner
+struct ReadTask {
+ /// \brief A future which will complete when the read is finished
+ Future<std::shared_ptr<RecordBatch>> task;
+ /// \brief An estimate of the number of data bytes that will
+ /// be read when all the reads have finished
+ ///
+ /// This may be lower than the requested number of data bytes
+ /// if the fragment is almost finished
+ ///
+ /// This may be higher than the requested number of data bytes
+ /// if the fragment had to read more to respect record batch
+ /// boundaries
+ uint64_t num_bytes;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan further ahead into the fragment
+ /// \param target_num_bytes The number of data bytes the scanner is ready to receive
+ /// \return Futures for each read operation that has been started
+ ///
+ /// `target_num_bytes` is a hint only and fragments do not need to match this
+ /// exactly. For example, when scanning a parquet file, reads have to align
+ /// with pages and so it is often neccesary to read more than asked for.
+ /// See ScanV2Options for more details.
+ ///
+ /// It is acceptable for a future to complete with an empty batch.
+ ///
+ /// This may be called before the previous read task has finished and the caller
+ /// should always check HasUnscannedData before each call.
+ virtual ReadTask ScanMore(uint64_t target_num_bytes) = 0;
+
+ /// \brief True if there is still data that has not been scanned
+ ///
+ /// This should return true when there is no more data to start scanning.
+ /// Some scan operations may still be in progress and that is fine.
+ virtual bool HasUnscannedData() = 0;
+};
+
+struct InspectedFragment {
Review Comment:
Added
##########
cpp/src/arrow/dataset/dataset.cc:
##########
@@ -238,5 +277,112 @@ Result<FragmentIterator> UnionDataset::GetFragmentsImpl(compute::Expression pred
return GetFragmentsFromDatasets(children_, predicate);
}
+namespace {
+
+class BasicFragmentEvolution : public FragmentEvolutionStrategy {
+ public:
+ BasicFragmentEvolution(std::vector<int> ds_to_frag_map, Schema* dataset_schema)
+ : ds_to_frag_map(std::move(ds_to_frag_map)), dataset_schema(dataset_schema) {}
+
+ virtual Result<compute::Expression> GetGuarantee(
+ const std::vector<FieldPath>& dataset_schema_selection) const override {
+ std::vector<compute::Expression> missing_fields;
+ for (const FieldPath& path : dataset_schema_selection) {
+ int top_level_field_idx = path[0];
+ if (ds_to_frag_map[top_level_field_idx] < 0) {
+ missing_fields.push_back(compute::equal(
Review Comment:
Changed to `is_null`
--
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] lidavidm commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r937901142
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,11 +38,107 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath column_reference;
+ /// \brief The requested type of the column
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. In this case the scan node will cast the column
+ /// to the appropriate type later.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
Review Comment:
Could it be treated as part of the 'schema evolution' functionality, since that's already being used to fill in missing fields?
--
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] lidavidm commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r974201441
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan a batch of data from the file
+ /// \param batch_number The index of the batch to read
+ virtual Future<std::shared_ptr<RecordBatch>> ScanBatch(int batch_number) = 0;
+ /// \brief Calculate an estimate of how many data bytes the given batch will represent
+ ///
+ /// "Data bytes" should be the total size of all the buffers once the data has been
+ /// decoded into the Arrow format.
+ virtual int64_t EstimatedDataBytes(int batch_number) = 0;
+ /// \brief The number of batches in the fragment to scan
+ virtual int NumBatches() = 0;
Review Comment:
Hmm, and I guess `(fragment_index, batch_index, is_last)` pushes too much complexity into the ordered execution bits?
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r980546757
##########
cpp/src/arrow/dataset/dataset.cc:
##########
@@ -238,5 +277,112 @@ Result<FragmentIterator> UnionDataset::GetFragmentsImpl(compute::Expression pred
return GetFragmentsFromDatasets(children_, predicate);
}
+namespace {
+
+class BasicFragmentEvolution : public FragmentEvolutionStrategy {
+ public:
+ BasicFragmentEvolution(std::vector<int> ds_to_frag_map, Schema* dataset_schema)
+ : ds_to_frag_map(std::move(ds_to_frag_map)), dataset_schema(dataset_schema) {}
+
+ virtual Result<compute::Expression> GetGuarantee(
+ const std::vector<FieldPath>& dataset_schema_selection) const override {
+ std::vector<compute::Expression> missing_fields;
+ for (const FieldPath& path : dataset_schema_selection) {
+ int top_level_field_idx = path[0];
+ if (ds_to_frag_map[top_level_field_idx] < 0) {
+ missing_fields.push_back(compute::equal(
+ compute::field_ref(top_level_field_idx),
+ compute::literal(
+ MakeNullScalar(dataset_schema->fields()[top_level_field_idx]->type()))));
+ }
+ }
+ if (missing_fields.empty()) {
+ return compute::literal(true);
+ }
+ if (missing_fields.size() == 1) {
+ return missing_fields[0];
+ }
+ return compute::and_(missing_fields);
Review Comment:
Added
##########
cpp/src/arrow/dataset/scan_node.cc:
##########
@@ -0,0 +1,653 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/dataset/scanner.h"
+
+#include <functional>
+#include <iostream>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "arrow/compute/exec/exec_plan.h"
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/scanner.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/make_unique.h"
+#include "arrow/util/tracing_internal.h"
+#include "arrow/util/unreachable.h"
+
+namespace cp = arrow::compute;
+
+namespace arrow {
+
+using internal::checked_cast;
+
+namespace dataset {
+
+namespace {
+
+Result<std::shared_ptr<Schema>> OutputSchemaFromOptions(const ScanV2Options& options) {
+ return FieldPath::GetAll(*options.dataset->schema(), options.columns);
+}
+
+class BatchOutputStrategy {
+ public:
+ BatchOutputStrategy(compute::ExecNode* self, compute::ExecNode* output)
+ : self_(self), output_(output) {}
+ virtual ~BatchOutputStrategy() = default;
+ virtual void DeliverBatch(compute::ExecBatch batch) = 0;
+ virtual void InputFinished() {
+ // std::cout << "BatchOutputFinished: " + std::to_string(batch_count_.load()) + "\n";
Review Comment:
All commented code should now be removed
--
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] ursabot commented on pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
ursabot commented on PR #13782:
URL: https://github.com/apache/arrow/pull/13782#issuecomment-1265535614
Benchmark runs are scheduled for baseline = 89c0214fa43f8d1bf2e19e3bae0fc3009df51e15 and contender = ec579df631deaa8f6186208ed2a4ebec00581dfa. ec579df631deaa8f6186208ed2a4ebec00581dfa is a master commit associated with this PR. Results will be available as each benchmark for each run completes.
Conbench compare runs links:
[Finished :arrow_down:0.0% :arrow_up:0.0%] [ec2-t3-xlarge-us-east-2](https://conbench.ursa.dev/compare/runs/b1e30b1cf1a846fa9892b5d98272eda4...757428c57b274461a28163f012481ea7/)
[Failed :arrow_down:0.03% :arrow_up:0.0%] [test-mac-arm](https://conbench.ursa.dev/compare/runs/07a24e3b60764c1bbe193b0850e58c06...514f5a4797c14133a1aec775dfd3c553/)
[Failed :arrow_down:0.27% :arrow_up:0.27%] [ursa-i9-9960x](https://conbench.ursa.dev/compare/runs/09a08cb4ff0d4f1a912a0deb577be8f6...7fe29ead92fa4d8594c332f6dcee1f1f/)
[Finished :arrow_down:1.45% :arrow_up:0.0%] [ursa-thinkcentre-m75q](https://conbench.ursa.dev/compare/runs/a98d6ee8952f4c85b70b332b96dcb9fb...86712dfdfae54be38adc5f8f7d2f88d4/)
Buildkite builds:
[Finished] [`ec579df6` ec2-t3-xlarge-us-east-2](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ec2-t3-xlarge-us-east-2/builds/1609)
[Failed] [`ec579df6` test-mac-arm](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-test-mac-arm/builds/1628)
[Failed] [`ec579df6` ursa-i9-9960x](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-i9-9960x/builds/1611)
[Finished] [`ec579df6` ursa-thinkcentre-m75q](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-thinkcentre-m75q/builds/1622)
[Finished] [`89c0214f` ec2-t3-xlarge-us-east-2](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ec2-t3-xlarge-us-east-2/builds/1608)
[Finished] [`89c0214f` test-mac-arm](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-test-mac-arm/builds/1627)
[Failed] [`89c0214f` ursa-i9-9960x](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-i9-9960x/builds/1610)
[Finished] [`89c0214f` ursa-thinkcentre-m75q](https://buildkite.com/apache-arrow/arrow-bci-benchmark-on-ursa-thinkcentre-m75q/builds/1621)
Supported benchmarks:
ec2-t3-xlarge-us-east-2: Supported benchmark langs: Python, R. Runs only benchmarks with cloud = True
test-mac-arm: Supported benchmark langs: C++, Python, R
ursa-i9-9960x: Supported benchmark langs: Python, R, JavaScript
ursa-thinkcentre-m75q: Supported benchmark langs: C++, Java
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r980547896
##########
cpp/src/arrow/util/async_util.cc:
##########
@@ -274,7 +282,9 @@ class AsyncTaskSchedulerImpl : public AsyncTaskScheduler {
AbortUnlocked(submit_result.status(), std::move(lk));
return;
}
- submit_result->AddCallback([this, cost](const Status& st) {
+ // FIXME(C++17, move into lambda?)
Review Comment:
Yep. Although, in retrospect, I'm not sure how it works, since `std::function` is copyable (maybe it works just because we don't happen to copy 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 merged pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace merged PR #13782:
URL: https://github.com/apache/arrow/pull/13782
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r974180407
##########
cpp/src/arrow/compute/exec/expression.h:
##########
@@ -277,6 +279,53 @@ ARROW_EXPORT Expression or_(Expression lhs, Expression rhs);
ARROW_EXPORT Expression or_(const std::vector<Expression>&);
ARROW_EXPORT Expression not_(Expression operand);
+/// Modify an Expression with pre-order and post-order visitation.
+/// `pre` will be invoked on each Expression. `pre` will visit Calls before their
+/// arguments, `post_call` will visit Calls (and no other Expressions) after their
+/// arguments. Visitors should return the Identical expression to indicate no change; this
+/// will prevent unnecessary construction in the common case where a modification is not
+/// possible/necessary/...
+///
+/// If an argument was modified, `post_call` visits a reconstructed Call with the modified
+/// arguments but also receives a pointer to the unmodified Expression as a second
+/// argument. If no arguments were modified the unmodified Expression* will be nullptr.
+template <typename PreVisit, typename PostVisitCall>
+Result<Expression> Modify(Expression expr, const PreVisit& pre,
Review Comment:
If I didn't then I ended up needing `expression_internal.h` in `partition.cc`. I'm not entirely sure why but including `expression_internal.h` in two different .cc files in the same module caused duplicate symbol errors. Not on `Modify` (which should be fine because it is templated) but on things like `KnownFieldValues` and `CallNotNull`. Adding `Modify` didn't require any additional includes in `expression.h` and so it doesn't seem too heavyweight an addition but I could be missing some reason we didn't want it exposed.
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #13782:
URL: https://github.com/apache/arrow/pull/13782#issuecomment-1203069034
:warning: Ticket **has not been started in JIRA**, please click 'Start Progress'.
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r980545312
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,11 +38,107 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
Review Comment:
Added
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,11 +38,107 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath column_reference;
+ /// \brief The requested type of the column
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. In this case the scan node will cast the column
+ /// to the appropriate type later.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
Review Comment:
Added (here and elsewhere)
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r937901786
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -59,6 +156,17 @@ class ARROW_DS_EXPORT Fragment : public std::enable_shared_from_this<Fragment> {
virtual Result<RecordBatchGenerator> ScanBatchesAsync(
const std::shared_ptr<ScanOptions>& options) = 0;
+ /// \brief Inspect a fragment to learn basic information
Review Comment:
Yes, exactly. I'm not a huge fan of this because now fragment scanning is broken into three steps...
FragmentScanner: Inspect fragment (new step)
Scanner: Create evolution
FragmentScanner: Open reader (where we previously determined column names / schema)
Scanner: Start iteration task
FragmentScanner: Scan
I could push the evolution creation into the fragment scanner but that makes a new step that every fragment scanner has to do and so I think that would be worse.
Your point on "actual parquet footer" and "the schema" is also correct. My "prototypical future evolution strategy" that I am trying to make sure we plan for is to resolve field references by parquet column id. So it would work in this fashion:
* User specifies numeric field references against the dataset schema
* Evolution strategy has a lookup table from numeric position in dataset schema to column id
* On inspection, physical column ids are inserted into the schema and passed to the evolution
* Evolution uses the lookup table combined with the physical column ids to determine the physical columns that are being asked for using
So this way the inspect step would need to return something with the column IDs. This could be the parquet footer. However, it also could be the schema extracted from the footer as we have a mechanism for encoding those column IDs into the schema today. This way we could leave the door open to someone adding column IDs to IPC files as well.
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r975585346
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan a batch of data from the file
+ /// \param batch_number The index of the batch to read
+ virtual Future<std::shared_ptr<RecordBatch>> ScanBatch(int batch_number) = 0;
+ /// \brief Calculate an estimate of how many data bytes the given batch will represent
+ ///
+ /// "Data bytes" should be the total size of all the buffers once the data has been
+ /// decoded into the Arrow format.
+ virtual int64_t EstimatedDataBytes(int batch_number) = 0;
+ /// \brief The number of batches in the fragment to scan
+ virtual int NumBatches() = 0;
+};
+
+struct InspectedFragment {
+ explicit InspectedFragment(std::vector<std::string> column_names)
+ : column_names(std::move(column_names)) {}
+ std::vector<std::string> column_names;
Review Comment:
My original thinking was that, with CSV, we can't know the types at this point. Plus, for the basic evolution strategy, we don't need to know the types. With that strategy all we need to know is the column names.
However, looking at this with fresh eyes, I think we do know the types with CSV. We don't re-infer with each new fragment and we assert that the schema matches the dataset schema (you cannot do a type-widening evolution with CSV for example).
In that case maybe I can get rid of `InspectedFragment` entirely.
The other reason for `InspectedFragment` was for the case where we know more than types+names. For example, we might know the parquet column ids for each column. However, we should be able to encode that information in a schema.
--
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] lidavidm commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r974199255
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan a batch of data from the file
+ /// \param batch_number The index of the batch to read
+ virtual Future<std::shared_ptr<RecordBatch>> ScanBatch(int batch_number) = 0;
+ /// \brief Calculate an estimate of how many data bytes the given batch will represent
+ ///
+ /// "Data bytes" should be the total size of all the buffers once the data has been
+ /// decoded into the Arrow format.
+ virtual int64_t EstimatedDataBytes(int batch_number) = 0;
+ /// \brief The number of batches in the fragment to scan
+ virtual int NumBatches() = 0;
Review Comment:
There is no way to know the number of batches, no. And I'd like to avoid reimplementing parallel scans for Flight/ADBC data sources…
--
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] lidavidm commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r936983246
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,11 +38,107 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
Review Comment:
```suggestion
struct ARROW_DS_EXPORT FragmentSelectionColumn {
```
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,11 +38,107 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath column_reference;
+ /// \brief The requested type of the column
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. In this case the scan node will cast the column
+ /// to the appropriate type later.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
Review Comment:
```suggestion
struct ARROW_DS_EXPORT FragmentScanRequest {
```
##########
cpp/src/arrow/dataset/scan_node.cc:
##########
@@ -0,0 +1,653 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/dataset/scanner.h"
+
+#include <functional>
+#include <iostream>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "arrow/compute/exec/exec_plan.h"
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/scanner.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/make_unique.h"
+#include "arrow/util/tracing_internal.h"
+#include "arrow/util/unreachable.h"
+
+namespace cp = arrow::compute;
+
+namespace arrow {
+
+using internal::checked_cast;
+
+namespace dataset {
+
+namespace {
+
+Result<std::shared_ptr<Schema>> OutputSchemaFromOptions(const ScanV2Options& options) {
+ return FieldPath::GetAll(*options.dataset->schema(), options.columns);
+}
+
+class BatchOutputStrategy {
+ public:
+ BatchOutputStrategy(compute::ExecNode* self, compute::ExecNode* output)
+ : self_(self), output_(output) {}
+ virtual ~BatchOutputStrategy() = default;
+ virtual void DeliverBatch(compute::ExecBatch batch) = 0;
+ virtual void InputFinished() {
+ // std::cout << "BatchOutputFinished: " + std::to_string(batch_count_.load()) + "\n";
+ output_->InputFinished(self_, batch_count_.load());
+ }
+
+ protected:
+ void SliceAndDoDeliverBatch(compute::ExecBatch batch) {
+ // FIXME - Add slicing
+ batch_count_++;
+ output_->InputReceived(self_, std::move(batch));
+ }
+
+ std::atomic<int32_t> batch_count_{0};
+ compute::ExecNode* self_;
+ compute::ExecNode* output_;
+};
+
+class UnorderedBatchOutputStrategy : public BatchOutputStrategy {
+ public:
+ UnorderedBatchOutputStrategy(compute::ExecNode* self, compute::ExecNode* output)
+ : BatchOutputStrategy(self, output){};
+ ~UnorderedBatchOutputStrategy() = default;
+ void DeliverBatch(compute::ExecBatch batch) override {
+ return SliceAndDoDeliverBatch(std::move(batch));
+ }
+};
+
+// Unlike other nodes the scanner has to synchronize a number of resources
+// that are external to the plan itself. These things may start tasks that
+// are invisible to the exec plan.
+class ScannerSynchronization {
+ public:
+ ScannerSynchronization(int fragment_readahead)
+ : fragment_readahead_(fragment_readahead) {}
+
+ bool MarkListingCompleted() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ finished_listing_ = true;
+ return fragments_active_ == 0;
+ }
+
+ void MarkListingFailed(Status err) {
+ std::lock_guard<std::mutex> lg(mutex_);
+ finished_listing_ = true;
+ cancelled_ = true;
+ error_ = std::move(err);
+ }
+
+ void Cancel() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (!cancelled_) {
+ cancelled_ = true;
+ error_ = Status::Cancelled("Plan cancelled before scan completed");
+ }
+ }
+
+ void Pause() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ paused_ = true;
+ }
+
+ void ReportFailedFragment(Status st) {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (!cancelled_) {
+ cancelled_ = true;
+ error_ = std::move(st);
+ fragments_active_--;
+ }
+ }
+
+ struct NextFragment {
+ std::shared_ptr<Fragment> fragment;
+ bool should_mark_input_finished;
+ };
+
+ NextFragment FinishFragmentAndGetNext() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (cancelled_ || paused_ || fragments_to_scan_.empty()) {
+ fragments_active_--;
+ // std::cout << "Not continuing fragment inline fragments_active=" +
+ // std::to_string(fragments_active_) + "\n";
+ return {nullptr, /*should_mark_input_finished=*/finished_listing_};
+ }
+ // std::cout << "Continuing fragment inline " + std::to_string(fragments_active_) +
+ // "\n";
+ std::shared_ptr<Fragment> next = std::move(fragments_to_scan_.front());
+ fragments_to_scan_.pop();
+ return {next, /*should_mark_input_finished=*/false};
+ }
+
+ bool QueueFragment(std::shared_ptr<Fragment> fragment) {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (cancelled_) {
+ return false;
+ }
+ if (!paused_ && fragments_active_ < fragment_readahead_) {
+ // If we have space start scanning the fragment immediately
+ fragments_active_++;
+ // std::cout << "Start scanning immediately: " + std::to_string(fragments_active_) +
+ // "\n";
+ return true;
+ }
+ fragments_to_scan_.push(std::move(fragment));
+ return false;
+ }
+
+ private:
+ int fragment_readahead_;
+ std::mutex mutex_;
+ std::queue<std::shared_ptr<Fragment>> fragments_to_scan_;
+ Status error_;
+ bool cancelled_ = false;
+ bool finished_listing_ = false;
+ int fragments_active_ = 0;
+ bool paused_ = false;
+};
+
+// In the future we should support async scanning of fragments. The
+// Dataset class doesn't support this yet but we pretend it does here to
+// ease future adoption of the feature.
+AsyncGenerator<std::shared_ptr<Fragment>> GetFragments(Dataset* dataset,
+ cp::Expression predicate) {
+ // In the future the dataset should be responsible for figuring out
+ // the I/O context. This will allow different I/O contexts to be used
+ // when scanning different datasets. For example, if we are scanning a
+ // union of a remote dataset and a local dataset.
+ const auto& io_context = io::default_io_context();
+ auto io_executor = io_context.executor();
+ Future<std::shared_ptr<FragmentIterator>> fragments_it_fut =
+ DeferNotOk(io_executor->Submit(
+ [dataset, predicate]() -> Result<std::shared_ptr<FragmentIterator>> {
+ // std::cout << "dataset->GetFragments\n";
+ ARROW_ASSIGN_OR_RAISE(FragmentIterator fragments_iter,
+ dataset->GetFragments(predicate));
+ return std::make_shared<FragmentIterator>(std::move(fragments_iter));
+ }));
+ Future<AsyncGenerator<std::shared_ptr<Fragment>>> fragments_gen_fut =
+ fragments_it_fut.Then([](const std::shared_ptr<FragmentIterator>& fragments_it)
+ -> Result<AsyncGenerator<std::shared_ptr<Fragment>>> {
+ ARROW_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<Fragment>> fragments,
+ fragments_it->ToVector());
+ return MakeVectorGenerator(std::move(fragments));
+ });
+ return MakeFromFuture(std::move(fragments_gen_fut));
+}
+
+/// \brief A node that scans a dataset
+///
+/// The scan node has three groups of io-tasks and one task.
+///
+/// The first io-task (listing) fetches the fragments from the dataset. This may be a
+/// simple iteration of paths or, if the dataset is described with wildcards, this may
+/// involve I/O for listing and walking directory paths. There is one listing io-task per
+/// dataset.
+///
+/// Ths next step is to fetch the metadata for the fragment. For some formats (e.g. CSV)
+/// this may be quite simple (get the size of the file). For other formats (e.g. parquet)
+/// this is more involved and requires reading data. There is one metadata io-task per
+/// fragment. The metadata io-task creates an AsyncGenerator<RecordBatch> from the
+/// fragment.
+///
+/// Once the metadata io-task is done we can issue read io-tasks. Each read io-task
+/// requests a single batch of data from the disk by pulling the next Future from the
+/// generator.
+///
+/// Finally, when the future is fulfilled, we issue a pipeline task to drive the batch
+/// through the pipeline.
+///
+/// Most of these tasks are io-tasks. They take very few CPU resources and they run on
+/// the I/O thread pool. These io-tasks are invisible to the exec plan and so we need to
+/// do some custom scheduling. We limit how many fragments we read from at any one time.
+/// This is referred to as "fragment readahead".
+///
+/// Within a fragment there is usually also some amount of "row readahead". This row
+/// readahead is handled by the fragment (and not the scanner) because the exact details
+/// of how it is performed depend on the underlying format.
+///
+/// When a scan node is aborted (StopProducing) we send a cancel signal to any active
+/// fragments. On destruction we continue consuming the fragments until they complete
+/// (which should be fairly quick since we cancelled the fragment). This ensures the
+/// I/O work is completely finished before the node is destroyed.
+class ScanNode : public cp::ExecNode {
+ public:
+ ScanNode(cp::ExecPlan* plan, ScanV2Options options,
+ std::shared_ptr<Schema> output_schema)
+ : cp::ExecNode(plan, {}, {}, std::move(output_schema),
+ /*num_outputs=*/1),
+ options_(options),
+ synchronization_(options_.fragment_readahead) {}
+
+ static Result<ScanV2Options> NormalizeAndValidate(const ScanV2Options& options,
+ compute::ExecContext* ctx) {
+ ScanV2Options normalized(options);
+ if (!normalized.dataset) {
+ return Status::Invalid("Scan options must include a dataset");
+ }
+
+ if (normalized.filter.IsBound()) {
+ // There is no easy way to make sure a filter was bound agaisnt the same
+ // function registry as the one in ctx so we just require it to be unbound
+ // FIXME - Do we care if it was bound to a different function registry?
+ return Status::Invalid("Scan filter must be unbound");
+ } else {
+ ARROW_ASSIGN_OR_RAISE(normalized.filter,
+ normalized.filter.Bind(*options.dataset->schema(), ctx));
+ }
+
+ return std::move(normalized);
+ }
+
+ static Result<cp::ExecNode*> Make(cp::ExecPlan* plan, std::vector<cp::ExecNode*> inputs,
+ const cp::ExecNodeOptions& options) {
+ RETURN_NOT_OK(ValidateExecNodeInputs(plan, inputs, 0, "ScanNode"));
+ const auto& scan_options = checked_cast<const ScanV2Options&>(options);
+ ARROW_ASSIGN_OR_RAISE(ScanV2Options normalized_options,
+ NormalizeAndValidate(scan_options, plan->exec_context()));
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Schema> output_schema,
+ OutputSchemaFromOptions(normalized_options));
+ return plan->EmplaceNode<ScanNode>(plan, std::move(normalized_options),
+ std::move(output_schema));
+ }
+
+ const char* kind_name() const override { return "ScanNode"; }
+
+ [[noreturn]] static void NoInputs() {
+ Unreachable("no inputs; this should never be called");
+ }
+ [[noreturn]] void InputReceived(cp::ExecNode*, cp::ExecBatch) override { NoInputs(); }
+ [[noreturn]] void ErrorReceived(cp::ExecNode*, Status) override { NoInputs(); }
+ [[noreturn]] void InputFinished(cp::ExecNode*, int) override { NoInputs(); }
+
+ Status Init() override {
+ batch_output_ =
+ ::arrow::internal::make_unique<UnorderedBatchOutputStrategy>(this, outputs_[0]);
+ return Status::OK();
+ }
+
+ Status StartProducing() override {
+ START_COMPUTE_SPAN(span_, std::string(kind_name()) + ":" + label(),
+ {{"node.kind", kind_name()},
+ {"node.label", label()},
+ {"node.output_schema", output_schema()->ToString()},
+ {"node.detail", ToString()}});
+ END_SPAN_ON_FUTURE_COMPLETION(span_, finished_);
+ // std::cout << "StartProducing\n";
+ ARROW_ASSIGN_OR_RAISE(Future<> list_task, plan_->BeginExternalTask());
+ if (!list_task.is_valid()) {
+ // Cancelled before we even started
+ return Status::OK();
+ }
+ // std::cout << "LIST TASK: BEGIN\n";
+
+ AsyncGenerator<std::shared_ptr<Fragment>> frag_gen =
+ GetFragments(options_.dataset.get(), options_.filter);
+ Future<> visit_task = VisitAsyncGenerator(
+ std::move(frag_gen), [this](const std::shared_ptr<Fragment>& fragment) {
+ // std::cout << "Queuing fragment: " + fragment->ToString() + "\n";
+ QueueOrStartFragment(fragment);
+ return Status::OK();
+ });
+ visit_task
+ .Then(
+ [this]() {
+ // std::cout << "Marking listing completed\n";
+ if (synchronization_.MarkListingCompleted()) {
+ // It's possible, but probably unlikely, that all fragment scans have
+ // already finished and so we need to report the total batch count here in
+ // this case
+ batch_output_->InputFinished();
+ finished_.MarkFinished();
+ }
+ },
+ [this](const Status& err) {
+ synchronization_.MarkListingFailed(err);
+ outputs_[0]->ErrorReceived(this, err);
+ })
+ .AddCallback([list_task](const Status& st) mutable {
+ // std::cout << "LIST TASK: END\n";
+ list_task.MarkFinished();
+ });
+
+ return Status::OK();
+ }
+
+ void PauseProducing(ExecNode* output, int32_t counter) override {
+ // FIXME(TODO)
+ // Need to ressurect AsyncToggle and then all fragment scanners
+ // should share the same toggle
+ }
+
+ void ResumeProducing(ExecNode* output, int32_t counter) override {
+ // FIXME(TODO)
+ }
+
+ void StopProducing(ExecNode* output) override {
+ DCHECK_EQ(output, outputs_[0]);
+ StopProducing();
+ }
+
+ void StopProducing() override { synchronization_.Cancel(); }
+
+ private:
+ struct FragmentScanTask;
+ using ScanTaskPtr = std::list<FragmentScanTask>::iterator;
+ struct FragmentScanTask {
+ std::unique_ptr<FragmentEvolutionStrategy> fragment_evolution;
+ std::shared_ptr<FragmentScanner> fragment_scanner;
+ uint64_t unused_readahead_bytes;
+ bool iterating_tasks = false;
+ // Transitions to true when we've finished listing fragments for this task
+ bool finished = false;
+ // Transitions to true when we receieve an error or cancellation is requested
Review Comment:
Might it make sense to have an enum instead of individual flags?
##########
cpp/src/arrow/dataset/dataset.cc:
##########
@@ -238,5 +277,112 @@ Result<FragmentIterator> UnionDataset::GetFragmentsImpl(compute::Expression pred
return GetFragmentsFromDatasets(children_, predicate);
}
+namespace {
+
+class BasicFragmentEvolution : public FragmentEvolutionStrategy {
+ public:
+ BasicFragmentEvolution(std::vector<int> ds_to_frag_map, Schema* dataset_schema)
+ : ds_to_frag_map(std::move(ds_to_frag_map)), dataset_schema(dataset_schema) {}
+
+ virtual Result<compute::Expression> GetGuarantee(
+ const std::vector<FieldPath>& dataset_schema_selection) const override {
+ std::vector<compute::Expression> missing_fields;
+ for (const FieldPath& path : dataset_schema_selection) {
+ int top_level_field_idx = path[0];
+ if (ds_to_frag_map[top_level_field_idx] < 0) {
+ missing_fields.push_back(compute::equal(
Review Comment:
I think the expression simplification only recognizes `is_null(x)`
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,11 +38,107 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath column_reference;
+ /// \brief The requested type of the column
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. In this case the scan node will cast the column
+ /// to the appropriate type later.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+/// \brief A collection of reads issued by a FragmentScanner
+struct ReadTask {
+ /// \brief A future which will complete when the read is finished
+ Future<std::shared_ptr<RecordBatch>> task;
+ /// \brief An estimate of the number of data bytes that will
+ /// be read when all the reads have finished
+ ///
+ /// This may be lower than the requested number of data bytes
+ /// if the fragment is almost finished
+ ///
+ /// This may be higher than the requested number of data bytes
+ /// if the fragment had to read more to respect record batch
+ /// boundaries
+ uint64_t num_bytes;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan further ahead into the fragment
+ /// \param target_num_bytes The number of data bytes the scanner is ready to receive
+ /// \return Futures for each read operation that has been started
+ ///
+ /// `target_num_bytes` is a hint only and fragments do not need to match this
+ /// exactly. For example, when scanning a parquet file, reads have to align
+ /// with pages and so it is often neccesary to read more than asked for.
+ /// See ScanV2Options for more details.
+ ///
+ /// It is acceptable for a future to complete with an empty batch.
+ ///
+ /// This may be called before the previous read task has finished and the caller
+ /// should always check HasUnscannedData before each call.
+ virtual ReadTask ScanMore(uint64_t target_num_bytes) = 0;
+
+ /// \brief True if there is still data that has not been scanned
+ ///
+ /// This should return true when there is no more data to start scanning.
+ /// Some scan operations may still be in progress and that is fine.
+ virtual bool HasUnscannedData() = 0;
+};
+
+struct InspectedFragment {
Review Comment:
nit: docstring?
##########
cpp/src/arrow/dataset/dataset.cc:
##########
@@ -238,5 +277,112 @@ Result<FragmentIterator> UnionDataset::GetFragmentsImpl(compute::Expression pred
return GetFragmentsFromDatasets(children_, predicate);
}
+namespace {
+
+class BasicFragmentEvolution : public FragmentEvolutionStrategy {
+ public:
+ BasicFragmentEvolution(std::vector<int> ds_to_frag_map, Schema* dataset_schema)
+ : ds_to_frag_map(std::move(ds_to_frag_map)), dataset_schema(dataset_schema) {}
+
+ virtual Result<compute::Expression> GetGuarantee(
+ const std::vector<FieldPath>& dataset_schema_selection) const override {
+ std::vector<compute::Expression> missing_fields;
+ for (const FieldPath& path : dataset_schema_selection) {
+ int top_level_field_idx = path[0];
+ if (ds_to_frag_map[top_level_field_idx] < 0) {
+ missing_fields.push_back(compute::equal(
+ compute::field_ref(top_level_field_idx),
+ compute::literal(
+ MakeNullScalar(dataset_schema->fields()[top_level_field_idx]->type()))));
+ }
+ }
+ if (missing_fields.empty()) {
+ return compute::literal(true);
+ }
+ if (missing_fields.size() == 1) {
+ return missing_fields[0];
+ }
+ return compute::and_(missing_fields);
Review Comment:
nit: std::move?
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,11 +38,107 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath column_reference;
+ /// \brief The requested type of the column
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. In this case the scan node will cast the column
+ /// to the appropriate type later.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
Review Comment:
Hmm, wouldn't we want to keep this responsibility in a more common place?
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,11 +38,107 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath column_reference;
+ /// \brief The requested type of the column
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. In this case the scan node will cast the column
+ /// to the appropriate type later.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
Review Comment:
etc.
##########
cpp/src/arrow/dataset/scan_node.cc:
##########
@@ -0,0 +1,653 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/dataset/scanner.h"
+
+#include <functional>
+#include <iostream>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "arrow/compute/exec/exec_plan.h"
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/scanner.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/make_unique.h"
+#include "arrow/util/tracing_internal.h"
+#include "arrow/util/unreachable.h"
+
+namespace cp = arrow::compute;
+
+namespace arrow {
+
+using internal::checked_cast;
+
+namespace dataset {
+
+namespace {
+
+Result<std::shared_ptr<Schema>> OutputSchemaFromOptions(const ScanV2Options& options) {
+ return FieldPath::GetAll(*options.dataset->schema(), options.columns);
+}
+
+class BatchOutputStrategy {
+ public:
+ BatchOutputStrategy(compute::ExecNode* self, compute::ExecNode* output)
+ : self_(self), output_(output) {}
+ virtual ~BatchOutputStrategy() = default;
+ virtual void DeliverBatch(compute::ExecBatch batch) = 0;
+ virtual void InputFinished() {
+ // std::cout << "BatchOutputFinished: " + std::to_string(batch_count_.load()) + "\n";
+ output_->InputFinished(self_, batch_count_.load());
+ }
+
+ protected:
+ void SliceAndDoDeliverBatch(compute::ExecBatch batch) {
+ // FIXME - Add slicing
+ batch_count_++;
+ output_->InputReceived(self_, std::move(batch));
+ }
+
+ std::atomic<int32_t> batch_count_{0};
+ compute::ExecNode* self_;
+ compute::ExecNode* output_;
+};
+
+class UnorderedBatchOutputStrategy : public BatchOutputStrategy {
+ public:
+ UnorderedBatchOutputStrategy(compute::ExecNode* self, compute::ExecNode* output)
+ : BatchOutputStrategy(self, output){};
+ ~UnorderedBatchOutputStrategy() = default;
+ void DeliverBatch(compute::ExecBatch batch) override {
+ return SliceAndDoDeliverBatch(std::move(batch));
+ }
+};
+
+// Unlike other nodes the scanner has to synchronize a number of resources
+// that are external to the plan itself. These things may start tasks that
+// are invisible to the exec plan.
+class ScannerSynchronization {
+ public:
+ ScannerSynchronization(int fragment_readahead)
+ : fragment_readahead_(fragment_readahead) {}
+
+ bool MarkListingCompleted() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ finished_listing_ = true;
+ return fragments_active_ == 0;
+ }
+
+ void MarkListingFailed(Status err) {
+ std::lock_guard<std::mutex> lg(mutex_);
+ finished_listing_ = true;
+ cancelled_ = true;
+ error_ = std::move(err);
+ }
+
+ void Cancel() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (!cancelled_) {
+ cancelled_ = true;
+ error_ = Status::Cancelled("Plan cancelled before scan completed");
+ }
+ }
+
+ void Pause() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ paused_ = true;
+ }
+
+ void ReportFailedFragment(Status st) {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (!cancelled_) {
+ cancelled_ = true;
+ error_ = std::move(st);
+ fragments_active_--;
+ }
+ }
+
+ struct NextFragment {
+ std::shared_ptr<Fragment> fragment;
+ bool should_mark_input_finished;
+ };
+
+ NextFragment FinishFragmentAndGetNext() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (cancelled_ || paused_ || fragments_to_scan_.empty()) {
+ fragments_active_--;
+ // std::cout << "Not continuing fragment inline fragments_active=" +
+ // std::to_string(fragments_active_) + "\n";
+ return {nullptr, /*should_mark_input_finished=*/finished_listing_};
+ }
+ // std::cout << "Continuing fragment inline " + std::to_string(fragments_active_) +
+ // "\n";
+ std::shared_ptr<Fragment> next = std::move(fragments_to_scan_.front());
+ fragments_to_scan_.pop();
+ return {next, /*should_mark_input_finished=*/false};
+ }
+
+ bool QueueFragment(std::shared_ptr<Fragment> fragment) {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (cancelled_) {
+ return false;
+ }
+ if (!paused_ && fragments_active_ < fragment_readahead_) {
+ // If we have space start scanning the fragment immediately
+ fragments_active_++;
+ // std::cout << "Start scanning immediately: " + std::to_string(fragments_active_) +
+ // "\n";
+ return true;
+ }
+ fragments_to_scan_.push(std::move(fragment));
+ return false;
+ }
+
+ private:
+ int fragment_readahead_;
+ std::mutex mutex_;
+ std::queue<std::shared_ptr<Fragment>> fragments_to_scan_;
+ Status error_;
+ bool cancelled_ = false;
+ bool finished_listing_ = false;
+ int fragments_active_ = 0;
+ bool paused_ = false;
+};
+
+// In the future we should support async scanning of fragments. The
+// Dataset class doesn't support this yet but we pretend it does here to
+// ease future adoption of the feature.
+AsyncGenerator<std::shared_ptr<Fragment>> GetFragments(Dataset* dataset,
+ cp::Expression predicate) {
+ // In the future the dataset should be responsible for figuring out
+ // the I/O context. This will allow different I/O contexts to be used
+ // when scanning different datasets. For example, if we are scanning a
+ // union of a remote dataset and a local dataset.
+ const auto& io_context = io::default_io_context();
+ auto io_executor = io_context.executor();
+ Future<std::shared_ptr<FragmentIterator>> fragments_it_fut =
+ DeferNotOk(io_executor->Submit(
+ [dataset, predicate]() -> Result<std::shared_ptr<FragmentIterator>> {
+ // std::cout << "dataset->GetFragments\n";
+ ARROW_ASSIGN_OR_RAISE(FragmentIterator fragments_iter,
+ dataset->GetFragments(predicate));
+ return std::make_shared<FragmentIterator>(std::move(fragments_iter));
+ }));
+ Future<AsyncGenerator<std::shared_ptr<Fragment>>> fragments_gen_fut =
+ fragments_it_fut.Then([](const std::shared_ptr<FragmentIterator>& fragments_it)
+ -> Result<AsyncGenerator<std::shared_ptr<Fragment>>> {
+ ARROW_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<Fragment>> fragments,
+ fragments_it->ToVector());
+ return MakeVectorGenerator(std::move(fragments));
+ });
+ return MakeFromFuture(std::move(fragments_gen_fut));
+}
+
+/// \brief A node that scans a dataset
Review Comment:
Thanks for the great explanation here!
##########
cpp/src/arrow/dataset/scan_node.cc:
##########
@@ -0,0 +1,653 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/dataset/scanner.h"
+
+#include <functional>
+#include <iostream>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "arrow/compute/exec/exec_plan.h"
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/scanner.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/make_unique.h"
+#include "arrow/util/tracing_internal.h"
+#include "arrow/util/unreachable.h"
+
+namespace cp = arrow::compute;
+
+namespace arrow {
+
+using internal::checked_cast;
+
+namespace dataset {
+
+namespace {
+
+Result<std::shared_ptr<Schema>> OutputSchemaFromOptions(const ScanV2Options& options) {
+ return FieldPath::GetAll(*options.dataset->schema(), options.columns);
+}
+
+class BatchOutputStrategy {
+ public:
+ BatchOutputStrategy(compute::ExecNode* self, compute::ExecNode* output)
+ : self_(self), output_(output) {}
+ virtual ~BatchOutputStrategy() = default;
+ virtual void DeliverBatch(compute::ExecBatch batch) = 0;
+ virtual void InputFinished() {
+ // std::cout << "BatchOutputFinished: " + std::to_string(batch_count_.load()) + "\n";
Review Comment:
nit: commented code (well, I guess there's lots of that for debugging)
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -59,6 +156,17 @@ class ARROW_DS_EXPORT Fragment : public std::enable_shared_from_this<Fragment> {
virtual Result<RecordBatchGenerator> ScanBatchesAsync(
const std::shared_ptr<ScanOptions>& options) = 0;
+ /// \brief Inspect a fragment to learn basic information
Review Comment:
This is basically to read the footer ahead of time?
So, ParquetFileFragment would attach the actual Parquet footer in the structure it returns, the Arrow fragment would return the schema, the CSV fragment would throw its hands up, etc.?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow] westonpace commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r937901786
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -59,6 +156,17 @@ class ARROW_DS_EXPORT Fragment : public std::enable_shared_from_this<Fragment> {
virtual Result<RecordBatchGenerator> ScanBatchesAsync(
const std::shared_ptr<ScanOptions>& options) = 0;
+ /// \brief Inspect a fragment to learn basic information
Review Comment:
Yes, exactly. I'm not a huge fan of this because now fragment scanning is broken into three steps...
FragmentScanner: Inspect fragment (new step)
Scanner: Create evolution
FragmentScanner: Open reader (where we previously determined column names / schema)
Scanner: Start iteration task
FragmentScanner: Scan
I could push the evolution creation into the fragment scanner but that makes a new step that every fragment scanner has to do and so I think that would be worse.
Your point on "actual parquet footer" and "the schema" is also correct. My "prototypical future evolution strategy" that I am trying to make sure we plan for is to resolve field references by parquet column id. So it would work in this fashion:
* User specifies numeric field references against the dataset schema
* Evolution strategy has a lookup table from numeric position in dataset schema to column id
* On inspection, physical column ids are inserted into the schema and passed to the evolution
* Evolution uses the lookup table combined with the physical column ids to determine the physical columns that are being asked for
So this way the inspect step would need to return something with the column IDs. This could be the parquet footer. However, it also could be the schema extracted from the footer as we have a mechanism for encoding those column IDs into the schema today. This way we could leave the door open to someone adding column IDs to IPC files as well.
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r974188037
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan a batch of data from the file
+ /// \param batch_number The index of the batch to read
+ virtual Future<std::shared_ptr<RecordBatch>> ScanBatch(int batch_number) = 0;
+ /// \brief Calculate an estimate of how many data bytes the given batch will represent
+ ///
+ /// "Data bytes" should be the total size of all the buffers once the data has been
+ /// decoded into the Arrow format.
+ virtual int64_t EstimatedDataBytes(int batch_number) = 0;
+ /// \brief The number of batches in the fragment to scan
+ virtual int NumBatches() = 0;
Review Comment:
> How would something like CSV implement this?
It's # of batches instead of # of rows so it should just be `ceil(file_size_bytes / block_size)`. That being said, this is a change from the previous fragment implementation which didn't require fragments to know the # of batches ahead of time. It will be a problem for something like the python iterator-based scanner.
It is possible to make this work without knowing the # of batches up front. However, it creates a bit more complexity for two reasons:
1. It makes it harder to know the batch index (I'm also working on ordered execution) because you can't know the batch index until previous fragments have been fully scanned (where this implementation just requires that previous fragments have been inspected).
2. It makes it difficult to know when we should stop spawning tasks to read further in the current fragment and start spawning tasks to read the next fragment. If we do not know `NumBatches` we have to rely on a fragment scanner returning an end-of-fragment future quickly. However, this should also be possible.
My hope is that an iterator-based dataset could easily just be an iterator-based source node and so it's ok for the scan node to have this expectation. If we were adapting flight to the scan node do you know if this is something that could cause a problem? Does a flight stream have a known # of batches before we even start consuming it? Or do we not know the # of batches until we have finished reading 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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r980547246
##########
cpp/src/arrow/compute/exec/expression.h:
##########
@@ -277,6 +279,53 @@ ARROW_EXPORT Expression or_(Expression lhs, Expression rhs);
ARROW_EXPORT Expression or_(const std::vector<Expression>&);
ARROW_EXPORT Expression not_(Expression operand);
+/// Modify an Expression with pre-order and post-order visitation.
+/// `pre` will be invoked on each Expression. `pre` will visit Calls before their
+/// arguments, `post_call` will visit Calls (and no other Expressions) after their
+/// arguments. Visitors should return the Identical expression to indicate no change; this
+/// will prevent unnecessary construction in the common case where a modification is not
+/// possible/necessary/...
+///
+/// If an argument was modified, `post_call` visits a reconstructed Call with the modified
+/// arguments but also receives a pointer to the unmodified Expression as a second
+/// argument. If no arguments were modified the unmodified Expression* will be nullptr.
+template <typename PreVisit, typename PostVisitCall>
+Result<Expression> Modify(Expression expr, const PreVisit& pre,
Review Comment:
I'll see if I can get it to work with appropriate addition of `static`
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
Review Comment:
Added
--
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] lidavidm commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r974206038
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan a batch of data from the file
+ /// \param batch_number The index of the batch to read
+ virtual Future<std::shared_ptr<RecordBatch>> ScanBatch(int batch_number) = 0;
+ /// \brief Calculate an estimate of how many data bytes the given batch will represent
+ ///
+ /// "Data bytes" should be the total size of all the buffers once the data has been
+ /// decoded into the Arrow format.
+ virtual int64_t EstimatedDataBytes(int batch_number) = 0;
+ /// \brief The number of batches in the fragment to scan
+ virtual int NumBatches() = 0;
Review Comment:
Actually I guess I'm curious how this will work with iterator-based sources in general. Flight/ADBC would give you a set of iterators, where there is no parallelism within an iterator. And if you wanted to support Arrow stream files for some reason, that'd have the same structure.
On the other hand, Flight/ADBC have a fixed number of fragments up front.
Maybe a separate source node for those sources might make sense, with an optional 'merge' step if you want to use ordered execution with them? It would be kind of a shame to force them outside of the Dataset framework entirely, IMO, but having two different interfaces would also be annoying. I'm kind of just typing out things on the fly 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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r974736847
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan a batch of data from the file
+ /// \param batch_number The index of the batch to read
+ virtual Future<std::shared_ptr<RecordBatch>> ScanBatch(int batch_number) = 0;
+ /// \brief Calculate an estimate of how many data bytes the given batch will represent
+ ///
+ /// "Data bytes" should be the total size of all the buffers once the data has been
+ /// decoded into the Arrow format.
+ virtual int64_t EstimatedDataBytes(int batch_number) = 0;
+ /// \brief The number of batches in the fragment to scan
+ virtual int NumBatches() = 0;
Review Comment:
I think the worst case would be if the fragments are large and there is no parallelism within an iterator and the user wants ordered execution. This will be a problem regardless of approach. We will need to queue a large number of batches in memory for any kind of resequencing and, as a result, will probably fallback to serial execution.
> Hmm, and I guess (fragment_index, batch_index, is_last) pushes too much complexity into the ordered execution bits?
It might not be too bad. I'll take another look.
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on PR #13782:
URL: https://github.com/apache/arrow/pull/13782#issuecomment-1203071080
This is still very much a draft. However, I have the basic path working. I will need to migrate over the existing scanner tests and test the various failure paths (as well as stress test for race conditions, there is potential for plenty here).
I'm very curious what people think about the new scan options as well as the new evolution interfaces.
--
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] lidavidm commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r937904517
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -59,6 +156,17 @@ class ARROW_DS_EXPORT Fragment : public std::enable_shared_from_this<Fragment> {
virtual Result<RecordBatchGenerator> ScanBatchesAsync(
const std::shared_ptr<ScanOptions>& options) = 0;
+ /// \brief Inspect a fragment to learn basic information
Review Comment:
Sounds good to me.
The Parquet/Arrow readers, IIRC, let you open a file and provide an already-parsed footer to at least skip that step.
--
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] lidavidm commented on a diff in pull request #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
lidavidm commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r973584495
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
Review Comment:
ARROW_DS_EXPORT/docstring here?
##########
cpp/src/arrow/dataset/scanner_test.cc:
##########
@@ -54,6 +57,751 @@ using internal::Iota;
namespace dataset {
+// The basic evolution strategy doesn't really need any info from the dataset
+// or the fragment other than the schema so we just make a dummy dataset/fragment
+// here.
+std::unique_ptr<Dataset> MakeDatasetFromSchema(std::shared_ptr<Schema> sch) {
+ return ::arrow::internal::make_unique<InMemoryDataset>(std::move(sch),
Review Comment:
We can use std::make_unique now!
##########
cpp/src/arrow/compute/exec/expression.h:
##########
@@ -277,6 +279,53 @@ ARROW_EXPORT Expression or_(Expression lhs, Expression rhs);
ARROW_EXPORT Expression or_(const std::vector<Expression>&);
ARROW_EXPORT Expression not_(Expression operand);
+/// Modify an Expression with pre-order and post-order visitation.
+/// `pre` will be invoked on each Expression. `pre` will visit Calls before their
+/// arguments, `post_call` will visit Calls (and no other Expressions) after their
+/// arguments. Visitors should return the Identical expression to indicate no change; this
+/// will prevent unnecessary construction in the common case where a modification is not
+/// possible/necessary/...
+///
+/// If an argument was modified, `post_call` visits a reconstructed Call with the modified
+/// arguments but also receives a pointer to the unmodified Expression as a second
+/// argument. If no arguments were modified the unmodified Expression* will be nullptr.
+template <typename PreVisit, typename PostVisitCall>
+Result<Expression> Modify(Expression expr, const PreVisit& pre,
Review Comment:
nit: did we have to move this out of the internal header?
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan a batch of data from the file
+ /// \param batch_number The index of the batch to read
+ virtual Future<std::shared_ptr<RecordBatch>> ScanBatch(int batch_number) = 0;
+ /// \brief Calculate an estimate of how many data bytes the given batch will represent
+ ///
+ /// "Data bytes" should be the total size of all the buffers once the data has been
+ /// decoded into the Arrow format.
+ virtual int64_t EstimatedDataBytes(int batch_number) = 0;
+ /// \brief The number of batches in the fragment to scan
+ virtual int NumBatches() = 0;
Review Comment:
How would something like CSV implement this?
##########
cpp/src/arrow/dataset/scan_node.cc:
##########
@@ -0,0 +1,372 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <functional>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "arrow/compute/exec/exec_plan.h"
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/scanner.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/make_unique.h"
+#include "arrow/util/tracing_internal.h"
+#include "arrow/util/unreachable.h"
+
+namespace cp = arrow::compute;
+
+namespace arrow {
+
+using internal::checked_cast;
+
+namespace dataset {
+
+namespace {
+
+Result<std::shared_ptr<Schema>> OutputSchemaFromOptions(const ScanV2Options& options) {
+ return FieldPath::GetAll(*options.dataset->schema(), options.columns);
+}
+
+// In the future we should support async scanning of fragments. The
+// Dataset class doesn't support this yet but we pretend it does here to
+// ease future adoption of the feature.
+AsyncGenerator<std::shared_ptr<Fragment>> GetFragments(Dataset* dataset,
+ cp::Expression predicate) {
+ // In the future the dataset should be responsible for figuring out
+ // the I/O context. This will allow different I/O contexts to be used
+ // when scanning different datasets. For example, if we are scanning a
+ // union of a remote dataset and a local dataset.
+ const auto& io_context = io::default_io_context();
+ auto io_executor = io_context.executor();
+ Future<std::shared_ptr<FragmentIterator>> fragments_it_fut =
+ DeferNotOk(io_executor->Submit(
+ [dataset, predicate]() -> Result<std::shared_ptr<FragmentIterator>> {
+ ARROW_ASSIGN_OR_RAISE(FragmentIterator fragments_iter,
+ dataset->GetFragments(predicate));
+ return std::make_shared<FragmentIterator>(std::move(fragments_iter));
+ }));
+ Future<AsyncGenerator<std::shared_ptr<Fragment>>> fragments_gen_fut =
+ fragments_it_fut.Then([](const std::shared_ptr<FragmentIterator>& fragments_it)
+ -> Result<AsyncGenerator<std::shared_ptr<Fragment>>> {
+ ARROW_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<Fragment>> fragments,
+ fragments_it->ToVector());
+ return MakeVectorGenerator(std::move(fragments));
+ });
+ return MakeFromFuture(std::move(fragments_gen_fut));
+}
+
+/// \brief A node that scans a dataset
+///
+/// The scan node has three groups of io-tasks and one task.
+///
+/// The first io-task (listing) fetches the fragments from the dataset. This may be a
+/// simple iteration of paths or, if the dataset is described with wildcards, this may
+/// involve I/O for listing and walking directory paths. There is one listing io-task per
+/// dataset.
+///
+/// Ths next step is to fetch the metadata for the fragment. For some formats (e.g. CSV)
+/// this may be quite simple (get the size of the file). For other formats (e.g. parquet)
+/// this is more involved and requires reading data. There is one metadata io-task per
+/// fragment. The metadata io-task creates an AsyncGenerator<RecordBatch> from the
+/// fragment.
+///
+/// Once the metadata io-task is done we can issue read io-tasks. Each read io-task
+/// requests a single batch of data from the disk by pulling the next Future from the
+/// generator.
+///
+/// Finally, when the future is fulfilled, we issue a pipeline task to drive the batch
+/// through the pipeline.
+///
+/// Most of these tasks are io-tasks. They take very few CPU resources and they run on
+/// the I/O thread pool. These io-tasks are invisible to the exec plan and so we need to
+/// do some custom scheduling. We limit how many fragments we read from at any one time.
+/// This is referred to as "fragment readahead".
+///
+/// Within a fragment there is usually also some amount of "row readahead". This row
+/// readahead is handled by the fragment (and not the scanner) because the exact details
+/// of how it is performed depend on the underlying format.
+///
+/// When a scan node is aborted (StopProducing) we send a cancel signal to any active
+/// fragments. On destruction we continue consuming the fragments until they complete
+/// (which should be fairly quick since we cancelled the fragment). This ensures the
+/// I/O work is completely finished before the node is destroyed.
+class ScanNode : public cp::ExecNode {
+ public:
+ ScanNode(cp::ExecPlan* plan, ScanV2Options options,
+ std::shared_ptr<Schema> output_schema)
+ : cp::ExecNode(plan, {}, {}, std::move(output_schema),
+ /*num_outputs=*/1),
+ options_(options),
+ fragments_throttle_(
+ util::AsyncTaskScheduler::MakeThrottle(options_.fragment_readahead + 1)),
+ batches_throttle_(
+ util::AsyncTaskScheduler::MakeThrottle(options_.target_bytes_readahead + 1)) {
+ }
+
+ static Result<ScanV2Options> NormalizeAndValidate(const ScanV2Options& options,
+ compute::ExecContext* ctx) {
+ ScanV2Options normalized(options);
+ if (!normalized.dataset) {
+ return Status::Invalid("Scan options must include a dataset");
+ }
+
+ if (options.fragment_readahead < 0) {
+ return Status::Invalid(
+ "Fragment readahead may not be less than 0. Set to 0 to disable readahead");
+ }
+
+ if (options.target_bytes_readahead < 0) {
+ return Status::Invalid(
+ "Batch readahead may not be less than 0. Set to 0 to disable readahead");
+ }
+
+ if (!normalized.filter.is_valid()) {
+ normalized.filter = compute::literal(true);
+ }
+
+ if (normalized.filter.call() && normalized.filter.IsBound()) {
+ // There is no easy way to make sure a filter was bound agaisnt the same
+ // function registry as the one in ctx so we just require it to be unbound
+ // FIXME - Do we care if it was bound to a different function registry?
+ return Status::Invalid("Scan filter must be unbound");
+ } else if (!normalized.filter.IsBound()) {
+ ARROW_ASSIGN_OR_RAISE(normalized.filter,
+ normalized.filter.Bind(*options.dataset->schema(), ctx));
+ } // Else we must have some simple filter like literal(true) which might be bound
+ // but we don't care
+
+ return std::move(normalized);
+ }
+
+ static Result<cp::ExecNode*> Make(cp::ExecPlan* plan, std::vector<cp::ExecNode*> inputs,
+ const cp::ExecNodeOptions& options) {
+ RETURN_NOT_OK(ValidateExecNodeInputs(plan, inputs, 0, "ScanNode"));
+ const auto& scan_options = checked_cast<const ScanV2Options&>(options);
+ ARROW_ASSIGN_OR_RAISE(ScanV2Options normalized_options,
+ NormalizeAndValidate(scan_options, plan->exec_context()));
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Schema> output_schema,
+ OutputSchemaFromOptions(normalized_options));
+ return plan->EmplaceNode<ScanNode>(plan, std::move(normalized_options),
+ std::move(output_schema));
+ }
+
+ const char* kind_name() const override { return "ScanNode"; }
+
+ [[noreturn]] static void NoInputs() {
+ Unreachable("no inputs; this should never be called");
+ }
+ [[noreturn]] void InputReceived(cp::ExecNode*, cp::ExecBatch) override { NoInputs(); }
+ [[noreturn]] void ErrorReceived(cp::ExecNode*, Status) override { NoInputs(); }
+ [[noreturn]] void InputFinished(cp::ExecNode*, int) override { NoInputs(); }
+
+ Status Init() override {
+ // batch_output_ =
+ // ::arrow::internal::make_unique<UnorderedBatchOutputStrategy>(this,
+ // outputs_[0]);
+ return Status::OK();
+ }
+
+ struct ScanState {
+ std::mutex mutex;
+ std::shared_ptr<FragmentScanner> fragment_scanner;
+ std::unique_ptr<FragmentEvolutionStrategy> fragment_evolution;
+ FragmentScanRequest scan_request;
+ };
+
+ struct ScanBatchTask : util::AsyncTaskScheduler::Task {
+ ScanBatchTask(ScanNode* node, ScanState* scan_state, int batch_index)
+ : node_(node), scan_(scan_state), batch_index_(batch_index) {
+ int64_t cost = scan_state->fragment_scanner->EstimatedDataBytes(batch_index_);
+ // It's possible, though probably a bad idea, for a single batch of a fragment
+ // to be larger than 2GiB. In that case, it doesn't matter much if we underestimate
+ // because the largest the throttle can be is 2GiB and thus we will be in "one batch
+ // at a time" mode anyways which is the best we can do in this case.
+ cost_ = static_cast<int>(
+ std::min(cost, static_cast<int64_t>(std::numeric_limits<int>::max())));
+ }
+
+ Result<Future<>> operator()(util::AsyncTaskScheduler* scheduler) override {
+ // Prevent concurrent calls to ScanBatch which might not be thread safe
+ std::lock_guard<std::mutex> lk(scan_->mutex);
+ return scan_->fragment_scanner->ScanBatch(batch_index_)
+ .Then([this](const std::shared_ptr<RecordBatch> batch) {
Review Comment:
nit: was this meant to be const&?
##########
cpp/src/arrow/util/async_util.cc:
##########
@@ -274,7 +282,9 @@ class AsyncTaskSchedulerImpl : public AsyncTaskScheduler {
AbortUnlocked(submit_result.status(), std::move(lk));
return;
}
- submit_result->AddCallback([this, cost](const Status& st) {
+ // FIXME(C++17, move into lambda?)
Review Comment:
should be able to do this now?
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan a batch of data from the file
+ /// \param batch_number The index of the batch to read
+ virtual Future<std::shared_ptr<RecordBatch>> ScanBatch(int batch_number) = 0;
+ /// \brief Calculate an estimate of how many data bytes the given batch will represent
+ ///
+ /// "Data bytes" should be the total size of all the buffers once the data has been
+ /// decoded into the Arrow format.
+ virtual int64_t EstimatedDataBytes(int batch_number) = 0;
+ /// \brief The number of batches in the fragment to scan
+ virtual int NumBatches() = 0;
Review Comment:
Should NumBatches and EstimatedDataBytes be `const`?
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,17 +38,91 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath path;
+ /// \brief The type of the column in the dataset schema
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. There is no point in doing anything special.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+ /// \brief The index in the output selection of this column
+ int selection_index;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
+ std::vector<FragmentSelectionColumn> columns;
+ /// \brief Options specific to the format being scanned
+ FragmentScanOptions* format_scan_options;
+};
+
+class FragmentScanner {
+ public:
+ /// This instance will only be destroyed after all ongoing scan futures
+ /// have been completed.
+ ///
+ /// This means any callbacks created as part of the scan can safely
+ /// capture `this`
+ virtual ~FragmentScanner() = default;
+ /// \brief Scan a batch of data from the file
+ /// \param batch_number The index of the batch to read
+ virtual Future<std::shared_ptr<RecordBatch>> ScanBatch(int batch_number) = 0;
+ /// \brief Calculate an estimate of how many data bytes the given batch will represent
+ ///
+ /// "Data bytes" should be the total size of all the buffers once the data has been
+ /// decoded into the Arrow format.
+ virtual int64_t EstimatedDataBytes(int batch_number) = 0;
+ /// \brief The number of batches in the fragment to scan
+ virtual int NumBatches() = 0;
+};
+
+struct InspectedFragment {
+ explicit InspectedFragment(std::vector<std::string> column_names)
+ : column_names(std::move(column_names)) {}
+ std::vector<std::string> column_names;
Review Comment:
Hmm, why don't we care about the types here?
##########
cpp/src/arrow/dataset/scan_node.cc:
##########
@@ -0,0 +1,372 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <functional>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "arrow/compute/exec/exec_plan.h"
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/scanner.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/make_unique.h"
+#include "arrow/util/tracing_internal.h"
+#include "arrow/util/unreachable.h"
+
+namespace cp = arrow::compute;
+
+namespace arrow {
+
+using internal::checked_cast;
+
+namespace dataset {
+
+namespace {
+
+Result<std::shared_ptr<Schema>> OutputSchemaFromOptions(const ScanV2Options& options) {
+ return FieldPath::GetAll(*options.dataset->schema(), options.columns);
+}
+
+// In the future we should support async scanning of fragments. The
+// Dataset class doesn't support this yet but we pretend it does here to
+// ease future adoption of the feature.
+AsyncGenerator<std::shared_ptr<Fragment>> GetFragments(Dataset* dataset,
+ cp::Expression predicate) {
+ // In the future the dataset should be responsible for figuring out
+ // the I/O context. This will allow different I/O contexts to be used
+ // when scanning different datasets. For example, if we are scanning a
+ // union of a remote dataset and a local dataset.
+ const auto& io_context = io::default_io_context();
+ auto io_executor = io_context.executor();
+ Future<std::shared_ptr<FragmentIterator>> fragments_it_fut =
+ DeferNotOk(io_executor->Submit(
+ [dataset, predicate]() -> Result<std::shared_ptr<FragmentIterator>> {
+ ARROW_ASSIGN_OR_RAISE(FragmentIterator fragments_iter,
+ dataset->GetFragments(predicate));
+ return std::make_shared<FragmentIterator>(std::move(fragments_iter));
+ }));
+ Future<AsyncGenerator<std::shared_ptr<Fragment>>> fragments_gen_fut =
+ fragments_it_fut.Then([](const std::shared_ptr<FragmentIterator>& fragments_it)
+ -> Result<AsyncGenerator<std::shared_ptr<Fragment>>> {
+ ARROW_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<Fragment>> fragments,
+ fragments_it->ToVector());
+ return MakeVectorGenerator(std::move(fragments));
+ });
+ return MakeFromFuture(std::move(fragments_gen_fut));
+}
+
+/// \brief A node that scans a dataset
+///
+/// The scan node has three groups of io-tasks and one task.
+///
+/// The first io-task (listing) fetches the fragments from the dataset. This may be a
+/// simple iteration of paths or, if the dataset is described with wildcards, this may
+/// involve I/O for listing and walking directory paths. There is one listing io-task per
+/// dataset.
+///
+/// Ths next step is to fetch the metadata for the fragment. For some formats (e.g. CSV)
+/// this may be quite simple (get the size of the file). For other formats (e.g. parquet)
+/// this is more involved and requires reading data. There is one metadata io-task per
+/// fragment. The metadata io-task creates an AsyncGenerator<RecordBatch> from the
+/// fragment.
+///
+/// Once the metadata io-task is done we can issue read io-tasks. Each read io-task
+/// requests a single batch of data from the disk by pulling the next Future from the
+/// generator.
+///
+/// Finally, when the future is fulfilled, we issue a pipeline task to drive the batch
+/// through the pipeline.
+///
+/// Most of these tasks are io-tasks. They take very few CPU resources and they run on
+/// the I/O thread pool. These io-tasks are invisible to the exec plan and so we need to
+/// do some custom scheduling. We limit how many fragments we read from at any one time.
+/// This is referred to as "fragment readahead".
+///
+/// Within a fragment there is usually also some amount of "row readahead". This row
+/// readahead is handled by the fragment (and not the scanner) because the exact details
+/// of how it is performed depend on the underlying format.
+///
+/// When a scan node is aborted (StopProducing) we send a cancel signal to any active
+/// fragments. On destruction we continue consuming the fragments until they complete
+/// (which should be fairly quick since we cancelled the fragment). This ensures the
+/// I/O work is completely finished before the node is destroyed.
+class ScanNode : public cp::ExecNode {
+ public:
+ ScanNode(cp::ExecPlan* plan, ScanV2Options options,
+ std::shared_ptr<Schema> output_schema)
+ : cp::ExecNode(plan, {}, {}, std::move(output_schema),
+ /*num_outputs=*/1),
+ options_(options),
+ fragments_throttle_(
+ util::AsyncTaskScheduler::MakeThrottle(options_.fragment_readahead + 1)),
+ batches_throttle_(
+ util::AsyncTaskScheduler::MakeThrottle(options_.target_bytes_readahead + 1)) {
+ }
+
+ static Result<ScanV2Options> NormalizeAndValidate(const ScanV2Options& options,
+ compute::ExecContext* ctx) {
+ ScanV2Options normalized(options);
+ if (!normalized.dataset) {
+ return Status::Invalid("Scan options must include a dataset");
+ }
+
+ if (options.fragment_readahead < 0) {
+ return Status::Invalid(
+ "Fragment readahead may not be less than 0. Set to 0 to disable readahead");
+ }
+
+ if (options.target_bytes_readahead < 0) {
+ return Status::Invalid(
+ "Batch readahead may not be less than 0. Set to 0 to disable readahead");
+ }
+
+ if (!normalized.filter.is_valid()) {
+ normalized.filter = compute::literal(true);
+ }
+
+ if (normalized.filter.call() && normalized.filter.IsBound()) {
+ // There is no easy way to make sure a filter was bound agaisnt the same
+ // function registry as the one in ctx so we just require it to be unbound
+ // FIXME - Do we care if it was bound to a different function registry?
+ return Status::Invalid("Scan filter must be unbound");
+ } else if (!normalized.filter.IsBound()) {
+ ARROW_ASSIGN_OR_RAISE(normalized.filter,
+ normalized.filter.Bind(*options.dataset->schema(), ctx));
+ } // Else we must have some simple filter like literal(true) which might be bound
+ // but we don't care
+
+ return std::move(normalized);
+ }
+
+ static Result<cp::ExecNode*> Make(cp::ExecPlan* plan, std::vector<cp::ExecNode*> inputs,
+ const cp::ExecNodeOptions& options) {
+ RETURN_NOT_OK(ValidateExecNodeInputs(plan, inputs, 0, "ScanNode"));
+ const auto& scan_options = checked_cast<const ScanV2Options&>(options);
+ ARROW_ASSIGN_OR_RAISE(ScanV2Options normalized_options,
+ NormalizeAndValidate(scan_options, plan->exec_context()));
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Schema> output_schema,
+ OutputSchemaFromOptions(normalized_options));
+ return plan->EmplaceNode<ScanNode>(plan, std::move(normalized_options),
+ std::move(output_schema));
+ }
+
+ const char* kind_name() const override { return "ScanNode"; }
+
+ [[noreturn]] static void NoInputs() {
+ Unreachable("no inputs; this should never be called");
+ }
+ [[noreturn]] void InputReceived(cp::ExecNode*, cp::ExecBatch) override { NoInputs(); }
+ [[noreturn]] void ErrorReceived(cp::ExecNode*, Status) override { NoInputs(); }
+ [[noreturn]] void InputFinished(cp::ExecNode*, int) override { NoInputs(); }
+
+ Status Init() override {
+ // batch_output_ =
Review Comment:
Commented code?
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on PR #13782:
URL: https://github.com/apache/arrow/pull/13782#issuecomment-1203072225
CC @save-buffer @marsupialtail who might be interested in this as well as we have spoken on similar topics
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #13782:
URL: https://github.com/apache/arrow/pull/13782#issuecomment-1203068987
https://issues.apache.org/jira/browse/ARROW-17287
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r937894460
##########
cpp/src/arrow/dataset/dataset.h:
##########
@@ -37,11 +38,107 @@ namespace dataset {
using RecordBatchGenerator = std::function<Future<std::shared_ptr<RecordBatch>>()>;
+/// \brief Description of a column to scan
+struct FragmentSelectionColumn {
+ /// \brief The path to the column to load
+ FieldPath column_reference;
+ /// \brief The requested type of the column
+ ///
+ /// A format may choose to ignore this field completely. For example, when
+ /// reading from IPC the reader can just return the column in the data type
+ /// that is stored on disk. In this case the scan node will cast the column
+ /// to the appropriate type later.
+ ///
+ /// However, some formats may be capable of casting on the fly. For example,
+ /// when reading from CSV, if we know the target type of the column, we can
+ /// convert from string to the target type as we read.
+ DataType* requested_type;
+};
+/// \brief Instructions for scanning a particular fragment
+///
+/// The fragment scan request is dervied from ScanV2Options. The main
+/// difference is that the scan options are based on the dataset schema
+/// while the fragment request is based on the fragment schema.
+struct FragmentScanRequest {
+ /// \brief A row filter
+ ///
+ /// The filter expression should be written against the fragment schema.
+ ///
+ /// \see ScanV2Options for details on how this filter should be applied
+ compute::Expression filter = compute::literal(true);
+
+ /// \brief The columns to scan
+ ///
+ /// These indices refer to the fragment schema
+ ///
+ /// Note: This is NOT a simple list of top-level column indices.
+ /// For more details \see ScanV2Options
+ ///
+ /// If possible a fragment should only read from disk the data needed
+ /// to satisfy these columns. If a format cannot partially read a nested
+ /// column (e.g. JSON) then it must apply the column selection (in memory)
+ /// before returning the scanned batch.
Review Comment:
I suppose the scanner can do this "field pruning" just before the batch is sent for evolution. I'm not entirely sure how the scanner would know which fields need to be pruned though. The fragment scanner could maybe have a boolean flag whether it will do the pruning or not do the pruning.
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r980547547
##########
cpp/src/arrow/dataset/scan_node.cc:
##########
@@ -0,0 +1,372 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <functional>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "arrow/compute/exec/exec_plan.h"
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/scanner.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/make_unique.h"
+#include "arrow/util/tracing_internal.h"
+#include "arrow/util/unreachable.h"
+
+namespace cp = arrow::compute;
+
+namespace arrow {
+
+using internal::checked_cast;
+
+namespace dataset {
+
+namespace {
+
+Result<std::shared_ptr<Schema>> OutputSchemaFromOptions(const ScanV2Options& options) {
+ return FieldPath::GetAll(*options.dataset->schema(), options.columns);
+}
+
+// In the future we should support async scanning of fragments. The
+// Dataset class doesn't support this yet but we pretend it does here to
+// ease future adoption of the feature.
+AsyncGenerator<std::shared_ptr<Fragment>> GetFragments(Dataset* dataset,
+ cp::Expression predicate) {
+ // In the future the dataset should be responsible for figuring out
+ // the I/O context. This will allow different I/O contexts to be used
+ // when scanning different datasets. For example, if we are scanning a
+ // union of a remote dataset and a local dataset.
+ const auto& io_context = io::default_io_context();
+ auto io_executor = io_context.executor();
+ Future<std::shared_ptr<FragmentIterator>> fragments_it_fut =
+ DeferNotOk(io_executor->Submit(
+ [dataset, predicate]() -> Result<std::shared_ptr<FragmentIterator>> {
+ ARROW_ASSIGN_OR_RAISE(FragmentIterator fragments_iter,
+ dataset->GetFragments(predicate));
+ return std::make_shared<FragmentIterator>(std::move(fragments_iter));
+ }));
+ Future<AsyncGenerator<std::shared_ptr<Fragment>>> fragments_gen_fut =
+ fragments_it_fut.Then([](const std::shared_ptr<FragmentIterator>& fragments_it)
+ -> Result<AsyncGenerator<std::shared_ptr<Fragment>>> {
+ ARROW_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<Fragment>> fragments,
+ fragments_it->ToVector());
+ return MakeVectorGenerator(std::move(fragments));
+ });
+ return MakeFromFuture(std::move(fragments_gen_fut));
+}
+
+/// \brief A node that scans a dataset
+///
+/// The scan node has three groups of io-tasks and one task.
+///
+/// The first io-task (listing) fetches the fragments from the dataset. This may be a
+/// simple iteration of paths or, if the dataset is described with wildcards, this may
+/// involve I/O for listing and walking directory paths. There is one listing io-task per
+/// dataset.
+///
+/// Ths next step is to fetch the metadata for the fragment. For some formats (e.g. CSV)
+/// this may be quite simple (get the size of the file). For other formats (e.g. parquet)
+/// this is more involved and requires reading data. There is one metadata io-task per
+/// fragment. The metadata io-task creates an AsyncGenerator<RecordBatch> from the
+/// fragment.
+///
+/// Once the metadata io-task is done we can issue read io-tasks. Each read io-task
+/// requests a single batch of data from the disk by pulling the next Future from the
+/// generator.
+///
+/// Finally, when the future is fulfilled, we issue a pipeline task to drive the batch
+/// through the pipeline.
+///
+/// Most of these tasks are io-tasks. They take very few CPU resources and they run on
+/// the I/O thread pool. These io-tasks are invisible to the exec plan and so we need to
+/// do some custom scheduling. We limit how many fragments we read from at any one time.
+/// This is referred to as "fragment readahead".
+///
+/// Within a fragment there is usually also some amount of "row readahead". This row
+/// readahead is handled by the fragment (and not the scanner) because the exact details
+/// of how it is performed depend on the underlying format.
+///
+/// When a scan node is aborted (StopProducing) we send a cancel signal to any active
+/// fragments. On destruction we continue consuming the fragments until they complete
+/// (which should be fairly quick since we cancelled the fragment). This ensures the
+/// I/O work is completely finished before the node is destroyed.
+class ScanNode : public cp::ExecNode {
+ public:
+ ScanNode(cp::ExecPlan* plan, ScanV2Options options,
+ std::shared_ptr<Schema> output_schema)
+ : cp::ExecNode(plan, {}, {}, std::move(output_schema),
+ /*num_outputs=*/1),
+ options_(options),
+ fragments_throttle_(
+ util::AsyncTaskScheduler::MakeThrottle(options_.fragment_readahead + 1)),
+ batches_throttle_(
+ util::AsyncTaskScheduler::MakeThrottle(options_.target_bytes_readahead + 1)) {
+ }
+
+ static Result<ScanV2Options> NormalizeAndValidate(const ScanV2Options& options,
+ compute::ExecContext* ctx) {
+ ScanV2Options normalized(options);
+ if (!normalized.dataset) {
+ return Status::Invalid("Scan options must include a dataset");
+ }
+
+ if (options.fragment_readahead < 0) {
+ return Status::Invalid(
+ "Fragment readahead may not be less than 0. Set to 0 to disable readahead");
+ }
+
+ if (options.target_bytes_readahead < 0) {
+ return Status::Invalid(
+ "Batch readahead may not be less than 0. Set to 0 to disable readahead");
+ }
+
+ if (!normalized.filter.is_valid()) {
+ normalized.filter = compute::literal(true);
+ }
+
+ if (normalized.filter.call() && normalized.filter.IsBound()) {
+ // There is no easy way to make sure a filter was bound agaisnt the same
+ // function registry as the one in ctx so we just require it to be unbound
+ // FIXME - Do we care if it was bound to a different function registry?
+ return Status::Invalid("Scan filter must be unbound");
+ } else if (!normalized.filter.IsBound()) {
+ ARROW_ASSIGN_OR_RAISE(normalized.filter,
+ normalized.filter.Bind(*options.dataset->schema(), ctx));
+ } // Else we must have some simple filter like literal(true) which might be bound
+ // but we don't care
+
+ return std::move(normalized);
+ }
+
+ static Result<cp::ExecNode*> Make(cp::ExecPlan* plan, std::vector<cp::ExecNode*> inputs,
+ const cp::ExecNodeOptions& options) {
+ RETURN_NOT_OK(ValidateExecNodeInputs(plan, inputs, 0, "ScanNode"));
+ const auto& scan_options = checked_cast<const ScanV2Options&>(options);
+ ARROW_ASSIGN_OR_RAISE(ScanV2Options normalized_options,
+ NormalizeAndValidate(scan_options, plan->exec_context()));
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Schema> output_schema,
+ OutputSchemaFromOptions(normalized_options));
+ return plan->EmplaceNode<ScanNode>(plan, std::move(normalized_options),
+ std::move(output_schema));
+ }
+
+ const char* kind_name() const override { return "ScanNode"; }
+
+ [[noreturn]] static void NoInputs() {
+ Unreachable("no inputs; this should never be called");
+ }
+ [[noreturn]] void InputReceived(cp::ExecNode*, cp::ExecBatch) override { NoInputs(); }
+ [[noreturn]] void ErrorReceived(cp::ExecNode*, Status) override { NoInputs(); }
+ [[noreturn]] void InputFinished(cp::ExecNode*, int) override { NoInputs(); }
+
+ Status Init() override {
+ // batch_output_ =
Review Comment:
Removed
##########
cpp/src/arrow/dataset/scan_node.cc:
##########
@@ -0,0 +1,372 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <functional>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "arrow/compute/exec/exec_plan.h"
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/scanner.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/make_unique.h"
+#include "arrow/util/tracing_internal.h"
+#include "arrow/util/unreachable.h"
+
+namespace cp = arrow::compute;
+
+namespace arrow {
+
+using internal::checked_cast;
+
+namespace dataset {
+
+namespace {
+
+Result<std::shared_ptr<Schema>> OutputSchemaFromOptions(const ScanV2Options& options) {
+ return FieldPath::GetAll(*options.dataset->schema(), options.columns);
+}
+
+// In the future we should support async scanning of fragments. The
+// Dataset class doesn't support this yet but we pretend it does here to
+// ease future adoption of the feature.
+AsyncGenerator<std::shared_ptr<Fragment>> GetFragments(Dataset* dataset,
+ cp::Expression predicate) {
+ // In the future the dataset should be responsible for figuring out
+ // the I/O context. This will allow different I/O contexts to be used
+ // when scanning different datasets. For example, if we are scanning a
+ // union of a remote dataset and a local dataset.
+ const auto& io_context = io::default_io_context();
+ auto io_executor = io_context.executor();
+ Future<std::shared_ptr<FragmentIterator>> fragments_it_fut =
+ DeferNotOk(io_executor->Submit(
+ [dataset, predicate]() -> Result<std::shared_ptr<FragmentIterator>> {
+ ARROW_ASSIGN_OR_RAISE(FragmentIterator fragments_iter,
+ dataset->GetFragments(predicate));
+ return std::make_shared<FragmentIterator>(std::move(fragments_iter));
+ }));
+ Future<AsyncGenerator<std::shared_ptr<Fragment>>> fragments_gen_fut =
+ fragments_it_fut.Then([](const std::shared_ptr<FragmentIterator>& fragments_it)
+ -> Result<AsyncGenerator<std::shared_ptr<Fragment>>> {
+ ARROW_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<Fragment>> fragments,
+ fragments_it->ToVector());
+ return MakeVectorGenerator(std::move(fragments));
+ });
+ return MakeFromFuture(std::move(fragments_gen_fut));
+}
+
+/// \brief A node that scans a dataset
+///
+/// The scan node has three groups of io-tasks and one task.
+///
+/// The first io-task (listing) fetches the fragments from the dataset. This may be a
+/// simple iteration of paths or, if the dataset is described with wildcards, this may
+/// involve I/O for listing and walking directory paths. There is one listing io-task per
+/// dataset.
+///
+/// Ths next step is to fetch the metadata for the fragment. For some formats (e.g. CSV)
+/// this may be quite simple (get the size of the file). For other formats (e.g. parquet)
+/// this is more involved and requires reading data. There is one metadata io-task per
+/// fragment. The metadata io-task creates an AsyncGenerator<RecordBatch> from the
+/// fragment.
+///
+/// Once the metadata io-task is done we can issue read io-tasks. Each read io-task
+/// requests a single batch of data from the disk by pulling the next Future from the
+/// generator.
+///
+/// Finally, when the future is fulfilled, we issue a pipeline task to drive the batch
+/// through the pipeline.
+///
+/// Most of these tasks are io-tasks. They take very few CPU resources and they run on
+/// the I/O thread pool. These io-tasks are invisible to the exec plan and so we need to
+/// do some custom scheduling. We limit how many fragments we read from at any one time.
+/// This is referred to as "fragment readahead".
+///
+/// Within a fragment there is usually also some amount of "row readahead". This row
+/// readahead is handled by the fragment (and not the scanner) because the exact details
+/// of how it is performed depend on the underlying format.
+///
+/// When a scan node is aborted (StopProducing) we send a cancel signal to any active
+/// fragments. On destruction we continue consuming the fragments until they complete
+/// (which should be fairly quick since we cancelled the fragment). This ensures the
+/// I/O work is completely finished before the node is destroyed.
+class ScanNode : public cp::ExecNode {
+ public:
+ ScanNode(cp::ExecPlan* plan, ScanV2Options options,
+ std::shared_ptr<Schema> output_schema)
+ : cp::ExecNode(plan, {}, {}, std::move(output_schema),
+ /*num_outputs=*/1),
+ options_(options),
+ fragments_throttle_(
+ util::AsyncTaskScheduler::MakeThrottle(options_.fragment_readahead + 1)),
+ batches_throttle_(
+ util::AsyncTaskScheduler::MakeThrottle(options_.target_bytes_readahead + 1)) {
+ }
+
+ static Result<ScanV2Options> NormalizeAndValidate(const ScanV2Options& options,
+ compute::ExecContext* ctx) {
+ ScanV2Options normalized(options);
+ if (!normalized.dataset) {
+ return Status::Invalid("Scan options must include a dataset");
+ }
+
+ if (options.fragment_readahead < 0) {
+ return Status::Invalid(
+ "Fragment readahead may not be less than 0. Set to 0 to disable readahead");
+ }
+
+ if (options.target_bytes_readahead < 0) {
+ return Status::Invalid(
+ "Batch readahead may not be less than 0. Set to 0 to disable readahead");
+ }
+
+ if (!normalized.filter.is_valid()) {
+ normalized.filter = compute::literal(true);
+ }
+
+ if (normalized.filter.call() && normalized.filter.IsBound()) {
+ // There is no easy way to make sure a filter was bound agaisnt the same
+ // function registry as the one in ctx so we just require it to be unbound
+ // FIXME - Do we care if it was bound to a different function registry?
+ return Status::Invalid("Scan filter must be unbound");
+ } else if (!normalized.filter.IsBound()) {
+ ARROW_ASSIGN_OR_RAISE(normalized.filter,
+ normalized.filter.Bind(*options.dataset->schema(), ctx));
+ } // Else we must have some simple filter like literal(true) which might be bound
+ // but we don't care
+
+ return std::move(normalized);
+ }
+
+ static Result<cp::ExecNode*> Make(cp::ExecPlan* plan, std::vector<cp::ExecNode*> inputs,
+ const cp::ExecNodeOptions& options) {
+ RETURN_NOT_OK(ValidateExecNodeInputs(plan, inputs, 0, "ScanNode"));
+ const auto& scan_options = checked_cast<const ScanV2Options&>(options);
+ ARROW_ASSIGN_OR_RAISE(ScanV2Options normalized_options,
+ NormalizeAndValidate(scan_options, plan->exec_context()));
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Schema> output_schema,
+ OutputSchemaFromOptions(normalized_options));
+ return plan->EmplaceNode<ScanNode>(plan, std::move(normalized_options),
+ std::move(output_schema));
+ }
+
+ const char* kind_name() const override { return "ScanNode"; }
+
+ [[noreturn]] static void NoInputs() {
+ Unreachable("no inputs; this should never be called");
+ }
+ [[noreturn]] void InputReceived(cp::ExecNode*, cp::ExecBatch) override { NoInputs(); }
+ [[noreturn]] void ErrorReceived(cp::ExecNode*, Status) override { NoInputs(); }
+ [[noreturn]] void InputFinished(cp::ExecNode*, int) override { NoInputs(); }
+
+ Status Init() override {
+ // batch_output_ =
+ // ::arrow::internal::make_unique<UnorderedBatchOutputStrategy>(this,
+ // outputs_[0]);
+ return Status::OK();
+ }
+
+ struct ScanState {
+ std::mutex mutex;
+ std::shared_ptr<FragmentScanner> fragment_scanner;
+ std::unique_ptr<FragmentEvolutionStrategy> fragment_evolution;
+ FragmentScanRequest scan_request;
+ };
+
+ struct ScanBatchTask : util::AsyncTaskScheduler::Task {
+ ScanBatchTask(ScanNode* node, ScanState* scan_state, int batch_index)
+ : node_(node), scan_(scan_state), batch_index_(batch_index) {
+ int64_t cost = scan_state->fragment_scanner->EstimatedDataBytes(batch_index_);
+ // It's possible, though probably a bad idea, for a single batch of a fragment
+ // to be larger than 2GiB. In that case, it doesn't matter much if we underestimate
+ // because the largest the throttle can be is 2GiB and thus we will be in "one batch
+ // at a time" mode anyways which is the best we can do in this case.
+ cost_ = static_cast<int>(
+ std::min(cost, static_cast<int64_t>(std::numeric_limits<int>::max())));
+ }
+
+ Result<Future<>> operator()(util::AsyncTaskScheduler* scheduler) override {
+ // Prevent concurrent calls to ScanBatch which might not be thread safe
+ std::lock_guard<std::mutex> lk(scan_->mutex);
+ return scan_->fragment_scanner->ScanBatch(batch_index_)
+ .Then([this](const std::shared_ptr<RecordBatch> batch) {
Review Comment:
Yes. Thanks, fixed.
--
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 #13782: ARROW-17287: [C++] Create scan node that doesn't rely on the merged generator
Posted by GitBox <gi...@apache.org>.
westonpace commented on code in PR #13782:
URL: https://github.com/apache/arrow/pull/13782#discussion_r980547075
##########
cpp/src/arrow/dataset/scan_node.cc:
##########
@@ -0,0 +1,653 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include "arrow/dataset/scanner.h"
+
+#include <functional>
+#include <iostream>
+#include <list>
+#include <memory>
+#include <mutex>
+#include <string>
+#include <vector>
+
+#include "arrow/compute/exec/exec_plan.h"
+#include "arrow/compute/exec/expression.h"
+#include "arrow/dataset/scanner.h"
+#include "arrow/record_batch.h"
+#include "arrow/result.h"
+#include "arrow/status.h"
+#include "arrow/type.h"
+#include "arrow/util/checked_cast.h"
+#include "arrow/util/make_unique.h"
+#include "arrow/util/tracing_internal.h"
+#include "arrow/util/unreachable.h"
+
+namespace cp = arrow::compute;
+
+namespace arrow {
+
+using internal::checked_cast;
+
+namespace dataset {
+
+namespace {
+
+Result<std::shared_ptr<Schema>> OutputSchemaFromOptions(const ScanV2Options& options) {
+ return FieldPath::GetAll(*options.dataset->schema(), options.columns);
+}
+
+class BatchOutputStrategy {
+ public:
+ BatchOutputStrategy(compute::ExecNode* self, compute::ExecNode* output)
+ : self_(self), output_(output) {}
+ virtual ~BatchOutputStrategy() = default;
+ virtual void DeliverBatch(compute::ExecBatch batch) = 0;
+ virtual void InputFinished() {
+ // std::cout << "BatchOutputFinished: " + std::to_string(batch_count_.load()) + "\n";
+ output_->InputFinished(self_, batch_count_.load());
+ }
+
+ protected:
+ void SliceAndDoDeliverBatch(compute::ExecBatch batch) {
+ // FIXME - Add slicing
+ batch_count_++;
+ output_->InputReceived(self_, std::move(batch));
+ }
+
+ std::atomic<int32_t> batch_count_{0};
+ compute::ExecNode* self_;
+ compute::ExecNode* output_;
+};
+
+class UnorderedBatchOutputStrategy : public BatchOutputStrategy {
+ public:
+ UnorderedBatchOutputStrategy(compute::ExecNode* self, compute::ExecNode* output)
+ : BatchOutputStrategy(self, output){};
+ ~UnorderedBatchOutputStrategy() = default;
+ void DeliverBatch(compute::ExecBatch batch) override {
+ return SliceAndDoDeliverBatch(std::move(batch));
+ }
+};
+
+// Unlike other nodes the scanner has to synchronize a number of resources
+// that are external to the plan itself. These things may start tasks that
+// are invisible to the exec plan.
+class ScannerSynchronization {
+ public:
+ ScannerSynchronization(int fragment_readahead)
+ : fragment_readahead_(fragment_readahead) {}
+
+ bool MarkListingCompleted() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ finished_listing_ = true;
+ return fragments_active_ == 0;
+ }
+
+ void MarkListingFailed(Status err) {
+ std::lock_guard<std::mutex> lg(mutex_);
+ finished_listing_ = true;
+ cancelled_ = true;
+ error_ = std::move(err);
+ }
+
+ void Cancel() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (!cancelled_) {
+ cancelled_ = true;
+ error_ = Status::Cancelled("Plan cancelled before scan completed");
+ }
+ }
+
+ void Pause() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ paused_ = true;
+ }
+
+ void ReportFailedFragment(Status st) {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (!cancelled_) {
+ cancelled_ = true;
+ error_ = std::move(st);
+ fragments_active_--;
+ }
+ }
+
+ struct NextFragment {
+ std::shared_ptr<Fragment> fragment;
+ bool should_mark_input_finished;
+ };
+
+ NextFragment FinishFragmentAndGetNext() {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (cancelled_ || paused_ || fragments_to_scan_.empty()) {
+ fragments_active_--;
+ // std::cout << "Not continuing fragment inline fragments_active=" +
+ // std::to_string(fragments_active_) + "\n";
+ return {nullptr, /*should_mark_input_finished=*/finished_listing_};
+ }
+ // std::cout << "Continuing fragment inline " + std::to_string(fragments_active_) +
+ // "\n";
+ std::shared_ptr<Fragment> next = std::move(fragments_to_scan_.front());
+ fragments_to_scan_.pop();
+ return {next, /*should_mark_input_finished=*/false};
+ }
+
+ bool QueueFragment(std::shared_ptr<Fragment> fragment) {
+ std::lock_guard<std::mutex> lg(mutex_);
+ if (cancelled_) {
+ return false;
+ }
+ if (!paused_ && fragments_active_ < fragment_readahead_) {
+ // If we have space start scanning the fragment immediately
+ fragments_active_++;
+ // std::cout << "Start scanning immediately: " + std::to_string(fragments_active_) +
+ // "\n";
+ return true;
+ }
+ fragments_to_scan_.push(std::move(fragment));
+ return false;
+ }
+
+ private:
+ int fragment_readahead_;
+ std::mutex mutex_;
+ std::queue<std::shared_ptr<Fragment>> fragments_to_scan_;
+ Status error_;
+ bool cancelled_ = false;
+ bool finished_listing_ = false;
+ int fragments_active_ = 0;
+ bool paused_ = false;
+};
+
+// In the future we should support async scanning of fragments. The
+// Dataset class doesn't support this yet but we pretend it does here to
+// ease future adoption of the feature.
+AsyncGenerator<std::shared_ptr<Fragment>> GetFragments(Dataset* dataset,
+ cp::Expression predicate) {
+ // In the future the dataset should be responsible for figuring out
+ // the I/O context. This will allow different I/O contexts to be used
+ // when scanning different datasets. For example, if we are scanning a
+ // union of a remote dataset and a local dataset.
+ const auto& io_context = io::default_io_context();
+ auto io_executor = io_context.executor();
+ Future<std::shared_ptr<FragmentIterator>> fragments_it_fut =
+ DeferNotOk(io_executor->Submit(
+ [dataset, predicate]() -> Result<std::shared_ptr<FragmentIterator>> {
+ // std::cout << "dataset->GetFragments\n";
+ ARROW_ASSIGN_OR_RAISE(FragmentIterator fragments_iter,
+ dataset->GetFragments(predicate));
+ return std::make_shared<FragmentIterator>(std::move(fragments_iter));
+ }));
+ Future<AsyncGenerator<std::shared_ptr<Fragment>>> fragments_gen_fut =
+ fragments_it_fut.Then([](const std::shared_ptr<FragmentIterator>& fragments_it)
+ -> Result<AsyncGenerator<std::shared_ptr<Fragment>>> {
+ ARROW_ASSIGN_OR_RAISE(std::vector<std::shared_ptr<Fragment>> fragments,
+ fragments_it->ToVector());
+ return MakeVectorGenerator(std::move(fragments));
+ });
+ return MakeFromFuture(std::move(fragments_gen_fut));
+}
+
+/// \brief A node that scans a dataset
+///
+/// The scan node has three groups of io-tasks and one task.
+///
+/// The first io-task (listing) fetches the fragments from the dataset. This may be a
+/// simple iteration of paths or, if the dataset is described with wildcards, this may
+/// involve I/O for listing and walking directory paths. There is one listing io-task per
+/// dataset.
+///
+/// Ths next step is to fetch the metadata for the fragment. For some formats (e.g. CSV)
+/// this may be quite simple (get the size of the file). For other formats (e.g. parquet)
+/// this is more involved and requires reading data. There is one metadata io-task per
+/// fragment. The metadata io-task creates an AsyncGenerator<RecordBatch> from the
+/// fragment.
+///
+/// Once the metadata io-task is done we can issue read io-tasks. Each read io-task
+/// requests a single batch of data from the disk by pulling the next Future from the
+/// generator.
+///
+/// Finally, when the future is fulfilled, we issue a pipeline task to drive the batch
+/// through the pipeline.
+///
+/// Most of these tasks are io-tasks. They take very few CPU resources and they run on
+/// the I/O thread pool. These io-tasks are invisible to the exec plan and so we need to
+/// do some custom scheduling. We limit how many fragments we read from at any one time.
+/// This is referred to as "fragment readahead".
+///
+/// Within a fragment there is usually also some amount of "row readahead". This row
+/// readahead is handled by the fragment (and not the scanner) because the exact details
+/// of how it is performed depend on the underlying format.
+///
+/// When a scan node is aborted (StopProducing) we send a cancel signal to any active
+/// fragments. On destruction we continue consuming the fragments until they complete
+/// (which should be fairly quick since we cancelled the fragment). This ensures the
+/// I/O work is completely finished before the node is destroyed.
+class ScanNode : public cp::ExecNode {
+ public:
+ ScanNode(cp::ExecPlan* plan, ScanV2Options options,
+ std::shared_ptr<Schema> output_schema)
+ : cp::ExecNode(plan, {}, {}, std::move(output_schema),
+ /*num_outputs=*/1),
+ options_(options),
+ synchronization_(options_.fragment_readahead) {}
+
+ static Result<ScanV2Options> NormalizeAndValidate(const ScanV2Options& options,
+ compute::ExecContext* ctx) {
+ ScanV2Options normalized(options);
+ if (!normalized.dataset) {
+ return Status::Invalid("Scan options must include a dataset");
+ }
+
+ if (normalized.filter.IsBound()) {
+ // There is no easy way to make sure a filter was bound agaisnt the same
+ // function registry as the one in ctx so we just require it to be unbound
+ // FIXME - Do we care if it was bound to a different function registry?
+ return Status::Invalid("Scan filter must be unbound");
+ } else {
+ ARROW_ASSIGN_OR_RAISE(normalized.filter,
+ normalized.filter.Bind(*options.dataset->schema(), ctx));
+ }
+
+ return std::move(normalized);
+ }
+
+ static Result<cp::ExecNode*> Make(cp::ExecPlan* plan, std::vector<cp::ExecNode*> inputs,
+ const cp::ExecNodeOptions& options) {
+ RETURN_NOT_OK(ValidateExecNodeInputs(plan, inputs, 0, "ScanNode"));
+ const auto& scan_options = checked_cast<const ScanV2Options&>(options);
+ ARROW_ASSIGN_OR_RAISE(ScanV2Options normalized_options,
+ NormalizeAndValidate(scan_options, plan->exec_context()));
+ ARROW_ASSIGN_OR_RAISE(std::shared_ptr<Schema> output_schema,
+ OutputSchemaFromOptions(normalized_options));
+ return plan->EmplaceNode<ScanNode>(plan, std::move(normalized_options),
+ std::move(output_schema));
+ }
+
+ const char* kind_name() const override { return "ScanNode"; }
+
+ [[noreturn]] static void NoInputs() {
+ Unreachable("no inputs; this should never be called");
+ }
+ [[noreturn]] void InputReceived(cp::ExecNode*, cp::ExecBatch) override { NoInputs(); }
+ [[noreturn]] void ErrorReceived(cp::ExecNode*, Status) override { NoInputs(); }
+ [[noreturn]] void InputFinished(cp::ExecNode*, int) override { NoInputs(); }
+
+ Status Init() override {
+ batch_output_ =
+ ::arrow::internal::make_unique<UnorderedBatchOutputStrategy>(this, outputs_[0]);
+ return Status::OK();
+ }
+
+ Status StartProducing() override {
+ START_COMPUTE_SPAN(span_, std::string(kind_name()) + ":" + label(),
+ {{"node.kind", kind_name()},
+ {"node.label", label()},
+ {"node.output_schema", output_schema()->ToString()},
+ {"node.detail", ToString()}});
+ END_SPAN_ON_FUTURE_COMPLETION(span_, finished_);
+ // std::cout << "StartProducing\n";
+ ARROW_ASSIGN_OR_RAISE(Future<> list_task, plan_->BeginExternalTask());
+ if (!list_task.is_valid()) {
+ // Cancelled before we even started
+ return Status::OK();
+ }
+ // std::cout << "LIST TASK: BEGIN\n";
+
+ AsyncGenerator<std::shared_ptr<Fragment>> frag_gen =
+ GetFragments(options_.dataset.get(), options_.filter);
+ Future<> visit_task = VisitAsyncGenerator(
+ std::move(frag_gen), [this](const std::shared_ptr<Fragment>& fragment) {
+ // std::cout << "Queuing fragment: " + fragment->ToString() + "\n";
+ QueueOrStartFragment(fragment);
+ return Status::OK();
+ });
+ visit_task
+ .Then(
+ [this]() {
+ // std::cout << "Marking listing completed\n";
+ if (synchronization_.MarkListingCompleted()) {
+ // It's possible, but probably unlikely, that all fragment scans have
+ // already finished and so we need to report the total batch count here in
+ // this case
+ batch_output_->InputFinished();
+ finished_.MarkFinished();
+ }
+ },
+ [this](const Status& err) {
+ synchronization_.MarkListingFailed(err);
+ outputs_[0]->ErrorReceived(this, err);
+ })
+ .AddCallback([list_task](const Status& st) mutable {
+ // std::cout << "LIST TASK: END\n";
+ list_task.MarkFinished();
+ });
+
+ return Status::OK();
+ }
+
+ void PauseProducing(ExecNode* output, int32_t counter) override {
+ // FIXME(TODO)
+ // Need to ressurect AsyncToggle and then all fragment scanners
+ // should share the same toggle
+ }
+
+ void ResumeProducing(ExecNode* output, int32_t counter) override {
+ // FIXME(TODO)
+ }
+
+ void StopProducing(ExecNode* output) override {
+ DCHECK_EQ(output, outputs_[0]);
+ StopProducing();
+ }
+
+ void StopProducing() override { synchronization_.Cancel(); }
+
+ private:
+ struct FragmentScanTask;
+ using ScanTaskPtr = std::list<FragmentScanTask>::iterator;
+ struct FragmentScanTask {
+ std::unique_ptr<FragmentEvolutionStrategy> fragment_evolution;
+ std::shared_ptr<FragmentScanner> fragment_scanner;
+ uint64_t unused_readahead_bytes;
+ bool iterating_tasks = false;
+ // Transitions to true when we've finished listing fragments for this task
+ bool finished = false;
+ // Transitions to true when we receieve an error or cancellation is requested
Review Comment:
This code went away (and the complexity went into the async task scheduler for the most part)
--
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