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]]]