From 753ea69c3d95708cb39371976dbab41cee7be282 Mon Sep 17 00:00:00 2001 From: "xiaolei.zl" Date: Wed, 1 Apr 2026 16:42:27 +0800 Subject: [PATCH] init commit for list impl MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit support to_list expression in compiler Committed-by: Xiaoli Zhou from Dev container resolve some comments use owned property to hold the string memory temporarily when passing it to storage fix default value of list type in ddl Committed-by: Xiaoli Zhou from Dev container fix fix fold constant bugs Committed-by: Xiaoli Zhou from Dev container Committed-by: Xiaoli Zhou from Dev container todo: resolve the test cases minor fix Revert "fix fold constant bugs" This reverts commit 287e2a9cf26ae981f560241612d02bc4e7f67581. Committed-by: Xiaoli Zhou from Dev container convert list const value to to_list expression Committed-by: Xiaoli Zhou from Dev container Committed-by: Xiaoli Zhou from Dev container Committed-by: Xiaoli Zhou from Dev container fix bug fix add more test minor fix for schema yaml parsing format python format remove dummy pb_utils changes minor fix fix python format replace property with value update doc and format update column feat: complete List type support with Value-native column interface - Replace ColumnBase/RefColumnBase/EdgeDataAccessor Property interface with execution::Value (get_value/set_value/set_any) - Implement ListColumn as child-column model with append-only updates and dump-time compaction (matching string column pattern) - Add List support to Arrow-based bulk loader and EdgeTable unbundled path - Fix EdgeSchema::is_bundled() to route List properties to unbundled CSR - Add DumpChildren/OpenChildren for recursive list persistence (方案B) - Delete dead code: value_to_property, property_to_value, ValueToListBlob, listViewToValue, Property list APIs, Schema::get_default_properties - Migrate all tests to Value API (test_table, test_acid, test_update_transaction, test_indexer, test_edge_table, etc.) Co-Authored-By: Claude Opus 4.6 Remove ListView move list column to nest_column.h refactoring add test resolve some issue fix format fix compile --- AGENTS.md | 4 + doc/source/cypher_manual/ddl_clause.md | 3 + include/neug/common/types.h | 3 + include/neug/compiler/gopt/g_ddl_converter.h | 6 +- include/neug/compiler/gopt/g_expr_converter.h | 15 +- .../neug/compiler/gopt/g_physical_convertor.h | 12 +- .../neug/compiler/gopt/g_query_converter.h | 5 +- include/neug/compiler/gopt/g_scalar_type.h | 2 - include/neug/compiler/gopt/g_type_utils.h | 13 + include/neug/storages/README.md | 44 ++ include/neug/storages/loader/loader_utils.h | 7 + include/neug/utils/id_indexer.h | 1 - include/neug/utils/property/column.h | 7 + include/neug/utils/property/nest_column.h | 343 ++++++++++ include/neug/utils/property/types.h | 43 +- proto/cypher_ddl.proto | 3 + src/common/types.cc | 18 + src/compiler/gopt/g_ddl_converter.cpp | 20 +- src/compiler/gopt/g_expr_converter.cpp | 126 +--- src/compiler/gopt/g_query_converter.cpp | 17 +- src/compiler/planner/gopt_planner.cc | 2 +- src/execution/common/types/value.cc | 30 +- .../execute/ops/batch/batch_update_utils.cc | 161 ++++- src/storages/graph/edge_table.cc | 43 +- src/storages/graph/property_graph.cc | 1 + src/storages/graph/schema.cc | 3 +- src/storages/graph/vertex_table.cc | 21 +- src/storages/loader/loader_utils.cc | 93 +++ src/transaction/insert_transaction.cc | 35 +- src/transaction/update_transaction.cc | 11 + src/utils/pb_utils.cc | 70 +- src/utils/property/column.cc | 20 +- src/utils/property/nest_column.cc | 56 ++ src/utils/writer/writer.cc | 8 +- src/utils/yaml_utils.cc | 5 + tests/compiler/gopt_test.h | 6 +- tests/execution/test_value.cc | 35 +- tests/storage/CMakeLists.txt | 2 + tests/storage/test_ddl.cc | 9 + tests/storage/test_list_column.cc | 623 ++++++++++++++++++ tests/utils/test_table.cc | 5 +- tools/python_bind/tests/test_db_list.py | 130 +++- tools/python_bind/tests/test_ddl.py | 132 ++++ tools/python_bind/tests/test_load.py | 214 ++++++ 44 files changed, 2210 insertions(+), 197 deletions(-) create mode 100644 include/neug/utils/property/nest_column.h create mode 100644 src/utils/property/nest_column.cc create mode 100644 tests/storage/test_list_column.cc diff --git a/AGENTS.md b/AGENTS.md index 65243bc7b..5d6250c07 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -96,6 +96,10 @@ tools/python_bind/ Cypher → ANTLR Parser → Binder → Logical Plan → gopt Converter → Physical Plan → Execution ``` +## Known Limitations + +- **CSV-based bulk load does not support List type**: The CSV-based `COPY FROM` path does not support `List` type properties because Arrow CSV reader cannot natively parse list values from CSV. The Arrow RecordBatch-based bulk loader (used by the programmatic API) does support List type properties. + ## Code Style - **C++**: C++20, clang-format (style=file) diff --git a/doc/source/cypher_manual/ddl_clause.md b/doc/source/cypher_manual/ddl_clause.md index de1aa9da8..e7fbf928e 100644 --- a/doc/source/cypher_manual/ddl_clause.md +++ b/doc/source/cypher_manual/ddl_clause.md @@ -16,6 +16,9 @@ The following table lists the recommended syntax for defining default values for | `DATE` | `prop DATE DEFAULT DATE('1970-01-01')` | `DATE('1970-01-01')` | | `TIMESTAMP` | `prop TIMESTAMP DEFAULT TIMESTAMP('1970-01-01')` | `TIMESTAMP('1970-01-01')` | | `INTERVAL` | `prop INTERVAL DEFAULT INTERVAL('0 year 0 month 0 day')` | `INTERVAL('0 year 0 month 0 day')` | +| `[]` | `prop INT64[] DEFAULT [1, 2, 3]` | `[]` (empty list) | + +List types are declared by appending `[]` to any supported element type (e.g., `STRING[]`, `DOUBLE[]`). Nesting is supported — `[][]` declares a list of lists, and deeper nesting follows the same pattern. Default values use list literal syntax: `DEFAULT ['a', 'b']` or `DEFAULT [[1, 2], [3]]` for nested lists. Please refer to the following examples for more usages. diff --git a/include/neug/common/types.h b/include/neug/common/types.h index 1fcd7e91b..461359f6b 100644 --- a/include/neug/common/types.h +++ b/include/neug/common/types.h @@ -150,6 +150,9 @@ struct DataType { std::string ToString() const; + static std::string ToYAMLString(const DataType& type); + static DataType FromYAMLString(const std::string& str); + private: DataTypeId id_; std::shared_ptr type_info_; diff --git a/include/neug/compiler/gopt/g_ddl_converter.h b/include/neug/compiler/gopt/g_ddl_converter.h index a8da60fcc..c077223c5 100644 --- a/include/neug/compiler/gopt/g_ddl_converter.h +++ b/include/neug/compiler/gopt/g_ddl_converter.h @@ -21,6 +21,7 @@ #include "neug/compiler/gopt/g_catalog.h" #include "neug/compiler/gopt/g_expr_converter.h" #include "neug/compiler/gopt/g_type_converter.h" +#include "neug/compiler/main/client_context.h" #include "neug/compiler/planner/operator/ddl/logical_alter.h" #include "neug/compiler/planner/operator/ddl/logical_create_table.h" #include "neug/compiler/planner/operator/ddl/logical_drop.h" @@ -45,8 +46,9 @@ struct EdgeLabel { class GDDLConverter { public: explicit GDDLConverter(std::shared_ptr aliasManager, - neug::catalog::Catalog* catalog) - : catalog{catalog}, exprConverter(aliasManager) {} + neug::catalog::Catalog* catalog, + main::ClientContext* clientContext) + : catalog{catalog}, exprConverter(aliasManager, clientContext) {} virtual ~GDDLConverter() = default; diff --git a/include/neug/compiler/gopt/g_expr_converter.h b/include/neug/compiler/gopt/g_expr_converter.h index 2c88ead80..94f6dc832 100644 --- a/include/neug/compiler/gopt/g_expr_converter.h +++ b/include/neug/compiler/gopt/g_expr_converter.h @@ -33,6 +33,7 @@ #include "neug/compiler/gopt/g_precedence.h" #include "neug/compiler/gopt/g_scalar_type.h" #include "neug/compiler/gopt/g_type_converter.h" +#include "neug/compiler/main/client_context.h" #include "neug/config.h" #include "neug/generated/proto/plan/algebra.pb.h" #include "neug/generated/proto/plan/common.pb.h" @@ -44,8 +45,9 @@ namespace gopt { class GExprConverter { public: - GExprConverter(const std::shared_ptr aliasManager) - : aliasManager{std::move(aliasManager)} {} + GExprConverter(const std::shared_ptr aliasManager, + main::ClientContext* clientContext) + : aliasManager{std::move(aliasManager)}, ctx{clientContext} {} // Main conversion function std::unique_ptr<::common::Expression> convert( @@ -61,7 +63,7 @@ class GExprConverter { const binder::AggregateFunctionExpression& expr, const planner::LogicalOperator& child); std::unique_ptr<::common::Variable> convertDefaultVar(); - std::unique_ptr<::common::Value> convertDefaultValue( + std::unique_ptr<::common::Expression> convertDefaultValue( const binder::PropertyDefinition& propertyDef); std::unique_ptr<::common::Property> convertPropertyExpr( const std::string& propName); @@ -122,7 +124,7 @@ class GExprConverter { const std::vector& schemaAlias); // helper functions - std::unique_ptr<::common::Value> convertValue( + std::unique_ptr<::common::Expression> convertValue( const neug::common::Value& value); std::unique_ptr<::common::Variable> convertVarProperty( const std::string& aliasName, const std::string& propertyName, @@ -145,8 +147,6 @@ class GExprConverter { std::unique_ptr<::common::Expression> convertUDFFunc( const std::string& funcName, const binder::Expression& expr, size_t paramNum, const std::vector& schemaAlias); - std::unique_ptr<::common::Value> convertToLiteralArray( - const common::Value& value, const common::LogicalType& childType); std::unique_ptr<::common::Expression> convertRegexFunc( const binder::Expression& expr, const GScalarType& scalarType, const std::vector& schemaAlias); @@ -156,13 +156,14 @@ class GExprConverter { const binder::Expression& expr, const GScalarType& scalarType, const std::vector& schemaAlias); - std::unique_ptr<::common::Value> castLiteral( + std::unique_ptr<::common::Expression> castLiteral( const binder::Expression& castExpr); private: const std::shared_ptr aliasManager; gopt::GPhysicalTypeConverter typeConverter; gopt::GPrecedence preced; + main::ClientContext* ctx; }; } // namespace gopt diff --git a/include/neug/compiler/gopt/g_physical_convertor.h b/include/neug/compiler/gopt/g_physical_convertor.h index c3c8d5066..2c91c31f8 100644 --- a/include/neug/compiler/gopt/g_physical_convertor.h +++ b/include/neug/compiler/gopt/g_physical_convertor.h @@ -19,6 +19,7 @@ #include "neug/compiler/gopt/g_ddl_converter.h" #include "neug/compiler/gopt/g_physical_analyzer.h" #include "neug/compiler/gopt/g_query_converter.h" +#include "neug/compiler/main/client_context.h" #include "neug/compiler/planner/operator/logical_plan.h" #include "neug/compiler/planner/operator/simple/logical_extension.h" #include "neug/generated/proto/plan/physical.pb.h" @@ -29,8 +30,11 @@ namespace gopt { class GPhysicalConvertor { public: GPhysicalConvertor(std::shared_ptr aliasManager, - neug::catalog::Catalog* catalog) - : aliasManager{aliasManager}, catalog{catalog} {} + neug::catalog::Catalog* catalog, + main::ClientContext* clientContext) + : aliasManager{aliasManager}, + catalog{catalog}, + clientContext{clientContext} {} std::unique_ptr<::physical::PhysicalPlan> createEmptyPlan() { auto physicalPlan = std::make_unique<::physical::PhysicalPlan>(); @@ -83,13 +87,15 @@ class GPhysicalConvertor { private: std::unique_ptr<::physical::PhysicalPlan> convertQuery( const planner::LogicalPlan& plan, bool skipSink) { - auto converter = std::make_unique(aliasManager, catalog); + auto converter = + std::make_unique(aliasManager, catalog, clientContext); return converter->convert(plan, skipSink); } private: std::shared_ptr aliasManager; neug::catalog::Catalog* catalog; + main::ClientContext* clientContext; }; } // namespace gopt diff --git a/include/neug/compiler/gopt/g_query_converter.h b/include/neug/compiler/gopt/g_query_converter.h index c8c70c193..cb7002ed6 100644 --- a/include/neug/compiler/gopt/g_query_converter.h +++ b/include/neug/compiler/gopt/g_query_converter.h @@ -31,6 +31,7 @@ #include "neug/compiler/gopt/g_ddl_converter.h" #include "neug/compiler/gopt/g_expr_converter.h" #include "neug/compiler/gopt/g_type_converter.h" +#include "neug/compiler/main/client_context.h" #include "neug/compiler/planner/operator/extend/logical_extend.h" #include "neug/compiler/planner/operator/extend/logical_recursive_extend.h" #include "neug/compiler/planner/operator/logical_aggregate.h" @@ -79,7 +80,8 @@ struct EdgeLabelId { class GQueryConvertor { public: GQueryConvertor(std::shared_ptr aliasManager, - neug::catalog::Catalog* catalog); + neug::catalog::Catalog* catalog, + main::ClientContext* clientContext); std::unique_ptr<::physical::PhysicalPlan> convert( const planner::LogicalPlan& plan, bool skipSink); @@ -256,6 +258,7 @@ class GQueryConvertor { std::unique_ptr exprConvertor; std::unique_ptr typeConverter; neug::catalog::Catalog* catalog; + main::ClientContext* clientContext; neug::gopt::GDDLConverter ddlConverter; }; diff --git a/include/neug/compiler/gopt/g_scalar_type.h b/include/neug/compiler/gopt/g_scalar_type.h index 1f00b7c4d..91aa4cb01 100644 --- a/include/neug/compiler/gopt/g_scalar_type.h +++ b/include/neug/compiler/gopt/g_scalar_type.h @@ -117,10 +117,8 @@ class GScalarType { } else if (func.name == function::ListCreationFunction::name) { const auto& type = expr.getDataType(); if (type.getLogicalTypeID() == common::LogicalTypeID::LIST) { - LOG(INFO) << "type is list"; return ScalarType::TO_LIST; } else if (type.getLogicalTypeID() == common::LogicalTypeID::STRUCT) { - LOG(INFO) << "type is struct"; return ScalarType::TO_TUPLE; } THROW_EXCEPTION_WITH_FILE_LINE("Invalid data type: " + type.toString() + diff --git a/include/neug/compiler/gopt/g_type_utils.h b/include/neug/compiler/gopt/g_type_utils.h index 2d6bb6b43..9de0bf507 100644 --- a/include/neug/compiler/gopt/g_type_utils.h +++ b/include/neug/compiler/gopt/g_type_utils.h @@ -98,6 +98,9 @@ class GTypeUtils { } } auto arrayType = node["array"]; + if (!arrayType) { + arrayType = node["list"]; + } if (arrayType && arrayType.IsMap()) { auto componentType = arrayType["component_type"]; CHECK(componentType.IsDefined()) @@ -147,6 +150,16 @@ class GTypeUtils { return YAML_NODE_TEMPORAL_DATETIME(); case neug::common::LogicalTypeID::INTERVAL: return YAML_NODE_TEMPORAL_INTERVAL(); + case neug::common::LogicalTypeID::LIST: { + auto extraInfo = type.getExtraTypeInfo(); + if (!extraInfo) { + THROW_RUNTIME_ERROR("List type should have extra info"); + } + auto listType = extraInfo->constPtrCast(); + YAML::Node n; + n["array"]["component_type"] = toYAML(listType->getChildType()); + return n; + } default: LOG(WARNING) << "Unsupported type in YAML: " << static_cast(type.getLogicalTypeID()); diff --git a/include/neug/storages/README.md b/include/neug/storages/README.md index 7de825533..4dc96ab06 100644 --- a/include/neug/storages/README.md +++ b/include/neug/storages/README.md @@ -84,6 +84,50 @@ Unlike the property table of vertices, the property table of edges is not column Fow now, only one property is supported for edges, but developers can define a struct with multiple fields to store multiple properties. +## 5. List Property Storage + +List properties are stored using a binary encoding in [VarLenColumn](../utils/property/column.h), +which extends the dual-buffer storage pattern used for strings. The encoding differs based on +whether element types are POD (fixed-size) or non-POD (variable-size like strings or nested lists). + +### 5.1 Binary Encoding + +**POD Lists** (integers, floats, dates, booleans): +``` +┌──────────────────────────────────────────┐ +│ T[0] T[1] … T[count-1] │ +└──────────────────────────────────────────┘ +Total bytes: count * sizeof(T) (empty list = 0 bytes) +``` +`count` is not stored; it is derived from the buffer size as +`data_.size() / pod_type_size(child_id)`. + +**Non-POD Lists** (strings, nested lists): +``` +┌──────────┬──────────────────────────────┬────────────────────┐ +│ count │ off[0] off[1] … off[count-1] │ data[0] data[1] … │ +│ uint32_t │ uint32_t × count │ │ +└──────────┴──────────────────────────────┴────────────────────┘ +Empty list: 4 bytes (count=0). +``` +`off[i]` is relative to the start of the data region (first byte after the +last offset entry). The data region size is derived from the buffer size: +`data_.size() - sizeof(uint32_t) - count * sizeof(uint32_t)`. The length of +element `i` is `off[i+1] - off[i]` for `i < count-1`, and +`data_region_size - off[count-1]` for the last element. + +### 5.2 Thread Safety + +VarLenColumn uses `std::atomic pos_` for concurrent offset allocation via `fetch_add`. +- Concurrent writes to different indices are thread-safe +- The `insert_safe` parameter controls resize behavior: when `false`, throws on insufficient space; + when `true`, caller must provide external synchronization during resize + +### 5.3 Nested Lists + +Nested lists use non-POD encoding: each inner list is serialized as a blob, then stored +as a variable-length element in the outer list's data region. + ## 6. Durability diff --git a/include/neug/storages/loader/loader_utils.h b/include/neug/storages/loader/loader_utils.h index ac7b8a680..c6d6dd21d 100644 --- a/include/neug/storages/loader/loader_utils.h +++ b/include/neug/storages/loader/loader_utils.h @@ -23,11 +23,14 @@ #include #include +#include #include #include #include #include +#include "neug/common/types.h" +#include "neug/execution/common/types/value.h" #include "neug/storages/loader/loading_config.h" #include "neug/utils/exception/exception.h" #include "neug/utils/string_utils.h" @@ -244,4 +247,8 @@ void set_properties_column(std::shared_ptr col, const std::vector& vids, std::shared_mutex& mutex); +execution::Value arrow_element_to_value( + const std::shared_ptr& arr, int64_t idx, + const DataType& neug_type); + } // namespace neug diff --git a/include/neug/utils/id_indexer.h b/include/neug/utils/id_indexer.h index 39fe05964..e72915394 100644 --- a/include/neug/utils/id_indexer.h +++ b/include/neug/utils/id_indexer.h @@ -380,7 +380,6 @@ class LFIndexer { while (true) { INDEX_T ind = indices_ptr[index]; if (ind == LFIndexer::sentinel) { - VLOG(10) << "cannot find " << oid.to_string() << " in lf_indexer"; return ind; } else if (keys_->get_any(ind) == oid) { return ind; diff --git a/include/neug/utils/property/column.h b/include/neug/utils/property/column.h index 127959a14..eab394d93 100644 --- a/include/neug/utils/property/column.h +++ b/include/neug/utils/property/column.h @@ -49,6 +49,8 @@ namespace neug { class Table; +class ModuleBroker; +class CheckpointManifest; std::string_view truncate_utf8(std::string_view str, size_t length); @@ -73,6 +75,11 @@ class ColumnBase : public Module { virtual execution::Value get_any(size_t index) const = 0; virtual void ingest(uint32_t index, OutArchive& arc) = 0; + + virtual void DumpTo(Checkpoint& ckp, CheckpointManifest& meta, + const std::string& key); + + virtual void RestoreChildren(ModuleBroker& store, const std::string& key) {} }; template diff --git a/include/neug/utils/property/nest_column.h b/include/neug/utils/property/nest_column.h new file mode 100644 index 000000000..701f67947 --- /dev/null +++ b/include/neug/utils/property/nest_column.h @@ -0,0 +1,343 @@ +/** Copyright 2020 Alibaba Group Holding Limited. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#pragma once + +#include +#include +#include +#include +#include + +#include "neug/utils/property/column.h" + +namespace neug { + +struct list_entry { + uint64_t offset; + uint64_t length; +}; + +class ListColumn : public ColumnBase { + public: + static constexpr uint16_t DEFAULT_AVG_LEN = 64; + + ListColumn() : size_(0), child_frontier_(0), list_type_(DataType{}) {} + explicit ListColumn(const DataType& list_type) + : size_(0), child_frontier_(0), list_type_(list_type) { + if (list_type_.id() == DataTypeId::kList) { + auto child_type = ListType::GetChildType(list_type_); + child_column_ = CreateColumn(child_type); + } + } + ListColumn(ListColumn&& rhs) + : items_buffer_(std::move(rhs.items_buffer_)), + child_column_(std::move(rhs.child_column_)), + size_(rhs.size_), + child_frontier_(rhs.child_frontier_.load()), + list_type_(rhs.list_type_) { + rhs.size_ = 0; + rhs.child_frontier_.store(0); + } + + ~ListColumn() = default; + + void Open(Checkpoint& ckp, const ModuleDescriptor& desc, + MemoryLevel level) override { + auto offsets_path = desc.get_path("offsets"); + if (offsets_path.has_value() && !offsets_path.value().empty()) { + auto offsets_buf = ckp.OpenFile(offsets_path.value(), level); + size_t num_offsets = offsets_buf->GetDataSize() / sizeof(uint64_t); + if (num_offsets == 0) { + size_ = 0; + child_frontier_.store(0); + } else { + size_ = num_offsets - 1; + auto* offsets = + reinterpret_cast(offsets_buf->GetData()); + child_frontier_.store(offsets[size_]); + + items_buffer_ = ckp.OpenFile("", level); + items_buffer_->Resize(size_ * sizeof(list_entry)); + auto* entries = reinterpret_cast(items_buffer_->GetData()); + for (size_t i = 0; i < size_; ++i) { + entries[i] = {offsets[i], offsets[i + 1] - offsets[i]}; + } + } + } else { + items_buffer_ = ckp.OpenFile("", level); + size_ = 0; + child_frontier_.store(0); + } + + if (child_column_ && + !desc.get_path(ModuleDescriptor::kDataPath).has_value()) { + child_column_->Open(ckp, ModuleDescriptor{}, level); + } + } + + void Close() { + items_buffer_.reset(); + child_column_.reset(); + } + + ModuleDescriptor Dump(Checkpoint& ckp) override { + ModuleDescriptor desc; + desc.module_type = ModuleTypeName(); + if (!items_buffer_) { + THROW_RUNTIME_ERROR("List column items buffer not initialized for dump"); + } + compact(ckp); + + // Write cumulative offsets to disk + auto offsets_buf = ckp.OpenFile("", MemoryLevel::kInMemory); + offsets_buf->Resize((size_ + 1) * sizeof(uint64_t)); + auto* offsets = reinterpret_cast(offsets_buf->GetData()); + offsets[0] = 0; + for (size_t i = 0; i < size_; ++i) { + auto entry = get_entry(i); + offsets[i + 1] = offsets[i] + entry.length; + } + + desc.set_path("offsets", ckp.Commit(*offsets_buf)); + return desc; + } + + void DumpTo(Checkpoint& ckp, CheckpointManifest& meta, + const std::string& key) override; + + void RestoreChildren(ModuleBroker& store, const std::string& key) override; + + size_t size() const override { return size_; } + + void resize(size_t size) override { + items_buffer_->Resize(size * sizeof(list_entry)); + if (child_column_) { + uint64_t needed = child_frontier_.load() + + (size > size_ ? (size - size_) : 0) * DEFAULT_AVG_LEN; + if (needed > child_column_->size()) { + child_column_->resize(needed); + } + } + size_ = size; + } + + void resize(size_t size, const execution::Value& default_value) override { + size_t old_size = size_; + items_buffer_->Resize(size * sizeof(list_entry)); + size_ = size; + + if (old_size >= size) { + return; + } + + if (default_value.IsNull() || + default_value.type().id() != DataTypeId::kList) { + for (size_t i = old_size; i < size; ++i) { + set_entry(i, {0, 0}); + } + return; + } + + const auto& children = execution::ListValue::GetChildren(default_value); + uint64_t L = children.size(); + + if (L == 0) { + for (size_t i = old_size; i < size; ++i) { + set_entry(i, {0, 0}); + } + return; + } + + uint64_t shared_start = child_frontier_.fetch_add(L); + ensure_child_capacity(shared_start + L); + for (uint64_t j = 0; j < L; ++j) { + child_column_->set_any(shared_start + j, children[j], true); + } + for (size_t i = old_size; i < size; ++i) { + set_entry(i, {shared_start, L}); + } + } + + DataTypeId type() const override { return DataTypeId::kList; } + DataType list_data_type() const { return list_type_; } + + void set_any(size_t idx, const execution::Value& value, + bool insert_safe) override { + if (idx >= size_) { + THROW_RUNTIME_ERROR("Index out of range"); + } + if (value.IsNull()) { + set_entry(idx, {0, 0}); + return; + } + const auto& children = execution::ListValue::GetChildren(value); + uint64_t L = children.size(); + if (L == 0) { + set_entry(idx, {0, 0}); + return; + } + uint64_t start = child_frontier_.fetch_add(L); + if (start + L > child_column_->size()) { + if (insert_safe) { + size_t new_cap = std::max(static_cast(start + L), + child_column_->size() * 3 / 2); + child_column_->resize(new_cap); + } else { + THROW_STORAGE_EXCEPTION( + "Not enough child capacity and insert_safe is false"); + } + } + for (uint64_t j = 0; j < L; ++j) { + child_column_->set_any(start + j, children[j], true); + } + set_entry(idx, {start, L}); + } + + execution::Value get_any(size_t idx) const override { + assert(idx < size_); + auto entry = get_entry(idx); + if (entry.length == 0) { + return execution::Value::LIST(ListType::GetChildType(list_type_), {}); + } + std::vector children; + children.reserve(entry.length); + for (uint64_t j = 0; j < entry.length; ++j) { + children.push_back(child_column_->get_any(entry.offset + j)); + } + return execution::Value::LIST(ListType::GetChildType(list_type_), + std::move(children)); + } + + void ingest(uint32_t index, OutArchive& arc) override { + execution::Value v; + arc >> v; + set_any(index, v, true); + } + + const DataType& list_type() const { return list_type_; } + void SetListType(const DataType& t) { + list_type_ = t; + if (!child_column_ && list_type_.id() == DataTypeId::kList) { + child_column_ = CreateColumn(ListType::GetChildType(list_type_)); + } + } + void SetChildColumn(std::unique_ptr col) { + child_column_ = std::move(col); + } + + ColumnBase* child_column() const { return child_column_.get(); } + uint64_t child_frontier() const { return child_frontier_.load(); } + + std::string ModuleTypeName() const override { return type_name(); } + + static std::string type_name() { return "column"; } + + private: + void compact(Checkpoint& ckp) { + if (!child_column_ || size_ == 0) { + return; + } + uint64_t frontier = child_frontier_.load(); + uint64_t total_len = 0; + for (size_t i = 0; i < size_; ++i) { + total_len += get_entry(i).length; + } + if (total_len == frontier && is_sequential()) { + return; + } + + auto child_type = ListType::GetChildType(list_type_); + auto fresh = CreateColumn(child_type); + fresh->Open(ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + fresh->resize(total_len); + + uint64_t new_offset = 0; + for (size_t i = 0; i < size_; ++i) { + auto entry = get_entry(i); + if (entry.length == 0) { + set_entry(i, {new_offset, 0}); + continue; + } + for (uint64_t j = 0; j < entry.length; ++j) { + fresh->set_any(new_offset + j, child_column_->get_any(entry.offset + j), + true); + } + set_entry(i, {new_offset, entry.length}); + new_offset += entry.length; + } + + child_column_ = std::move(fresh); + child_frontier_.store(new_offset); + } + + bool is_sequential() const { + uint64_t expected_offset = 0; + for (size_t i = 0; i < size_; ++i) { + auto entry = get_entry(i); + if (entry.offset != expected_offset) { + return false; + } + expected_offset += entry.length; + } + return true; + } + + void ensure_child_capacity(uint64_t needed) { + if (!child_column_) { + THROW_RUNTIME_ERROR("ListColumn child_column_ is null"); + } + if (needed > child_column_->size()) { + child_column_->resize( + std::max(static_cast(needed), child_column_->size() * 3 / 2)); + } + } + + inline list_entry get_entry(size_t idx) const { + assert(idx < size_); + return reinterpret_cast(items_buffer_->GetData())[idx]; + } + + inline void set_entry(size_t idx, const list_entry& item) { + assert(idx < size_); + reinterpret_cast(items_buffer_->GetData())[idx] = item; + } + + std::unique_ptr items_buffer_; + std::unique_ptr child_column_; + size_t size_; + std::atomic child_frontier_; + DataType list_type_; +}; + +class ListRefColumn : public RefColumnBase { + public: + explicit ListRefColumn(const ListColumn& column) : column_(column) {} + ~ListRefColumn() override = default; + + execution::Value get_any(size_t index) const override { + return column_.get_any(index); + } + + DataTypeId type() const override { return DataTypeId::kList; } + + DataType list_data_type() const { return column_.list_type(); } + + ColType col_type() const override { return ColType::kInternal; } + + private: + const ListColumn& column_; +}; + +} // namespace neug diff --git a/include/neug/utils/property/types.h b/include/neug/utils/property/types.h index ab17fffcd..c5c00dcef 100644 --- a/include/neug/utils/property/types.h +++ b/include/neug/utils/property/types.h @@ -587,6 +587,20 @@ struct convert { } } else if (config["date"]) { property_type = neug::DataTypeId::kDate; + } else if (config["list"]) { + auto array_node = config["list"]; + if (array_node["component_type"]) { + neug::DataType child_type; + if (!convert::decode(array_node["component_type"], + child_type)) { + LOG(ERROR) << "Failed to decode array component_type"; + return false; + } + property_type = neug::DataType::List(child_type); + } else { + LOG(ERROR) << "list type requires component_type"; + return false; + } } else { LOG(ERROR) << "Unrecognized property type: " << config; return false; @@ -596,22 +610,35 @@ struct convert { static Node encode(const neug::DataType& type) { YAML::Node node; - if (type == neug::DataTypeId::kBoolean || - type == neug::DataTypeId::kInt32 || type == neug::DataTypeId::kUInt32 || - type == neug::DataTypeId::kFloat || type == neug::DataTypeId::kInt64 || - type == neug::DataTypeId::kUInt64 || - type == neug::DataTypeId::kDouble) { + auto id = type.id(); + if (id == neug::DataTypeId::kBoolean || id == neug::DataTypeId::kInt32 || + id == neug::DataTypeId::kUInt32 || id == neug::DataTypeId::kFloat || + id == neug::DataTypeId::kInt64 || id == neug::DataTypeId::kUInt64 || + id == neug::DataTypeId::kDouble) { node["primitive_type"] = - neug::config_parsing::PrimitivePropertyTypeToString(type.id()); - } else if (type == neug::DataTypeId::kVarchar) { + neug::config_parsing::PrimitivePropertyTypeToString(id); + } else if (id == neug::DataTypeId::kVarchar) { const auto* extra_type_info = type.RawExtraTypeInfo(); const auto* string_type_info = dynamic_cast(extra_type_info); node["string"]["varchar"]["max_length"] = string_type_info ? string_type_info->max_length : neug::STRING_DEFAULT_MAX_LENGTH; - } else if (type == neug::DataTypeId::kDate) { + } else if (id == neug::DataTypeId::kDate) { node["temporal"]["datetime"] = ""; + } else if (id == neug::DataTypeId::kTimestampMs) { + node["temporal"]["timestamp"] = ""; + } else if (id == neug::DataTypeId::kInterval) { + node["temporal"]["interval"] = ""; + } else if (id == neug::DataTypeId::kList) { + const auto* extra_type_info = type.RawExtraTypeInfo(); + const auto* list_type_info = + dynamic_cast(extra_type_info); + if (list_type_info) { + node["list"]["component_type"] = encode(list_type_info->child_type); + } else { + LOG(ERROR) << "List type missing ListTypeInfo"; + } } else { LOG(ERROR) << "Unrecognized property type: " << type.ToString(); } diff --git a/proto/cypher_ddl.proto b/proto/cypher_ddl.proto index b72a24a07..7f575d3ac 100644 --- a/proto/cypher_ddl.proto +++ b/proto/cypher_ddl.proto @@ -19,6 +19,7 @@ package physical; import "common.proto"; import "basic_type.proto"; +import "expr.proto"; // The create/drop/alter operations for edges are all based on the triplet type . message EdgeType { @@ -30,7 +31,9 @@ message EdgeType { message PropertyDef { string name = 1; common.DataType type = 2; + // to be deprecated as we need to support nested list common.Value default_value = 3; + common.Expression default_expr = 4; } // When the operation does not meet the data requirements, an exception is thrown by default; diff --git a/src/common/types.cc b/src/common/types.cc index d4f172475..25b2e0ec2 100644 --- a/src/common/types.cc +++ b/src/common/types.cc @@ -23,10 +23,12 @@ #include "neug/common/types.h" +#include #include "neug/common/extra_type_info.h" #include "neug/generated/proto/plan/common.pb.h" #include "neug/generated/proto/plan/type.pb.h" #include "neug/utils/exception/exception.h" +#include "neug/utils/property/types.h" namespace neug { @@ -272,4 +274,20 @@ std::string DataType::ToString() const { } } +std::string DataType::ToYAMLString(const DataType& type) { + YAML::Node node = YAML::convert::encode(type); + YAML::Emitter emitter; + emitter << YAML::Flow << node; + return std::string(emitter.c_str()); +} + +DataType DataType::FromYAMLString(const std::string& str) { + YAML::Node node = YAML::Load(str); + DataType type; + if (!YAML::convert::decode(node, type)) { + THROW_RUNTIME_ERROR("Failed to parse DataType from YAML: " + str); + } + return type; +} + } // namespace neug \ No newline at end of file diff --git a/src/compiler/gopt/g_ddl_converter.cpp b/src/compiler/gopt/g_ddl_converter.cpp index eafe8987c..747f2a56c 100644 --- a/src/compiler/gopt/g_ddl_converter.cpp +++ b/src/compiler/gopt/g_ddl_converter.cpp @@ -171,9 +171,9 @@ GDDLConverter::convertToCreateVertexSchema( } auto* propertyDef = create_vertex->add_properties(); propertyDef->set_name(prop.getName()); - auto irType = typeConverter.convertSimpleLogicalType(prop.getType()); + auto irType = typeConverter.convertLogicalType(prop.getType()); *propertyDef->mutable_type() = std::move(*irType->mutable_data_type()); - propertyDef->set_allocated_default_value( + propertyDef->set_allocated_default_expr( exprConverter.convertDefaultValue(prop).release()); } @@ -265,9 +265,9 @@ GDDLConverter::convertToCreateEdgeGroupSchema( } auto* propertyDef = create_edge->add_properties(); propertyDef->set_name(prop.getName()); - auto irType = typeConverter.convertSimpleLogicalType(prop.getType()); + auto irType = typeConverter.convertLogicalType(prop.getType()); *propertyDef->mutable_type() = std::move(*irType->mutable_data_type()); - propertyDef->set_allocated_default_value( + propertyDef->set_allocated_default_expr( exprConverter.convertDefaultValue(prop).release()); } @@ -314,9 +314,9 @@ GDDLConverter::convertToCreateEdgeSchema( } auto* propertyDef = create_edge->add_properties(); propertyDef->set_name(prop.getName()); - auto irType = typeConverter.convertSimpleLogicalType(prop.getType()); + auto irType = typeConverter.convertLogicalType(prop.getType()); *propertyDef->mutable_type() = std::move(*irType->mutable_data_type()); - propertyDef->set_allocated_default_value( + propertyDef->set_allocated_default_expr( exprConverter.convertDefaultValue(prop).release()); } @@ -412,9 +412,9 @@ GDDLConverter::convertToAddVertexPropertySchema( // Add property definition auto* property = add_property->add_properties(); property->set_name(propertyDef.getName()); - auto irType = typeConverter.convertSimpleLogicalType(propertyDef.getType()); + auto irType = typeConverter.convertLogicalType(propertyDef.getType()); *property->mutable_type() = std::move(*irType->mutable_data_type()); - property->set_allocated_default_value( + property->set_allocated_default_expr( exprConverter.convertDefaultValue(propertyDef).release()); // Set conflict action @@ -455,9 +455,9 @@ GDDLConverter::convertToAddEdgePropertySchema(const planner::LogicalAlter& op) { // Add property definition auto* property = add_property->add_properties(); property->set_name(propertyDef.getName()); - auto irType = typeConverter.convertSimpleLogicalType(propertyDef.getType()); + auto irType = typeConverter.convertLogicalType(propertyDef.getType()); *property->mutable_type() = std::move(*irType->mutable_data_type()); - property->set_allocated_default_value( + property->set_allocated_default_expr( exprConverter.convertDefaultValue(propertyDef).release()); // Set conflict action diff --git a/src/compiler/gopt/g_expr_converter.cpp b/src/compiler/gopt/g_expr_converter.cpp index 30df2d76a..6c31039c7 100644 --- a/src/compiler/gopt/g_expr_converter.cpp +++ b/src/compiler/gopt/g_expr_converter.cpp @@ -28,6 +28,7 @@ #include "neug/compiler/binder/expression/rel_expression.h" #include "neug/compiler/binder/expression/scalar_function_expression.h" #include "neug/compiler/binder/expression/variable_expression.h" +#include "neug/compiler/binder/expression_evaluator_utils.h" #include "neug/compiler/common/enums/expression_type.h" #include "neug/compiler/common/string_utils.h" #include "neug/compiler/common/types/date_t.h" @@ -226,7 +227,7 @@ std::unique_ptr<::algebra::IndexPredicate> GExprConverter::convertPrimaryKey( return indexPB; } -std::unique_ptr<::common::Value> GExprConverter::castLiteral( +std::unique_ptr<::common::Expression> GExprConverter::castLiteral( const binder::Expression& castExpr) { GScalarType type(castExpr); if (type.getType() != ScalarType::CAST) { @@ -267,7 +268,7 @@ std::unique_ptr<::common::Value> GExprConverter::castLiteral( } // set default value for property definition -std::unique_ptr<::common::Value> GExprConverter::convertDefaultValue( +std::unique_ptr<::common::Expression> GExprConverter::convertDefaultValue( const binder::PropertyDefinition& propertyDef) { std::shared_ptr defaultExpr = propertyDef.boundExpr; // the query default value of temporal type (date, datetime, interval) is @@ -290,27 +291,32 @@ std::unique_ptr<::common::Value> GExprConverter::convertDefaultValue( defaultExpr = funcExpr->getChild(0); } } - auto valuePB = convert(*defaultExpr, {}); - if (valuePB->operators_size() == 0) { - THROW_EXCEPTION_WITH_FILE_LINE( - "Default value expression should not be empty"); - } - auto oprPB = valuePB->operators(0); - if (!oprPB.has_const_()) { - THROW_EXCEPTION_WITH_FILE_LINE( - "Default value expression should be a constant"); - } - return std::unique_ptr<::common::Value>(oprPB.release_const_()); + return convert(*defaultExpr, {}); } -std::unique_ptr<::common::Value> GExprConverter::convertValue( +std::unique_ptr<::common::Expression> GExprConverter::convertValue( const neug::common::Value& value) { - std::unique_ptr<::common::Value> valuePB = - std::make_unique<::common::Value>(); if (value.isNull()) { + auto valuePB = std::make_unique<::common::Value>(); valuePB->set_allocated_none(new ::common::None()); - return valuePB; + auto exprPB = std::make_unique<::common::Expression>(); + exprPB->add_operators()->set_allocated_const_(valuePB.release()); + return exprPB; + } + if (value.getDataType().getLogicalTypeID() == common::LogicalTypeID::ARRAY || + value.getDataType().getLogicalTypeID() == common::LogicalTypeID::LIST) { + auto toListPB = std::make_unique<::common::ToList>(); + for (const auto& child : value.children) { + toListPB->mutable_fields()->AddAllocated(convertValue(*child).release()); + } + auto exprPB = std::make_unique<::common::Expression>(); + auto oprPB = exprPB->add_operators(); + oprPB->set_allocated_to_list(toListPB.release()); + oprPB->set_allocated_node_type( + typeConverter.convertLogicalType(value.getDataType()).release()); + return exprPB; } + auto valuePB = std::make_unique<::common::Value>(); switch (value.getDataType().getLogicalTypeID()) { case common::LogicalTypeID::BOOL: valuePB->set_boolean(value.getValue()); @@ -348,29 +354,13 @@ std::unique_ptr<::common::Value> GExprConverter::convertValue( valuePB->set_str(neug::common::Interval::toString( value.getValue())); break; - case common::LogicalTypeID::ARRAY: { - auto extraInfo = value.getDataType().getExtraTypeInfo(); - if (extraInfo == nullptr) { - THROW_EXCEPTION_WITH_FILE_LINE("List type should have extra info"); - } - auto arrayInfo = extraInfo->constPtrCast(); - auto& childType = arrayInfo->getChildType(); - return convertToLiteralArray(value, childType); - } - case common::LogicalTypeID::LIST: { - auto extraInfo = value.getDataType().getExtraTypeInfo(); - if (extraInfo == nullptr) { - THROW_EXCEPTION_WITH_FILE_LINE("List type should have extra info"); - } - auto listInfo = extraInfo->constPtrCast(); - auto& childType = listInfo->getChildType(); - return convertToLiteralArray(value, childType); - } default: THROW_EXCEPTION_WITH_FILE_LINE("Unsupported value type " + value.getDataType().toString()); } - return valuePB; + auto exprPB = std::make_unique<::common::Expression>(); + exprPB->add_operators()->set_allocated_const_(valuePB.release()); + return exprPB; } std::string GExprConverter::convertRegexValue(const std::string& regex, @@ -422,56 +412,6 @@ std::unique_ptr<::common::Expression> GExprConverter::convertRegexFunc( return convertChildren(expr, schemaAlias); } -std::unique_ptr<::common::Value> GExprConverter::convertToLiteralArray( - const common::Value& value, const common::LogicalType& childType) { - if (value.children.empty()) { - THROW_EXCEPTION_WITH_FILE_LINE( - "Array function should have at least one child"); - } - auto valuePB = std::make_unique<::common::Value>(); - switch (childType.getLogicalTypeID()) { - case common::LogicalTypeID::INT32: { - auto i32Array = valuePB->mutable_i32_array(); - for (auto& child : value.children) { - i32Array->add_item(child->getValue()); - } - break; - } - case common::LogicalTypeID::INT64: { - auto i64Array = valuePB->mutable_i64_array(); - for (auto& child : value.children) { - i64Array->add_item(child->getValue()); - } - break; - } - case common::LogicalTypeID::FLOAT: { - auto f32Array = valuePB->mutable_f64_array(); - for (auto& child : value.children) { - f32Array->add_item(child->getValue()); - } - break; - } - case common::LogicalTypeID::DOUBLE: { - auto f64Array = valuePB->mutable_f64_array(); - for (auto& child : value.children) { - f64Array->add_item(child->getValue()); - } - break; - } - case common::LogicalTypeID::STRING: { - auto strArray = valuePB->mutable_str_array(); - for (auto& child : value.children) { - strArray->add_item(child->getValue()); - } - break; - } - default: - THROW_EXCEPTION_WITH_FILE_LINE("Unsupported value type " + - childType.toString()); - } - return valuePB; -} - std::unique_ptr<::common::NameOrId> GExprConverter::convertAlias( common::alias_id_t aliasId) { auto alias = std::make_unique<::common::NameOrId>(); @@ -495,10 +435,7 @@ std::unique_ptr<::common::Expression> GExprConverter::convertParam( std::unique_ptr<::common::Expression> GExprConverter::convertLiteral( const binder::LiteralExpression& expr) { - auto result = std::make_unique<::common::Expression>(); - auto literal = result->add_operators(); - literal->set_allocated_const_(convertValue(expr.getValue()).release()); - return result; + return convertValue(expr.getValue()); } std::unique_ptr<::common::Variable> GExprConverter::convertDefaultVar() { @@ -946,7 +883,7 @@ std::unique_ptr<::common::ExprOpr> GExprConverter::convertOperator( return result; } -::std::unique_ptr<::common::Expression> GExprConverter::convertCast( +std::unique_ptr<::common::Expression> GExprConverter::convertCast( const binder::Expression& expr, const std::vector& schemaAlias) { if (expr.expressionType != common::ExpressionType::FUNCTION) { @@ -962,12 +899,7 @@ ::std::unique_ptr<::common::Expression> GExprConverter::convertCast( auto sourceExpr = children[0]; switch (sourceExpr->expressionType) { case common::ExpressionType::LITERAL: { - auto valuePB = castLiteral(expr); - if (valuePB) { - auto exprPB = std::make_unique<::common::Expression>(); - exprPB->add_operators()->set_allocated_const_(valuePB.release()); - return exprPB; - } + return castLiteral(expr); } case common::ExpressionType::PARAMETER: { // cast dynamic param by // setting its meta data with diff --git a/src/compiler/gopt/g_query_converter.cpp b/src/compiler/gopt/g_query_converter.cpp index 14e356254..8d54ceea1 100644 --- a/src/compiler/gopt/g_query_converter.cpp +++ b/src/compiler/gopt/g_query_converter.cpp @@ -76,12 +76,15 @@ namespace neug { namespace gopt { GQueryConvertor::GQueryConvertor(std::shared_ptr aliasManager, - neug::catalog::Catalog* catalog) - : ddlConverter(aliasManager, catalog), - aliasManager(aliasManager), + neug::catalog::Catalog* catalog, + main::ClientContext* clientContext) + : aliasManager(std::move(aliasManager)), + exprConvertor( + std::make_unique(this->aliasManager, clientContext)), + typeConverter(std::make_unique()), catalog(catalog), - exprConvertor(std::make_unique(aliasManager)), - typeConverter(std::make_unique()) {} + clientContext(clientContext), + ddlConverter(this->aliasManager, catalog, this->clientContext) {} std::unique_ptr<::physical::PhysicalPlan> GQueryConvertor::convert( const planner::LogicalPlan& plan, bool skipSink) { @@ -1984,7 +1987,7 @@ common::TableType GQueryConvertor::getTableType( void GQueryConvertor::convertCrossProduct( const planner::LogicalCrossProduct& cross, ::physical::PhysicalPlan* plan) { auto joinPB = std::make_unique<::physical::Join>(); - GPhysicalConvertor convertor(aliasManager, catalog); + GPhysicalConvertor convertor(aliasManager, catalog, clientContext); // convert left plan planner::LogicalPlan leftPlan; leftPlan.setLastOperator(cross.getChild(0)); @@ -2039,7 +2042,7 @@ void GQueryConvertor::extractJoinKeys( void GQueryConvertor::convertHashJoin(const planner::LogicalHashJoin& join, ::physical::PhysicalPlan* plan) { auto joinPB = std::make_unique<::physical::Join>(); - GPhysicalConvertor convertor(aliasManager, catalog); + GPhysicalConvertor convertor(aliasManager, catalog, clientContext); auto leftOp = join.getChild(0); // convert left plan to pre query before the join, and set empty plan as the // join left branch diff --git a/src/compiler/planner/gopt_planner.cc b/src/compiler/planner/gopt_planner.cc index ccc656768..ac7ecd200 100644 --- a/src/compiler/planner/gopt_planner.cc +++ b/src/compiler/planner/gopt_planner.cc @@ -51,7 +51,7 @@ result> GOptPlanner::compilePlan( auto aliasManager = std::make_shared(*statement->logicalPlan); neug::gopt::GPhysicalConvertor converter(aliasManager, - database->getCatalog()); + database->getCatalog(), ctx.get()); auto physicalPlan = converter.convert(*statement->logicalPlan); VLOG(10) << "got plan: " << physicalPlan->DebugString(); diff --git a/src/execution/common/types/value.cc b/src/execution/common/types/value.cc index b43d41f90..89a8d8ff1 100644 --- a/src/execution/common/types/value.cc +++ b/src/execution/common/types/value.cc @@ -26,6 +26,12 @@ #include "neug/utils/serialization/out_archive.h" namespace neug { + +// Defined in schema.cc; declared here to avoid pulling in the heavy schema.h +// header. Needed by the kList branch of Value serialization below. +InArchive& operator<<(InArchive& arc, const DataType& type); +OutArchive& operator>>(OutArchive& arc, DataType& type); + namespace execution { enum class ExtraValueInfoType : uint8_t { INVALID_TYPE_INFO = 0, @@ -228,7 +234,7 @@ Value Value::LIST(const DataType& child_type, std::vector&& values) { Value Value::LIST(std::vector&& values) { if (values.empty()) { - throw std::runtime_error("Cannot create LIST Value with empty values"); + return Value::LIST(DataType(DataTypeId::kUnknown), std::move(values)); } const auto& type = values[0].type(); return Value::LIST(type, std::move(values)); @@ -890,6 +896,14 @@ InArchive& operator<<(InArchive& in_archive, const execution::Value& value) { auto interval = value.GetValue(); in_archive << type_id << interval.months << interval.days << interval.micros; + } else if (type_id == DataTypeId::kList) { + const auto& child_type = ListType::GetChildType(value.type()); + const auto& children = execution::ListValue::GetChildren(value); + in_archive << type_id << child_type + << static_cast(children.size()); + for (const auto& child : children) { + in_archive << child; + } } else { THROW_NOT_SUPPORTED_EXCEPTION( std::string("Value serialization not supported for type: ") + @@ -949,6 +963,18 @@ OutArchive& operator>>(OutArchive& out_archive, execution::Value& value) { Interval interval; out_archive >> interval.months >> interval.days >> interval.micros; value = execution::Value::INTERVAL(interval); + } else if (type_id == DataTypeId::kList) { + DataType child_type; + uint32_t count; + out_archive >> child_type >> count; + std::vector children; + children.reserve(count); + for (uint32_t i = 0; i < count; ++i) { + execution::Value child; + out_archive >> child; + children.push_back(std::move(child)); + } + value = execution::Value::LIST(child_type, std::move(children)); } else { THROW_NOT_SUPPORTED_EXCEPTION( std::string("Value deserialization not supported for type: ") + @@ -957,4 +983,4 @@ OutArchive& operator>>(OutArchive& out_archive, execution::Value& value) { return out_archive; } -} // namespace neug \ No newline at end of file +} // namespace neug diff --git a/src/execution/execute/ops/batch/batch_update_utils.cc b/src/execution/execute/ops/batch/batch_update_utils.cc index 959179007..43a520f0a 100644 --- a/src/execution/execute/ops/batch/batch_update_utils.cc +++ b/src/execution/execute/ops/batch/batch_update_utils.cc @@ -15,6 +15,7 @@ #include "neug/execution/execute/ops/batch/batch_update_utils.h" +#include #include #include #include @@ -405,38 +406,164 @@ create_record_batch_supplier_from_arrow_stream_column( THROW_RUNTIME_ERROR("No valid column mappings found."); } +static void append_value_to_builder(arrow::ArrayBuilder* builder, + const Value& val, + const DataType& elem_type) { + if (val.IsNull()) { + auto s = builder->AppendNull(); + CHECK(s.ok()) << "AppendNull failed: " << s.ToString(); + return; + } + arrow::Status s; + switch (elem_type.id()) { + case DataTypeId::kInt32: + s = static_cast(builder)->Append( + val.GetValue()); + break; + case DataTypeId::kInt64: + s = static_cast(builder)->Append( + val.GetValue()); + break; + case DataTypeId::kUInt32: + s = static_cast(builder)->Append( + val.GetValue()); + break; + case DataTypeId::kUInt64: + s = static_cast(builder)->Append( + val.GetValue()); + break; + case DataTypeId::kFloat: + s = static_cast(builder)->Append( + val.GetValue()); + break; + case DataTypeId::kDouble: + s = static_cast(builder)->Append( + val.GetValue()); + break; + case DataTypeId::kBoolean: + s = static_cast(builder)->Append( + val.GetValue()); + break; + case DataTypeId::kVarchar: { + auto& sv = StringValue::Get(val); + s = static_cast(builder)->Append(sv); + break; + } + case DataTypeId::kList: { + auto* list_builder = static_cast(builder); + s = list_builder->Append(); + CHECK(s.ok()) << "ListBuilder::Append failed: " << s.ToString(); + auto child_type = ListType::GetChildType(elem_type); + const auto& children = ListValue::GetChildren(val); + for (const auto& child : children) { + append_value_to_builder(list_builder->value_builder(), child, child_type); + } + return; + } + default: + THROW_RUNTIME_ERROR("Unsupported type for arrow builder: " + + elem_type.ToString()); + } + CHECK(s.ok()) << "Append failed: " << s.ToString(); +} + +static std::shared_ptr value_column_to_arrow_array( + const std::shared_ptr& column) { + auto arrow_type = PropertyTypeToArrowType(column->elem_type()); + auto builder_result = arrow::MakeBuilder(arrow_type); + CHECK(builder_result.ok()) << "Failed to create arrow builder"; + auto builder = std::move(builder_result).ValueUnsafe(); + size_t n = column->size(); + auto reserve_s = builder->Reserve(n); + CHECK(reserve_s.ok()) << "Reserve failed: " << reserve_s.ToString(); + for (size_t i = 0; i < n; ++i) { + append_value_to_builder(builder.get(), column->get_elem(i), + column->elem_type()); + } + auto result = builder->Finish(); + CHECK(result.ok()) << "Failed to finish arrow builder"; + return std::move(result).ValueUnsafe(); +} + std::vector> create_record_batch_supplier( const Context& ctx, const std::vector>& prop_mappings) { - // We expect all columns are of same type. - ContextColumnType column_type = ContextColumnType::kNone; + bool has_arrow_array = false; + bool has_arrow_stream = false; + bool has_value = false; for (const auto& mapping : prop_mappings) { auto tag_id = mapping.first; auto column = ctx.get(tag_id); if (column == nullptr) { - LOG(ERROR) << "Column not found for tag id: " << tag_id; THROW_RUNTIME_ERROR("Column not found for tag id: " + std::to_string(tag_id)); } - if (column_type == ContextColumnType::kNone) { - column_type = column->column_type(); - } else if (column_type != column->column_type()) { - LOG(ERROR) << "Column type mismatch for tag id: " << tag_id; - THROW_RUNTIME_ERROR("Column type mismatch for tag id: " + - std::to_string(tag_id)); + switch (column->column_type()) { + case ContextColumnType::kArrowArray: + has_arrow_array = true; + break; + case ContextColumnType::kArrowStream: + has_arrow_stream = true; + break; + case ContextColumnType::kValue: + has_value = true; + break; + default: + THROW_RUNTIME_ERROR( + "Unsupported column type: " + + std::to_string(static_cast(column->column_type()))); } } - if (column_type == ContextColumnType::kArrowArray) { - return create_record_batch_supplier_from_arrow_array_column(ctx, - prop_mappings); - } else if (column_type == ContextColumnType::kArrowStream) { + + if (has_arrow_stream && !has_arrow_array && !has_value) { return create_record_batch_supplier_from_arrow_stream_column(ctx, prop_mappings); - } else { - LOG(ERROR) << "Unsupported column type: " << static_cast(column_type); - THROW_RUNTIME_ERROR("Unsupported column type: " + - std::to_string(static_cast(column_type))); } + if (has_arrow_array && !has_value) { + return create_record_batch_supplier_from_arrow_array_column(ctx, + prop_mappings); + } + + // Mixed kArrowArray + kValue, or all kValue: build arrays from all columns. + std::vector> suppliers; + std::vector>> arrays; + std::vector> fields; + arrays.resize(prop_mappings.size()); + + for (size_t i = 0; i < prop_mappings.size(); ++i) { + auto tag_id = prop_mappings[i].first; + auto prop_name = prop_mappings[i].second; + auto column = ctx.get(tag_id); + + if (column->column_type() == ContextColumnType::kArrowArray) { + auto arrow_column = + std::dynamic_pointer_cast(column); + for (auto& array : arrow_column->GetColumns()) { + arrays[i].emplace_back(array); + } + fields.emplace_back(std::make_shared( + prop_name, arrow_column->GetArrowType(), true)); + } else { + auto arr = value_column_to_arrow_array(column); + arrays[i].emplace_back(arr); + fields.emplace_back( + std::make_shared(prop_name, arr->type(), true)); + } + } + + if (!arrays.empty()) { + size_t batch_size = arrays[0].size(); + for (size_t i = 1; i < arrays.size(); ++i) { + if (arrays[i].size() != batch_size) { + THROW_INTERNAL_EXCEPTION("Array size mismatch for tag id: " + + std::to_string(prop_mappings[i].first)); + } + } + } + auto schema = std::make_shared(fields); + suppliers.emplace_back( + std::make_shared(arrays, schema)); + return suppliers; } void to_arrow_csv_options( diff --git a/src/storages/graph/edge_table.cc b/src/storages/graph/edge_table.cc index 2e1935f2e..c45b3de73 100644 --- a/src/storages/graph/edge_table.cc +++ b/src/storages/graph/edge_table.cc @@ -14,6 +14,8 @@ */ #include "neug/storages/graph/edge_table.h" +#include "neug/execution/common/types/value.h" +#include "neug/utils/property/nest_column.h" #include "neug/storages/checkpoint_manifest.h" #include "neug/storages/module/module_broker.h" @@ -300,12 +302,10 @@ static std::vector get_row_from_recordbatch( for (int i = 0; i < rb->num_columns(); ++i) { auto array = rb->column(i); if (!array->type()->Equals(expected_types[i])) { - // Except for large string and string if ((expected_types[i]->Equals(arrow::utf8()) && array->type()->Equals(arrow::large_utf8())) || (expected_types[i]->Equals(arrow::large_utf8()) && array->type()->Equals(arrow::utf8()))) { - // pass } else { THROW_INVALID_ARGUMENT_EXCEPTION( std::string("property type not match recordbatch column type: ") + @@ -373,6 +373,21 @@ static std::vector get_row_from_recordbatch( } break; } + case DataTypeId::kList: { + auto list_arr = std::static_pointer_cast(array); + auto child_type = ListType::GetChildType(prop_types[i]); + auto values_arr = list_arr->values(); + auto start = list_arr->value_offset(row_idx); + auto length = list_arr->value_length(row_idx); + std::vector children; + children.reserve(length); + for (int64_t j = 0; j < length; ++j) { + children.push_back( + arrow_element_to_value(values_arr, start + j, child_type)); + } + row.push_back(execution::Value::LIST(child_type, std::move(children))); + break; + } default: THROW_NOT_SUPPORTED_EXCEPTION("not support type " + array->type()->ToString()); @@ -748,7 +763,8 @@ void EdgeTable::AddProperties( // NOTE: Rather than check meta_->is_bundled(),we check whether the table // is empty. if (meta_->properties.size() == 1 && - meta_->properties[0].id() != DataTypeId::kVarchar) { + meta_->properties[0].id() != DataTypeId::kVarchar && + meta_->properties[0].id() != DataTypeId::kList) { dropAndCreateNewBundledCSR(ckp, nullptr); } else { dropAndCreateNewUnbundledCSR(ckp, false); @@ -795,7 +811,8 @@ void EdgeTable::DeleteProperties(Checkpoint& ckp, dropAndCreateNewUnbundledCSR(ckp, true); } else if (table_->col_num() == 1) { auto remaining_col = table_->get_column_by_id(0); - if (remaining_col->type() != DataTypeId::kVarchar) { + if (remaining_col->type() != DataTypeId::kVarchar && + remaining_col->type() != DataTypeId::kList) { dropAndCreateNewBundledCSR(ckp, remaining_col); } } @@ -893,6 +910,7 @@ void EdgeTable::BatchAddEdges( } EnsureCapacity(new_cap); } + if (meta_->is_bundled()) { std::vector flat_edge_data; assert(meta_->properties.size() == 1); @@ -1151,9 +1169,17 @@ EdgeTable EdgeTable::OpenFrom(Checkpoint& ckp, if (!es->is_bundled()) { auto table = std::make_unique(es->property_names, es->properties); for (size_t i = 0; i < es->properties.size(); ++i) { + const std::string key = KeyProperty(src, edge, dst, i); + auto col = store.TakeModule(key); + if (es->properties[i].id() == DataTypeId::kList) { + auto* list_col = dynamic_cast(col.get()); + if (list_col) { + list_col->SetListType(es->properties[i]); + } + } + col->RestoreChildren(store, key); table->SetColumn(static_cast(i), - std::shared_ptr(store.TakeModule( - KeyProperty(src, edge, dst, i)))); + std::shared_ptr(std::move(col))); } et.SetTable(std::move(table)); et.SetTableIdx( @@ -1182,8 +1208,9 @@ void EdgeTable::DisassembleTo(ModuleBroker& store, CheckpointManifest& meta, if (!meta_->is_bundled()) { auto table = TakeTable(); for (size_t i = 0; i < table->col_num(); ++i) { - meta.set_module(KeyProperty(src, edge, dst, i), - table->get_column_by_id(i)->Dump(ckp)); + const std::string key = KeyProperty(src, edge, dst, i); + auto col = table->get_column_by_id(i); + col->DumpTo(ckp, meta, key); } meta.SetScalar(ScalarKey(src, edge, dst, "table_idx"), std::to_string(GetTableIdx())); diff --git a/src/storages/graph/property_graph.cc b/src/storages/graph/property_graph.cc index a0b3ad7ef..9ede543a6 100644 --- a/src/storages/graph/property_graph.cc +++ b/src/storages/graph/property_graph.cc @@ -994,6 +994,7 @@ bool PropertyGraph::get_lid(label_t label, const execution::Value& oid, execution::Value PropertyGraph::GetOid(label_t label, vid_t lid, timestamp_t ts) const { + schema_.ensure_vertex_label_valid(label); return vertex_tables_[label].GetOid(lid, ts); } diff --git a/src/storages/graph/schema.cc b/src/storages/graph/schema.cc index 06882541b..88a929406 100644 --- a/src/storages/graph/schema.cc +++ b/src/storages/graph/schema.cc @@ -237,7 +237,8 @@ bool EdgeSchema::is_bundled() const { if (properties.empty()) { return true; } else if (properties.size() == 1 && - properties[0].id() == DataTypeId::kVarchar) { + (properties[0].id() == DataTypeId::kVarchar || + properties[0].id() == DataTypeId::kList)) { return false; } else if (properties.size() > 1) { return false; diff --git a/src/storages/graph/vertex_table.cc b/src/storages/graph/vertex_table.cc index 32b133ae0..e40cf6683 100644 --- a/src/storages/graph/vertex_table.cc +++ b/src/storages/graph/vertex_table.cc @@ -14,11 +14,13 @@ */ #include "neug/storages/graph/vertex_table.h" - +#include "neug/execution/common/types/value.h" #include "neug/storages/checkpoint_manifest.h" #include "neug/storages/module/module_broker.h" #include "neug/storages/module_descriptor.h" +#include "neug/utils/file_utils.h" #include "neug/utils/likely.h" +#include "neug/utils/property/nest_column.h" namespace neug { @@ -117,6 +119,7 @@ bool VertexTable::AddVertex(const execution::Value& id, return true; } }()); + table_->insert(vid, props, insert_safe); return true; } @@ -305,9 +308,17 @@ VertexTable VertexTable::OpenFrom(Checkpoint& ckp, auto table = std::make_unique
(vs->property_names, vs->property_types); for (size_t i = 0; i < vs->property_types.size(); ++i) { + const std::string key = KeyProperty(lbl, i); + auto col = store.TakeModule(key); + if (vs->property_types[i].id() == DataTypeId::kList) { + auto* list_col = dynamic_cast(col.get()); + if (list_col) { + list_col->SetListType(vs->property_types[i]); + } + } + col->RestoreChildren(store, key); table->SetColumn(static_cast(i), - std::shared_ptr( - store.TakeModule(KeyProperty(lbl, i)))); + std::shared_ptr(std::move(col))); } vt.SetTable(std::move(table)); vt.SetVertexTimestamp( @@ -332,7 +343,9 @@ void VertexTable::DisassembleTo(ModuleBroker& store, CheckpointManifest& meta, auto table = TakeTable(); for (size_t i = 0; i < table->col_num(); ++i) { - meta.set_module(KeyProperty(lbl, i), table->get_column_by_id(i)->Dump(ckp)); + const std::string key = KeyProperty(lbl, i); + auto col = table->get_column_by_id(i); + col->DumpTo(ckp, meta, key); } store.SetModule(KeyVertexTimestamp(lbl), TakeVertexTimestamp()); } diff --git a/src/storages/loader/loader_utils.cc b/src/storages/loader/loader_utils.cc index 391ee6958..01bdd62b3 100644 --- a/src/storages/loader/loader_utils.cc +++ b/src/storages/loader/loader_utils.cc @@ -27,8 +27,10 @@ #include #include +#include "neug/common/types.h" #include "neug/utils/arrow_utils.h" #include "neug/utils/exception/exception.h" +#include "neug/utils/property/nest_column.h" #include "neug/utils/string_utils.h" namespace neug { @@ -875,6 +877,94 @@ void set_column_from_string_array(std::shared_ptr col, } } +execution::Value arrow_element_to_value( + const std::shared_ptr& arr, int64_t idx, + const DataType& neug_type) { + if (arr->IsNull(idx)) { + return execution::Value(DataType::SQLNULL); + } + switch (neug_type.id()) { + case DataTypeId::kBoolean: + return execution::Value::CreateValue( + std::static_pointer_cast(arr)->Value(idx)); + case DataTypeId::kInt32: + return execution::Value::CreateValue( + std::static_pointer_cast(arr)->Value(idx)); + case DataTypeId::kUInt32: + return execution::Value::CreateValue( + std::static_pointer_cast(arr)->Value(idx)); + case DataTypeId::kInt64: + return execution::Value::CreateValue( + std::static_pointer_cast(arr)->Value(idx)); + case DataTypeId::kUInt64: + return execution::Value::CreateValue( + std::static_pointer_cast(arr)->Value(idx)); + case DataTypeId::kFloat: + return execution::Value::CreateValue( + std::static_pointer_cast(arr)->Value(idx)); + case DataTypeId::kDouble: + return execution::Value::CreateValue( + std::static_pointer_cast(arr)->Value(idx)); + case DataTypeId::kVarchar: { + auto str = std::static_pointer_cast(arr)->GetView(idx); + return execution::Value::STRING(std::string(str)); + } + case DataTypeId::kDate: + return execution::Value::CreateValue( + Date(std::static_pointer_cast(arr)->Value(idx))); + case DataTypeId::kTimestampMs: + return execution::Value::CreateValue(DateTime( + std::static_pointer_cast(arr)->Value(idx))); + case DataTypeId::kList: { + auto list_arr = std::static_pointer_cast(arr); + auto child_type = ListType::GetChildType(neug_type); + auto values_arr = list_arr->values(); + auto start = list_arr->value_offset(idx); + auto length = list_arr->value_length(idx); + std::vector children; + children.reserve(length); + for (int64_t j = 0; j < length; ++j) { + children.push_back( + arrow_element_to_value(values_arr, start + j, child_type)); + } + return execution::Value::LIST(child_type, std::move(children)); + } + default: + THROW_NOT_SUPPORTED_EXCEPTION("Unsupported list child type: " + + neug_type.ToString()); + } +} + +void set_column_from_list_array(std::shared_ptr col, + std::shared_ptr array, + const std::vector& vids) { + auto* list_col = dynamic_cast(col.get()); + CHECK(list_col != nullptr) << "Expected ListColumn"; + auto neug_type = list_col->list_type(); + for (int j = 0; j < array->num_chunks(); ++j) { + auto chunk = array->chunk(j); + auto list_arr = std::static_pointer_cast(chunk); + for (int64_t k = 0; k < list_arr->length(); ++k) { + if (vids[k] >= std::numeric_limits::max()) { + continue; + } + auto child_type = ListType::GetChildType(neug_type); + auto values_arr = list_arr->values(); + auto start = list_arr->value_offset(k); + auto length = list_arr->value_length(k); + std::vector children; + children.reserve(length); + for (int64_t i = 0; i < length; ++i) { + children.push_back( + arrow_element_to_value(values_arr, start + i, child_type)); + } + col->set_any(vids[k], + execution::Value::LIST(child_type, std::move(children)), + true); + } + } +} + void set_properties_column(std::shared_ptr col, std::shared_ptr array, const std::vector& vids, @@ -902,6 +992,9 @@ void set_properties_column(std::shared_ptr col, case DataTypeId::kVarchar: set_column_from_string_array(col, array, vids, mutex, true); break; + case DataTypeId::kList: + set_column_from_list_array(col, array, vids); + break; default: THROW_NOT_SUPPORTED_EXCEPTION("Not support type: " + type->ToString()); } diff --git a/src/transaction/insert_transaction.cc b/src/transaction/insert_transaction.cc index 3d615cc16..04e8a8517 100644 --- a/src/transaction/insert_transaction.cc +++ b/src/transaction/insert_transaction.cc @@ -58,10 +58,11 @@ bool InsertTransaction::GetVertexIndex(label_t label, if (graph_.get_lid(label, id, index, timestamp_)) { return true; } - if (added_vertices_.size() > label && added_vertices_[label] != nullptr && - added_vertices_[label]->get_index(id, index)) { - index += added_vertices_base_[label]; - return true; + if (added_vertices_.size() > label && added_vertices_[label] != nullptr) { + if (added_vertices_[label]->get_index(id, index)) { + index += added_vertices_base_[label]; + return true; + } } return false; } @@ -111,6 +112,19 @@ Status InsertTransaction::AddVertex(label_t label, const execution::Value& id, types[col_i].ToString() + ", but got " + prop.type().ToString()); } + if (types[col_i].id() == DataTypeId::kList && + !(prop.type() == types[col_i])) { + std::string label_name = graph_.schema().get_vertex_label_name(label); + LOG(ERROR) << "Vertex [" << label_name << "][" << col_i + << "] list child type not match, expected " + << types[col_i].ToString() << ", but got " + << prop.type().ToString(); + return Status(StatusCode::ERR_INVALID_ARGUMENT, + "Vertex [" + label_name + "][" + std::to_string(col_i) + + "] list child type not match, expected " + + types[col_i].ToString() + ", but got " + + prop.type().ToString()); + } } create_id_indexer_if_not_exists(label); if (!GetVertexIndex(label, id, vid)) { @@ -151,6 +165,19 @@ Status InsertTransaction::AddEdge( " type not match, expected " + types[i].ToString() + ", got " + properties[i].type().ToString()); } + if (types[i].id() == DataTypeId::kList && + !(properties[i].type() == types[i])) { + std::string label_name = graph_.schema().get_edge_label_name(edge_label); + LOG(ERROR) << "Edge property " << label_name + << " list child type not match, expected " + << types[i].ToString() << ", got " + << properties[i].type().ToString(); + return Status(StatusCode::ERR_INVALID_ARGUMENT, + "Edge property " + label_name + + " list child type not match, expected " + + types[i].ToString() + ", got " + + properties[i].type().ToString()); + } } InsertEdgeRedo::Serialize(arc_, src_label, src, dst_label, dst, edge_label, properties); diff --git a/src/transaction/update_transaction.cc b/src/transaction/update_transaction.cc index 77c78d34c..d9fab93b6 100644 --- a/src/transaction/update_transaction.cc +++ b/src/transaction/update_transaction.cc @@ -578,6 +578,13 @@ Status UpdateTransaction::AddVertex(label_t label, const execution::Value& oid, std::to_string(col_i) + " for vertex of label " + graph_.schema().get_vertex_label_name(label)); } + if (types[col_i].id() == DataTypeId::kList && + !(props[col_i].type() == types[col_i])) { + return Status(StatusCode::ERR_INVALID_ARGUMENT, + "List child type mismatch at column " + + std::to_string(col_i) + " for vertex of label " + + graph_.schema().get_vertex_label_name(label)); + } } const auto& v_table = graph_.get_vertex_table(label); @@ -810,6 +817,10 @@ bool UpdateTransaction::UpdateVertexProperty(label_t label, vid_t lid, if (types[col_id].id() != value.type().id()) { return false; } + if (types[col_id].id() == DataTypeId::kList && + !(value.type() == types[col_id])) { + return false; + } UpdateVertexPropRedo::Serialize(arc_, label, GetVertexId(label, lid), col_id, value); diff --git a/src/utils/pb_utils.cc b/src/utils/pb_utils.cc index 1e15dca08..b07dbac33 100644 --- a/src/utils/pb_utils.cc +++ b/src/utils/pb_utils.cc @@ -231,8 +231,19 @@ bool data_type_to_property_type(const common::DataType& data_type, return temporal_type_to_property_type(data_type.temporal(), out_type); } case common::DataType::kArray: { - LOG(ERROR) << "Array type is not supported"; - return false; + // A List/Array property: recursively resolve the element type. + const auto& array_type = data_type.array(); + if (!array_type.has_component_type()) { + LOG(ERROR) << "Array type missing component type: " + << data_type.DebugString(); + return false; + } + DataType child_type; + if (!data_type_to_property_type(array_type.component_type(), child_type)) { + return false; + } + out_type = DataType::List(child_type); + break; } case common::DataType::kMap: { LOG(ERROR) << "Map type is not supported"; @@ -246,6 +257,7 @@ bool data_type_to_property_type(const common::DataType& data_type, LOG(ERROR) << "Unknown data type: " << data_type.DebugString(); return false; } + return true; } bool common_value_to_value(const DataType& type, const common::Value& value, @@ -306,6 +318,50 @@ bool common_value_to_value(const DataType& type, const common::Value& value, return true; } +// Convert a protobuf Expression (e.g. ToList, scalar const) into an +// execution::Value. This is needed because the compiler serialises default +// values for complex types (nested lists) into the `default_expr` field of +// PropertyDef rather than the flat `default_value` field. +static bool common_expr_to_value(const DataType& type, + const common::Expression& expr, + execution::Value& out_value) { + if (expr.operators_size() == 0) { + LOG(ERROR) << "Empty expression in default_expr"; + return false; + } + const auto& opr = expr.operators(0); + + if (opr.has_to_list()) { + if (type.id() != DataTypeId::kList) { + LOG(ERROR) << "ToList expression but type is not list: " + << type.ToString(); + return false; + } + DataType child_type = ListType::GetChildType(type); + const auto& to_list = opr.to_list(); + std::vector items; + items.reserve(to_list.fields_size()); + for (const auto& field_expr : to_list.fields()) { + execution::Value child_value(DataType::SQLNULL); + if (!common_expr_to_value(child_type, field_expr, child_value)) { + LOG(ERROR) << "Failed to convert ToList child expression"; + return false; + } + items.push_back(std::move(child_value)); + } + out_value = execution::Value::LIST(child_type, std::move(items)); + return true; + } + + if (opr.has_const_()) { + return common_value_to_value(type, opr.const_(), out_value); + } + + LOG(ERROR) << "Unsupported expression operator in default_expr: " + << expr.DebugString(); + return false; +} + neug::result>> property_defs_to_value( const google::protobuf::RepeatedPtrField& @@ -320,7 +376,15 @@ property_defs_to_value( "Invalid property type: " + property.DebugString())); } - if (property.has_default_value()) { + if (property.has_default_expr()) { + if (!common_expr_to_value(type, property.default_expr(), default_value)) { + RETURN_ERROR(Status(StatusCode::ERR_INVALID_ARGUMENT, + "Failed to convert default_expr for property: " + + property.DebugString())); + } + VLOG(10) << "Default expr convert to value success:" + << property.default_expr().DebugString(); + } else if (property.has_default_value()) { if (!common_value_to_value(type, property.default_value(), default_value)) { RETURN_ERROR( diff --git a/src/utils/property/column.cc b/src/utils/property/column.cc index 63e08323c..47e2ac82e 100644 --- a/src/utils/property/column.cc +++ b/src/utils/property/column.cc @@ -15,17 +15,18 @@ #include "neug/utils/property/column.h" -#include - -#include "neug/storages/container/container_utils.h" +#include "neug/storages/checkpoint_manifest.h" #include "neug/storages/module/module_factory.h" -#include "neug/utils/id_indexer.h" -#include "neug/utils/property/table.h" +#include "neug/utils/property/nest_column.h" #include "neug/utils/property/types.h" -#include "neug/utils/serialization/out_archive.h" namespace neug { +void ColumnBase::DumpTo(Checkpoint& ckp, CheckpointManifest& meta, + const std::string& key) { + meta.set_module(key, Dump(ckp)); +} + std::string_view truncate_utf8(std::string_view str, size_t length) { if (str.size() <= length) { return str; @@ -75,6 +76,9 @@ std::unique_ptr CreateColumn(DataType type) { case DataTypeId::kEmpty: { return std::make_unique>(); } + case DataTypeId::kList: { + return std::make_unique(type); + } default: { THROW_NOT_SUPPORTED_EXCEPTION("Unsupported type for column: " + type.ToString()); @@ -95,6 +99,10 @@ std::shared_ptr CreateRefColumn(const ColumnBase& column) { return std::make_shared>( dynamic_cast(column)); } + case DataTypeId::kList: { + return std::make_shared( + dynamic_cast(column)); + } default: { THROW_NOT_SUPPORTED_EXCEPTION("Unsupported type for reference column: " + std::to_string(type)); diff --git a/src/utils/property/nest_column.cc b/src/utils/property/nest_column.cc new file mode 100644 index 000000000..1990e1362 --- /dev/null +++ b/src/utils/property/nest_column.cc @@ -0,0 +1,56 @@ +/** Copyright 2020 Alibaba Group Holding Limited. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "neug/utils/property/nest_column.h" + +#include "neug/storages/checkpoint_manifest.h" +#include "neug/storages/module/module_broker.h" +#include "neug/storages/module/module_factory.h" + +namespace neug { + +NEUG_REGISTER_MODULE(ListColumn); + +void ListColumn::DumpTo(Checkpoint& ckp, CheckpointManifest& meta, + const std::string& key) { + meta.set_module(key, Dump(ckp)); + if (child_column_) { + const std::string child_key = key + "/child"; + child_column_->DumpTo(ckp, meta, child_key); + } +} + +void ListColumn::RestoreChildren(ModuleBroker& store, const std::string& key) { + const std::string child_key = key + "/child"; + if (!store.Contains(child_key)) { + return; + } + child_column_ = store.TakeModule(child_key); + + // Propagate child type info for nested lists before recursing. + if (list_type_.id() == DataTypeId::kList) { + auto child_type = ListType::GetChildType(list_type_); + if (child_type.id() == DataTypeId::kList) { + auto* child_list = dynamic_cast(child_column_.get()); + if (child_list) { + child_list->SetListType(child_type); + } + } + } + + child_column_->RestoreChildren(store, child_key); +} + +} // namespace neug diff --git a/src/utils/writer/writer.cc b/src/utils/writer/writer.cc index 771bbb0ed..b892bc37a 100644 --- a/src/utils/writer/writer.cc +++ b/src/utils/writer/writer.cc @@ -354,10 +354,10 @@ neug::Status CsvQueryExportWriter::writeTable( auto stream_result = fileSystem_->OpenOutputStream(schema_.paths[0]); if (!stream_result.ok()) { int err = arrow::internal::ErrnoFromStatus(stream_result.status()); - auto code = (err == EACCES || err == EPERM) - ? StatusCode::ERR_PERMISSION - : StatusCode::ERR_IO_ERROR; - return neug::Status(code, "Failed to open file stream: " + stream_result.status().ToString()); + auto code = (err == EACCES || err == EPERM) ? StatusCode::ERR_PERMISSION + : StatusCode::ERR_IO_ERROR; + return neug::Status(code, "Failed to open file stream: " + + stream_result.status().ToString()); } auto stream = stream_result.ValueOrDie(); diff --git a/src/utils/yaml_utils.cc b/src/utils/yaml_utils.cc index e6ca90f3f..c63da4e8a 100644 --- a/src/utils/yaml_utils.cc +++ b/src/utils/yaml_utils.cc @@ -76,6 +76,11 @@ YAML::Node property_type_to_yaml(const DataType& type) { case DataTypeId::kInterval: node["temporal"] = config_parsing::TemporalTypeToYAML(type.id()); break; + case DataTypeId::kList: { + auto child_type = ListType::GetChildType(type); + node["list"]["component_type"] = property_type_to_yaml(child_type); + break; + } default: THROW_INVALID_ARGUMENT_EXCEPTION( "Unrecognized property type for YAML encoding: " + type.ToString()); diff --git a/tests/compiler/gopt_test.h b/tests/compiler/gopt_test.h index 45778a5bc..e709091fe 100644 --- a/tests/compiler/gopt_test.h +++ b/tests/compiler/gopt_test.h @@ -223,7 +223,8 @@ class GOptTest : public ::testing::Test { std::unique_ptr<::physical::PhysicalPlan> planPhysical( const planner::LogicalPlan& plan, std::shared_ptr aliasManager) { - gopt::GPhysicalConvertor converter(aliasManager, database->getCatalog()); + gopt::GPhysicalConvertor converter(aliasManager, database->getCatalog(), + ctx.get()); auto physicalPlan = converter.convert(plan); return physicalPlan; } @@ -232,7 +233,8 @@ class GOptTest : public ::testing::Test { const planner::LogicalPlan& plan) { // Convert to physical plan auto aliasManager = std::make_shared(plan); - gopt::GPhysicalConvertor converter(aliasManager, database->getCatalog()); + gopt::GPhysicalConvertor converter(aliasManager, database->getCatalog(), + ctx.get()); auto physicalPlan = converter.convert(plan); return physicalPlan; } diff --git a/tests/execution/test_value.cc b/tests/execution/test_value.cc index 8a233d8f1..824535a76 100644 --- a/tests/execution/test_value.cc +++ b/tests/execution/test_value.cc @@ -15,6 +15,8 @@ #include #include "neug/execution/common/types/value.h" +#include "neug/utils/serialization/in_archive.h" +#include "neug/utils/serialization/out_archive.h" namespace neug { namespace execution { @@ -415,13 +417,38 @@ TEST_F(ValueTest, GetValueTemplate) { EXPECT_EQ(str_result, "hello"); } -TEST_F(ValueTest, PropertyConversion) { - // value_to_property/property_to_value have been removed; this test is - // intentionally left empty after the Property type retirement. +TEST_F(ValueTest, ArchiveRoundTrip) { + auto round_trip = [](const Value& original) { + InArchive in; + in << original; + OutArchive out; + out.SetSlice(in.GetBuffer(), in.GetSize()); + Value restored; + out >> restored; + EXPECT_TRUE(restored == original); + }; + + round_trip(Value::BOOLEAN(false)); + round_trip(Value::INT32(0x7fffffff)); + round_trip(Value::INT64(123456789012345LL)); + round_trip(Value::UINT32(42)); + round_trip(Value::UINT64(0xdeadbeefdeadbeefULL)); + round_trip(Value::FLOAT(2.71828f)); + round_trip(Value::DOUBLE(3.141592653589793)); + round_trip(Value::STRING("archive round trip")); + round_trip(Value::DATE(Date(std::string("2026-06-02")))); + round_trip(Value::TIMESTAMPMS(DateTime(987654321))); + round_trip(Value::INTERVAL(Interval(std::string("3years")))); + round_trip(Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(1), Value::INT32(2), Value::INT32(3)})); } TEST_F(ValueTest, EdgeCases) { - EXPECT_THROW({ Value::LIST(std::vector()); }, std::runtime_error); + { + auto empty_list = Value::LIST(std::vector()); + EXPECT_EQ(empty_list.type().id(), DataTypeId::kList); + EXPECT_TRUE(ListValue::GetChildren(empty_list).empty()); + } // LOG(FATAL) calls abort(); EXPECT_DEATH is unreliable under sanitizers. #if !defined(__SANITIZE_ADDRESS__) && !defined(__SANITIZE_THREAD__) && \ diff --git a/tests/storage/CMakeLists.txt b/tests/storage/CMakeLists.txt index 7e8833a40..9b92791ac 100644 --- a/tests/storage/CMakeLists.txt +++ b/tests/storage/CMakeLists.txt @@ -20,3 +20,5 @@ add_neug_test(test_vertex_table test_vertex_table.cc) add_neug_test(edge_table_test test_edge_table.cc) add_neug_test(graph_view_test test_graph_view.cc) + +add_neug_test(test_list_column test_list_column.cc) diff --git a/tests/storage/test_ddl.cc b/tests/storage/test_ddl.cc index 515256d57..70d9451cf 100644 --- a/tests/storage/test_ddl.cc +++ b/tests/storage/test_ddl.cc @@ -268,3 +268,12 @@ TEST(StorageDDLTest, CreateAndAlterTables) { neug::test::AssertInt64Column(table.response(), 0, {4}); } } + +// --------------------------------------------------------------------------- +// Primary key type validation: LIST should be rejected +// --------------------------------------------------------------------------- +TEST_F(DDLTestDBFixture, ListAsPrimaryKeyRejected) { + // Attempt to create table with LIST as primary key - should fail + EXPECT_FALSE( + conn->Query("CREATE NODE TABLE BadTable(id INT64[] PRIMARY KEY);")); +} diff --git a/tests/storage/test_list_column.cc b/tests/storage/test_list_column.cc new file mode 100644 index 000000000..c3d914066 --- /dev/null +++ b/tests/storage/test_list_column.cc @@ -0,0 +1,623 @@ +/** Copyright 2020 Alibaba Group Holding Limited. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include + +#include +#include +#include +#include +#include +#include + +#include "neug/common/types.h" +#include "neug/execution/common/types/value.h" +#include "neug/storages/checkpoint_manager.h" +#include "neug/storages/checkpoint_manifest.h" +#include "neug/storages/module/module_broker.h" +#include "neug/utils/property/column.h" +#include "neug/utils/property/nest_column.h" +#include "neug/utils/property/property.h" +#include "unittest/utils.h" + +using namespace neug; +using namespace neug::execution; + +// =========================================================================== +// Helpers +// =========================================================================== + +template +static void ExpectValueListEq(const Value& val, + std::initializer_list expected) { + ASSERT_EQ(val.type().id(), DataTypeId::kList); + const auto& children = ListValue::GetChildren(val); + ASSERT_EQ(children.size(), expected.size()); + size_t i = 0; + for (auto& e : expected) { + EXPECT_EQ(children[i].template GetValue(), e) << "index " << i; + ++i; + } +} + +static void ExpectValueStringListEq( + const Value& val, std::initializer_list expected) { + ASSERT_EQ(val.type().id(), DataTypeId::kList); + const auto& children = ListValue::GetChildren(val); + ASSERT_EQ(children.size(), expected.size()); + size_t i = 0; + for (auto& e : expected) { + EXPECT_EQ(StringValue::Get(children[i]), e) << "index " << i; + ++i; + } +} + +// Test fixture with CheckpointManager for ListColumn tests. +class ListColumnFixture : public ::testing::Test { + protected: + void SetUp() override { + temp_dir_ = std::filesystem::temp_directory_path() / + ("test_list_column_" + std::to_string(::getpid()) + "_" + + GetTestName()); + if (std::filesystem::exists(temp_dir_)) { + std::filesystem::remove_all(temp_dir_); + } + std::filesystem::create_directories(temp_dir_); + ws_.Open(temp_dir_.string()); + } + + void TearDown() override { + if (std::filesystem::exists(temp_dir_)) { + std::filesystem::remove_all(temp_dir_); + } + } + + std::shared_ptr MakeCheckpoint() { return make_checkpoint(ws_); } + + private: + std::filesystem::path temp_dir_; + CheckpointManager ws_; + + std::string GetTestName() const { + const testing::TestInfo* const test_info = + testing::UnitTest::GetInstance()->current_test_info(); + return std::string(test_info->name()); + } +}; + +// --------------------------------------------------------------------------- +// ListColumn — set / get via Value API +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, SetAndGetIntList) { + auto ckp = MakeCheckpoint(); + DataType list_type = DataType::List(DataType(DataTypeId::kInt32)); + ListColumn col(list_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + col.resize(4); + + col.set_any( + 0, + Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(10), Value::INT32(20), Value::INT32(30)}), + true); + col.set_any(1, Value::LIST(DataType(DataTypeId::kInt32), {}), true); + col.set_any(2, Value::LIST(DataType(DataTypeId::kInt32), {Value::INT32(99)}), + true); + col.set_any(3, + Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(-1), Value::INT32(-2)}), + true); + + ExpectValueListEq(col.get_any(0), {10, 20, 30}); + EXPECT_EQ(ListValue::GetChildren(col.get_any(1)).size(), 0u); + ExpectValueListEq(col.get_any(2), {99}); + ExpectValueListEq(col.get_any(3), {-1, -2}); +} + +TEST_F(ListColumnFixture, SetAndGetVarcharList) { + auto ckp = MakeCheckpoint(); + DataType list_type = DataType::List(DataType::Varchar(256)); + ListColumn col(list_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + col.resize(2); + + col.set_any( + 0, + Value::LIST(DataType::Varchar(256), {Value::STRING(std::string("alice")), + Value::STRING(std::string("bob"))}), + true); + col.set_any(1, + Value::LIST(DataType::Varchar(256), + {Value::STRING(std::string("single"))}), + true); + + ExpectValueStringListEq(col.get_any(0), {"alice", "bob"}); + ExpectValueStringListEq(col.get_any(1), {"single"}); +} + +// --------------------------------------------------------------------------- +// ListColumn — dump and reload (with child column round-trip via broker) +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, DumpAndReload) { + auto ckp = MakeCheckpoint(); + DataType list_type = DataType::List(DataType(DataTypeId::kInt64)); + + CheckpointManifest meta; + const std::string col_key = "test_col"; + { + ListColumn col(list_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + col.resize(3); + + col.set_any(0, + Value::LIST(DataType(DataTypeId::kInt64), + {Value::INT64(100), Value::INT64(200)}), + true); + col.set_any(1, + Value::LIST(DataType(DataTypeId::kInt64), {Value::INT64(300)}), + true); + col.set_any(2, Value::LIST(DataType(DataTypeId::kInt64), {}), true); + + col.DumpTo(*ckp, meta, col_key); + } + + { + ModuleBroker store; + store.Open(*ckp, meta, MemoryLevel::kInMemory); + auto col = store.TakeModule(col_key); + col->SetListType(list_type); + col->RestoreChildren(store, col_key); + + ExpectValueListEq(col->get_any(0), {100L, 200L}); + ExpectValueListEq(col->get_any(1), {300L}); + EXPECT_EQ(ListValue::GetChildren(col->get_any(2)).size(), 0u); + } +} + +// --------------------------------------------------------------------------- +// ListColumn — resize with non-empty default value +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, ResizeWithDefaultValue) { + auto ckp = MakeCheckpoint(); + DataType list_type = DataType::List(DataType(DataTypeId::kInt32)); + ListColumn col(list_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + + Value default_val = Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(7), Value::INT32(8)}); + col.resize(3, default_val); + + for (size_t i = 0; i < 3; ++i) { + ExpectValueListEq(col.get_any(i), {7, 8}); + } + + col.resize(5); + EXPECT_EQ(ListValue::GetChildren(col.get_any(3)).size(), 0u); + EXPECT_EQ(ListValue::GetChildren(col.get_any(4)).size(), 0u); + + Value default2 = + Value::LIST(DataType(DataTypeId::kInt32), {Value::INT32(99)}); + col.resize(1); + col.resize(4, default2); + + ExpectValueListEq(col.get_any(0), {7, 8}); + for (size_t i = 1; i < 4; ++i) { + ExpectValueListEq(col.get_any(i), {99}); + } + + Value empty_default = + Value::LIST(DataType(DataTypeId::kInt32), std::vector{}); + col.resize(6, empty_default); + for (size_t i = 4; i < 6; ++i) { + EXPECT_EQ(ListValue::GetChildren(col.get_any(i)).size(), 0u) << "row " << i; + } +} + +// --------------------------------------------------------------------------- +// Concurrent writes to ListColumn (child column model) +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, ConcurrentWrites) { + auto ckp = MakeCheckpoint(); + DataType list_type = DataType::List(DataType(DataTypeId::kInt32)); + ListColumn col(list_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + col.resize(100); + + std::atomic counter(0); + std::vector threads; + const int thread_num = 8; + + for (int i = 0; i < thread_num; ++i) { + threads.emplace_back([&]() { + while (true) { + size_t idx = counter.fetch_add(1); + if (idx >= 100) + break; + col.set_any(idx, + Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(static_cast(idx))}), + true); + } + }); + } + for (auto& th : threads) + th.join(); + + for (size_t i = 0; i < 100; ++i) { + ExpectValueListEq(col.get_any(i), {static_cast(i)}); + } +} + +// --------------------------------------------------------------------------- +// Mixed POD and non-POD concurrent writes +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, MixedPodNonPodConcurrent) { + auto ckp = MakeCheckpoint(); + DataType pod_type = DataType::List(DataType(DataTypeId::kInt64)); + DataType nonpod_type = DataType::List(DataType::Varchar(256)); + + ListColumn pod_col(pod_type); + ListColumn nonpod_col(nonpod_type); + pod_col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + nonpod_col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + pod_col.resize(50); + nonpod_col.resize(50); + + std::atomic counter(0); + std::vector threads; + const int thread_num = 8; + + for (int i = 0; i < thread_num; ++i) { + threads.emplace_back([&]() { + while (true) { + size_t idx = counter.fetch_add(1); + if (idx >= 50) + break; + pod_col.set_any( + idx, + Value::LIST(DataType(DataTypeId::kInt64), + {Value::INT64(static_cast(idx)), + Value::INT64(static_cast(idx * 2))}), + true); + nonpod_col.set_any( + idx, + Value::LIST(DataType::Varchar(256), + {Value::STRING("str_" + std::to_string(idx))}), + true); + } + }); + } + for (auto& th : threads) + th.join(); + + for (size_t i = 0; i < 50; ++i) { + auto val = pod_col.get_any(i); + const auto& children = ListValue::GetChildren(val); + ASSERT_EQ(children.size(), 2u) << "row " << i; + EXPECT_EQ(children[0].GetValue(), static_cast(i)); + EXPECT_EQ(children[1].GetValue(), static_cast(i * 2)); + + ExpectValueStringListEq(nonpod_col.get_any(i), + {"str_" + std::to_string(i)}); + } +} + +// --------------------------------------------------------------------------- +// Update same row grows child frontier (old data becomes hole) +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, UpdateGrowsChildFrontier) { + auto ckp = MakeCheckpoint(); + DataType list_type = DataType::List(DataType(DataTypeId::kInt32)); + ListColumn col(list_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + col.resize(2); + + col.set_any(0, Value::LIST(DataType(DataTypeId::kInt32), {Value::INT32(1)}), + true); + uint64_t frontier_after_first = col.child_frontier(); + + col.set_any(0, + Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(10), Value::INT32(20)}), + true); + uint64_t frontier_after_second = col.child_frontier(); + + EXPECT_GT(frontier_after_second, frontier_after_first); + ExpectValueListEq(col.get_any(0), {10, 20}); +} + +// --------------------------------------------------------------------------- +// Nested list: List> set/get round-trip +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, NestedListStorage) { + auto ckp = MakeCheckpoint(); + DataType inner_type = DataType::List(DataType(DataTypeId::kInt32)); + DataType outer_type = DataType::List(inner_type); + ListColumn col(outer_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + col.resize(3); + + Value row0 = Value::LIST( + inner_type, + {Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(1), Value::INT32(2)}), + Value::LIST(DataType(DataTypeId::kInt32), {Value::INT32(3)})}); + Value row1 = Value::LIST(inner_type, {}); + Value row2 = + Value::LIST(inner_type, {Value::LIST(DataType(DataTypeId::kInt32), {})}); + + col.set_any(0, row0, true); + col.set_any(1, row1, true); + col.set_any(2, row2, true); + + { + auto val = col.get_any(0); + const auto& outer = ListValue::GetChildren(val); + ASSERT_EQ(outer.size(), 2u); + ExpectValueListEq(outer[0], {1, 2}); + ExpectValueListEq(outer[1], {3}); + } + { + auto val = col.get_any(1); + EXPECT_EQ(ListValue::GetChildren(val).size(), 0u); + } + { + auto val = col.get_any(2); + const auto& outer = ListValue::GetChildren(val); + ASSERT_EQ(outer.size(), 1u); + EXPECT_EQ(ListValue::GetChildren(outer[0]).size(), 0u); + } + + ASSERT_NE(col.child_column(), nullptr); + auto* inner_lc = dynamic_cast(col.child_column()); + ASSERT_NE(inner_lc, nullptr); +} + +// --------------------------------------------------------------------------- +// Nested list dump and reload via ModuleBroker +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, NestedListDumpReload) { + auto ckp = MakeCheckpoint(); + DataType inner_type = DataType::List(DataType(DataTypeId::kInt32)); + DataType outer_type = DataType::List(inner_type); + + CheckpointManifest meta; + const std::string col_key = "nested_col"; + { + ListColumn col(outer_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + col.resize(2); + + col.set_any( + 0, + Value::LIST(inner_type, + {Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(10), Value::INT32(20)})}), + true); + col.set_any( + 1, + Value::LIST( + inner_type, + {Value::LIST(DataType(DataTypeId::kInt32), {Value::INT32(30)}), + Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(40), Value::INT32(50)})}), + true); + + col.DumpTo(*ckp, meta, col_key); + } + + { + ModuleBroker store; + store.Open(*ckp, meta, MemoryLevel::kInMemory); + auto col = store.TakeModule(col_key); + col->SetListType(outer_type); + col->RestoreChildren(store, col_key); + + auto val0 = col->get_any(0); + const auto& outer0 = ListValue::GetChildren(val0); + ASSERT_EQ(outer0.size(), 1u); + ExpectValueListEq(outer0[0], {10, 20}); + + auto val1 = col->get_any(1); + const auto& outer1 = ListValue::GetChildren(val1); + ASSERT_EQ(outer1.size(), 2u); + ExpectValueListEq(outer1[0], {30}); + ExpectValueListEq(outer1[1], {40, 50}); + } +} + +// --------------------------------------------------------------------------- +// Dump compacts child column by eliminating holes from updates +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, DumpCompactsChildHoles) { + auto ckp = MakeCheckpoint(); + DataType list_type = DataType::List(DataType(DataTypeId::kInt32)); + + CheckpointManifest meta; + const std::string col_key = "compact_col"; + { + ListColumn col(list_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + col.resize(3); + + col.set_any( + 0, + Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(1), Value::INT32(2), Value::INT32(3)}), + true); + col.set_any(1, + Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(10), Value::INT32(20)}), + true); + col.set_any(2, + Value::LIST(DataType(DataTypeId::kInt32), {Value::INT32(100)}), + true); + + // 6 live children, child_frontier == 6 + EXPECT_EQ(col.child_frontier(), 6u); + + // Update row 0 with a longer list — old 3 children become holes + col.set_any(0, + Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(7), Value::INT32(8)}), + true); + // child_frontier grew: 6 + 2 = 8, but only 5 children are live + EXPECT_EQ(col.child_frontier(), 8u); + + // Update row 1 — old 2 children become holes + col.set_any( + 1, Value::LIST(DataType(DataTypeId::kInt32), {Value::INT32(99)}), true); + // child_frontier: 8 + 1 = 9, live children: 4 + EXPECT_EQ(col.child_frontier(), 9u); + + col.DumpTo(*ckp, meta, col_key); + // After Dump (which runs compact), compaction should have run + EXPECT_EQ(col.child_frontier(), 4u); + } + + // Reload and verify data integrity + { + ModuleBroker store; + store.Open(*ckp, meta, MemoryLevel::kInMemory); + auto col = store.TakeModule(col_key); + col->SetListType(list_type); + col->RestoreChildren(store, col_key); + + ExpectValueListEq(col->get_any(0), {7, 8}); + ExpectValueListEq(col->get_any(1), {99}); + ExpectValueListEq(col->get_any(2), {100}); + EXPECT_EQ(col->child_frontier(), 4u); + } +} + +// --------------------------------------------------------------------------- +// Dump compaction expands shared entries into sequential layout for offset +// format +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, DumpCompactsSharedDefaults) { + auto ckp = MakeCheckpoint(); + DataType list_type = DataType::List(DataType(DataTypeId::kInt32)); + + CheckpointManifest meta; + const std::string col_key = "compact_dedup"; + { + ListColumn col(list_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + + // resize with a non-empty default — all rows share the same child segment + auto default_val = Value::LIST(DataType(DataTypeId::kInt32), + {Value::INT32(42), Value::INT32(43)}); + col.resize(5, default_val); + // All 5 rows point to the same {offset, 2} entry + // child_frontier == 2 (shared default written once) + EXPECT_EQ(col.child_frontier(), 2u); + + // Update row 2 — creates a new segment, old shared entry still valid for + // others + col.set_any( + 2, Value::LIST(DataType(DataTypeId::kInt32), {Value::INT32(99)}), true); + EXPECT_EQ(col.child_frontier(), 3u); + + col.DumpTo(*ckp, meta, col_key); + // After compaction: shared entries are fully expanded for sequential + // layout. Each row gets its own region: 2+2+1+2+2 = 9 total child elements. + EXPECT_EQ(col.child_frontier(), 9u); + } + + { + ModuleBroker store; + store.Open(*ckp, meta, MemoryLevel::kInMemory); + auto col = store.TakeModule(col_key); + col->SetListType(list_type); + col->RestoreChildren(store, col_key); + + for (size_t i : {0u, 1u, 3u, 4u}) { + ExpectValueListEq(col->get_any(i), {42, 43}); + } + ExpectValueListEq(col->get_any(2), {99}); + } +} + +// --------------------------------------------------------------------------- +// Large list (>65535 elements) — verifies 64-bit length support +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, LargeListExceeds16BitLimit) { + auto ckp = MakeCheckpoint(); + DataType list_type = DataType::List(DataType(DataTypeId::kInt32)); + ListColumn col(list_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + col.resize(1); + + const size_t N = 70000; // exceeds old 16-bit limit of 65535 + std::vector children; + children.reserve(N); + for (size_t i = 0; i < N; ++i) { + children.push_back(Value::INT32(static_cast(i))); + } + col.set_any(0, Value::LIST(DataType(DataTypeId::kInt32), std::move(children)), + true); + + auto val = col.get_any(0); + const auto& result = ListValue::GetChildren(val); + ASSERT_EQ(result.size(), N); + EXPECT_EQ(result[0].GetValue(), 0); + EXPECT_EQ(result[N - 1].GetValue(), static_cast(N - 1)); +} + +// --------------------------------------------------------------------------- +// Large list dump and reload round-trip (offset vector format) +// --------------------------------------------------------------------------- +TEST_F(ListColumnFixture, LargeListDumpReload) { + auto ckp = MakeCheckpoint(); + DataType list_type = DataType::List(DataType(DataTypeId::kInt32)); + + CheckpointManifest meta; + const std::string col_key = "large_list_col"; + const size_t N = 70000; + { + ListColumn col(list_type); + col.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory); + col.resize(2); + + std::vector children; + children.reserve(N); + for (size_t i = 0; i < N; ++i) { + children.push_back(Value::INT32(static_cast(i))); + } + col.set_any(0, + Value::LIST(DataType(DataTypeId::kInt32), std::move(children)), + true); + col.set_any( + 1, Value::LIST(DataType(DataTypeId::kInt32), {Value::INT32(42)}), true); + + col.DumpTo(*ckp, meta, col_key); + } + + { + ModuleBroker store; + store.Open(*ckp, meta, MemoryLevel::kInMemory); + auto col = store.TakeModule(col_key); + col->SetListType(list_type); + col->RestoreChildren(store, col_key); + + auto val0 = col->get_any(0); + const auto& result = ListValue::GetChildren(val0); + ASSERT_EQ(result.size(), N); + EXPECT_EQ(result[0].GetValue(), 0); + EXPECT_EQ(result[N - 1].GetValue(), static_cast(N - 1)); + + ExpectValueListEq(col->get_any(1), {42}); + } +} diff --git a/tests/utils/test_table.cc b/tests/utils/test_table.cc index f435cb6cc..e03693e1e 100644 --- a/tests/utils/test_table.cc +++ b/tests/utils/test_table.cc @@ -63,6 +63,9 @@ static const std::vector string_data = { namespace neug { namespace test { +using execution::StringValue; +using execution::Value; + // Test-side Open / Dump for Table: round-trips columns through ModuleBroker // + CheckpointManifest the same way the production OpenVertexTable flow does. // Pass an empty CheckpointManifest to initialize fresh columns, or one returned @@ -90,8 +93,6 @@ static void OpenTableLegacy(Table& t, Checkpoint& ckp, } static CheckpointManifest DumpTableLegacy(Table& t, Checkpoint& ckp) { - // Table holds columns by shared_ptr, so ownership can't be transferred - // into a unique_ptr-typed ModuleBroker — dump inline directly. CheckpointManifest meta; for (size_t i = 0; i < t.col_num(); ++i) { meta.set_module(TablePropKey(i), t.get_column_by_id(i)->Dump(ckp)); diff --git a/tools/python_bind/tests/test_db_list.py b/tools/python_bind/tests/test_db_list.py index 53d7a81d5..e2adb20a3 100644 --- a/tools/python_bind/tests/test_db_list.py +++ b/tools/python_bind/tests/test_db_list.py @@ -71,7 +71,9 @@ def test_return_multiple_lists(tmp_path): db.close() -@pytest.mark.skip(reason="list nesting is not supported") +@pytest.mark.skip( + reason="heterogeneous nested list (mixed child types) becomes STRUCT not LIST" +) def test_return_nesting_lists(tmp_path): db_dir = tmp_path / "return_nesting_lists" shutil.rmtree(db_dir, ignore_errors=True) @@ -92,3 +94,129 @@ def test_return_nesting_lists(tmp_path): conn.close() db.close() + + +def test_return_nested_list_literal(tmp_path): + """Homogeneous nested list literal: RETURN [[1],[2,3]]""" + db_dir = tmp_path / "return_nested_literal" + shutil.rmtree(db_dir, ignore_errors=True) + db_dir.mkdir() + db = Database(db_path=str(db_dir), mode="w") + conn = db.connect() + + result = conn.execute("RETURN [[1, 2], [3, 4]];") + result = list(result) + assert result[0][0] == [[1, 2], [3, 4]] + + conn.close() + db.close() + + +def test_nested_list(tmp_path): + db_dir = tmp_path / "nested_list" + shutil.rmtree(db_dir, ignore_errors=True) + db_dir.mkdir() + db = Database(db_path=str(db_dir), mode="w") + conn = db.connect() + + conn.execute( + "CREATE NODE TABLE PERSON(id INT64, string_prop STRING[][], PRIMARY KEY(id));" + ) + conn.execute("CREATE (p: PERSON {id: 0, string_prop: [['a', 'b'], ['c', 'd']]} );") + conn.execute("CREATE (p: PERSON {id: 1, string_prop: [['e', 'f'], ['g', 'h']]} );") + + result = conn.execute("MATCH (p: PERSON) RETURN p.string_prop;") + result = list(result) + assert result[0][0] == [["a", "b"], ["c", "d"]] + assert result[1][0] == [["e", "f"], ["g", "h"]] + + conn.close() + db.close() + + +def test_nested_list_default_value(tmp_path): + db_dir = tmp_path / "nested_list_default" + shutil.rmtree(db_dir, ignore_errors=True) + db_dir.mkdir() + db = Database(db_path=str(db_dir), mode="w") + conn = db.connect() + + conn.execute( + "CREATE NODE TABLE PERSON(" + "id INT64 PRIMARY KEY, " + "list_prop STRING[][] DEFAULT [['x'], ['y', 'z']]);" + ) + + conn.execute("CREATE (p: PERSON {id: 0});") + conn.execute("CREATE (p: PERSON {id: 1, list_prop: [['a', 'b'], ['c']]} );") + + result = conn.execute("MATCH (p: PERSON) RETURN p.id, p.list_prop ORDER BY p.id;") + rows = list(result) + assert rows[0] == [ + 0, + [["x"], ["y", "z"]], + ], f"expected nested default, got {rows[0]}" + assert rows[1] == [ + 1, + [["a", "b"], ["c"]], + ], f"expected explicit nested list, got {rows[1]}" + + conn.close() + db.close() + + +def test_list_as_primary_key_rejected(tmp_path): + """Test that LIST type cannot be used as primary key.""" + db_dir = tmp_path / "list_pk_rejection" + shutil.rmtree(db_dir, ignore_errors=True) + db_dir.mkdir() + db = Database(db_path=str(db_dir), mode="w") + conn = db.connect() + + with pytest.raises(Exception) as excinfo: + conn.execute("CREATE NODE TABLE BadTable(id INT64[] PRIMARY KEY);") + # Verify error message mentions invalid primary key type + assert "primary key" in str(excinfo.value).lower() + + conn.close() + db.close() + + +def test_list_wal_replay(tmp_path): + """List properties survive WAL replay after DB restart.""" + db_dir = tmp_path / "list_wal" + shutil.rmtree(db_dir, ignore_errors=True) + db_dir.mkdir() + + db = Database(db_path=str(db_dir), mode="w") + conn = db.connect() + + conn.execute( + "CREATE NODE TABLE ITEM(" + "id INT64 PRIMARY KEY, " + "tags STRING[], " + "scores DOUBLE[], " + "matrix INT64[][]);" + ) + conn.execute( + "CREATE (i: ITEM {id: 1, tags: ['a', 'b'], " + "scores: [1.5, 2.5], matrix: [[10, 20], [30]]});" + ) + conn.execute("CREATE (i: ITEM {id: 2, tags: ['c'], scores: [3.0], matrix: [[1]]});") + + conn.close() + db.close() + + db = Database(db_path=str(db_dir), mode="r") + conn = db.connect() + + result = list( + conn.execute( + "MATCH (i: ITEM) RETURN i.id, i.tags, i.scores, i.matrix ORDER BY i.id;" + ) + ) + assert result[0] == [1, ["a", "b"], [1.5, 2.5], [[10, 20], [30]]] + assert result[1] == [2, ["c"], [3.0], [[1]]] + + conn.close() + db.close() diff --git a/tools/python_bind/tests/test_ddl.py b/tools/python_bind/tests/test_ddl.py index 947f741f5..296c3acb3 100644 --- a/tools/python_bind/tests/test_ddl.py +++ b/tools/python_bind/tests/test_ddl.py @@ -560,3 +560,135 @@ def test_create_rel_table_with_options(tmp_path): # todo: check options in graph schema conn.close() db.close() + + +def test_list_type(): + db_dir = "/tmp/test_list_type" + shutil.rmtree(db_dir, ignore_errors=True) + db = Database(db_dir, "w") + conn = db.connect() + res = conn.execute("Return ['tag1', 'tag2'];") + assert list(res) == [[["tag1", "tag2"]]] + conn.execute( + "CREATE NODE TABLE TestNode(id INT64, tags STRING[], PRIMARY KEY(id));" + ) + conn.execute("CREATE (:TestNode {id: 1, tags: ['tag1', 'tag2']});") + res = conn.execute("Match (n:TestNode) Return n.tags;") + assert list(res) == [[["tag1", "tag2"]]] + conn.close() + db.close() + + +def test_nested_list_type(): + """Test creating and querying tables with various list element types.""" + db_dir = "/tmp/test_nested_list_type" + shutil.rmtree(db_dir, ignore_errors=True) + db = Database(db_dir, "w") + conn = db.connect() + + # INT64 list + conn.execute("CREATE NODE TABLE IntListNode(id INT64 PRIMARY KEY, values INT64[]);") + conn.execute("CREATE (:IntListNode {id: 1, values: [10, 20, 30]});") + res = conn.execute("MATCH (n:IntListNode) RETURN n.values;") + assert list(res) == [[[10, 20, 30]]] + + # DOUBLE list + conn.execute( + "CREATE NODE TABLE DoubleListNode(id INT64 PRIMARY KEY, scores DOUBLE[]);" + ) + conn.execute("CREATE (:DoubleListNode {id: 1, scores: [1.5, 2.5, 3.5]});") + res = conn.execute("MATCH (n:DoubleListNode) RETURN n.scores;") + assert list(res) == [[[1.5, 2.5, 3.5]]] + + # Multiple list columns on the same table + conn.execute( + "CREATE NODE TABLE MultiListNode(" + "id INT64 PRIMARY KEY, " + "names STRING[], " + "ages INT64[], " + "weights DOUBLE[]);" + ) + conn.execute( + "CREATE (:MultiListNode {id: 1, " + "names: ['Alice', 'Bob'], " + "ages: [30, 25], " + "weights: [55.5, 70.2]});" + ) + res = conn.execute("MATCH (n:MultiListNode) RETURN n.names, n.ages, n.weights;") + assert list(res) == [[["Alice", "Bob"], [30, 25], [55.5, 70.2]]] + + # Verify multiple rows + conn.execute("CREATE (:IntListNode {id: 2, values: [40, 50]});") + res = conn.execute("MATCH (n:IntListNode) RETURN n.values ORDER BY n.id;") + assert list(res) == [[[10, 20, 30]], [[40, 50]]] + + conn.close() + db.close() + + # Reopen and verify persistence + db2 = Database(db_dir, "r") + conn2 = db2.connect() + res2 = conn2.execute("MATCH (n:MultiListNode) RETURN n.names, n.ages, n.weights;") + assert list(res2) == [[["Alice", "Bob"], [30, 25], [55.5, 70.2]]] + conn2.close() + db2.close() + + +def test_list_type_with_default_values(): + """Test that list columns with DEFAULT values work correctly.""" + db_dir = "/tmp/test_list_type_default" + shutil.rmtree(db_dir, ignore_errors=True) + db = Database(db_dir, "w") + conn = db.connect() + + # Create table with a list column that has a default value + conn.execute( + "CREATE NODE TABLE TaggedNode(" + "id INT64 PRIMARY KEY, " + "tags STRING[] DEFAULT ['default_tag']);" + ) + + # Insert without specifying tags — should use default value + conn.execute("CREATE (:TaggedNode {id: 1});") + + # Insert with explicit tags — should override default + conn.execute("CREATE (:TaggedNode {id: 2, tags: ['custom1', 'custom2']});") + + # Insert with a different explicit value + conn.execute("CREATE (:TaggedNode {id: 3, tags: ['override']});") + + res = conn.execute("MATCH (n:TaggedNode) RETURN n.id, n.tags ORDER BY n.id;") + rows = list(res) + assert rows[0] == [1, ["default_tag"]], f"Expected default value, got {rows[0]}" + assert rows[1] == [ + 2, + ["custom1", "custom2"], + ], f"Expected custom values, got {rows[1]}" + assert rows[2] == [3, ["override"]], f"Expected override value, got {rows[2]}" + + # Also test INT64 list default + conn.execute( + "CREATE NODE TABLE ScoredNode(" + "id INT64 PRIMARY KEY, " + "scores INT64[] DEFAULT [0, 0, 0]);" + ) + conn.execute("CREATE (:ScoredNode {id: 1});") + conn.execute("CREATE (:ScoredNode {id: 2, scores: [100, 200]});") + res = conn.execute("MATCH (n:ScoredNode) RETURN n.id, n.scores ORDER BY n.id;") + rows = list(res) + assert rows[0] == [1, [0, 0, 0]], f"Expected default scores, got {rows[0]}" + assert rows[1] == [2, [100, 200]], f"Expected custom scores, got {rows[1]}" + + conn.close() + db.close() + + # Reopen and verify persistence of defaults + db2 = Database(db_dir, "r") + conn2 = db2.connect() + res2 = conn2.execute("MATCH (n:TaggedNode) RETURN n.id, n.tags ORDER BY n.id;") + rows2 = list(res2) + assert rows2[0] == [1, ["default_tag"]] + assert rows2[1] == [2, ["custom1", "custom2"]] + assert rows2[2] == [3, ["override"]] + conn2.close() + db2.close() diff --git a/tools/python_bind/tests/test_load.py b/tools/python_bind/tests/test_load.py index 973e7d512..71ace5fa8 100644 --- a/tools/python_bind/tests/test_load.py +++ b/tools/python_bind/tests/test_load.py @@ -2237,3 +2237,217 @@ def test_copy_from_csv_edge_no_header_positional(self): res = self.conn.execute("MATCH ()-[r:Knows]->() RETURN count(r);") count = next(res)[0] assert count == 1, f"Expected 1 edge, got {count}" + + def test_copy_from_node_with_int64_list_property(self): + """COPY FROM CSV into a node table with INT64[] list property.""" + csv_path = self.tmp_path / "persons_with_scores.csv" + csv_path.write_text( + "id|name|s1|s2|s3\n" + "1|Alice|10|20|30\n" + "2|Bob|40|50|60\n" + "3|Carol|70|80|90\n", + encoding="utf-8", + ) + + self.conn.execute( + "CREATE NODE TABLE PersonScores(" + "id INT64 PRIMARY KEY, name STRING, scores INT64[]);" + ) + self.conn.execute( + f"COPY PersonScores FROM (" + f' LOAD FROM "{csv_path}" (header=true, delim="|")' + f' RETURN CAST(id, "INT64") as id, name,' + f' [CAST(s1, "INT64"), CAST(s2, "INT64"), CAST(s3, "INT64")] as scores' + f")" + ) + + res = self.conn.execute( + "MATCH (p:PersonScores) RETURN p.id, p.name, p.scores ORDER BY p.id;" + ) + rows = list(res) + assert len(rows) == 3 + assert rows[0] == [1, "Alice", [10, 20, 30]] + assert rows[1] == [2, "Bob", [40, 50, 60]] + assert rows[2] == [3, "Carol", [70, 80, 90]] + + def test_copy_from_node_with_string_list_property(self): + """COPY FROM CSV into a node table with STRING[] list property.""" + csv_path = self.tmp_path / "persons_with_tags.csv" + csv_path.write_text( + "id|tag1|tag2\n" "1|engineer|python\n" "2|designer|figma\n", + encoding="utf-8", + ) + + self.conn.execute( + "CREATE NODE TABLE PersonTags(" "id INT64 PRIMARY KEY, tags STRING[]);" + ) + self.conn.execute( + f"COPY PersonTags FROM (" + f' LOAD FROM "{csv_path}" (header=true, delim="|")' + f' RETURN CAST(id, "INT64") as id, [tag1, tag2] as tags' + f")" + ) + + res = self.conn.execute( + "MATCH (p:PersonTags) RETURN p.id, p.tags ORDER BY p.id;" + ) + rows = list(res) + assert len(rows) == 2 + assert rows[0] == [1, ["engineer", "python"]] + assert rows[1] == [2, ["designer", "figma"]] + + def test_copy_from_node_with_multiple_list_properties(self): + """COPY FROM CSV into a node table with both INT64[] and STRING[] properties.""" + csv_path = self.tmp_path / "persons_multi_list.csv" + csv_path.write_text( + "id|name|s1|s2|t1|t2\n" "1|Alice|10|20|math|sci\n" "2|Bob|30|40|eng|art\n", + encoding="utf-8", + ) + + self.conn.execute( + "CREATE NODE TABLE PersonMultiList(" + "id INT64 PRIMARY KEY, name STRING, scores INT64[], subjects STRING[]);" + ) + self.conn.execute( + f"COPY PersonMultiList FROM (" + f' LOAD FROM "{csv_path}" (header=true, delim="|")' + f' RETURN CAST(id, "INT64") as id, name,' + f' [CAST(s1, "INT64"), CAST(s2, "INT64")] as scores,' + f" [t1, t2] as subjects" + f")" + ) + + res = self.conn.execute( + "MATCH (p:PersonMultiList) RETURN p.id, p.name, p.scores, p.subjects ORDER BY p.id;" + ) + rows = list(res) + assert len(rows) == 2 + assert rows[0] == [1, "Alice", [10, 20], ["math", "sci"]] + assert rows[1] == [2, "Bob", [30, 40], ["eng", "art"]] + + def test_copy_from_node_with_double_list_property(self): + """COPY FROM CSV into a node table with DOUBLE[] list property.""" + csv_path = self.tmp_path / "persons_with_weights.csv" + csv_path.write_text( + "id|w1|w2|w3\n" "1|1.1|2.2|3.3\n" "2|4.4|5.5|6.6\n", + encoding="utf-8", + ) + + self.conn.execute( + "CREATE NODE TABLE PersonWeights(" + "id INT64 PRIMARY KEY, weights DOUBLE[]);" + ) + self.conn.execute( + f"COPY PersonWeights FROM (" + f' LOAD FROM "{csv_path}" (header=true, delim="|")' + f' RETURN CAST(id, "INT64") as id,' + f' [CAST(w1, "DOUBLE"), CAST(w2, "DOUBLE"), CAST(w3, "DOUBLE")] as weights' + f")" + ) + + res = self.conn.execute( + "MATCH (p:PersonWeights) RETURN p.id, p.weights ORDER BY p.id;" + ) + rows = list(res) + assert len(rows) == 2 + assert rows[0][0] == 1 + assert len(rows[0][1]) == 3 + assert abs(rows[0][1][0] - 1.1) < 1e-9 + assert abs(rows[0][1][1] - 2.2) < 1e-9 + assert abs(rows[0][1][2] - 3.3) < 1e-9 + + def test_copy_from_edge_with_list_property(self): + """COPY FROM CSV into an edge table with INT64[] list property.""" + node_csv = self.tmp_path / "nodes_for_edge_list.csv" + node_csv.write_text("id|name\n1|Alice\n2|Bob\n3|Carol\n", encoding="utf-8") + edge_csv = self.tmp_path / "edges_with_list.csv" + edge_csv.write_text( + "src|dst|w1|w2\n" "1|2|10|20\n" "2|3|30|40\n" "1|3|50|60\n", + encoding="utf-8", + ) + + self.conn.execute( + "CREATE NODE TABLE PersonEL(id INT64 PRIMARY KEY, name STRING);" + ) + self.conn.execute( + "CREATE REL TABLE KnowsEL(FROM PersonEL TO PersonEL, weights INT64[]);" + ) + self.conn.execute(f'COPY PersonEL FROM "{node_csv}" (header=true, delim="|");') + self.conn.execute( + f"COPY KnowsEL FROM (" + f' LOAD FROM "{edge_csv}" (header=true, delim="|")' + f' RETURN CAST(src, "INT64") as src, CAST(dst, "INT64") as dst,' + f' [CAST(w1, "INT64"), CAST(w2, "INT64")] as weights' + f")" + ) + + res = self.conn.execute( + "MATCH (a:PersonEL)-[r:KnowsEL]->(b:PersonEL) " + "RETURN a.id, b.id, r.weights ORDER BY a.id, b.id;" + ) + rows = list(res) + assert len(rows) == 3 + assert rows[0] == [1, 2, [10, 20]] + assert rows[1] == [1, 3, [50, 60]] + assert rows[2] == [2, 3, [30, 40]] + + def test_copy_from_edge_with_string_list_property(self): + """COPY FROM CSV into an edge table with STRING[] list property.""" + node_csv = self.tmp_path / "nodes_for_str_edge.csv" + node_csv.write_text("id|name\n1|Alice\n2|Bob\n", encoding="utf-8") + edge_csv = self.tmp_path / "edges_str_list.csv" + edge_csv.write_text( + "src|dst|label1|label2\n" "1|2|friend|colleague\n", + encoding="utf-8", + ) + + self.conn.execute( + "CREATE NODE TABLE PersonSEL(id INT64 PRIMARY KEY, name STRING);" + ) + self.conn.execute( + "CREATE REL TABLE KnowsSEL(FROM PersonSEL TO PersonSEL, labels STRING[]);" + ) + self.conn.execute(f'COPY PersonSEL FROM "{node_csv}" (header=true, delim="|");') + self.conn.execute( + f"COPY KnowsSEL FROM (" + f' LOAD FROM "{edge_csv}" (header=true, delim="|")' + f' RETURN CAST(src, "INT64") as src, CAST(dst, "INT64") as dst,' + f" [label1, label2] as labels" + f")" + ) + + res = self.conn.execute( + "MATCH (a:PersonSEL)-[r:KnowsSEL]->(b:PersonSEL) " + "RETURN a.name, b.name, r.labels;" + ) + rows = list(res) + assert len(rows) == 1 + assert rows[0] == ["Alice", "Bob", ["friend", "colleague"]] + + def test_copy_from_node_with_nested_list_property(self): + """COPY FROM CSV into a node table with INT64[][] nested list property.""" + csv_path = self.tmp_path / "persons_nested_list.csv" + csv_path.write_text( + "id|a1|a2|b1|b2\n" "1|1|2|3|4\n" "2|5|6|7|8\n", + encoding="utf-8", + ) + + self.conn.execute( + "CREATE NODE TABLE PersonNested(" "id INT64 PRIMARY KEY, matrix INT64[][]);" + ) + self.conn.execute( + f"COPY PersonNested FROM (" + f' LOAD FROM "{csv_path}" (header=true, delim="|")' + f' RETURN CAST(id, "INT64") as id,' + f' [[CAST(a1, "INT64"), CAST(a2, "INT64")],' + f' [CAST(b1, "INT64"), CAST(b2, "INT64")]] as matrix' + f")" + ) + + res = self.conn.execute( + "MATCH (p:PersonNested) RETURN p.id, p.matrix ORDER BY p.id;" + ) + rows = list(res) + assert len(rows) == 2 + assert rows[0] == [1, [[1, 2], [3, 4]]] + assert rows[1] == [2, [[5, 6], [7, 8]]]