refactor: bulk build vetices/edges from CSV COPY inputs

include/neug/compiler/function/import/csv_read_function.h

@@ -36,6 +36,7 @@ struct CSVReadFunction {
         std::vector<common::DataTypeId>{common::DataTypeId::kVarchar};
     auto readFunction = std::make_unique<ReadFunction>(name, typeIDs);
     readFunction->execFunc = execFunc;
+    readFunction->sourceFunc = sourceFunc;
     readFunction->sniffFunc = sniffFunc;
     function_set functionSet;
     functionSet.push_back(std::move(readFunction));
@@ -111,27 +112,53 @@ struct CSVReadFunction {
     }
   }
 
-  static execution::Context execFunc(
-      std::shared_ptr<reader::ReadSharedState> state) {
-    validateAndConvertExecOptions(state);
-    const auto& vfs = neug::main::MetadataRegistry::getVFS();
-    const auto& fs = vfs->Provide(state->schema.file);
-    auto resolvedPaths = std::vector<std::string>();
+  static void resolvePaths(
+      const std::shared_ptr<reader::ReadSharedState>& state) {
+    const auto& fs =
+        neug::main::MetadataRegistry::getVFS()->Provide(state->schema.file);
+    std::vector<std::string> resolvedPaths;
     for (const auto& path : state->schema.file.paths) {
       const auto& resolved = fs->glob(path);
       resolvedPaths.insert(resolvedPaths.end(), resolved.begin(),
                            resolved.end());
     }
     state->schema.file.paths = std::move(resolvedPaths);
+  }
+
+  static std::shared_ptr<reader::CsvReader> createReader(
+      const std::shared_ptr<reader::ReadSharedState>& state) {
     auto optionsBuilder = std::make_unique<reader::CsvOptionsBuilder>(state);
-    auto reader =
-        std::make_unique<reader::CsvReader>(state, std::move(optionsBuilder));
+    return std::make_shared<reader::CsvReader>(state,
+                                               std::move(optionsBuilder));
+  }
+
+  static execution::Context execFunc(
+      std::shared_ptr<reader::ReadSharedState> state) {
+    validateAndConvertExecOptions(state);
+    resolvePaths(state);
+    auto reader = createReader(state);
     execution::Context ctx;
     auto localState = std::make_shared<reader::ReadLocalState>();
     reader->read(localState, ctx);
     return ctx;
   }
 
+  static std::unique_ptr<IDataChunkSource> sourceFunc(
+      std::shared_ptr<reader::ReadSharedState> state,
+      std::vector<int32_t> projected_columns) {
+    if (!state) {
+      THROW_INVALID_ARGUMENT_EXCEPTION("State is null");
+    }
+    // sourceFunc is speculative: a caller may create a source only to decide
+    // that the destination is ineligible, then execute the legacy path. Keep
+    // the original state pristine for that fallback.
+    auto source_state = std::make_shared<reader::ReadSharedState>(*state);
+    validateAndConvertExecOptions(source_state);
+    resolvePaths(source_state);
+    return createReader(source_state)
+        ->createChunkSource(std::move(projected_columns));
+  }
+
   static std::shared_ptr<reader::EntrySchema> sniffFunc(
       const reader::FileSchema& schema) {
     auto state = std::make_shared<reader::ReadSharedState>();
@@ -143,18 +170,8 @@ struct CSVReadFunction {
     validateAndConvertSniffOptions(externalSchema.file);
     externalSchema.file.options["BATCH_SIZE"] =
         std::to_string(reader::kSniffBlockSize);
-    const auto& vfs = neug::main::MetadataRegistry::getVFS();
-    const auto& fs = vfs->Provide(state->schema.file);
-    auto resolvedPaths = std::vector<std::string>();
-    for (const auto& path : state->schema.file.paths) {
-      const auto& resolved = fs->glob(path);
-      resolvedPaths.insert(resolvedPaths.end(), resolved.begin(),
-                           resolved.end());
-    }
-    state->schema.file.paths = std::move(resolvedPaths);
-    auto optionsBuilder = std::make_unique<reader::CsvOptionsBuilder>(state);
-    auto reader =
-        std::make_shared<reader::CsvReader>(state, std::move(optionsBuilder));
+    resolvePaths(state);
+    auto reader = createReader(state);
     auto sniffer = std::make_shared<reader::CsvSniffer>(reader);
     auto sniffResult = sniffer->sniff();
     if (sniffResult) {
@@ -170,9 +187,7 @@ struct CSVReadFunction {
     if (hasHeader) {
       options.insert({"SKIP_ROWS", "1"});
       options.insert({"AUTOGENERATE_COLUMN_NAMES", "TRUE"});
-      auto optionsBuilder2 = std::make_unique<reader::CsvOptionsBuilder>(state);
-      auto reader2 = std::make_shared<reader::CsvReader>(
-          state, std::move(optionsBuilder2));
+      auto reader2 = createReader(state);
       auto sniffer2 = std::make_shared<reader::CsvSniffer>(reader2);
       sniffResult = sniffer2->sniff();
       if (sniffResult) {
@@ -184,4 +199,4 @@ struct CSVReadFunction {
   }
 };
 }  // namespace function
-}  // namespace neug
+}  // namespace neug

include/neug/compiler/function/read_function.h

@@ -27,24 +27,32 @@
 #include "neug/utils/io/reader.h"
 
 namespace neug {
+class IDataChunkSource;
 namespace function {
 
 // The exec function invoked by data source operators to load data from external
 // data sources.
 using read_exec_func_t = std::function<execution::Context(
     std::shared_ptr<reader::ReadSharedState> state)>;
 
+/// Creates a configurable source for terminal ingestion. Storage selects the
+/// projection and concurrency plan before opening it once.
+using read_source_func_t = std::function<std::unique_ptr<IDataChunkSource>(
+    std::shared_ptr<reader::ReadSharedState> state,
+    std::vector<int32_t> projected_columns)>;
+
 // The function used to sniff/infer file column names and their types from
 // external data sources.
 using read_sniff_func_t = std::function<std::shared_ptr<reader::EntrySchema>(
     const reader::FileSchema& schema)>;
 
 struct ReadFunction : public TableFunction {
   read_exec_func_t execFunc = nullptr;
+  read_source_func_t sourceFunc = nullptr;
   read_sniff_func_t sniffFunc = nullptr;
 
   ReadFunction(std::string name, std::vector<common::DataTypeId> inputTypes)
       : TableFunction{std::move(name), std::move(inputTypes)} {}
 };
 }  // namespace function
-}  // namespace neug
+}  // namespace neug

include/neug/execution/execute/ops/batch/batch_insert_edge.h

@@ -40,6 +40,23 @@ class BatchInsertEdgeOprBuilder : public IOperatorBuilder {
   }
 };
 
+/// Fuses only a terminal, empty-sink COPY FROM plan. Storage chooses staged
+/// build or normal BatchAdd before opening the supplied source once.
+class BatchInsertEdgeFromSourceOprBuilder : public IOperatorBuilder {
+ public:
+  neug::result<OpBuildResultT> Build(const Schema& schema,
+                                     const ContextMeta& ctx_meta,
+                                     const physical::PhysicalPlan& plan,
+                                     int op_idx) override;
+
+  std::vector<physical::PhysicalOpr_Operator::OpKindCase> GetOpKinds()
+      const override {
+    return {physical::PhysicalOpr_Operator::OpKindCase::kSource,
+            physical::PhysicalOpr_Operator::OpKindCase::kLoadEdge,
+            physical::PhysicalOpr_Operator::OpKindCase::kSink};
+  }
+};
+
 }  // namespace ops
 }  // namespace execution
 }  // namespace neug

include/neug/execution/execute/ops/batch/batch_insert_vertex.h

@@ -39,6 +39,23 @@ class BatchInsertVertexOprBuilder : public IOperatorBuilder {
   }
 };
 
+/// Fuses only a terminal, empty-sink COPY FROM plan. Storage chooses staged
+/// build or normal BatchAdd before opening the supplied source once.
+class BatchInsertVertexFromSourceOprBuilder : public IOperatorBuilder {
+ public:
+  neug::result<OpBuildResultT> Build(const Schema& schema,
+                                     const ContextMeta& ctx_meta,
+                                     const physical::PhysicalPlan& plan,
+                                     int op_idx) override;
+
+  std::vector<physical::PhysicalOpr_Operator::OpKindCase> GetOpKinds()
+      const override {
+    return {physical::PhysicalOpr_Operator::OpKindCase::kSource,
+            physical::PhysicalOpr_Operator::OpKindCase::kLoadVertex,
+            physical::PhysicalOpr_Operator::OpKindCase::kSink};
+  }
+};
+
 }  // namespace ops
 }  // namespace execution
 }  // namespace neug

include/neug/execution/execute/ops/batch/batch_update_utils.h

@@ -18,11 +18,16 @@
 
 #include "neug/common/types/graph_types.h"
 #include "neug/execution/common/context.h"
+#include "neug/storages/loader/loader_utils.h"
 #include "neug/utils/property/types.h"
 
 namespace physical {
+class PhysicalPlan;
 class PropertyMapping;
-}
+}  // namespace physical
+namespace common {
+class NameOrId;
+}  // namespace common
 namespace google {
 namespace protobuf {
 template <typename T>
@@ -31,9 +36,14 @@ class RepeatedPtrField;
 }  // namespace google
 
 namespace neug {
-class IDataChunkSupplier;
 class Schema;
 class StorageReadInterface;
+namespace function {
+struct ReadFunction;
+}
+namespace reader {
+struct ReadSharedState;
+}
 namespace execution {
 
 namespace ops {
@@ -65,6 +75,29 @@ std::shared_ptr<IDataChunkSupplier> create_data_chunk_supplier(
     const Context& ctx,
     const std::vector<std::pair<int32_t, std::string>>& prop_mappings);
 
+bool resolve_vertex_label_id(const Schema& schema,
+                             const ::common::NameOrId& type, label_t& label_id);
+
+struct BatchInsertInput {
+  std::unique_ptr<IDataChunkSource> data;
+  Context output;
+};
+
+struct BatchInsertSource {
+  std::shared_ptr<reader::ReadSharedState> state;
+  function::ReadFunction* read_function;
+};
+
+bool is_terminal_batch_insert(const physical::PhysicalPlan& plan, int op_idx);
+
+BatchInsertSource build_batch_insert_source(const physical::PhysicalPlan& plan,
+                                            int op_idx);
+
+BatchInsertInput create_batch_insert_input(
+    const std::shared_ptr<reader::ReadSharedState>& shared_state,
+    const function::ReadFunction& read_function,
+    const std::vector<std::pair<int32_t, std::string>>& prop_mappings);
+
 std::vector<std::string> match_files_with_pattern(const std::string& file_path);
 
 std::vector<std::shared_ptr<IDataChunkSupplier>> create_csv_chunk_suppliers(

include/neug/main/query_processor.h

@@ -72,14 +72,15 @@ class QueryProcessor {
   result<std::pair<AccessMode, std::shared_ptr<execution::CacheValue>>>
   check_and_retrieve_pipeline(const PropertyGraph& pg,
                               const std::string& query_string,
-                              const std::string& access_mode,
-                              int32_t num_threads);
+                              const std::string& access_mode);
+
+  result<int32_t> resolve_thread_budget(int32_t requested_threads) const;
 
   result<QueryResult> execute_internal(
       SnapshotGuard& guard, const std::string& query_string,
       std::shared_ptr<execution::CacheValue> cache_value,
       AccessMode access_mode, const execution::ParamsMap& parameters = {},
-      int32_t num_threads = 0);
+      int32_t thread_budget = 0);
 
   result<QueryResult> execute_explain_mode(
       const std::string& query_string,

include/neug/storages/csr/mutable_csr.h

@@ -43,13 +43,23 @@
 
 namespace neug {
 
+namespace internal {
+class BundledEdgeCsrLoader;
+}
+
 // std::atomic<int> must have the same size as int on supported platforms
 // so that the degree_list buffer (persisted as int[]) can be safely
 // reinterpreted as atomic<int>[] for concurrent access.
 static_assert(
     sizeof(std::atomic<int>) == sizeof(int),
     "atomic<int> must have the same size as int on supported platforms");
 
+namespace mutable_csr_detail {
+
+int capacity_with_reserve(int degree);
+
+}  // namespace mutable_csr_detail
+
 template <typename EDATA_T>
 class MutableCsr : public TypedCsrBase<EDATA_T> {
  public:
@@ -235,6 +245,8 @@ class MutableCsr : public TypedCsrBase<EDATA_T> {
   }
 
  private:
+  friend class internal::BundledEdgeCsrLoader;
+
   std::unique_ptr<SpinLock[]> locks_;
   std::shared_ptr<IDataContainer> adj_list_buffer_;
   std::shared_ptr<IDataContainer> degree_list_;
@@ -382,6 +394,8 @@ class SingleMutableCsr : public TypedCsrBase<EDATA_T> {
   }
 
  private:
+  friend class internal::BundledEdgeCsrLoader;
+
   std::shared_ptr<IDataContainer> nbr_list_;
   std::atomic<uint64_t> edge_num_{0};
   CsrPrefetchPolicy prefetch_policy_;

include/neug/storages/graph/edge_table.h

@@ -29,6 +29,7 @@
 #include "neug/storages/csr/csr_base.h"
 #include "neug/storages/csr/csr_view.h"
 #include "neug/storages/graph/schema.h"
+#include "neug/storages/loader/loader_utils.h"
 #include "neug/storages/module/module.h"
 #include "neug/utils/indexers.h"
 #include "neug/utils/property/table.h"
@@ -40,8 +41,6 @@ class ModuleBroker;
 class CheckpointManifest;
 class PropertyGraph;
 
-class IDataChunkSupplier;
-
 class EdgeTable {
  public:
   EdgeTable(std::shared_ptr<const EdgeSchema> meta) : meta_(meta) {}
@@ -133,7 +132,8 @@ class EdgeTable {
 
   void BatchAddEdges(const IndexerType& src_indexer,
                      const IndexerType& dst_indexer,
-                     std::shared_ptr<IDataChunkSupplier> supplier);
+                     std::unique_ptr<IDataChunkSource> source,
+                     BulkLoadOptions options = {});
 
   // Add edges in batch to the edge table.
   void BatchAddEdges(const std::vector<vid_t>& src_lid_list,
@@ -190,6 +190,11 @@ class EdgeTable {
   void DetachInAdjlist(vid_t vid, Allocator& alloc);
 
  private:
+  bool TryBatchBuildEdges(const IndexerType& src_indexer,
+                          const IndexerType& dst_indexer,
+                          IDataChunkSource& source, vid_t src_vertex_capacity,
+                          vid_t dst_vertex_capacity, BulkLoadOptions options);
+
   void dropAndCreateNewBundledCSR(Checkpoint& ckp, ColumnBase* prev_data_col);
   void dropAndCreateNewUnbundledCSR(Checkpoint& ckp, bool delete_property);