feat(interactive): support long_text in interactive (#4522)

Fixes #4526 #4368

---------

Co-authored-by: [email protected] <[email protected]>

Author: liulx20

flex/interactive/sdk/python/gs_interactive/tests/conftest.py

@@ -1183,6 +1183,34 @@ def import_data_to_full_modern_graph(sess: Session, graph_id: str):
     assert wait_job_finish(sess, job_id)
 
 
+def import_long_string_data_data_to_vertex_only_modern_graph(
+    sess: Session, graph_id: str
+):
+    schema_mapping = SchemaMapping.from_dict(modern_graph_vertex_only_import_config)
+    # create a long string file under MODERN_GRAPH_DATA_DIR
+    long_string_file = os.path.join(MODERN_GRAPH_DATA_DIR, "person_long_string.csv")
+
+    # By default, the max length of a string property is 256,
+    # a string with length 4096 should be enough to test the long string feature
+    def generate_large_string(sample: str):
+        return sample * 4096
+
+    with open(long_string_file, "w") as f:
+        f.write("id|name|age\n")
+        f.write("1|" + generate_large_string("marko") + "|29\n")
+        f.write("2|" + generate_large_string("vadas") + "|27\n")
+        f.write("4|" + generate_large_string("josh") + "|32\n")
+        f.write("6|" + generate_large_string("peter") + "|35\n")
+
+    schema_mapping.vertex_mappings[0].inputs[0] = "person_long_string.csv"
+    resp = sess.bulk_loading(graph_id, schema_mapping)
+    assert resp.is_ok()
+    job_id = resp.get_value().job_id
+    assert wait_job_finish(sess, job_id)
+    # return a callable to clean up the long string file
+    os.remove(long_string_file)
+
+
 def import_data_to_full_graph_algo_graph(sess: Session, graph_id: str):
     schema_mapping = SchemaMapping.from_dict(graph_algo_graph_import_config)
     resp = sess.bulk_loading(graph_id, schema_mapping)

flex/interactive/sdk/python/gs_interactive/tests/test_robustness.py

@@ -38,6 +38,9 @@
 from gs_interactive.tests.conftest import (
     import_data_to_vertex_only_modern_graph_no_wait,
 )
+from gs_interactive.tests.conftest import (
+    import_long_string_data_data_to_vertex_only_modern_graph,
+)
 from gs_interactive.tests.conftest import run_cypher_test_suite
 from gs_interactive.tests.conftest import send_get_request_periodically
 from gs_interactive.tests.conftest import start_service_on_graph
@@ -540,3 +543,21 @@ def test_create_graph_with_temporal_type(
     result = neo4j_session.run("MATCH (n: person) return n.birthday AS birthday;")
     records = result.fetch(10)
     assert len(records) == 4
+
+
+def test_graph_with_long_text_property(
+    interactive_session, neo4j_session, create_vertex_only_modern_graph
+):
+    print("[Test graph with long text property]")
+    import_long_string_data_data_to_vertex_only_modern_graph(
+        interactive_session, create_vertex_only_modern_graph
+    )
+    start_service_on_graph(interactive_session, create_vertex_only_modern_graph)
+    ensure_compiler_schema_ready(
+        interactive_session, neo4j_session, create_vertex_only_modern_graph
+    )
+    result = neo4j_session.run("MATCH (n: person) return n.name AS name;")
+    records = result.fetch(10)
+    assert len(records) == 4
+    for record in records:
+        assert len(record["name"]) > 4096

flex/storages/rt_mutable_graph/loader/abstract_arrow_fragment_loader.cc

@@ -40,9 +40,14 @@ bool check_primary_key_type(std::shared_ptr<arrow::DataType> data_type) {
 
 void set_column_from_string_array(gs::ColumnBase* col,
                                   std::shared_ptr<arrow::ChunkedArray> array,
-                                  const std::vector<size_t>& offset) {
+                                  const std::vector<size_t>& offset,
+                                  bool enable_resize) {
   auto type = array->type();
   auto size = col->size();
+  auto typed_col = dynamic_cast<gs::TypedColumn<std::string_view>*>(col);
+  if (enable_resize) {
+    CHECK(typed_col != nullptr) << "Only support TypedColumn<std::string_view>";
+  }
   CHECK(type->Equals(arrow::large_utf8()) || type->Equals(arrow::utf8()))
       << "Inconsistent data type, expect string, but got " << type->ToString();
   size_t cur_ind = 0;
@@ -62,7 +67,11 @@ void set_column_from_string_array(gs::ColumnBase* col,
         if (offset[cur_ind] >= size) {
           cur_ind++;
         } else {
-          col->set_any(offset[cur_ind++], std::move(sw));
+          if (!enable_resize) {
+            col->set_any(offset[cur_ind++], std::move(sw));
+          } else {
+            typed_col->set_value_safe(offset[cur_ind++], std::move(sw));
+          }
         }
       }
     }
@@ -76,7 +85,11 @@ void set_column_from_string_array(gs::ColumnBase* col,
         if (offset[cur_ind] >= size) {
           cur_ind++;
         } else {
-          col->set_any(offset[cur_ind++], std::move(sw));
+          if (!enable_resize) {
+            col->set_any(offset[cur_ind++], std::move(sw));
+          } else {
+            typed_col->set_value_safe(offset[cur_ind++], std::move(sw));
+          }
         }
       }
     }
@@ -113,7 +126,7 @@ void set_properties_column(gs::ColumnBase* col,
   } else if (col_type.type_enum == impl::PropertyTypeImpl::kVarChar) {
     set_column_from_string_array(col, array, offset);
   } else if (col_type == PropertyType::kStringView) {
-    set_column_from_string_array(col, array, offset);
+    set_column_from_string_array(col, array, offset, true);
   } else {
     LOG(FATAL) << "Not support type: " << type->ToString();
   }

flex/storages/rt_mutable_graph/loader/abstract_arrow_fragment_loader.h

@@ -72,7 +72,8 @@ void set_column(gs::ColumnBase* col, std::shared_ptr<arrow::ChunkedArray> array,
 // For String types.
 void set_column_from_string_array(gs::ColumnBase* col,
                                   std::shared_ptr<arrow::ChunkedArray> array,
-                                  const std::vector<size_t>& offset);
+                                  const std::vector<size_t>& offset,
+                                  bool enable_resize = false);
 
 void set_column_from_timestamp_array(gs::ColumnBase* col,
                                      std::shared_ptr<arrow::ChunkedArray> array,

flex/utils/property/column.cc

@@ -218,6 +218,45 @@ std::shared_ptr<ColumnBase> CreateColumn(
   }
 }
 
+void TypedColumn<std::string_view>::set_value_safe(
+    size_t idx, const std::string_view& value) {
+  std::shared_lock<std::shared_mutex> lock(rw_mutex_);
+
+  if (idx >= basic_size_ && idx < basic_size_ + extra_size_) {
+    size_t offset = pos_.fetch_add(value.size());
+    if (pos_.load() > extra_buffer_.data_size()) {
+      lock.unlock();
+      std::unique_lock<std::shared_mutex> w_lock(rw_mutex_);
+      if (pos_.load() > extra_buffer_.data_size()) {
+        size_t new_avg_width =
+            (pos_.load() + idx - basic_size_) / (idx - basic_size_ + 1);
+        size_t new_len = std::max(extra_size_ * new_avg_width, pos_.load());
+        extra_buffer_.resize(extra_buffer_.size(), new_len);
+      }
+      w_lock.unlock();
+      lock.lock();
+    }
+    extra_buffer_.set(idx - basic_size_, offset, value);
+  } else if (idx < basic_size_) {
+    size_t offset = basic_pos_.fetch_add(value.size());
+    if (basic_pos_.load() > basic_buffer_.data_size()) {
+      lock.unlock();
+      std::unique_lock<std::shared_mutex> w_lock(rw_mutex_);
+      if (basic_pos_.load() > basic_buffer_.data_size()) {
+        size_t new_avg_width = (basic_pos_.load() + idx) / (idx + 1);
+        size_t new_len =
+            std::max(basic_size_ * new_avg_width, basic_pos_.load());
+        basic_buffer_.resize(basic_buffer_.size(), new_len);
+      }
+      w_lock.unlock();
+      lock.lock();
+    }
+    basic_buffer_.set(idx, offset, value);
+  } else {
+    LOG(FATAL) << "Index out of range";
+  }
+}
+
 std::shared_ptr<RefColumnBase> CreateRefColumn(
     std::shared_ptr<ColumnBase> column) {
   auto type = column->type();

flex/utils/property/column.h

@@ -16,6 +16,7 @@
 #ifndef GRAPHSCOPE_PROPERTY_COLUMN_H_
 #define GRAPHSCOPE_PROPERTY_COLUMN_H_
 
+#include <shared_mutex>
 #include <string>
 #include <string_view>
 #include "grape/utils/concurrent_queue.h"
@@ -345,9 +346,14 @@ class TypedColumn<grape::EmptyType> : public ColumnBase {
 template <>
 class TypedColumn<std::string_view> : public ColumnBase {
  public:
-  TypedColumn(StorageStrategy strategy,
-              uint16_t width = PropertyType::GetStringDefaultMaxLength())
-      : strategy_(strategy), width_(width) {}
+  TypedColumn(StorageStrategy strategy, uint16_t width)
+      : strategy_(strategy),
+        width_(width),
+        type_(PropertyType::Varchar(width_)) {}
+  TypedColumn(StorageStrategy strategy)
+      : strategy_(strategy),
+        width_(PropertyType::GetStringDefaultMaxLength()),
+        type_(PropertyType::kStringView) {}
   ~TypedColumn() { close(); }
 
   void open(const std::string& name, const std::string& snapshot_dir,
@@ -478,6 +484,7 @@ class TypedColumn<std::string_view> : public ColumnBase {
   size_t size() const override { return basic_size_ + extra_size_; }
 
   void resize(size_t size) override {
+    std::unique_lock<std::shared_mutex> lock(rw_mutex_);
     if (size < basic_buffer_.size()) {
       basic_size_ = size;
       extra_size_ = 0;
@@ -504,7 +511,7 @@ class TypedColumn<std::string_view> : public ColumnBase {
     }
   }
 
-  PropertyType type() const override { return PropertyType::Varchar(width_); }
+  PropertyType type() const override { return type_; }
 
   void set_value(size_t idx, const std::string_view& val) {
     auto copied_val = val;
@@ -547,6 +554,8 @@ class TypedColumn<std::string_view> : public ColumnBase {
     }
   }
 
+  void set_value_safe(size_t idx, const std::string_view& value);
+
   inline std::string_view get_view(size_t idx) const {
     return idx < basic_size_ ? basic_buffer_.get(idx)
                              : extra_buffer_.get(idx - basic_size_);
@@ -584,7 +593,9 @@ class TypedColumn<std::string_view> : public ColumnBase {
   std::atomic<size_t> pos_;
   std::atomic<size_t> basic_pos_;
   StorageStrategy strategy_;
+  std::shared_mutex rw_mutex_;
   uint16_t width_;
+  PropertyType type_;
 };
 
 using StringColumn = TypedColumn<std::string_view>;