refactor: stream ContextChunk between operators

doc/source/development/operator_streams.md

@@ -0,0 +1,69 @@
+# Operator batch streams
+
+Every concrete `IOperator::Eval` accepts and returns `Stream<ContextChunk>`
+directly. Initialization errors are deferred to `Next()`. A stream is a move-only synchronous pull
+interface. `Next()` produces one batch, EOF, or a terminal error. An empty
+batch retains its schema and is not EOF. Destroying the stream releases its
+cursor without reading the rest of its input.
+
+`Next()` returns `result<std::optional<ContextChunk>>` directly. ContextChunk
+owns its DataChunk and anonymous execution head; Stream has no additional Batch
+wrapper. Readers and storage suppliers keep their DataChunk interface and adapt
+once at the Source/BatchInsert boundary. Between operators, the same ContextChunk
+is moved directly through the kernels. Stream tag_ids retain output aliases.
+
+## Execution and state
+
+- Source copies its read configuration per execution and opens its reader on
+  first demand. CSV, JSON and JSONL provide the same supplier interface.
+- Project, Select, Unfold, vertex/edge expansion, path operations, intersection
+  and the triangle kernel use `map_chunks`. One pull invokes the kernel on one
+  batch and returns that batch directly; no Context is constructed.
+- Limit owns its cross-batch skip and remaining-row counters in the returned
+  stream. Once satisfied, it releases upstream without pulling another batch.
+- Union buffers its shared input for branch replay, then concatenates branch
+  streams on demand. It does not collect or flatten branch outputs.
+- The existing sorting, fused Project/OrderBy, grouping, deduplication and
+  general join kernels require complete input. Their explicit collection
+  boundaries operate on batches, without converting through Context. The
+  primary-key join can process each right-side batch independently.
+- Mutations retain read/write barriers so a downstream LIMIT cannot skip
+  writes and writes cannot invalidate data that is still being read. Edge
+  update/merge retains affected batches to refresh their property pointers.
+- BatchInsert consumes the same stream through the storage supplier API and
+  checks terminal reader errors before reporting success. Sink only sets tags.
+- DDL and administration execute once and pass through their stream.
+
+All mutable cursors and counters belong to the execution's stream, not the
+cached operator. `Pipeline::ExecuteStream` connects nested plans without
+materializing their results. Its caller must keep the pipeline, storage and
+PROFILE timer alive until consumption or destruction. The external
+`Pipeline::Execute` query boundary still materializes its public Context result.
+
+PROFILE charges production during `Next()` and excludes nested upstream time
+within a pipeline. Errors retain their producer's operator name while passing
+through downstream consumers.
+
+## Compatibility boundaries and remaining buffering
+
+Index-scan and export extension callbacks still take Context in their existing
+ABI; those operators convert only at that explicit callback boundary. Procedure
+and GDS callbacks still return Context, which is exposed as a stream once.
+Every registered reader must supply a supplier factory; there is no materialized
+reader callback or fallback. Parquet uses RecordBatchReader directly without a
+CountRows or ToTable pass. Reader classes expose only supplier creation and schema
+inference: their Context-based read methods, ReadLocalState, and the obsolete
+batch_read option have been removed. Reader tests consume suppliers directly.
+
+This does not replace every underlying algorithm with an incremental one.
+Graph scan kernels may still produce a large single batch. CSV counts a file
+before parsing; JSON array decoding still builds a document. Global sorting,
+grouping, deduplication and general join still buffer their inputs. Edge storage
+still accumulates endpoints and property batches, and mutation barriers retain
+input as described above. The interface is lazy, not asynchronous, and does not
+add parallel loading.
+
+Operators return Stream<ContextChunk> directly, and ExecuteStream only connects
+streams. Fallible execution initialization is deferred until the first Next();
+Next() is the stream error boundary and makes failures terminal. Plan builders
+and the materializing public Execute() still return result for their own work.

extension/parquet/include/parquet/arrow_reader.h

@@ -27,6 +27,7 @@
 #include "parquet/arrow_options.h"
 
 namespace neug {
+class IDataChunkSupplier;
 namespace reader {
 
 class DatasetBuilder {
@@ -49,8 +50,7 @@ class Reader {
         fileSystem(std::move(fileSystem)) {}
   virtual ~Reader() = default;
 
-  virtual void read(std::shared_ptr<ReadLocalState> localState,
-                    execution::Context& ctx) = 0;
+  virtual std::shared_ptr<IDataChunkSupplier> getDataChunkSupplier() = 0;
 
  protected:
   std::shared_ptr<ReadSharedState> sharedState;
@@ -74,20 +74,14 @@ class ArrowReader : public Reader<arrow::fs::FileSystem> {
         datasetBuilder(std::move(datasetBuilder)) {}
   ~ArrowReader() override = default;
 
-  void read(std::shared_ptr<ReadLocalState> localState,
-            execution::Context& ctx) override;
+  std::shared_ptr<IDataChunkSupplier> getDataChunkSupplier() override;
 
   arrow::Result<std::shared_ptr<arrow::Schema>> inferSchema();
 
  protected:
   std::shared_ptr<arrow::dataset::Scanner> createScanner(
       std::shared_ptr<arrow::fs::FileSystem> fs);
-  void full_read(std::shared_ptr<arrow::dataset::Scanner> scanner,
-                 execution::Context& output);
-  void batch_read(std::shared_ptr<arrow::dataset::Scanner> scanner,
-                  execution::Context& output);
 
-  DataChunk finishChunk(DataChunk chunk) const;
   bool filter_after_read_ = false;
 
   std::unique_ptr<ArrowOptionsBuilder> optionsBuilder;

extension/parquet/include/parquet_read_function.h

@@ -37,15 +37,15 @@ struct ParquetReadFunction {
   static function_set getFunctionSet() {
     auto typeIDs =
         std::vector<::neug::DataTypeId>{::neug::DataTypeId::kVarchar};
-    auto readFunction = std::make_unique<ReadFunction>(name, typeIDs);
-    readFunction->execFunc = execFunc;
+    auto readFunction =
+        std::make_unique<ReadFunction>(name, typeIDs, supplierFunc);
     readFunction->sniffFunc = sniffFunc;
     function_set functionSet;
     functionSet.push_back(std::move(readFunction));
     return functionSet;
   }
 
-  static execution::Context execFunc(
+  static std::shared_ptr<IDataChunkSupplier> supplierFunc(
       std::shared_ptr<reader::ReadSharedState> state) {
     const auto& vfs = neug::main::MetadataRegistry::getVFS();
     const auto& fs = vfs->Provide(state->schema.file);
@@ -64,10 +64,7 @@ struct ParquetReadFunction {
     auto reader = std::make_unique<reader::ArrowReader>(
         state, std::move(optionsBuilder), std::move(arrowFs));
 
-    execution::Context ctx;
-    auto localState = std::make_shared<reader::ReadLocalState>();
-    reader->read(localState, ctx);
-    return ctx;
+    return reader->getDataChunkSupplier();
   }
 
   static std::shared_ptr<reader::EntrySchema> sniffFunc(

extension/parquet/src/arrow_reader.cc

@@ -103,7 +103,7 @@ std::vector<std::string> fallbackProjection(const ReadSharedState& state) {
     }
   }
 
-  // Keep a stable scan order; finishChunk restores the requested output order.
+  // Keep a stable scan order; the supplier restores the requested output order.
   std::vector<std::string> columns;
   for (const auto& name : all_columns) {
     if (required.erase(name)) {
@@ -119,27 +119,48 @@ std::vector<std::string> fallbackProjection(const ReadSharedState& state) {
 
 }  // namespace
 
-void ArrowReader::read(std::shared_ptr<ReadLocalState> localState,
-                       execution::Context& ctx) {
-  if (!sharedState) {
-    THROW_INVALID_ARGUMENT_EXCEPTION("SharedState is null");
-  }
-
-  if (!fileSystem) {
-    THROW_INVALID_ARGUMENT_EXCEPTION("FileSystem is null");
-  }
-
+std::shared_ptr<IDataChunkSupplier> ArrowReader::getDataChunkSupplier() {
   auto scanner = createScanner(fileSystem);
-  NEUG_ASSERT(scanner != nullptr);
+  auto batches = scanner->ToRecordBatchReader();
+  if (!batches.ok()) {
+    THROW_IO_EXCEPTION("Failed to create RecordBatchReader: " +
+                       batches.status().message());
+  }
+  // Row count is unknown until consumption. Never scan the dataset merely to
+  // count rows before producing its first batch.
+  auto supplier =
+      std::make_shared<RecordBatchChunkSupplier>(batches.ValueOrDie(), -1);
+  if (!filter_after_read_) {
+    return supplier;
+  }
+  class FilteredSupplier final : public IDataChunkSupplier {
+   public:
+    FilteredSupplier(std::shared_ptr<IDataChunkSupplier> input,
+                     ReadSharedState state, std::vector<std::string> columns)
+        : input_(std::move(input)),
+          state_(std::move(state)),
+          columns_(std::move(columns)) {}
+
+    int64_t RowNum() const override { return -1; }
+
+    std::shared_ptr<DataChunk> GetNextChunk() override {
+      auto chunk = input_->GetNextChunk();
+      if (!chunk || chunk->col_num() == 0) {
+        return chunk;
+      }
+      auto filtered =
+          filter_chunk(*chunk, state_.skipRows, columns_, state_.parameters);
+      return std::make_shared<DataChunk>(
+          project_chunk(filtered, columns_, state_.projectColumns));
+    }
 
-  // Choose read mode: batch_read streams data, full_read loads entire dataset
-  const auto& fileSchema = sharedState->schema.file;
-  ReadOptions options;
-  if (options.batch_read.get(fileSchema.options)) {
-    batch_read(scanner, ctx);
-  } else {
-    full_read(scanner, ctx);
-  }
+   private:
+    std::shared_ptr<IDataChunkSupplier> input_;
+    ReadSharedState state_;
+    std::vector<std::string> columns_;
+  };
+  return std::make_shared<FilteredSupplier>(supplier, *sharedState,
+                                            fallback_columns_);
 }
 
 std::shared_ptr<arrow::dataset::Scanner> ArrowReader::createScanner(
@@ -272,90 +293,6 @@ std::shared_ptr<arrow::dataset::Scanner> ArrowReader::createScanner(
   return scanner_result.ValueOrDie();
 }
 
-void ArrowReader::full_read(std::shared_ptr<arrow::dataset::Scanner> scanner,
-                            execution::Context& output) {
-  if (!sharedState) {
-    THROW_INVALID_ARGUMENT_EXCEPTION("SharedState is null");
-  }
-  if (!scanner) {
-    THROW_INVALID_ARGUMENT_EXCEPTION("Scanner is null");
-  }
-
-  auto table_result = scanner->ToTable();
-  if (!table_result.ok()) {
-    LOG(ERROR) << "Failed to read table via scanner: "
-               << table_result.status().message();
-    THROW_IO_EXCEPTION("Failed to read table via scanner: " +
-                       table_result.status().message());
-  }
-  auto table = table_result.ValueOrDie();
-
-  int num_cols =
-      filter_after_read_ ? fallback_columns_.size() : sharedState->columnNum();
-  if (num_cols != table->num_columns()) {
-    THROW_IO_EXCEPTION(
-        "Column number mismatch between schema and table, schema: " +
-        std::to_string(num_cols) +
-        ", table: " + std::to_string(table->num_columns()));
-  }
-
-  output.clear();
-  DataChunk chunk;
-  for (int i = 0; i < num_cols; ++i) {
-    auto table_column = table->column(i);
-    chunk.set(i, arrow_arrays_to_value_column(table_column->chunks()));
-  }
-  output.append_chunk(finishChunk(std::move(chunk)));
-}
-
-void ArrowReader::batch_read(std::shared_ptr<arrow::dataset::Scanner> scanner,
-                             execution::Context& output) {
-  if (!sharedState) {
-    THROW_INVALID_ARGUMENT_EXCEPTION("SharedState is null");
-  }
-  if (!scanner) {
-    THROW_INVALID_ARGUMENT_EXCEPTION("Scanner is null");
-  }
-  auto row_num_result = scanner->CountRows();
-  int64_t row_num = 0;
-  if (!row_num_result.ok()) {
-    LOG(WARNING) << "Failed to count rows via scanner: "
-                 << row_num_result.status().message();
-    THROW_IO_EXCEPTION("Failed to count rows via scanner: " +
-                       row_num_result.status().message());
-  } else {
-    VLOG(10) << "Row count from scanner: " << row_num_result.ValueOrDie();
-    row_num = row_num_result.ValueOrDie();
-  }
-
-  auto batch_reader_result = scanner->ToRecordBatchReader();
-  if (!batch_reader_result.ok()) {
-    LOG(ERROR) << "Failed to create RecordBatchReader from scanner: "
-               << batch_reader_result.status().message();
-    THROW_IO_EXCEPTION("Failed to create RecordBatchReader from scanner: " +
-                       batch_reader_result.status().message());
-  }
-  auto batch_reader = batch_reader_result.ValueOrDie();
-
-  auto batch_supplier =
-      std::make_shared<RecordBatchChunkSupplier>(batch_reader, row_num);
-
-  output.clear();
-  while (auto chunk = batch_supplier->GetNextChunk()) {
-    output.append_chunk(finishChunk(std::move(*chunk)));
-  }
-}
-
-DataChunk ArrowReader::finishChunk(DataChunk chunk) const {
-  if (!filter_after_read_ || chunk.col_num() == 0) {
-    return chunk;
-  }
-  auto filtered = filter_chunk(chunk, sharedState->skipRows, fallback_columns_,
-                               sharedState->parameters);
-  return project_chunk(filtered, fallback_columns_,
-                       sharedState->projectColumns);
-}
-
 arrow::Result<std::shared_ptr<arrow::Schema>> ArrowReader::inferSchema() {
   if (!sharedState) {
     return arrow::Status::Invalid(neug::StatusCode::ERR_INVALID_ARGUMENT,

extension/parquet/src/parquet_options.cc

@@ -44,6 +44,9 @@ ArrowOptions ArrowParquetOptionsBuilder::build() const {
   // this the scanner always runs single-threaded, regardless of
   // ArrowReaderProperties::set_use_threads() below.
   ReadOptions readOpts;
+  ParquetParseOptions parquetOpts;
+  scanOptions->batch_size =
+      parquetOpts.row_batch_size.get(state->schema.file.options);
   scanOptions->use_threads =
       readOpts.use_threads.get(state->schema.file.options);
   // The scanner needs an IOContext with an executor for parallel scans.