Merge branch 'main' into dirty-bit-2

Author: liulx20

.github/workflows/build-and-test-common.yml

@@ -305,6 +305,7 @@ jobs:
         python3 -m pytest --cov=neug --cov-report=term --exitfirst --cov-append --cov-config=.coveragerc -sv tests/test_query.py
         python3 -m pytest --cov=neug --cov-report=term --exitfirst --cov-append --cov-config=.coveragerc -sv tests/test_tp_service.py
         python3 -m pytest --cov=neug --cov-report=term --exitfirst --cov-append --cov-config=.coveragerc -sv tests/test_load.py
+        python3 -m pytest --cov=neug --cov-report=term --exitfirst --cov-append --cov-config=.coveragerc -sv tests/test_load_array.py
         python3 -m pytest --cov=neug --cov-report=term --exitfirst --cov-append --cov-config=.coveragerc -sv tests/test_export.py
         python3 -m pytest --cov=neug --cov-report=term --exitfirst --cov-append --cov-config=.coveragerc -sv tests/test_copy_temp.py
 

.github/workflows/neug-extension-test.yml

@@ -140,6 +140,7 @@ jobs:
         export FLEX_DATA_DIR=${GITHUB_WORKSPACE}/example_dataset/comprehensive_graph
         GLOG_v=10 ${GITHUB_WORKSPACE}/build/tools/utils/bulk_loader -g ../../example_dataset/comprehensive_graph/graph.yaml -l ../../example_dataset/comprehensive_graph/import.yaml -d /tmp/comprehensive_graph
         NEUG_RUN_EXTENSION_TESTS=true python3 -m pytest -sv tests/test_load.py -k "parquet or httpfs"
+        NEUG_RUN_EXTENSION_TESTS=true python3 -m pytest -sv tests/test_load_array.py -k "parquet"
         NEUG_RUN_EXTENSION_TESTS=true python3 -m pytest -sv tests/test_export.py -k "parquet or httpfs"
         NEUG_RUN_EXTENSION_TESTS=true python3 -m pytest -sv tests/test_db_import_export.py -k "copy_from_parquet"
         NEUG_RUN_EXTENSION_TESTS=true python3 -m pytest -sv tests/test_sniffer.py -k "parquet"

doc/source/data_io/import_data.md

@@ -240,7 +240,25 @@ The following options control how CSV files are parsed during `COPY FROM`. These
 | `quoting`  | bool | `true` | Whether to enable quote processing            |
 | `escaping` | bool | `true` | Whether to enable escape character processing |
 
-> **Array limitation:** `COPY FROM` CSV currently does not auto-detect or directly materialize fixed-size `ARRAY` columns from bracketed CSV fields such as `"[1,2,3]"`. Use Cypher literals/parameters for array properties, or a typed non-CSV ingestion path when available.
+`COPY FROM` supports loading `ARRAY` data from CSV,
+JSON, and Parquet files. NeuG does not implicitly convert input values to
+`ARRAY`; use `LOAD FROM` and explicitly cast the column to a fixed-size
+`ARRAY` type.
+
+For example, given a `person_array.csv` file with an array column:
+
+```csv
+id,name,address
+1,Alice,"[Beijing,Hangzhou,Shanghai]"
+2,Bob,"[London,Paris,Berlin]"
+```
+
+```cypher
+COPY Person FROM (
+    LOAD FROM "person_array.csv" (delim=',')
+    RETURN id, name, CAST(address, 'STRING[3]') AS addresses
+);
+```
 
 ### JSON/JSONL
 

doc/source/data_io/load_data.md

@@ -46,7 +46,23 @@ LOAD FROM "person.csv" (delim=',', header=true)
 RETURN name, age;
 ```
 
-> **Array limitation:** `LOAD FROM` CSV currently does not parse CSV fields into fixed-size `ARRAY` values. A quoted field such as `"[1,2,3]"` is read as a `STRING`, and casting CSV columns to `INT64[3]`/other fixed-size array targets inside `LOAD FROM` is not supported yet.
+NeuG supports reading `ARRAY` data from CSV, JSON,
+and Parquet files. NeuG does not implicitly convert input values to `ARRAY`;
+explicitly cast the column to a fixed-size `ARRAY` type in the `RETURN`
+clause.
+
+For example, given a `person_array.csv` file with an array column:
+
+```csv
+id,name,address
+1,Alice,"[Beijing,Hangzhou,Shanghai]"
+2,Bob,"[London,Paris,Berlin]"
+```
+
+```cypher
+LOAD FROM "person_array.csv" (delim=',')
+RETURN id, name, CAST(address, 'STRING[3]') AS addresses;
+```
 
 ### JSON / JSONL
 

extension/parquet/src/arrow_column.cc

@@ -16,13 +16,181 @@
 #include "parquet/arrow_column.h"
 
 #include <arrow/array/array_binary.h>
+#include <arrow/array/array_nested.h>
 #include <arrow/type.h>
 
+#include "neug/common/columns/columns_utils.h"
 #include "neug/common/columns/value_columns.h"
+#include "neug/common/types/value.h"
 #include "neug/utils/exception/exception.h"
 
 namespace neug {
 
+static int64_t arrow_time_to_milliseconds(int64_t value,
+                                          arrow::TimeUnit::type unit) {
+  switch (unit) {
+  case arrow::TimeUnit::SECOND:
+    return value * Interval::MSECS_PER_SEC;
+  case arrow::TimeUnit::MILLI:
+    return value;
+  case arrow::TimeUnit::MICRO:
+    return value / Interval::MICROS_PER_MSEC;
+  case arrow::TimeUnit::NANO:
+    return value / (Interval::MICROS_PER_MSEC * 1000);
+  }
+  THROW_NOT_SUPPORTED_EXCEPTION("Unsupported Arrow time unit");
+}
+
+// Keep this mapping in sync with arrow_value_at when adding Arrow types.
+static DataType arrow_type_to_neug_type(const arrow::DataType& type) {
+  switch (type.id()) {
+  case arrow::Type::BOOL:
+    return DataType::BOOLEAN;
+  case arrow::Type::INT32:
+    return DataType::INT32;
+  case arrow::Type::INT64:
+    return DataType::INT64;
+  case arrow::Type::UINT32:
+    return DataType::UINT32;
+  case arrow::Type::UINT64:
+    return DataType::UINT64;
+  case arrow::Type::FLOAT:
+    return DataType::FLOAT;
+  case arrow::Type::DOUBLE:
+    return DataType::DOUBLE;
+  case arrow::Type::STRING:
+  case arrow::Type::LARGE_STRING:
+    return DataType::VARCHAR;
+  case arrow::Type::DATE32:
+  case arrow::Type::DATE64:
+    return DataType::DATE;
+  case arrow::Type::TIMESTAMP:
+    return DataType::TIMESTAMP_MS;
+  case arrow::Type::DURATION:
+    return DataType::INTERVAL;
+  case arrow::Type::FIXED_SIZE_LIST: {
+    const auto& array_type = static_cast<const arrow::FixedSizeListType&>(type);
+    return DataType::Array(arrow_type_to_neug_type(*array_type.value_type()),
+                           array_type.list_size());
+  }
+  case arrow::Type::LIST:
+  case arrow::Type::LARGE_LIST:
+    THROW_NOT_SUPPORTED_EXCEPTION(
+        "Parquet LIST is not supported as ARRAY. Specify a fixed-size Arrow "
+        "list / NeuG ARRAY type instead.");
+  default:
+    THROW_NOT_SUPPORTED_EXCEPTION("Unsupported arrow type: " + type.ToString());
+  }
+}
+
+static Value arrow_value_at(const arrow::Array& array, int64_t index,
+                            const DataType& type) {
+  if (array.IsNull(index)) {
+    return Value(type);
+  }
+  switch (array.type_id()) {
+  case arrow::Type::BOOL:
+    return Value::BOOLEAN(
+        static_cast<const arrow::BooleanArray&>(array).Value(index));
+  case arrow::Type::INT32:
+    return Value::INT32(
+        static_cast<const arrow::Int32Array&>(array).Value(index));
+  case arrow::Type::INT64:
+    return Value::INT64(
+        static_cast<const arrow::Int64Array&>(array).Value(index));
+  case arrow::Type::UINT32:
+    return Value::UINT32(
+        static_cast<const arrow::UInt32Array&>(array).Value(index));
+  case arrow::Type::UINT64:
+    return Value::UINT64(
+        static_cast<const arrow::UInt64Array&>(array).Value(index));
+  case arrow::Type::FLOAT:
+    return Value::FLOAT(
+        static_cast<const arrow::FloatArray&>(array).Value(index));
+  case arrow::Type::DOUBLE:
+    return Value::DOUBLE(
+        static_cast<const arrow::DoubleArray&>(array).Value(index));
+  case arrow::Type::STRING:
+    return Value::STRING(std::string(
+        static_cast<const arrow::StringArray&>(array).GetView(index)));
+  case arrow::Type::LARGE_STRING:
+    return Value::STRING(std::string(
+        static_cast<const arrow::LargeStringArray&>(array).GetView(index)));
+  case arrow::Type::DATE32: {
+    Date date;
+    date.from_num_days(
+        static_cast<const arrow::Date32Array&>(array).Value(index));
+    return Value::DATE(date);
+  }
+  case arrow::Type::DATE64: {
+    constexpr int64_t MILLIS_PER_DAY = 24LL * 60 * 60 * 1000;
+    Date date;
+    date.from_num_days(static_cast<int32_t>(
+        static_cast<const arrow::Date64Array&>(array).Value(index) /
+        MILLIS_PER_DAY));
+    return Value::DATE(date);
+  }
+  case arrow::Type::TIMESTAMP: {
+    const auto value =
+        static_cast<const arrow::TimestampArray&>(array).Value(index);
+    const auto& timestamp_type =
+        static_cast<const arrow::TimestampType&>(*array.type());
+    return Value::TIMESTAMPMS(
+        DateTime(arrow_time_to_milliseconds(value, timestamp_type.unit())));
+  }
+  case arrow::Type::DURATION: {
+    const auto value =
+        static_cast<const arrow::DurationArray&>(array).Value(index);
+    const auto& duration_type =
+        static_cast<const arrow::DurationType&>(*array.type());
+    Interval interval;
+    interval.from_mill_seconds(
+        arrow_time_to_milliseconds(value, duration_type.unit()));
+    return Value::INTERVAL(interval);
+  }
+  case arrow::Type::FIXED_SIZE_LIST: {
+    const auto& list = static_cast<const arrow::FixedSizeListArray&>(array);
+    const auto& array_type =
+        static_cast<const arrow::FixedSizeListType&>(*array.type());
+    const auto child_type = ArrayType::GetChildType(type);
+    std::vector<Value> values;
+    values.reserve(array_type.list_size());
+    const auto offset = list.value_offset(index);
+    for (int32_t i = 0; i < array_type.list_size(); ++i) {
+      values.push_back(arrow_value_at(*list.values(), offset + i, child_type));
+    }
+    return Value::ARRAY(type, std::move(values));
+  }
+  default:
+    THROW_NOT_SUPPORTED_EXCEPTION("Unsupported arrow type: " +
+                                  array.type()->ToString());
+  }
+}
+
+static std::shared_ptr<IContextColumn> convert_fixed_size_list_arrays(
+    const std::vector<std::shared_ptr<arrow::Array>>& arrays) {
+  const auto type = arrow_type_to_neug_type(*arrays.front()->type());
+  auto builder = ColumnsUtils::create_builder(type);
+  size_t size = 0;
+  for (const auto& array : arrays) {
+    size += array->length();
+  }
+  builder->reserve(size);
+  for (const auto& array : arrays) {
+    if (!array->type()->Equals(*arrays.front()->type())) {
+      THROW_SCHEMA_MISMATCH("Parquet ARRAY chunks have different types");
+    }
+    for (int64_t i = 0; i < array->length(); ++i) {
+      if (array->IsNull(i)) {
+        builder->push_back_null();
+        continue;
+      }
+      builder->push_back_elem(arrow_value_at(*array, i, type));
+    }
+  }
+  return builder->finish();
+}
+
 /// Convert numeric arrow arrays directly to ValueColumn<CppT>.
 template <typename ArrowArrayT, typename CppT>
 static std::shared_ptr<IContextColumn> convert_numeric_arrays(
@@ -89,7 +257,11 @@ static std::shared_ptr<IContextColumn> convert_date64_arrays(
       if (typed->IsNull(j)) {
         builder.push_back_null();
       } else {
-        builder.push_back_opt(Date(typed->Value(j)));
+        constexpr int64_t MILLIS_PER_DAY = 24LL * 60 * 60 * 1000;
+        Date date;
+        date.from_num_days(
+            static_cast<int32_t>(typed->Value(j) / MILLIS_PER_DAY));
+        builder.push_back_opt(date);
       }
     }
   }
@@ -106,7 +278,10 @@ static std::shared_ptr<IContextColumn> convert_timestamp_arrays(
       if (typed->IsNull(j)) {
         builder.push_back_null();
       } else {
-        builder.push_back_opt(DateTime(typed->Value(j)));
+        const auto& timestamp_type =
+            static_cast<const arrow::TimestampType&>(*typed->type());
+        builder.push_back_opt(DateTime(arrow_time_to_milliseconds(
+            typed->Value(j), timestamp_type.unit())));
       }
     }
   }
@@ -144,6 +319,13 @@ std::shared_ptr<IContextColumn> arrow_arrays_to_value_column(
     return convert_date64_arrays(arrays);
   case arrow::Type::TIMESTAMP:
     return convert_timestamp_arrays(arrays);
+  case arrow::Type::FIXED_SIZE_LIST:
+    return convert_fixed_size_list_arrays(arrays);
+  case arrow::Type::LIST:
+  case arrow::Type::LARGE_LIST:
+    THROW_NOT_SUPPORTED_EXCEPTION(
+        "Parquet LIST columns are not supported; use a fixed-size ARRAY "
+        "type such as FLOAT[3]");
   default:
     THROW_NOT_SUPPORTED_EXCEPTION("Unsupported arrow type: " +
                                   arrow_type->ToString());

include/neug/storages/loader/loader_utils.h

@@ -99,17 +99,17 @@ void fillVertexReaderMeta(label_t v_label, const std::string& v_label_name,
                           const std::string& v_file,
                           const LoadingConfig& loading_config,
                           const std::vector<std::string>& vertex_property_names,
-                          const std::vector<DataTypeId>& vertex_property_types,
-                          DataTypeId pk_type, const std::string& pk_name,
+                          const std::vector<DataType>& vertex_property_types,
+                          DataType pk_type, const std::string& pk_name,
                           size_t pk_ind, CsvReadConfig& config);
 
 void fillEdgeReaderMeta(label_t src_label_id, label_t dst_label_id,
                         label_t label_id, const std::string& edge_label_name,
                         const std::string& e_file,
                         const LoadingConfig& loading_config,
                         const std::vector<std::string>& edge_property_names,
-                        const std::vector<DataTypeId>& edge_property_types,
-                        DataTypeId src_pk_type, DataTypeId dst_pk_type,
+                        const std::vector<DataType>& edge_property_types,
+                        DataType src_pk_type, DataType dst_pk_type,
                         CsvReadConfig& config);
 
 void set_properties_from_context_column(

src/storages/loader/csv_property_graph_loader.cc

@@ -32,12 +32,12 @@ CSVPropertyGraphLoader::createVertexChunkSupplier(
     DataType pk_type, const std::string& pk_name, int pk_ind,
     const LoadingConfig& loading_config, int thread_id) const {
   auto vertex_property_names = schema_.get_vertex_property_names(v_label);
-  auto vertex_property_types = schema_.get_vertex_properties_id(v_label);
+  auto vertex_property_types = schema_.get_vertex_properties(v_label);
 
   CsvReadConfig config;
   fillVertexReaderMeta(v_label, v_label_name, v_file, loading_config,
-                       vertex_property_names, vertex_property_types,
-                       pk_type.id(), pk_name, pk_ind, config);
+                       vertex_property_names, vertex_property_types, pk_type,
+                       pk_name, pk_ind, config);
   return std::make_shared<CSVChunkSupplier>(v_file, std::move(config));
 }
 
@@ -49,7 +49,7 @@ CSVPropertyGraphLoader::createEdgeChunkSupplier(
   auto edge_property_names =
       schema_.get_edge_property_names(src_label_id, dst_label_id, e_label_id);
   auto edge_property_types =
-      schema_.get_edge_properties_id(src_label_id, dst_label_id, e_label_id);
+      schema_.get_edge_properties(src_label_id, dst_label_id, e_label_id);
   auto src_pk_type =
       std::get<0>(schema_.get_vertex_primary_key(src_label_id)[0]);
   auto dst_pk_type =
@@ -58,7 +58,7 @@ CSVPropertyGraphLoader::createEdgeChunkSupplier(
   fillEdgeReaderMeta(src_label_id, dst_label_id, e_label_id,
                      schema_.get_edge_label_name(e_label_id), e_file,
                      loading_config_, edge_property_names, edge_property_types,
-                     src_pk_type.id(), dst_pk_type.id(), config);
+                     src_pk_type, dst_pk_type, config);
   return std::make_shared<CSVChunkSupplier>(e_file, std::move(config));
 }