refactor: bulk build vetices/edges from CSV COPY inputs by zhanglei1949 · Pull Request #734 · alibaba/neug

zhanglei1949 · 2026-07-14T02:17:55Z

Optimize bulk load from CSV COPY inputs, by

Avoid reading all batches into memory by fusing DataSource & BatchInsert.
Parallelizing the bulk loading process with multiple threads.

try parallel bulk loading optimize for single

liulx20 · 2026-07-22T02:52:50Z

+
+ private:
+  const IContextColumn* column_;
+  const vector_t<EDATA_T>* values_ = nullptr;


I think there's no need to expose vector_t to the storage layer

liulx20 · 2026-07-22T03:54:26Z

+
+  /// Returns the repeatable source backing this supplier, when available.
+  /// Storage uses this hint to select staged bulk build internally.
+  virtual std::shared_ptr<IDataChunkSource> RepeatableSource() const {


Perhaps we could implement this by spilling intermediates. This would relax the "repeated" constraint and also avoid performing the oid→vid resolution twice.

liulx20 · 2026-07-22T03:58:18Z

+namespace chunk_pipeline_detail {
+
+inline int32_t hardware_worker_count() {
+  auto workers = static_cast<int32_t>(std::thread::hardware_concurrency());


should this be constrained by parameters such as max_thread_num?

Copilot

Pull request overview

Note

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

This PR refactors bulk loading from CSV COPY inputs to reduce memory usage and improve throughput by introducing repeatable chunk sources, parallel CSV chunk suppliers, and staged bulk-build paths for vertex/edge tables (Fix #735).

Changes:

Introduces IDataChunkSource/ChunkSourceOptions and fuses terminal COPY plans to feed storage directly from repeatable sources.
Adds parallel, record-boundary-aware CSV partition planning + range streaming to avoid loading full files into memory.
Implements staged bulk-build for vertices and bundled-edge CSR tables (two-pass count/fill) with cancellation/error propagation.

Reviewed changes

Copilot reviewed 32 out of 33 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
tests/utils/test_reader.h	Removes duplicate includes in test header.
tests/utils/test_reader.cc	Adds coverage for partitioned CSV scanning, projections, caching, and error propagation.
tests/unittest/utils.h	Adds reusable test chunk sources/suppliers and bulk-build size constant.
tests/storage/test_vertex_table.cc	Updates vertex insert API and adds bulk-build repeatable source test.
tests/storage/test_property_graph.cc	Fixes formatting/indent of closing namespace line.
tests/storage/test_edge_table.cc	Adds extensive bulk-build tests for bundled CSR and parallel/cancellation semantics.
tests/storage/test_copy_temp.cc	Adds env-driven integration test for selecting edge bulk-build path via COPY.
src/utils/io/read/csv/csv_reader.cc	Adds `CsvReader::createChunkSource()` to create repeatable CSV sources for terminal COPY.
src/utils/io/read/common/options.cc	Plumbs `use_threads` from read options into `CsvReadConfig`.
src/storages/loader/loader_utils.cc	Implements CSV file scanner, partition planning, range streams, chunk sources, and bulk-build source utilities.
src/storages/loader/bundled_edge_csr_loader.h	Declares bundled CSR bulk loader used by staged edge build.
src/storages/loader/bundled_edge_csr_loader.cc	Implements two-pass bundled CSR bulk build with parallel fill and cancellation behavior.
src/storages/graph/vertex_table.cc	Renames/extends vertex batch insert to staged bulk-build using repeatable sources.
src/storages/graph/property_graph.cc	Updates graph batch APIs to forward suppliers via move and new vertex method name.
src/storages/graph/graph_interface.cc	Rebuilds mutable view after batch vertex/edge updates on success.
src/storages/graph/edge_table.cc	Adds staged bundled-edge CSR bulk-build path and aligns fallback CSR sizing to indexer capacity.
src/storages/csr/mutable_csr.cc	Factors capacity reserve logic into shared helper.
src/execution/execute/plan_parser.cc	Registers new terminal fused batch-insert-from-source operator builders.
src/execution/execute/ops/batch/batch_update_utils.cc	Adds helpers for resolving labels and building `BatchInsertInput` from an optional repeatable source.
src/execution/execute/ops/batch/batch_insert_vertex.cc	Adds terminal fused-from-source path and improves control flow around supplier creation.
src/execution/execute/ops/batch/batch_insert_edge.cc	Adds terminal fused-from-source path and reuses shared vertex label resolution helper.
include/neug/utils/io/read/csv/csv_reader.h	Declares `createChunkSource()` API.
include/neug/utils/io/read/csv/csv_read_config.h	Adds `use_threads` to CSV config.
include/neug/storages/loader/loader_utils.h	Adds chunk source interfaces, bulk-build heuristics, CSV chunk source, and row count mode.
include/neug/storages/loader/chunk_pipeline_utils.h	Adds bounded queue + indexed-consumer pipeline helpers with cancellation/error handling.
include/neug/storages/graph/vertex_table.h	Renames insert method, forward-declares source/supplier, and adds staged build helpers.
include/neug/storages/graph/edge_table.h	Declares staged edge bulk-build helper API.
include/neug/storages/csr/mutable_csr.h	Exposes reserve-capacity helper and grants loader friend access.
include/neug/execution/execute/ops/batch/batch_update_utils.h	Declares new batch-insert helper structs/functions and forward declarations.
include/neug/execution/execute/ops/batch/batch_insert_vertex.h	Adds fused terminal-from-source operator builder.
include/neug/execution/execute/ops/batch/batch_insert_edge.h	Adds fused terminal-from-source operator builder.
include/neug/compiler/function/read_function.h	Adds `sourceFunc` for repeatable sources for terminal ingestion.
include/neug/compiler/function/import/csv_read_function.h	Implements CSV `sourceFunc` and factors shared reader/path resolution helpers.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

+  CsvScanResult scan_with_ranges(size_t file_size, int32_t requested_partitions,
+                                 int32_t scan_threads) const {
+    CsvScanResult result;
+    if (file_size == 0) {
+      return result;
+    }
+    const auto target_partitions = std::min(
+        file_size,
+        static_cast<size_t>(std::max<int32_t>(1, requested_partitions)));
+    const auto workers = static_cast<unsigned>(std::min(
+        file_size, static_cast<size_t>(std::max<int32_t>(1, scan_threads))));
+    if (workers == 1) {
+      result.row_count = count_single(file_size);
+      result.ranges.push_back({0, file_size, 0});
+      return result;
+    }
+    result = scan_parallel(file_size, workers, true);


+  explicit BulkEdgeDataReader(const std::shared_ptr<IContextColumn>& column)
+      : column_(column.get()),
+        value_column_(dynamic_cast<const ValueColumn<EDATA_T>*>(column.get())) {
+    CHECK(column_ != nullptr);
+  }


+  if (!resolve_vertex_label_id(graph.schema(), vertex_type_, vertex_label_id)) {
+    RETURN_STATUS_ERROR(StatusCode::ERR_INVALID_ARGUMENT,
+                        "Failed to resolve vertex type for BatchInsertVertex");
  }


+class ScopedEnvironmentVariable final {
+ public:
+  ScopedEnvironmentVariable(const char* key, const char* value) : key_(key) {
+    if (const char* previous = std::getenv(key); previous != nullptr) {
+      previous_ = previous;
+    }
+    if (::setenv(key, value, 1) != 0) {
+      throw std::runtime_error("Failed to set test environment variable");
+    }
+  }
+
+  ~ScopedEnvironmentVariable() {
+    if (previous_) {
+      ::setenv(key_.c_str(), previous_->c_str(), 1);
+    } else {
+      ::unsetenv(key_.c_str());
+    }
+  }


Copilot

Pull request overview

Copilot reviewed 32 out of 33 changed files in this pull request and generated 3 comments.

Comments suppressed due to low confidence (1)

tests/utils/test_reader.cc:1

ASSERT_NE(second, nullptr) can early-return the test before renaming the file back, potentially leaving the fixture directory in a mutated state and causing cascading failures in later tests. Use an RAII cleanup (scope guard) to restore the original filename unconditionally, or avoid ASSERT_* between rename/restore so cleanup always runs.

+  CHECK(supplier != nullptr);
+  const auto src_vertex_capacity = static_cast<vid_t>(src_indexer.capacity());
+  const auto dst_vertex_capacity = static_cast<vid_t>(dst_indexer.capacity());
+  auto source = supplier->RepeatableSource();
+  if (source && TryBatchBuildEdges(src_indexer, dst_indexer, source,
+                                   src_vertex_capacity, dst_vertex_capacity)) {
+    return;
+  }
+
+  // Keep fallback COPY paths aligned with the vertex table's actual capacity,
+  // while leaving completely unloaded edge tables lazy until persistence.
+  in_csr_->resize(dst_vertex_capacity);
+  out_csr_->resize(src_vertex_capacity);


+class ScopedEnvironmentVariable final {
+ public:
+  ScopedEnvironmentVariable(const char* key, const char* value) : key_(key) {
+    if (const char* previous = std::getenv(key); previous != nullptr) {
+      previous_ = previous;
+    }
+    if (::setenv(key, value, 1) != 0) {
+      throw std::runtime_error("Failed to set test environment variable");
+    }
+  }
+
+  ~ScopedEnvironmentVariable() {
+    if (previous_) {
+      ::setenv(key_.c_str(), previous_->c_str(), 1);
+    } else {
+      ::unsetenv(key_.c_str());
+    }
+  }


+BatchInsertSource build_batch_insert_source(const physical::PhysicalPlan& plan,
+                                            int op_idx) {
+  const auto& source = plan.plan(op_idx).opr().source();
+  ReadStateBuilder state_builder;
+  auto state = state_builder.build(source);
+  auto catalog = neug::main::MetadataRegistry::getCatalog();
+  auto registered_function =
+      catalog->getFunctionWithSignature(source.extension_name());
+  return {std::move(state),
+          registered_function->ptrCast<function::ReadFunction>()};
+}


Copilot

Pull request overview

Copilot reviewed 42 out of 43 changed files in this pull request and generated 3 comments.

 #include "neug/utils/io/file/file_utils.h"
 #include "neug/utils/property/types.h"

+#include "../loader/bundled_edge_csr_loader.h"


+  bool ParallelEnabled() const override { return config_.use_threads; }
+  bool ProvidesStableRowOrdinals() const override { return true; }


+  if (requested_threads < 0 || max_thread_num_ < 1) {
+    RETURN_ERROR(neug::Status(neug::StatusCode::ERR_INVALID_ARGUMENT,
+                              "Number of threads must be greater than 0"));
+  }
+  if (requested_threads == 0) {
+    return max_thread_num_;
+  }


zhanglei1949 mentioned this pull request Jul 14, 2026

[Feature] Optimize bulk load performance for CSV COPY inputs #735

Open

zhanglei1949 force-pushed the codex/copy-bulk-build-pipeline branch 4 times, most recently from e3eb453 to bf9a8b9 Compare July 14, 2026 11:01

feat: bulk build terminal CSV COPY inputs

4ad2827

try parallel bulk loading optimize for single

zhanglei1949 force-pushed the codex/copy-bulk-build-pipeline branch from 0d5fe27 to 4ad2827 Compare July 16, 2026 08:20

zhanglei1949 added 5 commits July 17, 2026 10:10

refactor code, reduce dummy code

78eb601

Merge branch 'main' into codex/copy-bulk-build-pipeline

0b5fcea

refactor and reduce code

fc674ed

fix vertex load issue

fa98077

Merge branch 'main' into codex/copy-bulk-build-pipeline

620d2b8

zhanglei1949 requested a review from liulx20 July 20, 2026 03:03

refactor

9fbec25

liulx20 reviewed Jul 22, 2026

View reviewed changes

avoid using vector_t in BulkEdgeDataReader

3a74fce

Copilot AI review requested due to automatic review settings July 22, 2026 03:49

Merge branch 'main' into codex/copy-bulk-build-pipeline

755029a

liulx20 reviewed Jul 22, 2026

View reviewed changes

Copilot AI reviewed Jul 22, 2026

View reviewed changes

Copilot AI review requested due to automatic review settings July 22, 2026 04:09

Copilot AI reviewed Jul 22, 2026

View reviewed changes

Copilot started reviewing on behalf of zhanglei1949 July 22, 2026 05:04 View session

use spill for csr loading to avoid read twice.

04ab362

Copilot AI review requested due to automatic review settings July 23, 2026 10:13

Copilot AI reviewed Jul 23, 2026

View reviewed changes

Copilot started reviewing on behalf of zhanglei1949 July 23, 2026 10:57 View session

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

refactor: bulk build vetices/edges from CSV COPY inputs#734

refactor: bulk build vetices/edges from CSV COPY inputs#734
zhanglei1949 wants to merge 10 commits into
alibaba:mainfrom
zhanglei1949:codex/copy-bulk-build-pipeline

zhanglei1949 commented Jul 14, 2026 •

edited

Loading

Uh oh!

liulx20 Jul 22, 2026 •

edited

Loading

Uh oh!

zhanglei1949 Jul 22, 2026

Uh oh!

liulx20 Jul 22, 2026

Uh oh!

liulx20 Jul 22, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		bool ParallelEnabled() const override { return config_.use_threads; }
		bool ProvidesStableRowOrdinals() const override { return true; }

Uh oh!

Conversation

zhanglei1949 commented Jul 14, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

liulx20 Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhanglei1949 Jul 22, 2026

Choose a reason for hiding this comment

Uh oh!

liulx20 Jul 22, 2026

Choose a reason for hiding this comment

Uh oh!

liulx20 Jul 22, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

zhanglei1949 commented Jul 14, 2026 •

edited

Loading

liulx20 Jul 22, 2026 •

edited

Loading