Execute instead of scan

cpp/src/arrow/dataset/file_parquet.cc

@@ -50,7 +50,7 @@ class ParquetScanTask : public ScanTask {
         column_projection_(std::move(column_projection)),
         reader_(reader) {}
 
-  Result<RecordBatchIterator> Scan() {
+  Result<RecordBatchIterator> Execute() {
     // The construction of parquet's RecordBatchReader is deferred here to
     // control the memory usage of consumers who materialize all ScanTasks
     // before dispatching them, e.g. for scheduling purposes.

cpp/src/arrow/dataset/file_parquet_test.cc

@@ -174,7 +174,7 @@ TEST_F(TestParquetFileFormat, ScanRecordBatchReader) {
 
   for (auto maybe_task : scan_task_it) {
     ASSERT_OK_AND_ASSIGN(auto task, std::move(maybe_task));
-    ASSERT_OK_AND_ASSIGN(auto rb_it, task->Scan());
+    ASSERT_OK_AND_ASSIGN(auto rb_it, task->Execute());
     for (auto maybe_batch : rb_it) {
       ASSERT_OK_AND_ASSIGN(auto batch, std::move(maybe_batch));
       row_count += batch->num_rows();
@@ -221,7 +221,7 @@ TEST_F(TestParquetFileFormat, ScanRecordBatchReaderProjected) {
 
   for (auto maybe_task : scan_task_it) {
     ASSERT_OK_AND_ASSIGN(auto task, std::move(maybe_task));
-    ASSERT_OK_AND_ASSIGN(auto rb_it, task->Scan());
+    ASSERT_OK_AND_ASSIGN(auto rb_it, task->Execute());
     for (auto maybe_batch : rb_it) {
       ASSERT_OK_AND_ASSIGN(auto batch, std::move(maybe_batch));
       row_count += batch->num_rows();
@@ -262,7 +262,7 @@ TEST_F(TestParquetFileFormat, ScanRecordBatchReaderProjectedMissingCols) {
 
   for (auto maybe_task : scan_task_it) {
     ASSERT_OK_AND_ASSIGN(auto task, std::move(maybe_task));
-    ASSERT_OK_AND_ASSIGN(auto rb_it, task->Scan());
+    ASSERT_OK_AND_ASSIGN(auto rb_it, task->Execute());
     for (auto maybe_batch : rb_it) {
       ASSERT_OK_AND_ASSIGN(auto batch, std::move(maybe_batch));
       row_count += batch->num_rows();
@@ -307,7 +307,7 @@ void CountRowsInScan(ScanTaskIterator& it, int64_t expected_rows,
 
   for (auto maybe_scan_task : it) {
     ASSERT_OK_AND_ASSIGN(auto scan_task, std::move(maybe_scan_task));
-    ASSERT_OK_AND_ASSIGN(auto rb_it, scan_task->Scan());
+    ASSERT_OK_AND_ASSIGN(auto rb_it, scan_task->Execute());
     for (auto maybe_record_batch : rb_it) {
       ASSERT_OK_AND_ASSIGN(auto record_batch, std::move(maybe_record_batch));
       actual_rows += record_batch->num_rows();
@@ -329,7 +329,7 @@ class TestParquetFileFormatPushDown : public TestParquetFileFormat {
     ASSERT_OK_AND_ASSIGN(auto it, fragment.Scan(ctx_));
     for (auto maybe_scan_task : it) {
       ASSERT_OK_AND_ASSIGN(auto scan_task, std::move(maybe_scan_task));
-      ASSERT_OK_AND_ASSIGN(auto rb_it, scan_task->Scan());
+      ASSERT_OK_AND_ASSIGN(auto rb_it, scan_task->Execute());
       for (auto maybe_record_batch : rb_it) {
         ASSERT_OK_AND_ASSIGN(auto record_batch, std::move(maybe_record_batch));
         actual_rows += record_batch->num_rows();

cpp/src/arrow/dataset/scanner.cc

@@ -46,7 +46,7 @@ ScanOptionsPtr ScanOptions::ReplaceSchema(std::shared_ptr<Schema> schema) const
   return copy;
 }
 
-Result<RecordBatchIterator> SimpleScanTask::Scan() {
+Result<RecordBatchIterator> SimpleScanTask::Execute() {
   return MakeVectorIterator(record_batches_);
 }
 
@@ -169,7 +169,7 @@ struct TableAggregator {
 
 struct ScanTaskPromise {
   Status operator()() {
-    ARROW_ASSIGN_OR_RAISE(auto it, task->Scan());
+    ARROW_ASSIGN_OR_RAISE(auto it, task->Execute());
     for (auto maybe_batch : it) {
       ARROW_ASSIGN_OR_RAISE(auto batch, std::move(maybe_batch));
       aggregator.Append(std::move(batch));

cpp/src/arrow/dataset/scanner.h

@@ -85,7 +85,7 @@ class ARROW_DS_EXPORT ScanTask {
   /// \brief Iterate through sequence of materialized record batches
   /// resulting from the Scan. Execution semantics are encapsulated in the
   /// particular ScanTask implementation
-  virtual Result<RecordBatchIterator> Scan() = 0;
+  virtual Result<RecordBatchIterator> Execute() = 0;
 
   virtual ~ScanTask() = default;
 
@@ -108,7 +108,7 @@ class ARROW_DS_EXPORT SimpleScanTask : public ScanTask {
       : ScanTask(std::move(options), std::move(context)),
         record_batches_(std::move(record_batches)) {}
 
-  Result<RecordBatchIterator> Scan() override;
+  Result<RecordBatchIterator> Execute() override;
 
  protected:
   std::vector<std::shared_ptr<RecordBatch>> record_batches_;

cpp/src/arrow/dataset/scanner_internal.h

@@ -53,8 +53,8 @@ class FilterAndProjectScanTask : public ScanTask {
   explicit FilterAndProjectScanTask(ScanTaskPtr task)
       : ScanTask(task->options(), task->context()), task_(std::move(task)) {}
 
-  Result<RecordBatchIterator> Scan() override {
-    ARROW_ASSIGN_OR_RAISE(auto it, task_->Scan());
+  Result<RecordBatchIterator> Execute() override {
+    ARROW_ASSIGN_OR_RAISE(auto it, task_->Execute());
     auto filter_it = FilterRecordBatch(std::move(it), *options_->evaluator,
                                        *options_->filter, context_->pool);
     return ProjectRecordBatch(std::move(filter_it), &task_->options()->projector,

cpp/src/arrow/dataset/test_util.h

@@ -87,7 +87,7 @@ class DatasetFixtureMixin : public ::testing::Test {
   /// record batches yielded by the data fragment.
   void AssertScanTaskEquals(RecordBatchReader* expected, ScanTask* task,
                             bool ensure_drained = true) {
-    ASSERT_OK_AND_ASSIGN(auto it, task->Scan());
+    ASSERT_OK_AND_ASSIGN(auto it, task->Execute());
     ARROW_EXPECT_OK(it.Visit([expected](std::shared_ptr<RecordBatch> rhs) -> Status {
       std::shared_ptr<RecordBatch> lhs;
       RETURN_NOT_OK(expected->ReadNext(&lhs));

python/pyarrow/_dataset.pyx

@@ -544,7 +544,7 @@ cdef class ScanTask:
     cdef inline shared_ptr[CScanTask] unwrap(self):
         return self.wrapped
 
-    def scan(self):
+    def execute(self):
         """Iterate through sequence of materialized record batches.
 
         Execution semantics are encapsulated in the particular ScanTask
@@ -558,7 +558,7 @@ cdef class ScanTask:
             CRecordBatchIterator iterator
             shared_ptr[CRecordBatch] record_batch
 
-        iterator = move(GetResultValue(move(self.task.Scan())))
+        iterator = move(GetResultValue(move(self.task.Execute())))
 
         while True:
             record_batch = GetResultValue(iterator.Next())

python/pyarrow/includes/libarrow_dataset.pxd

@@ -153,7 +153,7 @@ cdef extern from "arrow/dataset/api.h" namespace "arrow::dataset" nogil:
         "arrow::dataset::ScanContextPtr"
 
     cdef cppclass CScanTask" arrow::dataset::ScanTask":
-        CResult[CRecordBatchIterator] Scan()
+        CResult[CRecordBatchIterator] Execute()
 
     cdef cppclass CSimpleScanTask "arrow::dataset::SimpleScanTask"(CScanTask):
         pass

python/pyarrow/tests/test_dataset.py

@@ -150,7 +150,7 @@ def test_dataset(simple_data_source, tree_data_source, record_batch, schema):
 
     for task in scanner.scan():
         assert isinstance(task, ds.ScanTask)
-        for batch in task.scan():
+        for batch in task.execute():
             assert batch.equals(record_batch)
 
     table = scanner.to_table()
@@ -187,7 +187,7 @@ def test_scanner_builder(dataset):
     assert isinstance(scanner, ds.Scanner)
     assert len(list(scanner.scan())) == 3
     for task in scanner.scan():
-        for batch in task.scan():
+        for batch in task.execute():
             assert isinstance(batch, pa.RecordBatch)
             # FIXME(kszucs)
             # assert batch.num_columns == 1
@@ -347,7 +347,7 @@ def test_file_system_discovery(mockfs, paths_or_selector, record_batch):
 
     for task in scanner.scan():
         assert isinstance(task, ds.ScanTask)
-        for batch in task.scan():
+        for batch in task.execute():
             # FIXME(kszucs)
             continue
             # assert batch.equals(record_batch)