refactor: Bulk loading performance regression (#707)

Fix #702 
Fix #703 
Fix #704 
FIx #705 
Fix #706

Author: zhanglei1949

include/neug/common/columns/value_columns.h

@@ -122,6 +122,7 @@ class ValueColumnBuilder : public IContextColumnBuilder {
   }
 
   inline void push_back_opt(const T& val) { data_.push_back(val); }
+  inline void push_back_opt(T&& val) { data_.push_back(std::move(val)); }
   inline void push_back_null() override {
     if (valid_.empty()) {
       valid_.reserve(data_.capacity());

include/neug/storages/graph/vertex_table.h

@@ -14,6 +14,7 @@
  */
 #pragma once
 
+#include "neug/common/columns/value_columns.h"
 #include "neug/common/types/value.h"
 #include "neug/storages/graph/schema.h"
 #include "neug/storages/graph/vertex_timestamp.h"
@@ -270,8 +271,18 @@ class VertexTable {
     size_t row_num = pk_col->size();
     std::vector<vid_t> vids;
     vids.resize(row_num);
+    bool is_string = std::is_same_v<PK_T, std::string_view> ||
+                     std::is_same_v<PK_T, std::string>;
+    // Fast path: direct data() access when the column is a ValueColumn<PK_T>,
+    // avoiding per-element virtual dispatch via get_elem().
+    auto value_col = std::dynamic_pointer_cast<ValueColumn<PK_T>>(pk_col);
     for (size_t j = 0; j < row_num; ++j) {
-      auto oid = pk_col->get_elem(j);
+      Value oid;
+      if (value_col) {
+        oid = Value::CreateValue<PK_T>(value_col->data()[j]);
+      } else {
+        oid = pk_col->get_elem(j);
+      }
       if (NEUG_UNLIKELY(indexer_->get_index(oid, vids[j]))) {
         if (NEUG_UNLIKELY(v_ts_->IsVertexValid(vids[j], MAX_TIMESTAMP))) {
           vids[j] = std::numeric_limits<vid_t>::max();
@@ -280,8 +291,6 @@ class VertexTable {
         }
         continue;
       }
-      bool is_string = std::is_same_v<PK_T, std::string_view> ||
-                       std::is_same_v<PK_T, std::string>;
       vids[j] = insert_vertex_pk(oid, 0, is_string);
     }
     return vids;
@@ -290,8 +299,8 @@ class VertexTable {
   template <typename PK_T>
   void insert_vertices_impl(std::shared_ptr<IDataChunkSupplier> supplier) {
     auto row_nums = supplier->RowNum();
-    if (row_nums <= 0) {
-      LOG(WARNING) << "Row number from supplier is negative, treat it as 0.";
+    if (row_nums < 0) {
+      VLOG(1) << "Row number from supplier is unknown, skip pre-reserve.";
       row_nums = 0;
     }
     size_t new_size = indexer_->size() + row_nums;

src/storages/graph/edge_table.cc

@@ -27,6 +27,7 @@
 #include <string_view>
 #include <utility>
 
+#include "neug/common/columns/value_columns.h"
 #include "neug/storages/checkpoint_manager.h"
 #include "neug/storages/csr/csr_view_utils.h"
 #include "neug/storages/csr/immutable_csr.h"
@@ -210,10 +211,20 @@ void insert_edges_bundled_typed_impl(
   edge_data.reserve(src_lid.size());
   size_t cur_index = 0;
   for (auto& col : data_cols) {
-    for (size_t i = 0; i < col->size(); ++i) {
-      if (valid_flags[cur_index++]) {
-        auto val = col->get_elem(i);
-        edge_data.push_back(val.template GetValue<EDATA_T>());
+    auto value_col = std::dynamic_pointer_cast<ValueColumn<EDATA_T>>(col);
+    if (value_col) {
+      for (size_t i = 0; i < value_col->size(); ++i) {
+        if (valid_flags[cur_index++]) {
+          edge_data.push_back(value_col->get_value(i));
+        }
+      }
+    } else {
+      // Fallback: virtual dispatch path (non-ValueColumn, rare).
+      for (size_t i = 0; i < col->size(); ++i) {
+        if (valid_flags[cur_index++]) {
+          auto val = col->get_elem(i);
+          edge_data.push_back(val.template GetValue<EDATA_T>());
+        }
       }
     }
   }
@@ -234,15 +245,58 @@ void insert_edges_separated_impl(TypedCsrBase<uint64_t>* out_csr,
   in_csr->batch_put_edges(dst_lid, src_lid, edge_data);
 }
 
-static std::vector<Value> get_row_from_data_chunks(
-    const std::vector<std::shared_ptr<IContextColumn>>& prop_cols,
-    size_t row_idx) {
-  std::vector<Value> row;
-  row.reserve(prop_cols.size());
-  for (auto& col : prop_cols) {
-    row.push_back(col->get_elem(row_idx));
+/// Type-erased inserter: writes ValueColumn<T>::data()[src_idx] to
+/// TypedColumn<T>::set_value(dst_idx), bypassing get_elem() + set_any().
+struct TypedColumnInserter {
+  const IContextColumn* src;
+  ColumnBase* dst;
+  void (*fn)(const TypedColumnInserter&, size_t dst_idx, size_t src_idx,
+             bool insert_safe);
+
+  inline void insert(size_t dst_idx, size_t src_idx, bool insert_safe) const {
+    fn(*this, dst_idx, src_idx, insert_safe);
+  }
+};
+
+/// Fixed-length types: direct set_value, no Value, no virtual dispatch.
+/// Null entries already have T() in data_, so set_value writes the same
+/// default that set_any would write for null.
+template <typename T>
+void insert_typed_impl(const TypedColumnInserter& ins, size_t dst_idx,
+                       size_t src_idx, bool /*insert_safe*/) {
+  auto* typed_dst = static_cast<TypedColumn<T>*>(ins.dst);
+  auto vc = static_cast<const ValueColumn<T>*>(ins.src);
+  typed_dst->set_value(dst_idx, vc->data()[src_idx]);
+}
+
+/// Varchar: source is ValueColumn<std::string>, dest is
+/// TypedColumn<std::string_view>. Needs set_any for buffer resize, but skips
+/// get_elem() virtual dispatch.
+void insert_varchar_impl(const TypedColumnInserter& ins, size_t dst_idx,
+                         size_t src_idx, bool insert_safe) {
+  auto* typed_dst = static_cast<TypedColumn<std::string_view>*>(ins.dst);
+  auto vc = static_cast<const ValueColumn<std::string>*>(ins.src);
+  typed_dst->set_any(dst_idx,
+                     Value::CreateValue<std::string>(vc->data()[src_idx]),
+                     insert_safe);
+}
+
+TypedColumnInserter make_inserter(const DataType& type,
+                                  const IContextColumn* src, ColumnBase* dst) {
+  switch (type.id()) {
+#define MAKE_INSERTER(enum_val, cpp_type) \
+  case DataTypeId::enum_val:              \
+    return {src, dst, &insert_typed_impl<cpp_type>};
+    FOR_EACH_DATA_TYPE_NO_STRING(MAKE_INSERTER)
+#undef MAKE_INSERTER
+  case DataTypeId::kVarchar:
+    return {src, dst, &insert_varchar_impl};
+  default:
+    THROW_NOT_SUPPORTED_EXCEPTION(
+        "Unsupported data type for column inserter: " +
+        std::to_string(static_cast<int>(type.id())));
+    return {};
   }
-  return row;
 }
 
 void batch_add_unbundled_edges_impl(
@@ -266,13 +320,30 @@ void batch_add_unbundled_edges_impl(
       if (c)
         prop_cols.push_back(c);
     }
+    // Pre-resolve typed inserters for each column (once per chunk).
+    std::vector<TypedColumnInserter> inserters;
+    inserters.reserve(prop_cols.size());
+    for (size_t j = 0; j < prop_cols.size(); ++j) {
+      inserters.push_back(make_inserter(prop_types[j], prop_cols[j].get(),
+                                        table_->get_column_by_id(j)));
+    }
+    // Pre-compute valid rows: column-major loop needs branch-free inner loop.
+    std::vector<size_t> valid_rows;
+    valid_rows.reserve(num_rows);
     for (size_t i = 0; i < num_rows; ++i) {
       assert(cur_index < valid_flags.size());
       if (valid_flags[cur_index++]) {
-        auto row = get_row_from_data_chunks(prop_cols, i);
-        table_->insert(offset++, row, true);
+        valid_rows.push_back(i);
+      }
+    }
+    // Column-major: same fn pointer per inner iteration (BTB-friendly),
+    // sequential dst writes (cache-friendly), no branch.
+    for (auto& ins : inserters) {
+      for (size_t k = 0; k < valid_rows.size(); ++k) {
+        ins.insert(offset + k, valid_rows[k], true);
       }
     }
+    offset += valid_rows.size();
   }
 }