diff --git a/AGENTS.md b/AGENTS.md
index 6ef00c9f0..547bacc88 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -98,6 +98,11 @@ tools/python_bind/
 Cypher → ANTLR Parser → Binder → Logical Plan → gopt Converter → Physical Plan → Execution
 ```
 
+## Known Limitations
+
+- **List literals require explicit CAST**: In `CREATE`, `SET`, and `MERGE` clauses, list values must be wrapped with `CAST(..., 'TYPE[]')` — bare list literals like `[1, 2, 3]` are rejected. Use `CAST([1, 2, 3], 'INT64[]')` instead.
+- **List types cannot be primary keys**: Declaring a `PRIMARY KEY` on a `T[]` column is rejected.
+
 ## 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 20e0db872..bc35e08d6 100644
--- a/doc/source/cypher_manual/ddl_clause.md
+++ b/doc/source/cypher_manual/ddl_clause.md
@@ -17,6 +17,9 @@ The following table lists the recommended syntax for defining default values for
 | `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')`         |
 | `ARRAY`           | `prop INT32[3] DEFAULT [1, 2, 3]` | child defaults repeated to the fixed length, for example `[0, 0, 0]` for `INT32[3]` |
+| `LIST`             | `prop INT64[] DEFAULT [1, 2, 3]`  | `[]` (empty list)                  |
+
+List types (`T[]`) are variable-length and can hold any number of elements, including zero. Array types (`T[N]`) have a fixed length `N` that must be a positive integer.
 
 Please refer to the following examples for more usages.
 
@@ -124,6 +127,54 @@ CREATE NODE TABLE Matrix(
 
 If an array property is omitted during `CREATE`, or explicitly set to `NULL` during `CREATE`, NeuG stores the declared default for that array. If no explicit default is declared, the system default repeats the child type's default value; for `INT32[3]`, that default is `[0, 0, 0]`. Setting an existing array property to `NULL` with `SET` is not supported yet.
 
+## List Properties
+
+Use `T[]` to declare a variable-length list property, where `T` is the child type. Unlike fixed-size arrays (`T[N]`), the number of elements is not constrained at the schema level.
+
+```cypher
+CREATE NODE TABLE Person(
+    id INT64,
+    tags STRING[],
+    scores INT64[],
+    PRIMARY KEY(id)
+);
+
+CREATE REL TABLE Knows(
+    FROM Person TO Person,
+    ratings DOUBLE[]
+);
+```
+
+Nested lists are declared by chaining `[]`. `STRING[][]` is a list of lists of strings; `STRING[][2][]` is a list of fixed-size-2 arrays of variable-length string lists:
+
+```cypher
+CREATE NODE TABLE Matrix(
+    id INT64,
+    grid INT64[][],
+    PRIMARY KEY(id)
+);
+```
+
+List values must be explicitly cast in `CREATE`, `SET`, and `MERGE` clauses — bare list literals are rejected:
+
+```cypher
+-- Correct: explicit CAST
+CREATE (p:Person {id: 1, tags: CAST(['a', 'b'], 'STRING[]')});
+
+-- Error: bare list literal is not accepted
+CREATE (p:Person {id: 2, tags: ['a', 'b']});
+```
+
+List properties are supported in CSV and JSON bulk loading via `COPY FROM`. List values in CSV use bracket syntax `[1, 2, 3]`, and nested lists nest the brackets:
+
+```
+id|tags
+1|[a,b,c]
+2|[]
+```
+
+For more details on array vs list type distinctions, refer to the [Array Properties](#array-properties) section above.
+
 ## Drop Node Type
 
 Delete a specified Node type. Use IF EXISTS to avoid errors when the type doesn't exist.
diff --git a/include/neug/storages/README.md b/include/neug/storages/README.md
index 7de825533..04640de2e 100644
--- a/include/neug/storages/README.md
+++ b/include/neug/storages/README.md
@@ -84,6 +84,57 @@ 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
+
+Variable-length list properties (`T[]`) are stored using [ListPropertyColumn](../utils/property/list_property_column.h),
+which decomposes each list into three separate sub-columns:
+
+- **offsets** (`ULongColumn`): `offsets_[i]` is the starting index of row `i`'s elements within the elements column.
+- **lengths** (`ULongColumn`): `lengths_[i]` is the number of elements in row `i`'s list.
+- **elements** (`ColumnBase`): a contiguous column holding all list elements across all rows.
+
+```
+Row 0:  offsets_[0] = 0,  lengths_[0] = 3   →  elements_[0], elements_[1], elements_[2]
+Row 1:  offsets_[1] = 3,  lengths_[1] = 0   →  (empty list)
+Row 2:  offsets_[2] = 3,  lengths_[2] = 2   →  elements_[3], elements_[4]
+```
+
+This design allows each sub-column to use the most efficient storage format for its own type
+(e.g., `TypedColumn<T>` for POD elements, `StringColumn` for string elements), and enables
+sequential access patterns during checkpoint dump.
+
+### 5.1 Write Path
+
+When setting a list value at row `i` (`set_any`):
+
+1. If the new element count equals the existing `lengths_[i]`, the elements are overwritten
+   in place at `offsets_[i]` — no offset or length update needed.
+2. If the element count differs, the new elements are **appended** to the tail of the elements
+   column, and `offsets_[i]` / `lengths_[i]` are updated to point to the new region. The old
+   elements become dead space and are reclaimed during checkpoint dump.
+
+### 5.2 Checkpoint Dump (Compaction)
+
+During `Dump`, the column is compacted: all live elements are written contiguously into a new
+compact elements column, eliminating dead space from in-place updates that changed list lengths.
+The offsets and lengths columns are rewritten to reflect the compacted layout.
+
+### 5.3 Nested Lists
+
+For nested list types (e.g., `STRING[][]`), the `elements` sub-column is itself a
+`ListPropertyColumn`, creating a recursive storage structure. Each level of nesting adds another
+layer of offset/length/elements decomposition.
+
+For list-of-array types (e.g., `STRING[][2][]`), the elements column is an `ArrayColumn`,
+which stores fixed-size arrays inline.
+
+### 5.4 Thread Safety
+
+The `insert_safe` parameter controls resize behavior, inherited from the `ColumnBase` interface:
+- When `false`, throws on insufficient space in the elements column.
+- When `true`, the elements column is resized as needed; the caller must provide external
+  synchronization during resize.
+
 
 ## 6. Durability
 
diff --git a/include/neug/utils/property/list_property_column.h b/include/neug/utils/property/list_property_column.h
new file mode 100644
index 000000000..3c38d4117
--- /dev/null
+++ b/include/neug/utils/property/list_property_column.h
@@ -0,0 +1,83 @@
+/** 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 <memory>
+#include <string>
+
+#include "neug/common/types.h"
+#include "neug/common/types/value.h"
+#include "neug/utils/property/column.h"
+
+namespace neug {
+
+class ListPropertyColumn : public ColumnBase {
+ public:
+  ListPropertyColumn() : size_(0) {}
+  explicit ListPropertyColumn(const DataType& list_type);
+  ~ListPropertyColumn() override = default;
+
+  void Open(Checkpoint& ckp, const ModuleDescriptor& desc,
+            MemoryLevel level) override;
+  void Open(Checkpoint& ckp, const CheckpointManifest& manifest,
+            const ModuleDescriptor& desc, MemoryLevel level) override;
+
+  void Dump(Checkpoint& ckp, CheckpointManifest& meta,
+            const std::string& key) override;
+
+  size_t size() const override { return size_; }
+  void resize(size_t size) override;
+  void resize(size_t size, const Value& default_value) override;
+
+  DataTypeId type() const override { return DataTypeId::kList; }
+  void set_any(size_t index, const Value& value, bool insert_safe) override;
+  Value get_any(size_t index) const override;
+
+  std::unique_ptr<Module> Clone() const override;
+  void Detach(Checkpoint& ckp, MemoryLevel level) override;
+
+  std::string ModuleTypeName() const override { return type_name(); }
+  static std::string type_name() { return "column<list>"; }
+
+  const DataType& list_type() const { return list_type_; }
+  const DataType& child_type() const { return child_type_; }
+
+ private:
+  void openInternal(Checkpoint& ckp, const CheckpointManifest* manifest,
+                    const ModuleDescriptor& desc, MemoryLevel level);
+  ModuleDescriptor dumpSelfDescriptor() const;
+
+  DataType list_type_;
+  DataType child_type_;
+  size_t size_;
+  std::unique_ptr<ULongColumn> offsets_;
+  std::unique_ptr<ULongColumn> lengths_;
+  std::unique_ptr<ColumnBase> elements_;
+};
+
+class ListPropertyRefColumn : public RefColumnBase {
+ public:
+  explicit ListPropertyRefColumn(const ListPropertyColumn& column)
+      : column_(column) {}
+
+  Value get_any(size_t index) const override { return column_.get_any(index); }
+  DataTypeId type() const override { return DataTypeId::kList; }
+  ColType col_type() const override { return ColType::kInternal; }
+
+ private:
+  const ListPropertyColumn& column_;
+};
+
+}  // namespace neug
diff --git a/include/neug/utils/property/types.h b/include/neug/utils/property/types.h
index 5a9d2cbfe..1e6e291e1 100644
--- a/include/neug/utils/property/types.h
+++ b/include/neug/utils/property/types.h
@@ -646,11 +646,18 @@ struct convert<neug::DataType> {
                            : neug::STRING_DEFAULT_MAX_LENGTH;
     } else if (id == neug::DataTypeId::kDate) {
       node["temporal"]["date"] = "";
+    } else if (id == neug::DataTypeId::kTimestampMs) {
+      node["temporal"]["timestamp"] = "";
+    } else if (id == neug::DataTypeId::kInterval) {
+      node["temporal"]["interval"] = "";
     } else if (id == neug::DataTypeId::kArray) {
       auto child_type = neug::ArrayType::GetChildType(type);
       uint64_t array_size = neug::ArrayType::GetNumElements(type);
       node["array"]["component_type"] = encode(child_type);
       node["array"]["max_length"] = array_size;
+    } else if (id == neug::DataTypeId::kList) {
+      node["list"]["component_type"] =
+          encode(neug::ListType::GetChildType(type));
     } else {
       LOG(ERROR) << "Unrecognized property type: " << type.ToString();
     }
diff --git a/src/compiler/binder/expression_binder.cpp b/src/compiler/binder/expression_binder.cpp
index 2931b1187..a2f0e8a6c 100644
--- a/src/compiler/binder/expression_binder.cpp
+++ b/src/compiler/binder/expression_binder.cpp
@@ -91,7 +91,12 @@ std::shared_ptr<Expression> ExpressionBinder::bindExpression(
         "bindExpression(" + ExpressionTypeUtil::toString(expressionType) +
         ").");
   }
-  if (ConstantExpressionVisitor::needFold(*expression)) {
+  // Keep ANY-typed constants unfolded (e.g. the empty list literal `[]`):
+  // CAST binding retypes the list-creation function in place, which is only
+  // reachable while the argument is still a function expression, not a
+  // folded literal.
+  if (ConstantExpressionVisitor::needFold(*expression) &&
+      !expression->getDataType().containsAny()) {
     return foldExpression(expression);
   }
   return expression;
diff --git a/src/compiler/function/cast/cast_array.cpp b/src/compiler/function/cast/cast_array.cpp
index 787cab617..7ba194bc6 100644
--- a/src/compiler/function/cast/cast_array.cpp
+++ b/src/compiler/function/cast/cast_array.cpp
@@ -35,7 +35,10 @@ bool CastArrayHelper::checkCompatibleNestedTypes(DataTypeId sourceTypeID,
     return true;
   }
   case DataTypeId::kList:
-  case DataTypeId::kArray:
+  case DataTypeId::kArray: {
+    return targetTypeID == DataTypeId::kList ||
+           targetTypeID == DataTypeId::kArray;
+  }
   case DataTypeId::kMap:
   case DataTypeId::kStruct: {
     return sourceTypeID == targetTypeID;
@@ -62,12 +65,12 @@ const DataType& getListLikeChildType(const DataType& type) {
 
 bool CastArrayHelper::requiresArrayEntryValidation(const DataType& srcType,
                                                    const DataType& dstType) {
-  if (srcType.id() == DataTypeId::kArray &&
-      dstType.id() == DataTypeId::kArray) {
-    return true;
-  }
-
   if (checkCompatibleNestedTypes(srcType.id(), dstType.id())) {
+    if (dstType.id() == DataTypeId::kArray &&
+        (srcType.id() == DataTypeId::kList ||
+         srcType.id() == DataTypeId::kArray)) {
+      return true;
+    }
     switch (getPhysicalType(srcType.id())) {
     case PhysicalTypeID::LIST: {
       return requiresArrayEntryValidation(getListLikeChildType(srcType),
@@ -109,14 +112,19 @@ void CastArrayHelper::validateArrayEntries(ValueVector* inputVector,
 
   switch (getPhysicalType(resultType.id())) {
   case PhysicalTypeID::ARRAY: {
-    if (getPhysicalType(inputType.id()) != PhysicalTypeID::ARRAY ||
-        ArrayType::GetNumElements(inputType) !=
-            ArrayType::GetNumElements(resultType)) {
+    auto input_physical_type = getPhysicalType(inputType.id());
+    if (input_physical_type != PhysicalTypeID::ARRAY &&
+        input_physical_type != PhysicalTypeID::LIST) {
       THROW_CONVERSION_EXCEPTION(
           stringFormat("Unsupported casting function from {} to {}.",
                        inputType.ToString(), resultType.ToString()));
     }
     auto listEntry = inputVector->getValue<list_entry_t>(pos);
+    if (listEntry.size != ArrayType::GetNumElements(resultType)) {
+      THROW_CONVERSION_EXCEPTION(
+          stringFormat("Unsupported casting function from {} to {}.",
+                       inputType.ToString(), resultType.ToString()));
+    }
     auto inputChildVector = ListVector::getDataVector(inputVector);
     for (auto i = listEntry.offset; i < listEntry.offset + listEntry.size;
          i++) {
@@ -125,7 +133,9 @@ void CastArrayHelper::validateArrayEntries(ValueVector* inputVector,
     }
   } break;
   case PhysicalTypeID::LIST: {
-    if (getPhysicalType(inputType.id()) == PhysicalTypeID::LIST) {
+    auto input_physical_type = getPhysicalType(inputType.id());
+    if (input_physical_type == PhysicalTypeID::LIST ||
+        input_physical_type == PhysicalTypeID::ARRAY) {
       auto listEntry = inputVector->getValue<list_entry_t>(pos);
       auto inputChildVector = ListVector::getDataVector(inputVector);
       for (auto i = listEntry.offset; i < listEntry.offset + listEntry.size;
diff --git a/src/compiler/function/vector_cast_functions.cpp b/src/compiler/function/vector_cast_functions.cpp
index 6ed5b634b..e896b9248 100644
--- a/src/compiler/function/vector_cast_functions.cpp
+++ b/src/compiler/function/vector_cast_functions.cpp
@@ -27,12 +27,14 @@
 #include "neug/common/types/value.h"
 #include "neug/compiler/binder/expression/expression_util.h"
 #include "neug/compiler/binder/expression/literal_expression.h"
+#include "neug/compiler/binder/expression/scalar_function_expression.h"
 #include "neug/compiler/catalog/catalog.h"
 #include "neug/compiler/common/types/types.h"
 #include "neug/compiler/function/built_in_function_utils.h"
 #include "neug/compiler/function/cast/functions/cast_array.h"
 #include "neug/compiler/function/cast/functions/cast_from_string_functions.h"
 #include "neug/compiler/function/cast/functions/cast_functions.h"
+#include "neug/compiler/function/list/vector_list_functions.h"
 #include "neug/compiler/function/neug_scalar_function.h"
 #include "neug/compiler/function/scalar_function.h"
 #include "neug/compiler/main/client_context.h"
@@ -74,11 +76,6 @@ static void resolveNestedVector(std::shared_ptr<ValueVector> inputVector,
          getPhysicalType(inputType->id()) == PhysicalTypeID::ARRAY) &&
         (getPhysicalType(resultType->id()) == PhysicalTypeID::LIST ||
          getPhysicalType(resultType->id()) == PhysicalTypeID::ARRAY)) {
-      if (inputType->id() != resultType->id()) {
-        THROW_CONVERSION_EXCEPTION(
-            stringFormat("Unsupported casting function from {} to {}.",
-                         inputType->ToString(), resultType->ToString()));
-      }
       // copy data and nullmask from input
       memcpy(resultVector->getData(), inputVector->getData(),
              numOfEntries * resultVector->getNumBytesPerValue());
@@ -691,6 +688,21 @@ static std::unique_ptr<FunctionBindData> castBindFunc(
       targetType.id() != DataTypeId::kStruct) {  // No need to cast.
     return nullptr;
   }
+  if (input.arguments[0]->expressionType == ExpressionType::FUNCTION) {
+    auto source = input.arguments[0]->ptrCast<ScalarFunctionExpression>();
+    if (source->getFunction().name == ListCreationFunction::name &&
+        source->getNumChildren() == 0) {
+      if (targetType.id() == DataTypeId::kArray) {
+        THROW_CONVERSION_EXCEPTION("ARRAY value length mismatch for type " +
+                                   targetType.ToString());
+      }
+      if (targetType.id() == DataTypeId::kList) {
+        source->dataType = targetType.copy();
+        source->getBindData()->resultType = targetType.copy();
+        return nullptr;
+      }
+    }
+  }
   if (ExpressionUtil::canCastStatically(*input.arguments[0], targetType) &&
       targetType.id() != DataTypeId::kStruct) {
     input.arguments[0]->cast(targetType);
@@ -709,39 +721,76 @@ static std::unique_ptr<FunctionBindData> castBindFunc(
   return bindData;
 }
 
-static neug::Value castFunc(const std::vector<neug::Value>& args) {
-  if (args.size() != 2) {
-    THROW_RUNTIME_ERROR("CAST(VAL, TYPE): expect exactly 2 argument, got " +
-                        std::to_string(args.size()));
+static Value castValue(const Value& input, const DataType& targetType) {
+  if (input.IsNull()) {
+    return Value(targetType);
+  }
+  if (input.type() == targetType) {
+    return input;
+  }
+  if (targetType.id() == DataTypeId::kList ||
+      targetType.id() == DataTypeId::kArray) {
+    const std::vector<Value>* sourceChildren = nullptr;
+    if (input.type().id() == DataTypeId::kList) {
+      sourceChildren = &ListValue::GetChildren(input);
+    } else if (input.type().id() == DataTypeId::kArray) {
+      sourceChildren = &ArrayValue::GetChildren(input);
+    } else {
+      THROW_CONVERSION_EXCEPTION("Unsupported casting function from " +
+                                 input.type().ToString() + " to " +
+                                 targetType.ToString());
+    }
+    if (targetType.id() == DataTypeId::kArray &&
+        sourceChildren->size() != ArrayType::GetNumElements(targetType)) {
+      THROW_CONVERSION_EXCEPTION("ARRAY value length mismatch for type " +
+                                 targetType.ToString());
+    }
+    const auto& childType = targetType.id() == DataTypeId::kList
+                                ? ListType::GetChildType(targetType)
+                                : ArrayType::GetChildType(targetType);
+    std::vector<Value> children;
+    children.reserve(sourceChildren->size());
+    for (const auto& child : *sourceChildren) {
+      children.push_back(castValue(child, childType));
+    }
+    if (targetType.id() == DataTypeId::kList) {
+      return Value::LIST(childType, std::move(children));
+    }
+    return Value::ARRAY(targetType, std::move(children));
   }
-  const auto& arg0 = args[0];
-  const auto& arg1 = args[1];
-  auto type = StringValue::Get(arg1);
-  auto targetType = common::convertFromString(std::string(type), nullptr);
   switch (targetType.id()) {
   case DataTypeId::kInt64:
-    return performCast<int64_t>(arg0);
+    return performCast<int64_t>(input);
   case DataTypeId::kInt32:
-    return performCast<int32_t>(arg0);
+    return performCast<int32_t>(input);
   case DataTypeId::kFloat:
-    return performCast<float>(arg0);
+    return performCast<float>(input);
   case DataTypeId::kDouble:
-    return performCast<double>(arg0);
+    return performCast<double>(input);
   case DataTypeId::kVarchar:
-    return performCastToString(arg0);
+    return performCastToString(input);
   case DataTypeId::kDate:
-    return performCast<date_t>(arg0);
+    return performCast<date_t>(input);
   case DataTypeId::kTimestampMs:
-    return performCast<timestamp_ms_t>(arg0);
+    return performCast<timestamp_ms_t>(input);
   case DataTypeId::kUInt32:
-    return performCast<uint32_t>(arg0);
+    return performCast<uint32_t>(input);
   case DataTypeId::kUInt64:
-    return performCast<uint64_t>(arg0);
+    return performCast<uint64_t>(input);
   default:
     THROW_RUNTIME_ERROR(std::string("Unsupported target type for CAST: ") +
-                        std::string(type));
+                        targetType.ToString());
+  }
+}
+
+static neug::Value castFunc(const std::vector<neug::Value>& args) {
+  if (args.size() != 2) {
+    THROW_RUNTIME_ERROR("CAST(VAL, TYPE): expect exactly 2 argument, got " +
+                        std::to_string(args.size()));
   }
-  return neug::Value(DataType::SQLNULL);
+  auto type = StringValue::Get(args[1]);
+  auto targetType = common::convertFromString(std::string(type), nullptr);
+  return castValue(args[0], targetType);
 }
 
 function_set CastAnyFunction::getFunctionSet() {
diff --git a/src/storages/graph/edge_table.cc b/src/storages/graph/edge_table.cc
index c7d3c9bc3..547688200 100644
--- a/src/storages/graph/edge_table.cc
+++ b/src/storages/graph/edge_table.cc
@@ -275,6 +275,14 @@ void insert_varchar_impl(const TypedColumnInserter& ins, size_t dst_idx,
                      insert_safe);
 }
 
+void insert_nested_impl(const TypedColumnInserter& ins, size_t dst_idx,
+                        size_t src_idx, bool insert_safe) {
+  auto value = ins.src->get_elem(src_idx);
+  if (!value.IsNull()) {
+    ins.dst->set_any(dst_idx, value, insert_safe);
+  }
+}
+
 TypedColumnInserter make_inserter(const DataType& type,
                                   const IContextColumn* src, ColumnBase* dst) {
   switch (type.id()) {
@@ -285,6 +293,9 @@ TypedColumnInserter make_inserter(const DataType& type,
 #undef MAKE_INSERTER
   case DataTypeId::kVarchar:
     return {src, dst, &insert_varchar_impl};
+  case DataTypeId::kArray:
+  case DataTypeId::kList:
+    return {src, dst, &insert_nested_impl};
   default:
     THROW_NOT_SUPPORTED_EXCEPTION(
         "Unsupported data type for column inserter: " +
diff --git a/src/storages/loader/loader_utils.cc b/src/storages/loader/loader_utils.cc
index 6636f6105..922e375f6 100644
--- a/src/storages/loader/loader_utils.cc
+++ b/src/storages/loader/loader_utils.cc
@@ -395,8 +395,25 @@ Value parse_array_element(std::string_view token, const DataType& data_type,
     }
     return Value::ARRAY(data_type, std::move(values));
   }
+  case DataTypeId::kList: {
+    auto trimmed = trim_array_token(token);
+    if (trimmed.size() < 2 || trimmed.front() != '[' || trimmed.back() != ']') {
+      THROW_CONVERSION_EXCEPTION("Expected list value for type " +
+                                 data_type.ToString() + ": " +
+                                 std::string(trimmed));
+    }
+    const auto& child_type = ListType::GetChildType(data_type);
+    auto elements = split_array_elements(trimmed.substr(1, trimmed.size() - 2));
+    std::vector<Value> values;
+    values.reserve(elements.size());
+    for (const auto& element : elements) {
+      values.push_back(
+          parse_array_element(element, child_type, true_values, false_values));
+    }
+    return Value::LIST(child_type, std::move(values));
+  }
   default:
-    THROW_NOT_SUPPORTED_EXCEPTION("Unsupported ARRAY element type: " +
+    THROW_NOT_SUPPORTED_EXCEPTION("Unsupported nested element type: " +
                                   data_type.ToString());
   }
 }
@@ -502,6 +519,7 @@ FieldAppender make_appender(const DataType& type, void* builder,
                             const std::string* column_name) {
   switch (type.id()) {
   case DataTypeId::kArray:
+  case DataTypeId::kList:
     return {builder, column_name, &type, &append_array_impl};
 #define MAKE_APPENDER(enum_val, cpp_type) \
   case DataTypeId::enum_val:              \
@@ -1546,7 +1564,8 @@ void set_properties_from_context_column(
                                                                 vids);
     break;
   }
-  case DataTypeId::kArray: {
+  case DataTypeId::kArray:
+  case DataTypeId::kList: {
     for (size_t k = 0; k < vids.size(); ++k) {
       if (vids[k] >= std::numeric_limits<vid_t>::max()) {
         continue;
diff --git a/src/utils/io/read/common/type_converter.cc b/src/utils/io/read/common/type_converter.cc
index 40f8fe4c5..b26f7528b 100644
--- a/src/utils/io/read/common/type_converter.cc
+++ b/src/utils/io/read/common/type_converter.cc
@@ -76,6 +76,8 @@ DataType NeuGTypeConverter::convert(const ::common::DataType& type) const {
     }
     return DataType::Array(childType, fixed_length);
   }
+  case ::common::DataType::kList:
+    return DataType::List(convert(type.list().component_type()));
   default:
     THROW_CONVERSION_EXCEPTION("Unsupported DataType for NeuG conversion");
   }
@@ -146,6 +148,13 @@ std::shared_ptr<::common::DataType> NeuGTypeConverter::inferCommonType(
     commonType->set_allocated_array(array_msg.release());
     break;
   }
+  case DataTypeId::kList: {
+    auto list_msg = std::make_unique<::common::List>();
+    auto child_common = inferCommonType(ListType::GetChildType(type));
+    *list_msg->mutable_component_type() = *child_common;
+    commonType->set_allocated_list(list_msg.release());
+    break;
+  }
   default:
     THROW_CONVERSION_EXCEPTION(
         "Unsupported NeuG DataType for common conversion");
diff --git a/src/utils/io/read/json/json_reader.cc b/src/utils/io/read/json/json_reader.cc
index fa7d0f933..4451e27f9 100644
--- a/src/utils/io/read/json/json_reader.cc
+++ b/src/utils/io/read/json/json_reader.cc
@@ -37,6 +37,7 @@
 #include "neug/utils/io/read/common/row_expression_filter.h"
 #include "neug/utils/io/read/common/schema.h"
 #include "neug/utils/io/read/common/type_converter.h"
+#include "neug/utils/property/default_value.h"
 #include "neug/utils/result.h"
 #include "neug/utils/service_utils.h"
 
@@ -164,10 +165,33 @@ Value parse_json_value(const rapidjson::Value& value,
     std::vector<Value> values;
     values.reserve(value.Size());
     for (const auto& item : value.GetArray()) {
-      values.push_back(parse_json_value(item, child_type));
+      // A nested JSON null normalizes to the declared child default,
+      // matching property-column write semantics.
+      if (item.IsNull()) {
+        values.push_back(get_default_value(child_type));
+      } else {
+        values.push_back(parse_json_value(item, child_type));
+      }
     }
     return Value::ARRAY(data_type, std::move(values));
   }
+  case DataTypeId::kList: {
+    if (!value.IsArray()) {
+      THROW_CONVERSION_EXCEPTION("Expected JSON array for LIST type: " +
+                                 data_type.ToString());
+    }
+    const auto& child_type = ListType::GetChildType(data_type);
+    std::vector<Value> values;
+    values.reserve(value.Size());
+    for (const auto& item : value.GetArray()) {
+      if (item.IsNull()) {
+        values.push_back(get_default_value(child_type));
+      } else {
+        values.push_back(parse_json_value(item, child_type));
+      }
+    }
+    return Value::LIST(child_type, std::move(values));
+  }
   default:
     if (value.IsString()) {
       return Value::STRING(value.GetString());
@@ -252,8 +276,13 @@ class JsonChunkSupplier : public IDataChunkSupplier {
               "Column '" + name +
               "' not found in JSON object in file: " + file_path_);
         }
-        builders[col]->push_back_elem(
-            parse_json_value(obj[name.c_str()], selected_types[col]));
+        const auto& json_value = obj[name.c_str()];
+        if (json_value.IsNull()) {
+          builders[col]->push_back_null();
+        } else {
+          builders[col]->push_back_elem(
+              parse_json_value(json_value, selected_types[col]));
+        }
       }
       ++rows_in_chunk;
     }
diff --git a/src/utils/pb_utils.cc b/src/utils/pb_utils.cc
index 4063fb15c..e7137696b 100644
--- a/src/utils/pb_utils.cc
+++ b/src/utils/pb_utils.cc
@@ -248,6 +248,16 @@ bool data_type_to_property_type(const common::DataType& data_type,
     out_type = DataType::Array(child_type, fixed_length);
     return true;
   }
+  case common::DataType::kList: {
+    DataType child_type;
+    if (!data_type_to_property_type(data_type.list().component_type(),
+                                    child_type)) {
+      LOG(ERROR) << "Failed to parse list component type";
+      return false;
+    }
+    out_type = DataType::List(child_type);
+    return true;
+  }
   case common::DataType::kMap: {
     LOG(ERROR) << "Map type is not supported";
     return false;
diff --git a/src/utils/property/column.cc b/src/utils/property/column.cc
index 238161b91..fabfd07d7 100644
--- a/src/utils/property/column.cc
+++ b/src/utils/property/column.cc
@@ -22,6 +22,7 @@
 #include "neug/storages/module/module_factory.h"
 #include "neug/utils/id_indexer.h"
 #include "neug/utils/property/array_column.h"
+#include "neug/utils/property/list_property_column.h"
 #include "neug/utils/property/table.h"
 #include "neug/utils/property/types.h"
 #include "neug/utils/property/vec_column.h"
@@ -78,6 +79,9 @@ std::unique_ptr<ColumnBase> CreateColumn(DataType type) {
   case DataTypeId::kArray: {
     return std::make_unique<ArrayColumn>(type);
   }
+  case DataTypeId::kList: {
+    return std::make_unique<ListPropertyColumn>(type);
+  }
   case DataTypeId::kEmpty: {
     return std::make_unique<TypedColumn<EmptyType>>();
   }
@@ -108,6 +112,10 @@ std::shared_ptr<RefColumnBase> CreateRefColumn(const ColumnBase& column) {
     return std::make_shared<ArrayRefColumn>(
         dynamic_cast<const ArrayColumn&>(column));
   }
+  case DataTypeId::kList: {
+    return std::make_shared<ListPropertyRefColumn>(
+        dynamic_cast<const ListPropertyColumn&>(column));
+  }
   default: {
     THROW_NOT_SUPPORTED_EXCEPTION("Unsupported type for reference column: " +
                                   std::to_string(type));
diff --git a/src/utils/property/list_property_column.cc b/src/utils/property/list_property_column.cc
new file mode 100644
index 000000000..2d82d56d4
--- /dev/null
+++ b/src/utils/property/list_property_column.cc
@@ -0,0 +1,350 @@
+/** 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/list_property_column.h"
+
+#include <optional>
+#include <vector>
+
+#include <yaml-cpp/yaml.h>
+
+#include "neug/storages/checkpoint_manifest.h"
+#include "neug/storages/module/module_factory.h"
+#include "neug/utils/exception/exception.h"
+#include "neug/utils/property/default_value.h"
+#include "neug/utils/property/types.h"
+
+namespace neug {
+namespace {
+
+constexpr const char* kOffsetsRef = "offsets";
+constexpr const char* kLengthsRef = "lengths";
+constexpr const char* kElementsRef = "elements";
+
+std::string ChildModuleKey(const std::string& parent, const std::string& role) {
+  return parent + "/" + role;
+}
+
+void MarkReferenced(CheckpointManifest& meta, const std::string& key) {
+  auto it = meta.mutable_modules().find(key);
+  if (it == meta.mutable_modules().end()) {
+    THROW_RUNTIME_ERROR(
+        "ListPropertyColumn::Dump: child column did not write "
+        "module '" +
+        key + "'");
+  }
+  it->second.mark_as_referenced_module();
+}
+
+const ModuleDescriptor& ResolveChild(const CheckpointManifest& manifest,
+                                     const ModuleDescriptor& parent,
+                                     const char* role,
+                                     std::optional<ModuleDescriptor>& storage) {
+  auto ref = parent.get_ref(role);
+  if (!ref.has_value()) {
+    THROW_RUNTIME_ERROR("ListPropertyColumn::Open: missing '" +
+                        std::string(role) + "' ref");
+  }
+  storage = manifest.module(*ref);
+  if (!storage.has_value()) {
+    THROW_RUNTIME_ERROR("ListPropertyColumn::Open: missing child module '" +
+                        *ref + "'");
+  }
+  return *storage;
+}
+
+}  // namespace
+
+ListPropertyColumn::ListPropertyColumn(const DataType& list_type)
+    : list_type_(list_type),
+      child_type_(ListType::GetChildType(list_type)),
+      size_(0),
+      offsets_(std::make_unique<ULongColumn>()),
+      lengths_(std::make_unique<ULongColumn>()),
+      elements_(CreateColumn(child_type_)) {}
+
+void ListPropertyColumn::Open(Checkpoint& ckp, const ModuleDescriptor& desc,
+                              MemoryLevel level) {
+  openInternal(ckp, nullptr, desc, level);
+}
+
+void ListPropertyColumn::Open(Checkpoint& ckp,
+                              const CheckpointManifest& manifest,
+                              const ModuleDescriptor& desc, MemoryLevel level) {
+  openInternal(ckp, &manifest, desc, level);
+}
+
+void ListPropertyColumn::openInternal(Checkpoint& ckp,
+                                      const CheckpointManifest* manifest,
+                                      const ModuleDescriptor& desc,
+                                      MemoryLevel level) {
+  if (list_type_.id() == DataTypeId::kInvalid) {
+    auto type_yaml = desc.get("list_type");
+    if (!type_yaml.has_value()) {
+      THROW_RUNTIME_ERROR(
+          "ListPropertyColumn::Open: missing list_type in descriptor");
+    }
+    auto node = YAML::Load(*type_yaml);
+    if (!YAML::convert<DataType>::decode(node, list_type_)) {
+      THROW_RUNTIME_ERROR(
+          "ListPropertyColumn::Open: failed to parse list_type");
+    }
+    if (list_type_.id() != DataTypeId::kList) {
+      THROW_RUNTIME_ERROR(
+          "ListPropertyColumn::Open: descriptor type is not LIST");
+    }
+    child_type_ = ListType::GetChildType(list_type_);
+    offsets_ = std::make_unique<ULongColumn>();
+    lengths_ = std::make_unique<ULongColumn>();
+    elements_ = CreateColumn(child_type_);
+  }
+
+  if (desc.module_type.empty()) {
+    offsets_->Open(ckp, ModuleDescriptor{}, level);
+    lengths_->Open(ckp, ModuleDescriptor{}, level);
+    elements_->Open(ckp, ModuleDescriptor{}, level);
+    size_ = 0;
+    return;
+  }
+
+  auto row_count = desc.get("list_row_count");
+  if (!row_count.has_value()) {
+    THROW_RUNTIME_ERROR(
+        "ListPropertyColumn::Open: missing list_row_count in descriptor");
+  }
+  size_ = std::stoull(*row_count);
+
+  const auto& resolver = manifest ? *manifest : ckp.GetMeta();
+  std::optional<ModuleDescriptor> offsets_desc;
+  std::optional<ModuleDescriptor> lengths_desc;
+  std::optional<ModuleDescriptor> elements_desc;
+  offsets_->Open(ckp, ResolveChild(resolver, desc, kOffsetsRef, offsets_desc),
+                 level);
+  lengths_->Open(ckp, ResolveChild(resolver, desc, kLengthsRef, lengths_desc),
+                 level);
+  elements_->Open(ckp, resolver,
+                  ResolveChild(resolver, desc, kElementsRef, elements_desc),
+                  level);
+
+  if (offsets_->size() != size_ || lengths_->size() != size_) {
+    THROW_RUNTIME_ERROR("ListPropertyColumn::Open: row metadata size mismatch");
+  }
+}
+
+ModuleDescriptor ListPropertyColumn::dumpSelfDescriptor() const {
+  ModuleDescriptor desc;
+  desc.module_type = ModuleTypeName();
+  desc.set("list_row_count", std::to_string(size_));
+  desc.set("list_type",
+           YAML::Dump(YAML::convert<DataType>::encode(list_type_)));
+  return desc;
+}
+
+void ListPropertyColumn::Dump(Checkpoint& ckp, CheckpointManifest& meta,
+                              const std::string& key) {
+  if (key.empty()) {
+    THROW_RUNTIME_ERROR(
+        "ListPropertyColumn::Dump: module key must not be empty");
+  }
+
+  ULongColumn compact_offsets;
+  ULongColumn compact_lengths;
+  compact_offsets.Open(ckp, ModuleDescriptor{}, MemoryLevel::kInMemory);
+  compact_lengths.Open(ckp, ModuleDescriptor{}, MemoryLevel::kInMemory);
+  compact_offsets.resize(size_);
+  compact_lengths.resize(size_);
+
+  auto compact_elements = CreateColumn(child_type_);
+  compact_elements->Open(ckp, ModuleDescriptor{}, MemoryLevel::kInMemory);
+
+  // Precompute total element count to resize once, avoiding O(n^2) copying
+  // from repeated resize calls inside the loop.
+  size_t total_elements = 0;
+  for (size_t row = 0; row < size_; ++row) {
+    total_elements += lengths_->get_view(row);
+  }
+  compact_elements->resize(total_elements);
+
+  size_t tail = 0;
+  for (size_t row = 0; row < size_; ++row) {
+    auto offset = offsets_->get_view(row);
+    auto length = lengths_->get_view(row);
+    compact_offsets.set_value(row, tail);
+    compact_lengths.set_value(row, length);
+    for (size_t i = 0; i < length; ++i) {
+      compact_elements->set_any(tail + i, elements_->get_any(offset + i), true);
+    }
+    tail += length;
+  }
+
+  auto offsets_key = ChildModuleKey(key, kOffsetsRef);
+  auto lengths_key = ChildModuleKey(key, kLengthsRef);
+  auto elements_key = ChildModuleKey(key, kElementsRef);
+  compact_offsets.Dump(ckp, meta, offsets_key);
+  compact_lengths.Dump(ckp, meta, lengths_key);
+  compact_elements->Dump(ckp, meta, elements_key);
+  MarkReferenced(meta, offsets_key);
+  MarkReferenced(meta, lengths_key);
+  MarkReferenced(meta, elements_key);
+
+  auto desc = dumpSelfDescriptor();
+  desc.set_ref(kOffsetsRef, std::move(offsets_key));
+  desc.set_ref(kLengthsRef, std::move(lengths_key));
+  desc.set_ref(kElementsRef, std::move(elements_key));
+  meta.set_module(key, std::move(desc));
+}
+
+void ListPropertyColumn::resize(size_t size) {
+  if (size <= size_) {
+    offsets_->resize(size);
+    lengths_->resize(size);
+    size_ = size;
+    return;
+  }
+
+  auto old_size = size_;
+  auto tail = elements_->size();
+  offsets_->resize(size);
+  lengths_->resize(size);
+  for (size_t i = old_size; i < size; ++i) {
+    offsets_->set_value(i, tail);
+    lengths_->set_value(i, 0);
+  }
+  size_ = size;
+}
+
+void ListPropertyColumn::resize(size_t size, const Value& default_value) {
+  if (size <= size_) {
+    resize(size);
+    return;
+  }
+  auto old_size = size_;
+  resize(size);
+  for (size_t i = old_size; i < size_; ++i) {
+    set_any(i, default_value, true);
+  }
+}
+
+void ListPropertyColumn::set_any(size_t index, const Value& value,
+                                 bool insert_safe) {
+  if (index >= size_) {
+    THROW_RUNTIME_ERROR("ListPropertyColumn::set_any: index " +
+                        std::to_string(index) +
+                        " out of range (size=" + std::to_string(size_) + ")");
+  }
+
+  Value default_value;
+  const Value* normalized = &value;
+  if (value.IsNull()) {
+    default_value = get_default_value(list_type_);
+    normalized = &default_value;
+  }
+  if (normalized->type() != list_type_) {
+    THROW_INVALID_ARGUMENT_EXCEPTION("ListPropertyColumn::set_any: expected " +
+                                     list_type_.ToString() + ", got " +
+                                     normalized->type().ToString());
+  }
+  const auto& children = ListValue::GetChildren(*normalized);
+  auto old_offset = offsets_->get_view(index);
+  auto old_length = lengths_->get_view(index);
+
+  auto write_child = [&](size_t child_index, const Value& child,
+                         bool child_insert_safe) {
+    Value child_default;
+    const Value* normalized_child = &child;
+    if (child.IsNull()) {
+      child_default = get_default_value(child_type_);
+      normalized_child = &child_default;
+    }
+    if (normalized_child->type() != child_type_) {
+      THROW_INVALID_ARGUMENT_EXCEPTION(
+          "ListPropertyColumn::set_any: expected child type " +
+          child_type_.ToString() + ", got " +
+          normalized_child->type().ToString());
+    }
+    elements_->set_any(child_index, *normalized_child, child_insert_safe);
+  };
+
+  if (children.size() == old_length) {
+    for (size_t i = 0; i < children.size(); ++i) {
+      write_child(old_offset + i, children[i], insert_safe);
+    }
+    return;
+  }
+
+  // List length changed: new elements must be appended to the tail of
+  // elements_, which requires a resize. Respect the insert_safe contract:
+  // when false, the caller expects no reallocation.
+  if (!insert_safe) {
+    THROW_STORAGE_EXCEPTION(
+        "ListPropertyColumn::set_any: list length changed from " +
+        std::to_string(old_length) + " to " + std::to_string(children.size()) +
+        ", which requires resizing elements_ but insert_safe is false");
+  }
+
+  auto new_offset = elements_->size();
+  elements_->resize(new_offset + children.size());
+  for (size_t i = 0; i < children.size(); ++i) {
+    write_child(new_offset + i, children[i], true);
+  }
+  offsets_->set_value(index, new_offset);
+  lengths_->set_value(index, children.size());
+}
+
+Value ListPropertyColumn::get_any(size_t index) const {
+  if (index >= size_) {
+    THROW_RUNTIME_ERROR("ListPropertyColumn::get_any: index " +
+                        std::to_string(index) +
+                        " out of range (size=" + std::to_string(size_) + ")");
+  }
+  auto offset = offsets_->get_view(index);
+  auto length = lengths_->get_view(index);
+  std::vector<Value> children;
+  children.reserve(length);
+  for (size_t i = 0; i < length; ++i) {
+    children.emplace_back(elements_->get_any(offset + i));
+  }
+  return Value::LIST(child_type_, std::move(children));
+}
+
+std::unique_ptr<Module> ListPropertyColumn::Clone() const {
+  auto clone = std::make_unique<ListPropertyColumn>();
+  clone->list_type_ = list_type_;
+  clone->child_type_ = child_type_;
+  clone->size_ = size_;
+  if (offsets_) {
+    clone->offsets_.reset(
+        static_cast<ULongColumn*>(offsets_->Clone().release()));
+  }
+  if (lengths_) {
+    clone->lengths_.reset(
+        static_cast<ULongColumn*>(lengths_->Clone().release()));
+  }
+  if (elements_) {
+    clone->elements_.reset(
+        static_cast<ColumnBase*>(elements_->Clone().release()));
+  }
+  return clone;
+}
+
+void ListPropertyColumn::Detach(Checkpoint& ckp, MemoryLevel level) {
+  offsets_->Detach(ckp, level);
+  lengths_->Detach(ckp, level);
+  elements_->Detach(ckp, level);
+}
+
+NEUG_REGISTER_MODULE(ListPropertyColumn);
+
+}  // namespace neug
diff --git a/src/utils/yaml_utils.cc b/src/utils/yaml_utils.cc
index 6d1a70b2e..bbe05f7f3 100644
--- a/src/utils/yaml_utils.cc
+++ b/src/utils/yaml_utils.cc
@@ -84,6 +84,10 @@ YAML::Node property_type_to_yaml(const DataType& type) {
     node["array"]["max_length"] = array_size;
     break;
   }
+  case DataTypeId::kList:
+    node["list"]["component_type"] =
+        property_type_to_yaml(ListType::GetChildType(type));
+    break;
   default:
     THROW_INVALID_ARGUMENT_EXCEPTION(
         "Unrecognized property type for YAML encoding: " + type.ToString());
diff --git a/tests/transaction/test_update_transaction.cc b/tests/transaction/test_update_transaction.cc
index ad7a224f2..1cd3638ab 100644
--- a/tests/transaction/test_update_transaction.cc
+++ b/tests/transaction/test_update_transaction.cc
@@ -2276,6 +2276,236 @@ TEST_F(UpdateTransactionTest, TestReplayWal) {
   }
 }
 
+TEST_F(UpdateTransactionTest, NestedListSnapshotAbortAndWalReplay) {
+  auto string_list_type = neug::DataType::List(neug::DataType::VARCHAR);
+  auto pair_type = neug::DataType::Array(string_list_type, 2);
+  auto nested_type = neug::DataType::List(pair_type);
+  auto strings = [](std::initializer_list<const char*> values) {
+    std::vector<neug::Value> children;
+    for (auto value : values) {
+      children.push_back(neug::Value::STRING(value));
+    }
+    return neug::Value::LIST(neug::DataType::VARCHAR, std::move(children));
+  };
+  auto pair = [&](neug::Value lhs, neug::Value rhs) {
+    return neug::Value::ARRAY(pair_type, {std::move(lhs), std::move(rhs)});
+  };
+  auto nested = [&](std::vector<neug::Value> values) {
+    return neug::Value::LIST(pair_type, std::move(values));
+  };
+  auto initial = nested({pair(strings({"a"}), strings({}))});
+  auto committed = nested({pair(strings({"b", "c"}), strings({"d"})),
+                           pair(strings({}), strings({"e"}))});
+  auto aborted = nested({pair(strings({"not"}), strings({"visible"}))});
+
+  neug::NeugDBConfig config(db_dir);
+  config.memory_level = neug::MemoryLevel::kInMemory;
+  config.max_thread_num = 2;
+  config.checkpoint_on_close = false;
+  config.checkpoint_on_recovery = true;
+  {
+    neug::NeugDB db;
+    db.Open(config);
+    auto svc = std::make_shared<neug::NeugDBService>(db);
+    {
+      auto slot = svc->AcquireExecutionSlot();
+      auto txn = slot->GetUpdateTransaction();
+      neug::StorageTPUpdateInterface interface(txn);
+      EXPECT_TRUE(interface.CreateVertexType(BuildCreateVertexTypeParam(
+          "list_holder",
+          {{"id", neug::Value::INT64(0)},
+           {"nested", neug::Value::LIST(pair_type, {})}},
+          {"id"})));
+      auto label = interface.schema().get_vertex_label_id("list_holder");
+      neug::vid_t vid;
+      EXPECT_TRUE(
+          interface.AddVertex(label, neug::Value::INT64(1), {initial}, vid));
+      EXPECT_TRUE(txn.Commit());
+    }
+
+    auto old_slot = svc->AcquireExecutionSlot();
+    auto old_txn = old_slot->GetReadTransaction();
+    neug::StorageReadInterface old_reader(old_txn.view(), old_txn.timestamp());
+    auto label = old_reader.schema().get_vertex_label_id("list_holder");
+    neug::vid_t vid;
+    ASSERT_TRUE(old_reader.GetVertexIndex(label, neug::Value::INT64(1), vid));
+    auto old_column = old_reader.GetVertexPropColumn(label, "nested");
+    ASSERT_NE(old_column, nullptr);
+    EXPECT_EQ(old_column->get_any(vid), initial);
+
+    {
+      auto slot = svc->AcquireExecutionSlot();
+      auto txn = slot->GetUpdateTransaction();
+      neug::StorageTPUpdateInterface interface(txn);
+      EXPECT_TRUE(interface.UpdateVertexProperty(label, vid, 0, committed));
+      EXPECT_EQ(old_column->get_any(vid), initial);
+      EXPECT_TRUE(txn.Commit());
+    }
+    EXPECT_EQ(old_column->get_any(vid), initial);
+    EXPECT_TRUE(old_txn.Commit());
+    old_slot = {};
+
+    {
+      auto slot = svc->AcquireExecutionSlot();
+      auto txn = slot->GetUpdateTransaction();
+      neug::StorageTPUpdateInterface interface(txn);
+      ASSERT_TRUE(txn.GetVertexIndex(label, neug::Value::INT64(1), vid));
+      EXPECT_TRUE(interface.UpdateVertexProperty(label, vid, 0, aborted));
+      txn.Abort();
+    }
+    {
+      auto slot = svc->AcquireExecutionSlot();
+      auto txn = slot->GetReadTransaction();
+      neug::StorageReadInterface reader(txn.view(), txn.timestamp());
+      auto column = reader.GetVertexPropColumn(label, "nested");
+      ASSERT_NE(column, nullptr);
+      EXPECT_EQ(column->get_any(vid), committed);
+    }
+    svc.reset();
+    db.Close();
+  }
+  {
+    neug::NeugDB db;
+    db.Open(config);
+    auto svc = std::make_shared<neug::NeugDBService>(db);
+    auto slot = svc->AcquireExecutionSlot();
+    auto txn = slot->GetReadTransaction();
+    neug::StorageReadInterface reader(txn.view(), txn.timestamp());
+    auto label = reader.schema().get_vertex_label_id("list_holder");
+    neug::vid_t vid;
+    ASSERT_TRUE(reader.GetVertexIndex(label, neug::Value::INT64(1), vid));
+    auto column = reader.GetVertexPropColumn(label, "nested");
+    ASSERT_NE(column, nullptr);
+    EXPECT_EQ(column->get_any(vid), committed);
+  }
+}
+
+TEST_F(UpdateTransactionTest, NestedListEdgeSnapshotAbortAndWalReplay) {
+  auto string_list_type = neug::DataType::List(neug::DataType::VARCHAR);
+  auto pair_type = neug::DataType::Array(string_list_type, 2);
+  auto strings = [](std::initializer_list<const char*> values) {
+    std::vector<neug::Value> children;
+    for (auto value : values) {
+      children.push_back(neug::Value::STRING(value));
+    }
+    return neug::Value::LIST(neug::DataType::VARCHAR, std::move(children));
+  };
+  auto pair = [&](neug::Value lhs, neug::Value rhs) {
+    return neug::Value::ARRAY(pair_type, {std::move(lhs), std::move(rhs)});
+  };
+  auto nested = [&](std::vector<neug::Value> values) {
+    return neug::Value::LIST(pair_type, std::move(values));
+  };
+  auto initial = nested({pair(strings({"a"}), strings({}))});
+  auto committed = nested({pair(strings({"b", "c"}), strings({"d"})),
+                           pair(strings({}), strings({"e"}))});
+  auto aborted = nested({pair(strings({"not"}), strings({"visible"}))});
+
+  auto read_edge_value = [](auto& reader, neug::label_t label,
+                            neug::label_t edge_label,
+                            neug::vid_t src_vid) -> neug::Value {
+    auto ed_accessor = reader.GetEdgeDataAccessor(label, label, edge_label, 0);
+    auto view = reader.GetGenericOutgoingGraphView(label, label, edge_label);
+    auto edges = view.get_edges(src_vid);
+    auto it = edges.begin();
+    EXPECT_NE(it, edges.end());
+    return ed_accessor.get_data(it);
+  };
+
+  neug::NeugDBConfig config(db_dir);
+  config.memory_level = neug::MemoryLevel::kInMemory;
+  config.max_thread_num = 2;
+  config.checkpoint_on_close = false;
+  config.checkpoint_on_recovery = true;
+  {
+    neug::NeugDB db;
+    db.Open(config);
+    auto svc = std::make_shared<neug::NeugDBService>(db);
+    neug::label_t label = 0;
+    neug::label_t edge_label = 0;
+    neug::vid_t v1 = 0;
+    neug::vid_t v2 = 0;
+    {
+      auto slot = svc->AcquireExecutionSlot();
+      auto txn = slot->GetUpdateTransaction();
+      neug::StorageTPUpdateInterface interface(txn);
+      EXPECT_TRUE(interface.CreateVertexType(BuildCreateVertexTypeParam(
+          "edge_holder", {{"id", neug::Value::INT64(0)}}, {"id"})));
+      label = interface.schema().get_vertex_label_id("edge_holder");
+      EXPECT_TRUE(interface.CreateEdgeType(BuildCreateEdgeTypeParam(
+          "edge_holder", "edge_holder", "carries",
+          {{"nested", neug::Value::LIST(pair_type, {})}})));
+      edge_label = interface.schema().get_edge_label_id("carries");
+      EXPECT_TRUE(interface.AddVertex(label, neug::Value::INT64(1), {}, v1));
+      EXPECT_TRUE(interface.AddVertex(label, neug::Value::INT64(2), {}, v2));
+      const void* edge_prop = nullptr;
+      EXPECT_TRUE(interface.AddEdge(label, v1, label, v2, edge_label, {initial},
+                                    edge_prop));
+      EXPECT_TRUE(txn.Commit());
+    }
+
+    auto old_slot = svc->AcquireExecutionSlot();
+    auto old_txn = old_slot->GetReadTransaction();
+    neug::StorageReadInterface old_reader(old_txn.view(), old_txn.timestamp());
+    EXPECT_EQ(read_edge_value(old_reader, label, edge_label, v1), initial);
+
+    {
+      auto slot = svc->AcquireExecutionSlot();
+      auto txn = slot->GetUpdateTransaction();
+      neug::StorageTPUpdateInterface interface(txn);
+      update_edge_property(
+          txn, label, label, edge_label, v1,
+          [](neug::vid_t dst_vid) { return true; },
+          [&](neug::vid_t dst_vid, int32_t oe_offset, int32_t ie_offset) {
+            EXPECT_TRUE(interface.UpdateEdgeProperty(label, v1, label, dst_vid,
+                                                     edge_label, oe_offset,
+                                                     ie_offset, 0, committed));
+          });
+      EXPECT_EQ(read_edge_value(old_reader, label, edge_label, v1), initial);
+      EXPECT_TRUE(txn.Commit());
+    }
+    EXPECT_EQ(read_edge_value(old_reader, label, edge_label, v1), initial);
+    EXPECT_TRUE(old_txn.Commit());
+    old_slot = {};
+
+    {
+      auto slot = svc->AcquireExecutionSlot();
+      auto txn = slot->GetUpdateTransaction();
+      neug::StorageTPUpdateInterface interface(txn);
+      update_edge_property(
+          txn, label, label, edge_label, v1,
+          [](neug::vid_t dst_vid) { return true; },
+          [&](neug::vid_t dst_vid, int32_t oe_offset, int32_t ie_offset) {
+            EXPECT_TRUE(interface.UpdateEdgeProperty(label, v1, label, dst_vid,
+                                                     edge_label, oe_offset,
+                                                     ie_offset, 0, aborted));
+          });
+      txn.Abort();
+    }
+    {
+      auto slot = svc->AcquireExecutionSlot();
+      auto txn = slot->GetReadTransaction();
+      neug::StorageReadInterface reader(txn.view(), txn.timestamp());
+      EXPECT_EQ(read_edge_value(reader, label, edge_label, v1), committed);
+    }
+    svc.reset();
+    db.Close();
+  }
+  {
+    neug::NeugDB db;
+    db.Open(config);
+    auto svc = std::make_shared<neug::NeugDBService>(db);
+    auto slot = svc->AcquireExecutionSlot();
+    auto txn = slot->GetReadTransaction();
+    neug::StorageReadInterface reader(txn.view(), txn.timestamp());
+    auto label = reader.schema().get_vertex_label_id("edge_holder");
+    auto edge_label = reader.schema().get_edge_label_id("carries");
+    neug::vid_t v1;
+    ASSERT_TRUE(reader.GetVertexIndex(label, neug::Value::INT64(1), v1));
+    EXPECT_EQ(read_edge_value(reader, label, edge_label, v1), committed);
+  }
+}
+
 TEST_F(UpdateTransactionTest, TestAPIAfterDeleteVertexLabel) {
   neug::NeugDB db;
   neug::NeugDBConfig config(db_dir);
diff --git a/tests/unittest/test_column.cc b/tests/unittest/test_column.cc
index 7a5ec6ff6..407d137d7 100644
--- a/tests/unittest/test_column.cc
+++ b/tests/unittest/test_column.cc
@@ -23,6 +23,7 @@
 #include "neug/utils/exception/exception.h"
 #include "neug/utils/property/array_column.h"
 #include "neug/utils/property/column.h"
+#include "neug/utils/property/list_property_column.h"
 #include "neug/utils/property/vec_column.h"
 #include "unittest/utils.h"
 
@@ -378,6 +379,124 @@ TEST(ArrayColumnTest, SetAnyRequiresArrayValue) {
   std::filesystem::remove_all(temp_dir);
 }
 
+TEST(ListPropertyColumnTest, RecursiveLifecycle) {
+  auto temp_dir =
+      std::filesystem::temp_directory_path() /
+      ("list_property_column_recursive_" +
+       std::to_string(
+           std::chrono::steady_clock::now().time_since_epoch().count()));
+  std::filesystem::remove_all(temp_dir);
+  std::filesystem::create_directories(temp_dir);
+  CheckpointManager checkpoint_mgr;
+  checkpoint_mgr.Open(temp_dir.string());
+  auto ckp = make_checkpoint(checkpoint_mgr);
+
+  auto string_list_type = DataType::List(DataType::VARCHAR);
+  auto pair_type = DataType::Array(string_list_type, 2);
+  auto outer_type = DataType::List(pair_type);
+  auto strings = [&](std::initializer_list<const char*> values) {
+    std::vector<Value> children;
+    for (auto value : values) {
+      children.push_back(Value::STRING(value));
+    }
+    return Value::LIST(DataType::VARCHAR, std::move(children));
+  };
+  auto pair = [&](Value lhs, Value rhs) {
+    std::vector<Value> children;
+    children.push_back(std::move(lhs));
+    children.push_back(std::move(rhs));
+    return Value::ARRAY(pair_type, std::move(children));
+  };
+  auto outer = [&](std::vector<Value> values) {
+    return Value::LIST(pair_type, std::move(values));
+  };
+
+  ListPropertyColumn column(outer_type);
+  column.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory);
+  column.resize(2);
+  auto initial = outer({pair(strings({"a"}), strings({})),
+                        pair(strings({"b", "c"}), strings({"d"}))});
+  column.set_any(0, initial, true);
+  column.set_any(1, outer({}), true);
+  EXPECT_EQ(column.get_any(0), initial);
+  EXPECT_EQ(ListValue::GetChildren(column.get_any(1)).size(), 0);
+
+  auto clone_module = column.Clone();
+  auto* clone = dynamic_cast<ListPropertyColumn*>(clone_module.get());
+  ASSERT_NE(clone, nullptr);
+  clone->Detach(*ckp, MemoryLevel::kInMemory);
+  auto clone_value = outer({pair(strings({"x", "y"}), strings({"z"}))});
+  clone->set_any(0, clone_value, true);
+  EXPECT_EQ(column.get_any(0), initial);
+  EXPECT_EQ(clone->get_any(0), clone_value);
+
+  auto final_value = outer({pair(strings({}), strings({"last"}))});
+  clone->set_any(0, final_value, true);
+  CheckpointManifest manifest;
+  clone->Dump(*ckp, manifest, "list");
+  ListPropertyColumn reopened;
+  reopened.Open(*ckp, manifest, *manifest.module("list"),
+                MemoryLevel::kInMemory);
+  EXPECT_EQ(reopened.list_type(), outer_type);
+  EXPECT_EQ(reopened.get_any(0), final_value);
+  EXPECT_EQ(ListValue::GetChildren(reopened.get_any(1)).size(), 0);
+
+  std::filesystem::remove_all(temp_dir);
+}
+
+TEST(ListPropertyColumnTest, ResizeDefaultAndTypeContract) {
+  auto temp_dir =
+      std::filesystem::temp_directory_path() /
+      ("list_property_column_resize_" +
+       std::to_string(
+           std::chrono::steady_clock::now().time_since_epoch().count()));
+  std::filesystem::remove_all(temp_dir);
+  std::filesystem::create_directories(temp_dir);
+  CheckpointManager checkpoint_mgr;
+  checkpoint_mgr.Open(temp_dir.string());
+  auto ckp = make_checkpoint(checkpoint_mgr);
+
+  auto list_type = DataType::List(DataType::INT32);
+  auto list = [](std::initializer_list<int32_t> values) {
+    std::vector<Value> children;
+    for (auto value : values) {
+      children.push_back(Value::INT32(value));
+    }
+    return Value::LIST(DataType::INT32, std::move(children));
+  };
+
+  ListPropertyColumn column(list_type);
+  column.Open(*ckp, ModuleDescriptor{}, MemoryLevel::kInMemory);
+  column.resize(3);
+  for (size_t i = 0; i < column.size(); ++i) {
+    EXPECT_TRUE(ListValue::GetChildren(column.get_any(i)).empty());
+  }
+
+  column.set_any(0, list({1, 2}), true);
+  column.set_any(0, list({3, 4}), false);
+  EXPECT_EQ(column.get_any(0), list({3, 4}));
+  column.set_any(0, list({5}), true);
+  EXPECT_EQ(column.get_any(0), list({5}));
+  column.set_any(0, list({}), true);
+  EXPECT_TRUE(ListValue::GetChildren(column.get_any(0)).empty());
+  column.set_any(1, Value(list_type), true);
+  EXPECT_TRUE(ListValue::GetChildren(column.get_any(1)).empty());
+
+  column.resize(1);
+  column.resize(3);
+  EXPECT_TRUE(ListValue::GetChildren(column.get_any(1)).empty());
+  EXPECT_TRUE(ListValue::GetChildren(column.get_any(2)).empty());
+
+  EXPECT_THROW(
+      column.set_any(0, Value::LIST(DataType::INT64, {Value::INT64(1)}), true),
+      exception::InvalidArgumentException);
+  EXPECT_THROW(
+      column.set_any(0, Value::LIST(DataType::INT32, {Value::INT64(1)}), true),
+      exception::InvalidArgumentException);
+
+  std::filesystem::remove_all(temp_dir);
+}
+
 TEST(VecColumnTest, AccessResizeCloneAndDumpOpen) {
   auto temp_dir =
       std::filesystem::temp_directory_path() /
diff --git a/tests/utils/json_test.cc b/tests/utils/json_test.cc
index d526bf189..afb7be6f7 100644
--- a/tests/utils/json_test.cc
+++ b/tests/utils/json_test.cc
@@ -85,6 +85,17 @@ class JsonTest : public ::testing::Test {
     return type;
   }
 
+  std::shared_ptr<::common::DataType> createNestedStringListType() {
+    auto type = std::make_shared<::common::DataType>();
+    auto* outer_list = type->mutable_list();
+    auto* pair = outer_list->mutable_component_type()->mutable_array();
+    pair->set_fixed_length(2);
+    auto* inner_list = pair->mutable_component_type()->mutable_list();
+    auto* string_type = inner_list->mutable_component_type()->mutable_string();
+    string_type->mutable_var_char();
+    return type;
+  }
+
   std::shared_ptr<reader::ReadSharedState> createSharedState(
       const std::string& jsonFile, const std::vector<std::string>& columnNames,
       const std::vector<std::shared_ptr<::common::DataType>>& columnTypes,
@@ -217,5 +228,48 @@ TEST_F(JsonTest, TestJsonArrayColumnRejectsNonArray) {
   }
 }
 
+TEST_F(JsonTest, TestJsonRecursiveListColumn) {
+  createJsonFile(
+      "test_json_recursive_list.json",
+      "[{\"id\":1,\"nested\":[[[\"a\"],[]],[[\"b\",\"c\"],[\"d\"]]]},"
+      "{\"id\":2,\"nested\":[]},{\"id\":3,\"nested\":null},"
+      "{\"id\":4,\"nested\":[[null,[\"h\"]],null]}]");
+  auto sharedState =
+      createSharedState("test_json_recursive_list.json", {"id", "nested"},
+                        {createUInt32Type(), createNestedStringListType()},
+                        {{"batch_read", "false"}});
+  auto reader = createJsonReader(sharedState);
+  execution::Context ctx;
+  reader->read(std::make_shared<reader::ReadLocalState>(), ctx);
+
+  ASSERT_EQ(ctx.row_num(), 4);
+  auto nested = ctx.chunk(0).columns()[1];
+  ASSERT_EQ(nested->elem_type().id(), DataTypeId::kList);
+  auto first_value = nested->get_elem(0);
+  const auto& first = ListValue::GetChildren(first_value);
+  ASSERT_EQ(first.size(), 2);
+  const auto& first_pair = ArrayValue::GetChildren(first[0]);
+  ASSERT_EQ(first_pair.size(), 2);
+  EXPECT_EQ(StringValue::Get(ListValue::GetChildren(first_pair[0])[0]), "a");
+  EXPECT_TRUE(ListValue::GetChildren(first_pair[1]).empty());
+  auto second_value = nested->get_elem(1);
+  EXPECT_TRUE(ListValue::GetChildren(second_value).empty());
+  EXPECT_FALSE(nested->has_value(2));
+  EXPECT_TRUE(nested->get_elem(2).IsNull());
+  // Nested nulls normalize to the declared defaults: null LIST child -> [],
+  // null ARRAY child -> [[], []].
+  auto fourth_value = nested->get_elem(3);
+  const auto& fourth = ListValue::GetChildren(fourth_value);
+  ASSERT_EQ(fourth.size(), 2);
+  const auto& fourth_first = ArrayValue::GetChildren(fourth[0]);
+  ASSERT_EQ(fourth_first.size(), 2);
+  EXPECT_TRUE(ListValue::GetChildren(fourth_first[0]).empty());
+  EXPECT_EQ(StringValue::Get(ListValue::GetChildren(fourth_first[1])[0]), "h");
+  const auto& fourth_second = ArrayValue::GetChildren(fourth[1]);
+  ASSERT_EQ(fourth_second.size(), 2);
+  EXPECT_TRUE(ListValue::GetChildren(fourth_second[0]).empty());
+  EXPECT_TRUE(ListValue::GetChildren(fourth_second[1]).empty());
+}
+
 }  // namespace test
 }  // namespace neug
diff --git a/tests/utils/test_utils.cc b/tests/utils/test_utils.cc
index 12d7524e0..b8e3c6ad3 100644
--- a/tests/utils/test_utils.cc
+++ b/tests/utils/test_utils.cc
@@ -22,6 +22,7 @@
 #include "neug/utils/bitset.h"
 #include "neug/utils/datetime_parsers.h"
 #include "neug/utils/encoder.h"
+#include "neug/utils/io/read/common/type_converter.h"
 #include "neug/utils/pb_utils.h"
 #include "neug/utils/string_view_vector.h"
 #include "neug/utils/yaml_utils.h"
@@ -1478,6 +1479,36 @@ TEST_F(YamlUtilsTest, PropertyTypeToYaml_TemporalTypes) {
   }
 }
 
+TEST_F(YamlUtilsTest, RecursiveListTypeRoundTrip) {
+  auto type =
+      DataType::List(DataType::Array(DataType::List(DataType::VARCHAR), 2));
+
+  auto encoded = YAML::convert<DataType>::encode(type);
+  DataType decoded;
+  ASSERT_TRUE(YAML::convert<DataType>::decode(encoded, decoded));
+  EXPECT_EQ(decoded, type);
+
+  auto schema_yaml = property_type_to_yaml(type);
+  DataType schema_decoded;
+  ASSERT_TRUE(YAML::convert<DataType>::decode(schema_yaml, schema_decoded));
+  EXPECT_EQ(schema_decoded, type);
+
+  reader::NeuGTypeConverter converter;
+  auto common_type = converter.convert(type);
+  ASSERT_NE(common_type, nullptr);
+  EXPECT_EQ(converter.convert(*common_type), type);
+
+  google::protobuf::RepeatedPtrField<::physical::PropertyDef> properties;
+  auto* property = properties.Add();
+  property->set_name("nested");
+  property->mutable_type()->CopyFrom(*common_type);
+  auto defaults = property_defs_to_value(properties);
+  ASSERT_TRUE(defaults.has_value());
+  ASSERT_EQ(defaults->size(), 1);
+  EXPECT_EQ(defaults->front().second.type(), type);
+  EXPECT_TRUE(ListValue::GetChildren(defaults->front().second).empty());
+}
+
 TEST_F(YamlUtilsTest, PropertyTypeToYaml_UnknownType_Throws) {
   DataType type(DataType::SQLNULL);
   EXPECT_THROW(property_type_to_yaml(type), std::exception);
diff --git a/tools/python_bind/tests/test_db_array.py b/tools/python_bind/tests/test_db_array.py
index 01f7472ab..09949ed20 100644
--- a/tools/python_bind/tests/test_db_array.py
+++ b/tools/python_bind/tests/test_db_array.py
@@ -643,8 +643,8 @@ def test_array_zero_size_rejected(tmp_path):
     db.close()
 
 
-def test_cast_does_not_convert_between_list_and_array(tmp_path):
-    """CAST should not normalize LIST and fixed-size ARRAY values."""
+def test_explicit_cast_converts_between_list_and_array(tmp_path):
+    """LIST and ARRAY convert only through an explicit CAST."""
     db = Database(db_path=str(tmp_path), mode="w")
     conn = db.connect()
 
@@ -653,16 +653,22 @@ def test_cast_does_not_convert_between_list_and_array(tmp_path):
     )
     conn.execute("CREATE (s:Sensor {id: 1, readings: [1, 2, 3]});")
 
-    with pytest.raises(Exception):
-        list(conn.execute("MATCH (s:Sensor) RETURN CAST(s.readings, 'INT32[]');"))
+    rows = list(conn.execute("MATCH (s:Sensor) RETURN CAST(s.readings, 'INT32[]');"))
+    assert _nested_list(rows[0][0]) == [1, 2, 3]
+
+    rows = list(
+        conn.execute(
+            "UNWIND [1, 2, 3] AS v "
+            "WITH collect(v) AS values "
+            "RETURN CAST(values, 'INT64[3]');"
+        )
+    )
+    assert _nested_list(rows[0][0]) == [1, 2, 3]
 
     with pytest.raises(Exception):
-        list(
-            conn.execute(
-                "UNWIND [1, 2, 3] AS v "
-                "WITH collect(v) AS values "
-                "RETURN CAST(values, 'INT64[3]');"
-            )
+        conn.execute(
+            "UNWIND [1, 2] AS v WITH collect(v) AS values "
+            "RETURN CAST(values, 'INT64[3]');"
         )
 
     conn.close()
diff --git a/tools/python_bind/tests/test_db_list.py b/tools/python_bind/tests/test_db_list.py
index 53d7a81d5..7a51e61cf 100644
--- a/tools/python_bind/tests/test_db_list.py
+++ b/tools/python_bind/tests/test_db_list.py
@@ -14,44 +14,177 @@
 # 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.
-#
-
-import os
-import shutil
-import sys
 
 import pytest
 
 from neug.database import Database
 
 
-def test_return_single_list(tmp_path):
-    db_dir = tmp_path / "return_list"
-    shutil.rmtree(db_dir, ignore_errors=True)
-    db_dir.mkdir()
-    db = Database(db_path=str(db_dir), mode="w")
+def _nested_list(value):
+    if isinstance(value, (str, bytes)):
+        return value
+    try:
+        return [_nested_list(item) for item in value]
+    except TypeError:
+        return value
+
+
+def test_list_cast_contract(tmp_path):
+    db = Database(db_path=str(tmp_path), mode="w", checkpoint_on_close=False)
+    conn = db.connect()
+
+    assert _nested_list(list(conn.execute("RETURN CAST([], 'STRING[]');"))[0][0]) == []
+    assert _nested_list(
+        list(conn.execute("RETURN CAST([1, 2, 3], 'INT64[]');"))[0][0]
+    ) == [1, 2, 3]
+    assert _nested_list(
+        list(
+            conn.execute(
+                "UNWIND [1, 2, 3] AS v WITH collect(v) AS values "
+                "RETURN CAST(values, 'INT64[3]');"
+            )
+        )[0][0]
+    ) == [1, 2, 3]
+
+    with pytest.raises(Exception):
+        conn.execute(
+            "UNWIND [1, 2] AS v WITH collect(v) AS values "
+            "RETURN CAST(values, 'INT64[3]');"
+        )
+    with pytest.raises(Exception):
+        conn.execute("RETURN CAST([], 'INT64[2]');")
+
+    conn.execute("CREATE NODE TABLE T(id INT64, values INT64[], PRIMARY KEY(id));")
+    with pytest.raises(Exception):
+        conn.execute("CREATE (:T {id: 1, values: [1, 2]});")
+    with pytest.raises(Exception):
+        conn.execute("CREATE NODE TABLE Bad(id INT64[], PRIMARY KEY(id));")
+
+    conn.close()
+    db.close()
+
+
+def test_list_point_edge_update_and_reopen(tmp_path):
+    db_path = str(tmp_path)
+    db = Database(db_path=db_path, mode="w")
     conn = db.connect()
 
     conn.execute(
-        "CREATE NODE TABLE PERSON(id INT64, name STRING, score FLOAT, PRIMARY KEY(id));"
+        "CREATE NODE TABLE Person("
+        "id INT64, tags STRING[], nested STRING[][2][], PRIMARY KEY(id));"
+    )
+    conn.execute("CREATE REL TABLE Knows(FROM Person TO Person, scores INT64[]);")
+    conn.execute(
+        "CREATE (:Person {id: 1, tags: CAST(['a'], 'STRING[]'), nested: "
+        "CAST([CAST([CAST(['x'], 'STRING[]'), CAST([], 'STRING[]')], "
+        "'STRING[][2]')], 'STRING[][2][]')});"
+    )
+    conn.execute(
+        "CREATE (:Person {id: 2, tags: CAST([], 'STRING[]'), "
+        "nested: CAST([], 'STRING[][2][]')});"
+    )
+    conn.execute(
+        "MATCH (a:Person {id: 1}), (b:Person {id: 2}) "
+        "CREATE (a)-[:Knows {scores: CAST([1, 2], 'INT64[]')}]->(b);"
     )
-    conn.execute("CREATE (p: PERSON {id: 0, name: 'Alice', score: 99.5});")
-    conn.execute("CREATE (p: PERSON {id: 1, name: 'Bob', score: 98.5});")
 
-    result = conn.execute("MATCH (p: PERSON) RETURN [p.id, p.name, p.score];")
-    result = list(result)
-    assert result[0][0] == [0, "Alice", 99.5]
-    assert result[1][0] == [1, "Bob", 98.5]
+    conn.execute("MATCH (p:Person {id: 1}) SET p.tags = CAST(['b'], 'STRING[]');")
+    conn.execute(
+        "MATCH (p:Person {id: 1}) " "SET p.tags = CAST(['c', 'd', 'e'], 'STRING[]');"
+    )
+    conn.execute(
+        "MATCH (:Person {id: 1})-[e:Knows]->(:Person {id: 2}) "
+        "SET e.scores = CAST([7], 'INT64[]');"
+    )
+    conn.execute(
+        "MERGE (p:Person {id: 2}) "
+        "ON MATCH SET p.tags = CAST(['merged'], 'STRING[]');"
+    )
+
+    row = list(
+        conn.execute("MATCH (p:Person {id: 1}) RETURN p.tags, p.tags[1], p.nested;")
+    )[0]
+    assert _nested_list(row[0]) == ["c", "d", "e"]
+    assert row[1] == "d"
+    assert _nested_list(row[2]) == [[["x"], []]]
+    assert _nested_list(
+        list(conn.execute("MATCH (p:Person) RETURN collect(p.tags);"))[0][0]
+    ) == [["c", "d", "e"], ["merged"]]
+    assert [
+        item[0]
+        for item in conn.execute(
+            "MATCH (p:Person {id: 1}) UNWIND p.tags AS tag " "RETURN tag ORDER BY tag;"
+        )
+    ] == ["c", "d", "e"]
+    assert [
+        item[0]
+        for item in conn.execute(
+            "MATCH (p:Person {id: 1}) UNWIND p.nested AS pair "
+            "UNWIND pair[0] AS value RETURN value;"
+        )
+    ] == ["x"]
 
     conn.close()
     db.close()
 
+    db = Database(db_path=db_path, mode="w", checkpoint_on_close=False)
+    conn = db.connect()
+    assert _nested_list(
+        list(conn.execute("MATCH (p:Person {id: 1}) RETURN p.tags;"))[0][0]
+    ) == ["c", "d", "e"]
+    assert _nested_list(
+        list(
+            conn.execute(
+                "MATCH (:Person {id: 1})-[e:Knows]->(:Person {id: 2}) "
+                "RETURN e.scores;"
+            )
+        )[0][0]
+    ) == [7]
 
-def test_return_multiple_lists(tmp_path):
-    db_dir = tmp_path / "return_multiple_lists"
-    shutil.rmtree(db_dir, ignore_errors=True)
-    db_dir.mkdir()
-    db = Database(db_path=str(db_dir), mode="w")
+    conn.close()
+    db.close()
+
+
+def test_copy_recursive_list_from_csv_and_json(tmp_path):
+    csv_path = tmp_path / "nested.csv"
+    csv_path.write_text(
+        "id|nested\n" "1|[[[a],[]],[[b,c],[d]]]\n" "2|[]\n",
+        encoding="utf-8",
+    )
+    json_path = tmp_path / "nested.json"
+    json_path.write_text(
+        '[{"id":3,"nested":[[["e"],["f","g"]]]},'
+        '{"id":4,"nested":[[null,["h"]],null]}]',
+        encoding="utf-8",
+    )
+
+    db = Database(db_path=str(tmp_path / "db"), mode="w", checkpoint_on_close=False)
+    conn = db.connect()
+    conn.execute(
+        "CREATE NODE TABLE T(id INT64, nested STRING[][2][], PRIMARY KEY(id));"
+    )
+    conn.execute(f'COPY T FROM "{csv_path}" (header = true, delim = "|");')
+    conn.execute(f'COPY T FROM "{json_path}";')
+
+    rows = list(conn.execute("MATCH (n:T) RETURN n.id, n.nested ORDER BY n.id;"))
+    assert [[row[0], _nested_list(row[1])] for row in rows] == [
+        [1, [[["a"], []], [["b", "c"], ["d"]]]],
+        [2, []],
+        [3, [[["e"], ["f", "g"]]]],
+        [4, [[[], ["h"]], [[], []]]],
+    ]
+
+    bad_json = tmp_path / "bad.json"
+    bad_json.write_text('[{"id":5,"nested":[[["x"],["y"],["z"]]]}]', encoding="utf-8")
+    with pytest.raises(Exception):
+        conn.execute(f'COPY T FROM "{bad_json}";')
+
+    conn.close()
+    db.close()
+
+
+def test_return_single_list(tmp_path):
+    db = Database(db_path=str(tmp_path), mode="w")
     conn = db.connect()
 
     conn.execute(
@@ -60,23 +193,19 @@ def test_return_multiple_lists(tmp_path):
     conn.execute("CREATE (p: PERSON {id: 0, name: 'Alice', score: 99.5});")
     conn.execute("CREATE (p: PERSON {id: 1, name: 'Bob', score: 98.5});")
 
-    result = conn.execute("MATCH (p: PERSON) RETURN [p.id], [p.name, p.score];")
+    result = conn.execute(
+        "MATCH (p: PERSON) RETURN [p.id, p.name, p.score] ORDER BY p.id;"
+    )
     result = list(result)
-    assert result[0][0] == [0]
-    assert result[0][1] == ["Alice", 99.5]
-    assert result[1][0] == [1]
-    assert result[1][1] == ["Bob", 98.5]
+    assert result[0][0] == [0, "Alice", 99.5]
+    assert result[1][0] == [1, "Bob", 98.5]
 
     conn.close()
     db.close()
 
 
-@pytest.mark.skip(reason="list nesting is not supported")
-def test_return_nesting_lists(tmp_path):
-    db_dir = tmp_path / "return_nesting_lists"
-    shutil.rmtree(db_dir, ignore_errors=True)
-    db_dir.mkdir()
-    db = Database(db_path=str(db_dir), mode="w")
+def test_return_multiple_lists(tmp_path):
+    db = Database(db_path=str(tmp_path), mode="w")
     conn = db.connect()
 
     conn.execute(
@@ -85,10 +214,14 @@ def test_return_nesting_lists(tmp_path):
     conn.execute("CREATE (p: PERSON {id: 0, name: 'Alice', score: 99.5});")
     conn.execute("CREATE (p: PERSON {id: 1, name: 'Bob', score: 98.5});")
 
-    result = conn.execute("MATCH (p: PERSON) RETURN [[p.id], [p.name, p.score]];")
+    result = conn.execute(
+        "MATCH (p: PERSON) RETURN [p.id], [p.name, p.score] ORDER BY p.id;"
+    )
     result = list(result)
-    assert result[0][0] == [[0], ["Alice", 99.5]]
-    assert result[1][0] == [[1], ["Bob", 98.5]]
+    assert result[0][0] == [0]
+    assert result[0][1] == ["Alice", 99.5]
+    assert result[1][0] == [1]
+    assert result[1][1] == ["Bob", 98.5]
 
     conn.close()
     db.close()