refactor: clarify module boundaries for property types and export IO

extension/gds/include/impl/leiden_impl.h

@@ -19,9 +19,9 @@
 #include <memory>
 #include <vector>
 
+#include "neug/common/types/property_types.h"
 #include "neug/execution/common/context.h"
 #include "neug/storages/graph/graph_interface.h"
-#include "neug/utils/property/types.h"
 
 namespace neug {
 namespace gds {

extension/gds/include/impl/louvain_impl.h

@@ -19,9 +19,9 @@
 #include <memory>
 #include <vector>
 
+#include "neug/common/types/property_types.h"
 #include "neug/execution/common/context.h"
 #include "neug/storages/graph/graph_interface.h"
-#include "neug/utils/property/types.h"
 
 namespace neug {
 namespace gds {

extension/gds/include/utils/parallel_utils.h

@@ -21,8 +21,8 @@
 #include <thread>
 #include <vector>
 
+#include "neug/common/types/property_types.h"
 #include "neug/storages/graph/vertex_table.h"
-#include "neug/utils/property/types.h"
 
 namespace neug {
 namespace gds {

extension/parquet/include/parquet/arrow_reader.h

@@ -15,12 +15,14 @@
 #pragma once
 
 #include <memory>
+#include <vector>
 
 #include <arrow/dataset/dataset.h>
 #include <arrow/dataset/scanner.h>
 #include <arrow/filesystem/filesystem.h>
 #include <arrow/type.h>
 
+#include "neug/common/types/data_chunk.h"
 #include "neug/utils/io/reader.h"
 #include "parquet/arrow_options.h"
 
@@ -47,8 +49,8 @@ class Reader {
         fileSystem(std::move(fileSystem)) {}
   virtual ~Reader() = default;
 
-  virtual void read(std::shared_ptr<ReadLocalState> localState,
-                    execution::Context& ctx) = 0;
+  virtual std::vector<DataChunk> read(
+      std::shared_ptr<ReadLocalState> localState) = 0;
 
  protected:
   std::shared_ptr<ReadSharedState> sharedState;
@@ -72,18 +74,18 @@ class ArrowReader : public Reader<arrow::fs::FileSystem> {
         datasetBuilder(std::move(datasetBuilder)) {}
   ~ArrowReader() override = default;
 
-  void read(std::shared_ptr<ReadLocalState> localState,
-            execution::Context& ctx) override;
+  std::vector<DataChunk> read(
+      std::shared_ptr<ReadLocalState> localState) override;
 
   arrow::Result<std::shared_ptr<arrow::Schema>> inferSchema();
 
  protected:
   std::shared_ptr<arrow::dataset::Scanner> createScanner(
       std::shared_ptr<arrow::fs::FileSystem> fs);
-  void full_read(std::shared_ptr<arrow::dataset::Scanner> scanner,
-                 execution::Context& output);
-  void batch_read(std::shared_ptr<arrow::dataset::Scanner> scanner,
-                  execution::Context& output);
+  std::vector<DataChunk> full_read(
+      std::shared_ptr<arrow::dataset::Scanner> scanner);
+  std::vector<DataChunk> batch_read(
+      std::shared_ptr<arrow::dataset::Scanner> scanner);
 
   std::unique_ptr<ArrowOptionsBuilder> optionsBuilder;
   std::shared_ptr<DatasetBuilder> datasetBuilder;

extension/parquet/include/parquet_export_function.h

@@ -38,7 +38,8 @@ class ArrowParquetExportWriter : public QueryExportWriter {
         fileSystem_(std::move(fileSystem)) {}
   ~ArrowParquetExportWriter() override = default;
 
-  neug::Status writeTable(const QueryResponse* table) override;
+  neug::Status write(const DataChunk& chunk,
+                     const std::vector<DataType>& source_types = {}) override;
 
  private:
   std::shared_ptr<arrow::fs::FileSystem> fileSystem_;

extension/parquet/include/parquet_read_function.h

@@ -66,7 +66,7 @@ struct ParquetReadFunction {
 
     execution::Context ctx;
     auto localState = std::make_shared<reader::ReadLocalState>();
-    reader->read(localState, ctx);
+    ctx.append_chunks(reader->read(localState));
     return ctx;
   }
 

extension/parquet/src/arrow_column.cc

@@ -146,7 +146,7 @@ static Value arrow_value_at(const arrow::Array& array, int64_t index,
     Interval interval;
     interval.from_mill_seconds(
         arrow_time_to_milliseconds(value, duration_type.unit()));
-    return Value::INTERVAL(interval);
+    return Value::INTERVAL(interval.normalize());
   }
   case arrow::Type::FIXED_SIZE_LIST: {
     const auto& list = static_cast<const arrow::FixedSizeListArray&>(array);

extension/parquet/src/arrow_options.cc

@@ -69,7 +69,10 @@ bool ArrowOptionsBuilder::projectColumns(ArrowOptions& options) {
     }
   }
 
-  auto dataset_schema = createSchema(entrySchema);
+  auto dataset_schema = options.scanOptions->dataset_schema;
+  if (!dataset_schema) {
+    dataset_schema = createSchema(entrySchema);
+  }
   auto project_desc =
       arrow::dataset::ProjectionDescr::FromNames(columns, *dataset_schema);
   if (!project_desc.ok()) {

extension/parquet/src/arrow_reader.cc

@@ -24,16 +24,60 @@
 #include "parquet/record_batch_supplier.h"
 
 #include "neug/compiler/common/assert.h"
-#include "neug/execution/common/context.h"
 #include "neug/storages/loader/loader_utils.h"
 #include "neug/utils/exception/exception.h"
 #include "neug/utils/io/read/common/options.h"
 
 namespace neug {
 namespace reader {
 
-void ArrowReader::read(std::shared_ptr<ReadLocalState> localState,
-                       execution::Context& ctx) {
+static std::shared_ptr<arrow::DataType> reconcileIntervalType(
+    const std::shared_ptr<arrow::DataType>& expected,
+    const std::shared_ptr<arrow::DataType>& file_type) {
+  if ((expected->id() == arrow::Type::STRING ||
+       expected->id() == arrow::Type::LARGE_STRING) &&
+      file_type->id() == arrow::Type::DURATION) {
+    return file_type;
+  }
+  if (expected->id() != arrow::Type::FIXED_SIZE_LIST ||
+      file_type->id() != arrow::Type::FIXED_SIZE_LIST) {
+    return expected;
+  }
+
+  const auto& expected_list =
+      static_cast<const arrow::FixedSizeListType&>(*expected);
+  const auto& file_list =
+      static_cast<const arrow::FixedSizeListType&>(*file_type);
+  if (expected_list.list_size() != file_list.list_size()) {
+    return expected;
+  }
+  auto child =
+      reconcileIntervalType(expected_list.value_type(), file_list.value_type());
+  if (child->Equals(expected_list.value_type())) {
+    return expected;
+  }
+  return arrow::fixed_size_list(child, expected_list.list_size());
+}
+
+static std::shared_ptr<arrow::Schema> reconcileIntervalSchema(
+    const std::shared_ptr<arrow::Schema>& expected,
+    const std::shared_ptr<arrow::Schema>& file_schema) {
+  std::vector<std::shared_ptr<arrow::Field>> fields;
+  fields.reserve(expected->num_fields());
+  for (const auto& field : expected->fields()) {
+    auto file_field = file_schema->GetFieldByName(field->name());
+    if (!file_field) {
+      fields.push_back(field);
+      continue;
+    }
+    auto type = reconcileIntervalType(field->type(), file_field->type());
+    fields.push_back(field->WithType(std::move(type)));
+  }
+  return arrow::schema(std::move(fields), expected->metadata());
+}
+
+std::vector<DataChunk> ArrowReader::read(
+    std::shared_ptr<ReadLocalState> localState) {
   if (!sharedState) {
     THROW_INVALID_ARGUMENT_EXCEPTION("SharedState is null");
   }
@@ -49,10 +93,9 @@ void ArrowReader::read(std::shared_ptr<ReadLocalState> localState,
   const auto& fileSchema = sharedState->schema.file;
   ReadOptions options;
   if (options.batch_read.get(fileSchema.options)) {
-    batch_read(scanner, ctx);
-  } else {
-    full_read(scanner, ctx);
+    return batch_read(scanner);
   }
+  return full_read(scanner);
 }
 
 std::shared_ptr<arrow::dataset::Scanner> ArrowReader::createScanner(
@@ -81,10 +124,6 @@ std::shared_ptr<arrow::dataset::Scanner> ArrowReader::createScanner(
     THROW_INVALID_ARGUMENT_EXCEPTION("Failed to build arrow options");
   }
 
-  if (!optionsBuilder->projectColumns(arrowOptions)) {
-    LOG(WARNING) << "Failed to set column projection, using all columns";
-  }
-
   if (!optionsBuilder->skipRows(arrowOptions)) {
     LOG(WARNING) << "Failed to set row filter, using no filter";
   }
@@ -110,6 +149,8 @@ std::shared_ptr<arrow::dataset::Scanner> ArrowReader::createScanner(
                                 fileSchema->ToString());
         }
       }
+      scan_opts->dataset_schema =
+          reconcileIntervalSchema(scan_opts->dataset_schema, fileSchema);
     }
     dataset_result = factory->Finish(scan_opts->dataset_schema);
   } else {
@@ -125,6 +166,10 @@ std::shared_ptr<arrow::dataset::Scanner> ArrowReader::createScanner(
   }
   auto dataset = dataset_result.ValueOrDie();
 
+  if (!optionsBuilder->projectColumns(arrowOptions)) {
+    LOG(WARNING) << "Failed to set column projection, using all columns";
+  }
+
   arrow::dataset::ScannerBuilder scanner_builder(dataset, scan_opts);
   auto scanner_result = scanner_builder.Finish();
   if (!scanner_result.ok()) {
@@ -136,8 +181,8 @@ std::shared_ptr<arrow::dataset::Scanner> ArrowReader::createScanner(
   return scanner_result.ValueOrDie();
 }
 
-void ArrowReader::full_read(std::shared_ptr<arrow::dataset::Scanner> scanner,
-                            execution::Context& output) {
+std::vector<DataChunk> ArrowReader::full_read(
+    std::shared_ptr<arrow::dataset::Scanner> scanner) {
   if (!sharedState) {
     THROW_INVALID_ARGUMENT_EXCEPTION("SharedState is null");
   }
@@ -162,17 +207,18 @@ void ArrowReader::full_read(std::shared_ptr<arrow::dataset::Scanner> scanner,
         ", table: " + std::to_string(table->num_columns()));
   }
 
-  output.clear();
   DataChunk chunk;
   for (int i = 0; i < num_cols; ++i) {
     auto table_column = table->column(i);
     chunk.set(i, arrow_arrays_to_value_column(table_column->chunks()));
   }
-  output.append_chunk(std::move(chunk));
+  std::vector<DataChunk> output;
+  output.push_back(std::move(chunk));
+  return output;
 }
 
-void ArrowReader::batch_read(std::shared_ptr<arrow::dataset::Scanner> scanner,
-                             execution::Context& output) {
+std::vector<DataChunk> ArrowReader::batch_read(
+    std::shared_ptr<arrow::dataset::Scanner> scanner) {
   if (!sharedState) {
     THROW_INVALID_ARGUMENT_EXCEPTION("SharedState is null");
   }
@@ -203,10 +249,11 @@ void ArrowReader::batch_read(std::shared_ptr<arrow::dataset::Scanner> scanner,
   auto batch_supplier =
       std::make_shared<RecordBatchChunkSupplier>(batch_reader, row_num);
 
-  output.clear();
+  std::vector<DataChunk> output;
   while (auto chunk = batch_supplier->GetNextChunk()) {
-    output.append_chunk(std::move(*chunk));
+    output.push_back(std::move(*chunk));
   }
+  return output;
 }
 
 arrow::Result<std::shared_ptr<arrow::Schema>> ArrowReader::inferSchema() {