Arrow 22 的 Scanner::Scan 为何仍串行调用 visitor？

结论

在 Arrow 22.0.0 的实现中，Scanner::Scan 会逐个调用 visitor，use_threads = true 不会让多个 ProcessBatch 调用自动重叠执行。要流式读取并并行处理批次，可以由一个线程消费扫描迭代器，将批次提交给独立的工作线程，同时限制在途任务数量。Arrow 22.0.0 扫描器源码

对于这条回调调用路径，继续增大 fragment_readahead 或重复设置 ScannerBuilder::UseThreads(true)，都无法解决业务回调串行的问题。

为什么日志始终相隔 5 秒

示例在 ProcessBatch 入口打印日志，然后执行：

std::this_thread::sleep_for(std::chrono::seconds(5));

这次调用返回后，消费循环才会进入下一次 visitor，所以入口时间戳约相隔 5 秒。相同的线程 ID 与这种执行方式一致，但日志只反映回调的调用情况，无法据此判断底层读取、解码等工作是否也全部串行。

Arrow 22.0.0 中的相关调用路径为：

AsyncScanner::Scan
  -> VisitBatchesAsync
  -> ScanBatchesAsync
  -> VisitAsyncGenerator

visitor 返回同步的 Status，消费循环会在它返回后继续处理下一批。文档提到回调可能由工作线程调用，但这不意味着多次回调一定会同时运行。Arrow 22.0.0 的 Scan 实现

几个设置控制的行为不同：

ScanBatchesUnordered() 允许批次无序返回，但如果消费代码仍然是取一批、处理完、再取下一批，业务处理依旧串行。Scanner API 说明

采用有并发上限的批次处理

可以保留原来的 Dataset 和 Scanner 构建代码，用下面的函数替换 scanner->Scan(ProcessBatch)。

示例使用 std::async(std::launch::async, ...) 展示调度方式，最多保留 4 个尚未回收的处理任务。它不调用 ToTable()，也不收集整个数据集。

#include <arrow/api.h>
#include <arrow/dataset/api.h>
#include <arrow/util/iterator.h>

#include <cstddef>
#include <deque>
#include <exception>
#include <future>
#include <memory>
#include <utility>

// 使用原示例中的实现；并发执行时需保证内部操作线程安全。
arrow::Status ProcessBatch(
    const arrow::dataset::TaggedRecordBatch& tagged_batch);

arrow::Status ProcessBatchesParallel(
    const std::shared_ptr<arrow::dataset::Scanner>& scanner,
    std::size_t max_in_flight = 4) {
  if (max_in_flight == 0) {
    return arrow::Status::Invalid("max_in_flight must be positive");
  }

  ARROW_ASSIGN_OR_RAISE(auto batches, scanner->ScanBatchesUnordered());

  std::deque<std::future<arrow::Status>> pending;
  arrow::Status first_error = arrow::Status::OK();

  auto remember_error = [&](const arrow::Status& status) {
    if (first_error.ok() && !status.ok()) {
      first_error = status;
    }
  };

  auto wait_one = [&]() {
    auto task = std::move(pending.front());
    pending.pop_front();
    try {
      remember_error(task.get());
    } catch (const std::exception& e) {
      remember_error(arrow::Status::UnknownError(e.what()));
    } catch (...) {
      remember_error(arrow::Status::UnknownError("Batch task failed"));
    }
  };

  while (first_error.ok()) {
    if (pending.size() >= max_in_flight) {
      wait_one();
      if (!first_error.ok()) {
        break;
      }
    }

    auto next = batches.Next();
    if (!next.ok()) {
      remember_error(next.status());
      break;
    }

    auto batch = std::move(next).ValueOrDie();
    if (arrow::IsIterationEnd(batch)) {
      break;
    }

    arrow::dataset::TaggedRecordBatch tagged{
        batch.record_batch.value, batch.fragment.value};

    try {
      pending.emplace_back(std::async(
          std::launch::async,
          [tagged]() { return ProcessBatch(tagged); }));
    } catch (const std::exception& e) {
      remember_error(arrow::Status::UnknownError(e.what()));
      break;
    }
  }

  // 返回前等待所有已提交任务，回收结果并检查错误。
  while (!pending.empty()) {
    wait_one();
  }
  return first_error;
}

在原来的 ScanWholeDataset 中，构建完 Scanner 后调用：

ARROW_ASSIGN_OR_RAISE(auto scanner, scan_builder.Finish());
return ProcessBatchesParallel(scanner, 4);

代码使用 EnumeratedRecordBatch 中的 record_batch.value 和 fragment.value 构造 TaggedRecordBatch，以便继续复用原来的 ProcessBatch。

示例按提交顺序等待任务，较早的慢任务可能暂时影响后续提交。如果需要持续处理大量小批次，可以采用固定大小的线程池和有界队列，避免为每个批次反复启动异步任务。

并行处理时需要注意什么

批次必须按值持有

示例中的 [tagged] 会复制批次及 fragment 的 shared_ptr，使工作任务执行期间数据仍然有效。不要捕获循环局部变量的引用，也不要在任务完成前释放任务依赖的外部资源。

多个线程读取批次数据时通常不需要修改数据；如果处理函数会修改共享缓冲区、累加结果或写入同一个输出对象，就需要自行同步。

限制任务数量还需要配合预读设置

max_in_flight = 4 限制的是应用保留的任务数量。Scanner 内部预读、Parquet 解码缓冲区以及处理函数自身的分配仍会占用内存，因此这不是进程内存的硬上限。

可以从较小的 fragment_readahead、batch_readahead 和适当的 batch_size 开始，根据实际吞吐与内存占用调整。对于约 5 GiB 的 Parquet 文件，也不能直接把文件大小当作解码后的内存需求。

检查错误和完成顺序

原示例忽略了 Scan 的返回状态。如果继续使用同步回调，至少应改为：

ARROW_RETURN_NOT_OK(scanner->Scan(ProcessBatch));

如果回调只负责提交异步任务，Scan 返回时任务可能尚未完成，必须另外等待并检查任务结果。上面的示例会在返回前回收所有已提交任务，检测到错误后停止继续提交，但不会取消已经启动的任务。

并行任务的完成顺序可能不同于批次返回顺序。即使改用有序的 ScanBatches()，也需要在输出端额外维护顺序，才能保证最终结果按输入顺序写出。

用重叠执行验证并发

在业务处理入口增加原子计数器，在退出时减一，并记录最大同时运行数量，可以直接验证处理是否重叠。日志只应在输出时短暂加锁，避免锁覆盖整个处理过程。

原来的 5 秒休眠适合观察调用是否重叠；实际并发度和性能收益仍需用真实处理逻辑测量。原日志中的 system_clock::time_point 直接流输出还依赖相应的 C++ 标准库支持，编译环境不支持时可改为输出 time_since_epoch() 的毫秒数。

备注：内容仅供参考。