Skip to content

refactor: introduce FileReader base class for CSV, JSON, and Parquet #648

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
luoxiaojian wants to merge 4 commits into
base: main
Choose a base branch
from
Open

refactor: introduce FileReader base class for CSV, JSON, and Parquet #648

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9e280df
refactor: introduce FileReader base class for CSV, JSON, and Parquet …
luoxiaojian Jun 25, 2026
f18287e
remove sniffer
luoxiaojian Jun 29, 2026
1f200ec
fmt
luoxiaojian Jun 29, 2026
65396bd
refactor: extract neug/columnar module and relocate IO pipeline boundary
luoxiaojian Jul 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 9 additions & 9 deletions bin/benchmark.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@

#include "neug/execution/common/context.h"
#include "neug/execution/common/operators/retrieve/sink.h"
#include "neug/execution/common/types/value.h"
#include "neug/columnar/value.h"
#include "neug/execution/execute/plan_parser.h"
#include "neug/main/neug_db.h"
#include "neug/main/query_request.h"
Expand Down Expand Up @@ -54,40 +54,40 @@ neug::execution::ParamsMap deserialize_string_kv_map(
switch (type.id()) {
case neug::DataTypeId::kInt32: {
map.emplace(iter.first,
neug::execution::Value::INT32(std::stoi(iter.second)));
neug::columnar::Value::INT32(std::stoi(iter.second)));
break;
}
case neug::DataTypeId::kInt64: {
map.emplace(iter.first,
neug::execution::Value::INT64(std::stoll(iter.second)));
neug::columnar::Value::INT64(std::stoll(iter.second)));
break;
}
case neug::DataTypeId::kUInt32: {
map.emplace(iter.first,
neug::execution::Value::UINT32(std::stoul(iter.second)));
neug::columnar::Value::UINT32(std::stoul(iter.second)));
break;
}
case neug::DataTypeId::kUInt64: {
map.emplace(iter.first,
neug::execution::Value::UINT64(std::stoull(iter.second)));
neug::columnar::Value::UINT64(std::stoull(iter.second)));
break;
}
case neug::DataTypeId::kBoolean: {
map.emplace(iter.first,
neug::execution::Value::BOOLEAN(iter.second == "true"));
neug::columnar::Value::BOOLEAN(iter.second == "true"));
break;
}
case neug::DataTypeId::kVarchar: {
map.emplace(iter.first, neug::execution::Value::STRING(iter.second));
map.emplace(iter.first, neug::columnar::Value::STRING(iter.second));
break;
}
case neug::DataTypeId::kTimestampMs: {
map.emplace(iter.first, neug::execution::Value::TIMESTAMPMS(
map.emplace(iter.first, neug::columnar::Value::TIMESTAMPMS(
neug::DateTime(std::stoll(iter.second))));
break;
}
case neug::DataTypeId::kDate: {
map.emplace(iter.first, neug::execution::Value::DATE(neug::Date(
map.emplace(iter.first, neug::columnar::Value::DATE(neug::Date(
int64_t(std::stoll(iter.second)))));
break;
default:
Expand Down
4 changes: 2 additions & 2 deletions doc/source/_scripts/generate_cpp_docs.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1507,8 +1507,8 @@ def _generate_category_index_md(self, categories: Dict[str, Any]):
```cpp
// Safe parameter passing prevents injection
neug::execution::ParamsMap params;
params["min_age"] = neug::execution::Value(25);
params["city"] = neug::execution::Value("Beijing");
params["min_age"] = neug::columnar::Value(25);
params["city"] = neug::columnar::Value("Beijing");

auto result = conn->Query(
"MATCH (p:Person) WHERE p.age > $min_age AND p.city = $city RETURN p",
Expand Down
2 changes: 1 addition & 1 deletion doc/source/reference/cpp_api/connection.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ Compiles and executes a Cypher query string against the database. The query is p
auto result = conn->Query("MATCH (n:Person) RETURN n.name", "read");
// Query with parameters
neug::execution::ParamsMap params;
params["min_age"] = neug::execution::Value(18);
params["min_age"] = neug::columnar::Value(18);
result = conn->Query("MATCH (p:Person) WHERE p.age > $min_age RETURN p",
"read", params);
// Process results
Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/bfs_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@

#include <memory>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/storages/graph/graph_interface.h"
namespace neug {
Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/bfs_pred_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#include <cstdint>
#include <memory>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/execution/expression/expr.h"
#include "neug/storages/graph/graph_interface.h"
Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/cdlp_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#include <cstdint>
#include <vector>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/execution/expression/expr.h"
#include "neug/execution/expression/predicates.h"
Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/cdlp_pred_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#include <cstdint>
#include <memory>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/execution/expression/expr.h"
#include "neug/storages/graph/graph_interface.h"
Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/kcore_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
#include <memory>
#include <vector>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/storages/graph/graph_interface.h"

Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/kcore_pred_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#include <cstdint>
#include <vector>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/execution/expression/expr.h"
#include "neug/storages/graph/graph_interface.h"
Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/lcc_directed_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#include <cstdint>
#include <vector>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/storages/graph/graph_interface.h"

Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/lcc_pred_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#include <cstdint>
#include <memory>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/execution/expression/expr.h"
#include "neug/storages/graph/graph_interface.h"
Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/lcc_undirected_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
#include <memory>
#include <vector>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/storages/graph/graph_interface.h"

Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/page_rank_directed_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#include <memory>
#include <vector>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/expression/expr.h"

namespace neug {
Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/page_rank_pred_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#include <cstdint>
#include <memory>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/execution/expression/expr.h"
#include "neug/storages/graph/graph_interface.h"
Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/page_rank_undirected_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#include <memory>
#include <vector>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/expression/expr.h"

namespace neug {
Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/sssp_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
#include <string>
#include <vector>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/storages/graph/graph_interface.h"

Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/sssp_pred_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
#include <memory>
#include <string>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/execution/expression/expr.h"
#include "neug/storages/graph/graph_interface.h"
Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/wcc_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#include <atomic>
#include <memory>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/storages/graph/graph_interface.h"

Expand Down
2 changes: 1 addition & 1 deletion extension/gds/include/impl/wcc_pred_impl.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
#include <cstdint>
#include <memory>

#include "neug/execution/common/columns/container_types.h"
#include "neug/columnar/container_types.h"
#include "neug/execution/common/context.h"
#include "neug/execution/expression/expr.h"
#include "neug/storages/graph/graph_interface.h"
Expand Down
31 changes: 16 additions & 15 deletions extension/gds/include/utils/path_utils.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,15 +21,15 @@
#include <utility>
#include <vector>

#include "neug/columnar/columns/path_columns.h"
#include "neug/columnar/graph_types.h"
#include "neug/common/extra_type_info.h"
#include "neug/common/types.h"
#include "neug/compiler/binder/expression/expression.h"
#include "neug/compiler/common/constants.h"
#include "neug/compiler/function/gds/gds_algo_function.h"
#include "neug/compiler/function/table/table_function.h"
#include "neug/execution/common/columns/path_columns.h"
#include "neug/execution/common/operators/retrieve/sink.h"
#include "neug/execution/common/types/graph_types.h"
#include "neug/storages/graph/graph_interface.h"

namespace neug {
Expand All @@ -38,26 +38,27 @@ namespace gds {
// Build a Path object from a predecessor chain, looking up real edge data
// pointers from the CSR graph view. The caller provides the vertex chain in
// source-to-target order.
inline execution::Path build_path_from_chain(
const std::vector<vid_t>& chain, label_t vertex_label, label_t edge_label,
bool directed, const StorageReadInterface& graph) {
inline columnar::Path build_path_from_chain(const std::vector<vid_t>& chain,
label_t vertex_label,
label_t edge_label, bool directed,
const StorageReadInterface& graph) {
if (chain.size() <= 1) {
return execution::Path(vertex_label, chain[0]);
return columnar::Path(vertex_label, chain[0]);
}

auto oe_view =
graph.GetGenericOutgoingGraphView(vertex_label, vertex_label, edge_label);
auto ie_view =
graph.GetGenericIncomingGraphView(vertex_label, vertex_label, edge_label);

std::vector<std::pair<execution::Direction, const void*>> edge_datas;
std::vector<std::pair<columnar::Direction, const void*>> edge_datas;
edge_datas.reserve(chain.size() - 1);

for (size_t i = 0; i + 1 < chain.size(); ++i) {
vid_t from = chain[i];
vid_t to = chain[i + 1];
const void* prop = nullptr;
execution::Direction dir = execution::Direction::kOut;
columnar::Direction dir = columnar::Direction::kOut;

// Try outgoing edges first
auto oe_edges = oe_view.get_edges(from);
Expand All @@ -74,7 +75,7 @@ inline execution::Path build_path_from_chain(
for (auto it = ie_edges.begin(); it != ie_edges.end(); ++it) {
if (*it == to) {
prop = it.get_data_ptr();
dir = execution::Direction::kIn;
dir = columnar::Direction::kIn;
break;
}
}
Expand All @@ -83,7 +84,7 @@ inline execution::Path build_path_from_chain(
edge_datas.push_back({dir, prop});
}

return execution::Path(vertex_label, edge_label, chain, edge_datas);
return columnar::Path(vertex_label, edge_label, chain, edge_datas);
}

// Reconstruct a path by walking backward from `target` to `source` using
Expand All @@ -92,11 +93,11 @@ inline execution::Path build_path_from_chain(
// vertex ID. This enables post-hoc path reconstruction from the distance
// array without storing predecessors during computation.
template <typename PredFinder>
inline execution::Path reconstruct_path(vid_t target, vid_t source,
const PredFinder& find_pred,
label_t vertex_label,
label_t edge_label, bool directed,
const StorageReadInterface& graph) {
inline columnar::Path reconstruct_path(vid_t target, vid_t source,
const PredFinder& find_pred,
label_t vertex_label, label_t edge_label,
bool directed,
const StorageReadInterface& graph) {
std::vector<vid_t> chain;
vid_t cur = target;
while (cur != source) {
Expand Down
12 changes: 6 additions & 6 deletions extension/gds/src/impl/bfs_impl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,8 @@
#include <thread>
#include <vector>

#include "neug/execution/common/columns/value_columns.h"
#include "neug/execution/common/columns/vertex_columns.h"
#include "neug/columnar/columns/value_columns.h"
#include "neug/columnar/columns/vertex_columns.h"
#include "neug/execution/common/context.h"
#include "utils/parallel_utils.h"
#include "utils/path_utils.h"
Expand Down Expand Up @@ -160,11 +160,11 @@ void BFS::compute() {
void BFS::sink(execution::Context& ctx, int node_alias, int distance_alias,
int path_alias) {
execution::MSVertexColumnBuilder node_builder(vertex_label_);
execution::ValueColumnBuilder<int64_t> distance_builder;
columnar::ValueColumnBuilder<int64_t> distance_builder;

distance_builder.reserve(vertices_.size());

std::shared_ptr<execution::IContextColumn> path_column;
std::shared_ptr<columnar::IColumn> path_column;
if (return_path_) {
auto oe_view = graph_.GetGenericOutgoingGraphView(
vertex_label_, vertex_label_, edge_label_);
Expand All @@ -189,7 +189,7 @@ void BFS::sink(execution::Context& ctx, int node_alias, int distance_alias,
return source_;
};

execution::PathColumnBuilder path_builder;
columnar::PathColumnBuilder path_builder;
for (vid_t v : vertices_) {
if (distances_[v] == std::numeric_limits<uint32_t>::max()) {
path_builder.push_back_null();
Expand All @@ -210,7 +210,7 @@ void BFS::sink(execution::Context& ctx, int node_alias, int distance_alias,
}
node_builder.append(vertex_label_, std::move(vertices_));

execution::DataChunk chunk;
columnar::DataChunk chunk;
chunk.set(node_alias, node_builder.finish());
chunk.set(distance_alias, distance_builder.finish());

Expand Down
12 changes: 6 additions & 6 deletions extension/gds/src/impl/bfs_pred_impl.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,8 +21,8 @@
#include <memory>
#include <vector>

#include "neug/execution/common/columns/value_columns.h"
#include "neug/execution/common/columns/vertex_columns.h"
#include "neug/columnar/columns/value_columns.h"
#include "neug/columnar/columns/vertex_columns.h"
#include "neug/execution/expression/predicates.h"
#include "utils/path_utils.h"

Expand Down Expand Up @@ -123,10 +123,10 @@ void BFSPred::compute() {
void BFSPred::sink(execution::Context& ctx, int node_alias, int distance_alias,
int path_alias) {
execution::MSVertexColumnBuilder node_builder(vertex_label_);
execution::ValueColumnBuilder<int64_t> distance_builder;
columnar::ValueColumnBuilder<int64_t> distance_builder;
distance_builder.reserve(vertices_.size());

std::shared_ptr<execution::IContextColumn> path_column;
std::shared_ptr<columnar::IColumn> path_column;
if (return_path_) {
auto oe_view = graph_.GetGenericOutgoingGraphView(
vertex_label_, vertex_label_, edge_label_);
Expand Down Expand Up @@ -164,7 +164,7 @@ void BFSPred::sink(execution::Context& ctx, int node_alias, int distance_alias,
return source_;
};

execution::PathColumnBuilder path_builder;
columnar::PathColumnBuilder path_builder;
for (vid_t v : vertices_) {
if (distances_[v] == std::numeric_limits<uint32_t>::max()) {
path_builder.push_back_null();
Expand All @@ -185,7 +185,7 @@ void BFSPred::sink(execution::Context& ctx, int node_alias, int distance_alias,
}
node_builder.append(vertex_label_, std::move(vertices_));

execution::DataChunk chunk;
columnar::DataChunk chunk;
chunk.set(node_alias, node_builder.finish());
chunk.set(distance_alias, distance_builder.finish());

Expand Down
Loading