Add parsing without quotes

include/_mgp.hpp

@@ -782,4 +782,50 @@ inline void func_result_set_value(mgp_func_result *res, mgp_value *value, mgp_me
   MgInvokeVoid(mgp_func_result_set_value, res, value, memory);
 }
 
+// Messages
+
+inline mgp_source_type message_source_type(struct mgp_message *message) {
+  return MgInvoke<mgp_source_type>(mgp_message_source_type, message);
+}
+
+inline const char *message_payload(struct mgp_message *message) {
+  return MgInvoke<const char *>(mgp_message_payload, message);
+}
+
+inline size_t message_payload_size(struct mgp_message *message) {
+  return MgInvoke<size_t>(mgp_message_payload_size, message);
+}
+
+inline const char *message_topic_name(struct mgp_message *message) {
+  return MgInvoke<const char *>(mgp_message_topic_name, message);
+}
+
+inline const char *message_key(struct mgp_message *message) { return MgInvoke<const char *>(mgp_message_key, message); }
+
+inline size_t message_key_size(struct mgp_message *message) { return MgInvoke<size_t>(mgp_message_key_size, message); }
+
+inline int64_t message_timestamp(struct mgp_message *message) {
+  return MgInvoke<int64_t>(mgp_message_timestamp, message);
+}
+
+inline int64_t message_offset(struct mgp_message *message) { return MgInvoke<int64_t>(mgp_message_offset, message); }
+
+inline size_t messages_size(struct mgp_messages *message) { return MgInvoke<size_t>(mgp_messages_size, message); }
+
+inline mgp_message *messages_at(struct mgp_messages *message, size_t index) {
+  return MgInvoke<mgp_message *>(mgp_messages_at, message, index);
+}
+
+// Transformation context
+
+inline const char *transformation_query(struct mgp_trans_context *ctx) {
+  return MgInvoke<const char *>(mgp_transformation_query, ctx);
+}
+
+// Transformation
+
+inline void module_add_transformation(struct mgp_module *module, const char *name, mgp_trans_cb cb) {
+  return MgInvokeVoid(mgp_module_add_transformation, module, name, cb);
+}
+
 }  // namespace mgp

include/mg_procedure.h

@@ -1592,12 +1592,18 @@ enum mgp_error mgp_messages_size(struct mgp_messages *message, size_t *result);
 /// Get the message from a messages list at given index
 enum mgp_error mgp_messages_at(struct mgp_messages *message, size_t index, struct mgp_message **result);
 
+struct mgp_trans_context;
+
+// Return transformation query, empty string if exists.
+enum mgp_error mgp_transformation_query(struct mgp_trans_context *ctx, const char **result);
+
 /// Entry-point for a module transformation, invoked through a stream transformation.
 ///
 /// Passed in arguments will not live longer than the callback's execution.
 /// Therefore, you must not store them globally or use the passed in mgp_memory
 /// to allocate global resources.
-typedef void (*mgp_trans_cb)(struct mgp_messages *, struct mgp_graph *, struct mgp_result *, struct mgp_memory *);
+typedef void (*mgp_trans_cb)(struct mgp_messages *, struct mgp_trans_context *, struct mgp_result *,
+                             struct mgp_memory *);
 
 /// Register a transformation with a module.
 ///

include/mgp.hpp

@@ -1406,6 +1406,8 @@ class Record {
   void Insert(const char *field_name, const Duration &duration);
   /// @brief Inserts a @ref Value value under field `field_name`, and then call appropriate insert.
   void Insert(const char *field_name, const Value &value);
+  /// @brief Inserts a @ref null value under field `field_name`.
+  void Insert(const char *field_name);
 
  private:
   mgp_result_record *record_;
@@ -1533,6 +1535,112 @@ enum class ProcedureType : uint8_t {
   Write,
 };
 
+enum class StreamSourceType : uint8_t { Kafka, Pulsar };
+
+class Message {
+ public:
+  explicit Message(mgp_message *ptr);
+  explicit Message(const mgp_message *const_ptr);
+
+  Message(const Message &other) noexcept;
+  Message(Message &&other) noexcept;
+
+  Message &operator=(const Message &other) noexcept;
+  Message &operator=(Message &&other) noexcept;
+
+  ~Message();
+
+  StreamSourceType SourceType() const;
+  std::string Payload() const;
+  size_t PayloadSize() const;
+  std::string TopicName() const;
+  std::string Key() const;
+  size_t KeySize() const;
+  int64_t Timestamp() const;
+  int64_t Offset() const;
+
+ private:
+  mgp_message *ptr_;
+};
+
+class Messages {
+ public:
+  explicit Messages(mgp_messages *ptr);
+  explicit Messages(const mgp_messages *const_ptr);
+
+  Messages(const Messages &other) noexcept;
+  Messages(Messages &&other) noexcept;
+
+  Messages &operator=(const Messages &other) noexcept;
+  Messages &operator=(Messages &&other) noexcept;
+
+  ~Messages();
+
+  /// @brief Returns the size of the list.
+  size_t Size() const;
+  /// @brief Returns whether the list is empty.
+  bool Empty() const;
+
+  /// @brief Returns the value at the given `index`.
+  const Message operator[](size_t index) const;
+
+  ///@brief Same as above, but non const value
+  Message operator[](size_t index);
+
+  class Iterator {
+   private:
+    friend class Messages;
+
+   public:
+    using value_type = Messages;
+    using difference_type = std::ptrdiff_t;
+    using pointer = const Messages *;
+    using reference = const Messages &;
+    using iterator_category = std::forward_iterator_tag;
+
+    bool operator==(const Iterator &other) const;
+
+    bool operator!=(const Iterator &other) const;
+
+    Iterator &operator++();
+
+    const Message operator*() const;
+
+   private:
+    Iterator(const Messages *iterable, size_t index);
+
+    const Messages *iterable_;
+    size_t index_;
+  };
+
+  Iterator begin() const;
+  Iterator end() const;
+
+  Iterator cbegin() const;
+  Iterator cend() const;
+
+  /// @exception std::runtime_error List contains value of unknown type.
+  bool operator==(const Messages &other) const;
+  /// @exception std::runtime_error List contains value of unknown type.
+  bool operator!=(const Messages &other) const;
+
+ private:
+  mgp_messages *ptr_;
+};
+
+/// @brief Wrapper class for @ref mgp_trans_context.
+class TransformationContext {
+ private:
+ public:
+  explicit TransformationContext(mgp_trans_context *ctx);
+
+  /// @brief Returns the transformation query (empty string if not exists)
+  std::string Query() const;
+
+ private:
+  mgp_trans_context *ctx_;
+};
+
 /// @brief Adds a procedure to the query module.
 /// @param callback - procedure callback
 /// @param name - procedure name
@@ -1568,6 +1676,8 @@ inline void AddBatchProcedure(mgp_proc_cb callback, mgp_proc_initializer initial
 inline void AddFunction(mgp_func_cb callback, std::string_view name, std::vector<Parameter> parameters,
                         mgp_module *module, mgp_memory *memory);
 
+inline void AddTransformation(mgp_trans_cb callback, std::string_view name, mgp_module *module);
+
 /* #endregion */
 
 namespace util {
@@ -1794,6 +1904,34 @@ inline bool ValuesEqual(mgp_value *value1, mgp_value *value2) {
   throw ValueException("Invalid value; does not match any Memgraph type.");
 }
 
+inline bool MessageEqual(mgp_message *message1, mgp_message *message2) {
+  if (mgp::message_source_type(message1) != mgp::message_source_type(message2)) {
+    return false;
+  }
+
+  if (mgp::message_payload_size(message1) != mgp::message_payload_size(message2)) {
+    return false;
+  }
+
+  return mgp::message_payload(message1) == mgp::message_payload(message2);
+}
+
+inline bool MessagesEqual(mgp_messages *messages1, mgp_messages *messages2) {
+  if (messages1 == messages2) {
+    return true;
+  }
+  if (mgp::messages_size(messages1) != mgp::messages_size(messages2)) {
+    return false;
+  }
+  const size_t len = mgp::messages_size(messages1);
+  for (size_t i = 0; i < len; ++i) {
+    if (!util::MessageEqual(mgp::messages_at(messages1, i), mgp::messages_at(messages2, i))) {
+      return false;
+    }
+  }
+  return true;
+}
+
 /// @brief Converts C++ API types to their MGP API equivalents.
 inline mgp_type *ToMGPType(Type type) {
   switch (type) {
@@ -3948,6 +4086,12 @@ inline const std::string Value::ToString() const {
 
 inline Record::Record(mgp_result_record *record) : record_(record) {}
 
+inline void Record::Insert(const char *field_name) {
+  auto null_value = mgp::MemHandlerCallback(value_make_null);
+  { mgp::result_record_insert(record_, field_name, null_value); }
+  mgp::value_destroy(null_value);
+}
+
 inline void Record::Insert(const char *field_name, bool value) {
   auto mgp_val = mgp::MemHandlerCallback(value_make_bool, value);
   { mgp::result_record_insert(record_, field_name, mgp_val); }
@@ -4034,6 +4178,8 @@ inline void Record::Insert(const char *field_name, const Duration &duration) {
 
 inline void Record::Insert(const char *field_name, const Value &value) {
   switch (value.Type()) {
+    case Type::Null:
+      return Insert(field_name);
     case Type::Bool:
       return Insert(field_name, value.ValueBool());
     case Type::Int:
@@ -4241,6 +4387,82 @@ inline mgp_type *Return::GetMGPType() const {
   return util::ToMGPType(type_);
 }
 
+// Message
+
+inline Message::Message(mgp_message *ptr) : ptr_(ptr) {}
+inline Message::Message(const mgp_message *const_ptr) : ptr_(const_cast<mgp_message *>(const_ptr)) {}
+inline Message::Message(const Message &other) noexcept : Message(other.ptr_) {}
+inline Message::Message(Message &&other) noexcept : ptr_(other.ptr_) { other.ptr_ = nullptr; }
+inline Message &Message::operator=(Message &&other) noexcept {
+  if (this != &other) {
+    ptr_ = other.ptr_;
+    other.ptr_ = nullptr;
+  }
+  return *this;
+}
+inline Message &Message::operator=(const Message &other) noexcept { return *this; }
+inline Message::~Message() { ptr_ = nullptr; }
+inline StreamSourceType Message::SourceType() const {
+  auto result = mgp::message_source_type(ptr_);
+  switch (result) {
+    case mgp_source_type::KAFKA:
+      return StreamSourceType::Kafka;
+    case mgp_source_type::PULSAR:
+      return StreamSourceType::Pulsar;
+  }
+}
+inline std::string Message::Payload() const { return std::string{mgp::message_payload(ptr_)}; }
+inline size_t Message::PayloadSize() const { return Payload().size(); }
+inline std::string Message::TopicName() const { return std::string{mgp::message_topic_name(ptr_)}; }
+inline std::string Message::Key() const { return std::string{mgp::message_key(ptr_)}; }
+inline size_t Message::KeySize() const { return Key().size(); }
+inline int64_t Message::Timestamp() const { return mgp::message_timestamp(ptr_); }
+inline int64_t Message::Offset() const { return mgp::message_offset(ptr_); }
+
+// Messages
+
+inline Messages::Messages(mgp_messages *ptr) : ptr_(ptr) {}
+inline Messages::Messages(const mgp_messages *const_ptr) : ptr_(const_cast<mgp_messages *>(const_ptr)) {}
+inline Messages::Messages(const Messages &other) noexcept : Messages(other.ptr_) {}
+inline Messages::Messages(Messages &&other) noexcept : Messages(other.ptr_) { other.ptr_ = nullptr; }
+inline Messages &Messages::operator=(const Messages &other) noexcept { return *this; }
+inline Messages &Messages::operator=(Messages &&other) noexcept {
+  if (this != &other) {
+    ptr_ = other.ptr_;
+    other.ptr_ = nullptr;
+  }
+  return *this;
+}
+inline Messages::~Messages() { ptr_ = nullptr; }
+inline size_t Messages::Size() const { return mgp::messages_size(ptr_); }
+inline bool Messages::Empty() const { return mgp::messages_size(ptr_) == 0; }
+inline const Message Messages::operator[](size_t index) const { return Message(mgp::messages_at(ptr_, index)); }
+inline Message Messages::operator[](size_t index) { return Message(mgp::messages_at(ptr_, index)); }
+
+inline Messages::Iterator Messages::begin() const { return Messages::Iterator(this, 0); }
+inline Messages::Iterator Messages::end() const { return Messages::Iterator(this, Size()); }
+inline Messages::Iterator Messages::cbegin() const { return Messages::Iterator(this, 0); }
+inline Messages::Iterator Messages::cend() const { return Messages::Iterator(this, Size()); }
+
+inline bool Messages::operator==(const Messages &other) const { return util::MessagesEqual(ptr_, other.ptr_); }
+
+inline bool Messages::operator!=(const Messages &other) const { return !(*this == other); }
+
+inline bool Messages::Iterator::operator==(const Messages::Iterator &other) const {
+  return this->iterable_ == other.iterable_ && this->index_ == other.index_;
+}
+inline bool Messages::Iterator::operator!=(const Messages::Iterator &other) const { return !(*this == other); }
+inline Messages::Iterator &Messages::Iterator::operator++() {
+  index_++;
+  return *this;
+}
+inline const Message Messages::Iterator::operator*() const { return (*iterable_)[index_]; }
+inline Messages::Iterator::Iterator(const Messages *iterable, size_t index) : iterable_(iterable), index_(index) {}
+
+// Transformation context
+inline TransformationContext::TransformationContext(mgp_trans_context *ctx) : ctx_(ctx) {}
+inline std::string TransformationContext::Query() const { return std::string{mgp::transformation_query(ctx_)}; }
+
 // do not enter
 namespace detail {
 inline void AddParamsReturnsToProc(mgp_proc *proc, std::vector<Parameter> &parameters,
@@ -4293,6 +4515,10 @@ void AddFunction(mgp_func_cb callback, std::string_view name, std::vector<Parame
   }
 }
 
+void AddTransformation(mgp_trans_cb callback, std::string_view name, mgp_module *module) {
+  mgp::module_add_transformation(module, name.data(), callback);
+}
+
 /* #endregion */
 
 }  // namespace mgp

query_modules/CMakeLists.txt

@@ -6,6 +6,10 @@ project(memgraph_query_modules)
 
 disallow_in_source_build()
 
+find_package(fmt REQUIRED)
+
+set(CMAKE_SHARED_LIBRARY_PREFIX "")
+
 # Everything that is installed here, should be under the "query_modules" component.
 set(CMAKE_INSTALL_DEFAULT_COMPONENT_NAME "query_modules")
 string(TOLOWER ${CMAKE_BUILD_TYPE} lower_build_type)
@@ -37,9 +41,35 @@ endif()
 install(PROGRAMS $<TARGET_FILE:example_cpp>
         DESTINATION lib/memgraph/query_modules
         RENAME example_cpp.so)
+
 # Also install the source of the example, so user can read it.
 install(FILES example.cpp DESTINATION lib/memgraph/query_modules/src)
 
+add_library(example_cpp_transformation SHARED example_transformation.cpp)
+target_include_directories(example_cpp_transformation PRIVATE ${CMAKE_SOURCE_DIR}/include)
+target_compile_options(example_cpp_transformation PRIVATE -Wall)
+# Strip C++ transformation example in release build.
+if (lower_build_type STREQUAL "release")
+  add_custom_command(TARGET example_cpp_transformation POST_BUILD
+                     COMMAND strip -s $<TARGET_FILE:example_cpp_transformation>
+                     COMMENT "Stripping symbols and sections from the C++ transformation example module")
+endif()
+install(PROGRAMS $<TARGET_FILE:example_cpp_transformation>
+        DESTINATION lib/memgraph/query_modules)
+
+# Also install the source of the example, so user can read it.
+install(FILES example_cpp_transformation.cpp DESTINATION lib/memgraph/query_modules/src)
+
+add_library(generic_cpp_transformation SHARED generic_transformation.cpp)
+target_include_directories(generic_cpp_transformation PRIVATE ${CMAKE_SOURCE_DIR}/include)
+target_compile_options(generic_cpp_transformation PRIVATE -Wall)
+target_link_libraries(generic_cpp_transformation json fmt::fmt)
+
+install(PROGRAMS $<TARGET_FILE:generic_cpp_transformation>
+        DESTINATION lib/memgraph/query_modules)
+# Also install the source of the example, so user can read it.
+install(FILES generic_cpp_transformation.cpp DESTINATION lib/memgraph/query_modules/src)
+
 # Install the Python example and modules
 install(FILES example.py DESTINATION lib/memgraph/query_modules RENAME py_example.py)
 install(FILES graph_analyzer.py DESTINATION lib/memgraph/query_modules)

query_modules/example_transformation.cpp

@@ -0,0 +1,64 @@
+// Copyright 2023 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include <exception>
+#include <mgp.hpp>
+
+static constexpr std::string_view kQuery = "query";
+static constexpr std::string_view kParameters = "parameters";
+
+std::string EscapeString(std::string s) {
+  std::string sign = "'";
+  std::string replace_sign;
+
+  size_t pos;
+  while ((pos = s.find(sign)) != std::string::npos) {
+    s.replace(pos, 1, replace_sign);
+  }
+
+  return s;
+}
+
+void Transformation(struct mgp_messages *messages, mgp_trans_context *ctx, mgp_result *result, mgp_memory *memory) {
+  mgp::MemoryDispatcherGuard guard(memory);
+  auto record_factory = mgp::RecordFactory(result);
+  try {
+    auto stream_messages = mgp::Messages(messages);
+
+    for (const mgp::Message &message : stream_messages) {
+      auto record = record_factory.NewRecord();
+
+      auto payload = EscapeString(message.Payload());
+
+      auto query = "CREATE (:Data {payload: '" + payload + "'});";
+      auto query_value = mgp::Value(query.data());
+
+      record.Insert(kQuery.data(), query_value);
+      record.Insert(kParameters.data(), mgp::Value());
+    }
+  } catch (std::exception &ex) {
+    record_factory.SetErrorMessage(ex.what());
+    return;
+  }
+}
+
+extern "C" int mgp_init_module(mgp_module *module, mgp_memory *memory) {
+  try {
+    mgp::MemoryDispatcherGuard guard(memory);
+    mgp::AddTransformation(Transformation, "transform", module);
+  } catch (const std::exception &e) {
+    return 1;
+  }
+
+  return 0;
+}
+
+extern "C" int mgp_shutdown_module() { return 0; }

query_modules/generic_transformation.cpp

@@ -0,0 +1,135 @@
+// Copyright 2023 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include <fmt/format.h>
+#include <exception>
+#include <json/json.hpp>
+#include <mgp.hpp>
+
+static constexpr std::string_view kQuery = "query";
+static constexpr std::string_view kParameters = "parameters";
+
+template <typename T>
+mgp::Value PropToValue(T prop) {
+  if (prop.is_string()) {
+    return mgp::Value(prop.template get<std::string>());
+  }
+
+  if (prop.is_number_integer()) {
+    return mgp::Value(prop.template get<int64_t>());
+  }
+  if (prop.is_number_float()) {
+    return mgp::Value(prop.template get<float>());
+  }
+  if (prop.is_boolean()) {
+    return mgp::Value(prop.template get<bool>());
+  }
+
+  return mgp::Value();
+}
+
+template <typename T>
+void ReplaceQueryParametersArray(T array, std::string key, std::string &query, mgp::Map &params) {
+  auto list = mgp::List();
+  for (size_t i = 0; i < array.size(); i++) {
+    auto array_element = array[i];
+    for (auto &element_prop : array_element) {
+      auto value = PropToValue(element_prop);
+      if (!value.IsNull()) {
+        list.AppendExtend(std::move(value));
+        continue;
+      }
+      // not all elements are consistent, just exit
+      return;
+    }
+  }
+
+  params.Insert(key, mgp::Value(std::move(list)));
+}
+
+template <typename T>
+void ReplaceQueryParameters(T object, std::string key, std::string &query, mgp::Map &params) {
+  for (auto &el : object.items()) {
+    auto new_key = fmt::format("{}__{}", key, el.key());
+    auto value = PropToValue(el.value());
+
+    if (!value.IsNull()) {
+      params.Insert(new_key, std::move(value));
+    } else {
+      if (el.value().is_object()) {
+        ReplaceQueryParameters(el.value(), new_key, query, params);
+        continue;
+      }
+      if (el.value().is_array()) {
+        ReplaceQueryParametersArray(el.value(), new_key, query, params);
+        continue;
+      }
+    }
+  }
+}
+
+void ReplaceQueryParameters(nlohmann::json &json, std::string &query, mgp::Map &params) {
+  for (auto &el : json.items()) {
+    auto key = fmt::format("{}", el.key());
+    auto value = PropToValue(el.value());
+
+    if (!value.IsNull()) {
+      params.Insert(key, std::move(value));
+    } else {
+      if (el.value().is_object()) {
+        ReplaceQueryParameters(el.value(), key, query, params);
+        continue;
+      }
+      if (el.value().is_array()) {
+        ReplaceQueryParametersArray(el.value(), key, query, params);
+        continue;
+      }
+    }
+  }
+}
+
+void Transformation(struct mgp_messages *messages, mgp_trans_context *ctx, mgp_result *result, mgp_memory *memory) {
+  mgp::MemoryDispatcherGuard guard(memory);
+  auto record_factory = mgp::RecordFactory(result);
+  auto transformation_context = mgp::TransformationContext(ctx);
+  try {
+    auto stream_messages = mgp::Messages(messages);
+    for (const mgp::Message &message : stream_messages) {
+      auto record = record_factory.NewRecord();
+
+      auto payload = nlohmann::json::parse(message.Payload());
+
+      std::string generic_query = transformation_context.Query();
+      auto params = mgp::Map();
+      ReplaceQueryParameters(payload, generic_query, params);
+      auto query_value = mgp::Value(generic_query.data());
+
+      record.Insert(kQuery.data(), query_value);
+      record.Insert(kParameters.data(), mgp::Value(params));
+    }
+  } catch (std::exception &ex) {
+    record_factory.SetErrorMessage(ex.what());
+    return;
+  }
+}
+
+extern "C" int mgp_init_module(mgp_module *module, mgp_memory *memory) {
+  try {
+    mgp::MemoryDispatcherGuard guard(memory);
+    mgp::AddTransformation(Transformation, "transform", module);
+  } catch (const std::exception &e) {
+    return 1;
+  }
+
+  return 0;
+}
+
+extern "C" int mgp_shutdown_module() { return 0; }

src/integrations/kafka/consumer.hpp

@@ -86,6 +86,7 @@ struct ConsumerInfo {
   std::string bootstrap_servers;
   std::chrono::milliseconds batch_interval;
   int64_t batch_size;
+  std::string transformation_query;
   std::unordered_map<std::string, std::string> public_configs{};
   std::unordered_map<std::string, std::string> private_configs{};
 };

src/integrations/pulsar/consumer.hpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -41,6 +41,7 @@ using ConsumerFunction = std::function<void(const std::vector<Message> &)>;
 struct ConsumerInfo {
   int64_t batch_size;
   std::chrono::milliseconds batch_interval;
+  std::string transformation_query;
   std::vector<std::string> topics;
   std::string consumer_name;
   std::string service_url;

src/query/frontend/ast/ast.hpp

@@ -3254,6 +3254,7 @@ class StreamQuery : public memgraph::query::Query {
   memgraph::query::Expression *batch_limit_{nullptr};
   memgraph::query::Expression *timeout_{nullptr};
   std::string transform_name_;
+  memgraph::query::Expression *transformation_query_{nullptr};
   memgraph::query::Expression *batch_interval_{nullptr};
   memgraph::query::Expression *batch_size_{nullptr};
   std::variant<memgraph::query::Expression *, std::vector<std::string>> topic_names_{nullptr};
@@ -3271,6 +3272,7 @@ class StreamQuery : public memgraph::query::Query {
     object->batch_limit_ = batch_limit_ ? batch_limit_->Clone(storage) : nullptr;
     object->timeout_ = timeout_ ? timeout_->Clone(storage) : nullptr;
     object->transform_name_ = transform_name_;
+    object->transformation_query_ = transformation_query_;
     object->batch_interval_ = batch_interval_ ? batch_interval_->Clone(storage) : nullptr;
     object->batch_size_ = batch_size_ ? batch_size_->Clone(storage) : nullptr;
     if (auto *topic_expression = std::get_if<Expression *>(&topic_names_)) {

src/query/frontend/ast/cypher_main_visitor.cpp

@@ -584,12 +584,14 @@ void MapConfig(auto &memory, const EnumUint8 auto &enum_key, auto &destination)
   memory.erase(key);
 }
 
-enum class CommonStreamConfigKey : uint8_t { TRANSFORM, BATCH_INTERVAL, BATCH_SIZE, END };
+enum class CommonStreamConfigKey : uint8_t { TRANSFORM, TRANSFORMATION_QUERY, BATCH_INTERVAL, BATCH_SIZE, END };
 
 std::string_view ToString(const CommonStreamConfigKey key) {
   switch (key) {
     case CommonStreamConfigKey::TRANSFORM:
       return "TRANSFORM";
+    case CommonStreamConfigKey::TRANSFORMATION_QUERY:
+      return "TRANSFORMATION_QUERY";
     case CommonStreamConfigKey::BATCH_INTERVAL:
       return "BATCH_INTERVAL";
     case CommonStreamConfigKey::BATCH_SIZE:
@@ -624,7 +626,9 @@ std::string_view ToString(const KafkaConfigKey key) {
 }
 
 void MapCommonStreamConfigs(auto &memory, StreamQuery &stream_query) {
-  MapConfig<true, std::string>(memory, CommonStreamConfigKey::TRANSFORM, stream_query.transform_name_);
+  MapConfig<false, std::string>(memory, CommonStreamConfigKey::TRANSFORM, stream_query.transform_name_);
+  MapConfig<false, Expression *>(memory, CommonStreamConfigKey::TRANSFORMATION_QUERY,
+                                 stream_query.transformation_query_);
   MapConfig<false, Expression *>(memory, CommonStreamConfigKey::BATCH_INTERVAL, stream_query.batch_interval_);
   MapConfig<false, Expression *>(memory, CommonStreamConfigKey::BATCH_SIZE, stream_query.batch_size_);
 }
@@ -666,6 +670,10 @@ antlrcpp::Any CypherMainVisitor::visitKafkaCreateStream(MemgraphCypher::KafkaCre
 
   MapCommonStreamConfigs(memory_, *stream_query);
 
+  if (stream_query->transform_name_.empty() && !stream_query->transformation_query_) {
+    throw SemanticException("Transformation name and transformation query can't be empty at the same time!");
+  }
+
   return stream_query;
 }
 
@@ -766,6 +774,10 @@ antlrcpp::Any CypherMainVisitor::visitPulsarCreateStream(MemgraphCypher::PulsarC
 
   MapCommonStreamConfigs(memory_, *stream_query);
 
+  if (stream_query->transform_name_.empty() && !stream_query->transformation_query_) {
+    throw SemanticException("Transformation name and transformation query can't be empty at the same time!");
+  }
+
   return stream_query;
 }
 
@@ -799,6 +811,16 @@ antlrcpp::Any CypherMainVisitor::visitCommonCreateStreamConfig(MemgraphCypher::C
     return {};
   }
 
+  if (ctx->TRANSFORMATION_QUERY()) {
+    ThrowIfExists(memory_, CommonStreamConfigKey::TRANSFORMATION_QUERY);
+    if (!ctx->transformationQuery->StringLiteral()) {
+      throw SemanticException("Transform query must be a string literal!");
+    }
+    const auto transform_query_key = static_cast<uint8_t>(CommonStreamConfigKey::TRANSFORMATION_QUERY);
+    memory_[transform_query_key] = std::any_cast<Expression *>(ctx->transformationQuery->accept(this));
+    return {};
+  }
+
   if (ctx->BATCH_INTERVAL()) {
     ThrowIfExists(memory_, CommonStreamConfigKey::BATCH_INTERVAL);
     if (!ctx->batchInterval->numberLiteral() || !ctx->batchInterval->numberLiteral()->integerLiteral()) {

src/query/frontend/opencypher/grammar/MemgraphCypher.g4

@@ -106,6 +106,7 @@ memgraphCypherKeyword : cypherKeyword
                       | TOPICS
                       | TRANSACTION
                       | TRANSFORM
+                      | TRANSFORMATION_QUERY
                       | TRIGGER
                       | TRIGGERS
                       | UNCOMMITTED
@@ -408,6 +409,7 @@ symbolicTopicNames : symbolicNameWithDotsAndMinus ( COMMA symbolicNameWithDotsAn
 topicNames : symbolicTopicNames | literal ;
 
 commonCreateStreamConfig : TRANSFORM transformationName=procedureName
+                         | TRANSFORMATION_QUERY transformationQuery=literal
                          | BATCH_INTERVAL batchInterval=literal
                          | BATCH_SIZE batchSize=literal
                          ;

src/query/frontend/opencypher/grammar/MemgraphCypherLexer.g4

@@ -129,6 +129,7 @@ TRANSACTION             : T R A N S A C T I O N ;
 TRANSACTION_MANAGEMENT  : T R A N S A C T I O N UNDERSCORE M A N A G E M E N T ;
 TRANSACTIONS            : T R A N S A C T I O N S ;
 TRANSFORM               : T R A N S F O R M ;
+TRANSFORMATION_QUERY    : T R A N S F O R M A T I O N UNDERSCORE Q U E R Y ;
 TRIGGER                 : T R I G G E R ;
 TRIGGERS                : T R I G G E R S ;
 UNCOMMITTED             : U N C O M M I T T E D ;

src/query/interpreter.cpp

@@ -772,7 +772,9 @@ stream::CommonStreamInfo GetCommonStreamInfo(StreamQuery *stream_query, Expressi
       .batch_interval = GetOptionalValue<std::chrono::milliseconds>(stream_query->batch_interval_, evaluator)
                             .value_or(stream::kDefaultBatchInterval),
       .batch_size = GetOptionalValue<int64_t>(stream_query->batch_size_, evaluator).value_or(stream::kDefaultBatchSize),
-      .transformation_name = stream_query->transform_name_};
+      .transformation_name = !stream_query->transform_name_.empty() ? stream_query->transform_name_
+                                                                    : stream::kGenericTransformation.data(),
+      .transformation_query = GetOptionalStringValue(stream_query->transformation_query_, evaluator).value_or("")};
 }
 
 std::vector<std::string> EvaluateTopicNames(ExpressionVisitor<TypedValue> &evaluator,
@@ -945,7 +947,9 @@ Callback HandleStreamQuery(StreamQuery *stream_query, const Parameters &paramete
       return callback;
     }
     case StreamQuery::Action::SHOW_STREAMS: {
-      callback.header = {"name", "type", "batch_interval", "batch_size", "transformation_name", "owner", "is running"};
+      callback.header = {
+          "name",  "type",      "batch_interval", "batch_size", "transformation_name", "transformation_query",
+          "owner", "is running"};
       callback.fn = [interpreter_context]() {
         auto streams_status = interpreter_context->streams.GetStreamInfo();
         std::vector<std::vector<TypedValue>> results;
@@ -954,6 +958,7 @@ Callback HandleStreamQuery(StreamQuery *stream_query, const Parameters &paramete
           typed_status.emplace_back(stream_info.batch_interval.count());
           typed_status.emplace_back(stream_info.batch_size);
           typed_status.emplace_back(stream_info.transformation_name);
+          typed_status.emplace_back(stream_info.transformation_query);
         };
 
         for (const auto &status : streams_status) {

src/query/procedure/mg_procedure_impl.cpp

@@ -3475,6 +3475,10 @@ mgp_error mgp_messages_at(mgp_messages *messages, size_t index, mgp_message **re
       result);
 }
 
+mgp_error mgp_transformation_query(mgp_trans_context *ctx, const char **result) {
+  return WrapExceptions([ctx] { return ctx->query.data(); }, result);
+}
+
 mgp_error mgp_module_add_transformation(mgp_module *module, const char *name, mgp_trans_cb cb) {
   return WrapExceptions([=] {
     if (!IsValidIdentifierName(name)) {

src/query/procedure/mg_procedure_impl.hpp

@@ -826,6 +826,12 @@ struct mgp_proc {
   ProcedureInfo info;
 };
 
+struct mgp_trans_context {
+  memgraph::query::DbAccessor *impl;
+  memgraph::storage::View view;
+  std::string query;
+};
+
 struct mgp_trans {
   using allocator_type = memgraph::utils::Allocator<mgp_trans>;
 
@@ -836,7 +842,7 @@ struct mgp_trans {
 
   /// @throw std::bad_alloc
   /// @throw std::length_error
-  mgp_trans(const char *name, std::function<void(mgp_messages *, mgp_graph *, mgp_result *, mgp_memory *)> cb,
+  mgp_trans(const char *name, std::function<void(mgp_messages *, mgp_trans_context *, mgp_result *, mgp_memory *)> cb,
             memgraph::utils::MemoryResource *memory)
       : name(name, memory), cb(cb), results(memory) {}
 
@@ -859,7 +865,7 @@ struct mgp_trans {
   /// Name of the transformation.
   memgraph::utils::pmr::string name;
   /// Entry-point for the transformation.
-  std::function<void(mgp_messages *, mgp_graph *, mgp_result *, mgp_memory *)> cb;
+  std::function<void(mgp_messages *, mgp_trans_context *, mgp_result *, mgp_memory *)> cb;
   /// Fields this transformation returns.
   memgraph::utils::pmr::map<memgraph::utils::pmr::string,
                             std::pair<const memgraph::query::procedure::CypherType *, bool>>

src/query/procedure/py_module.cpp

@@ -1315,8 +1315,9 @@ PyObject *PyQueryModuleAddTransformation(PyQueryModule *self, PyObject *cb) {
   auto *memory = self->module->transformations.get_allocator().GetMemoryResource();
   mgp_trans trans(
       name,
-      [py_cb](mgp_messages *msgs, mgp_graph *graph, mgp_result *result, mgp_memory *memory) {
+      [py_cb](mgp_messages *msgs, mgp_trans_context *ctx, mgp_result *result, mgp_memory *memory) {
-        CallPythonTransformation(py_cb, msgs, graph, result, memory);
+        auto graph = mgp_graph::NonWritableGraph(*(ctx->impl), ctx->view);
+        CallPythonTransformation(py_cb, msgs, &graph, result, memory);
       },
       memory);
   const auto [trans_it, did_insert] = self->module->transformations.emplace(name, std::move(trans));

src/query/stream/common.cpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -18,12 +18,14 @@ namespace {
 const std::string kBatchIntervalKey{"batch_interval"};
 const std::string kBatchSizeKey{"batch_size"};
 const std::string kTransformationName{"transformation_name"};
+const std::string kTransformationQuery{"transformation_query"};
 }  // namespace
 
 void to_json(nlohmann::json &data, CommonStreamInfo &&common_info) {
   data[kBatchIntervalKey] = common_info.batch_interval.count();
   data[kBatchSizeKey] = common_info.batch_size;
   data[kTransformationName] = common_info.transformation_name;
+  data[kTransformationQuery] = common_info.transformation_query;
 }
 
 void from_json(const nlohmann::json &data, CommonStreamInfo &common_info) {
@@ -41,5 +43,6 @@ void from_json(const nlohmann::json &data, CommonStreamInfo &common_info) {
   }
 
   data.at(kTransformationName).get_to(common_info.transformation_name);
+  data.at(kTransformationQuery).get_to(common_info.transformation_query);
 }
 }  // namespace memgraph::query::stream

src/query/stream/common.hpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -25,6 +25,7 @@ namespace memgraph::query::stream {
 
 inline constexpr std::chrono::milliseconds kDefaultBatchInterval{100};
 inline constexpr int64_t kDefaultBatchSize{1000};
+inline constexpr std::string_view kGenericTransformation{"generic_cpp_transformation.transform"};
 
 template <typename TMessage>
 using ConsumerFunction = std::function<void(const std::vector<TMessage> &)>;
@@ -33,6 +34,7 @@ struct CommonStreamInfo {
   std::chrono::milliseconds batch_interval;
   int64_t batch_size;
   std::string transformation_name;
+  std::string transformation_query;
 };
 
 template <typename T>

src/query/stream/sources.cpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -25,6 +25,7 @@ KafkaStream::KafkaStream(std::string stream_name, StreamInfo stream_info,
       .bootstrap_servers = std::move(stream_info.bootstrap_servers),
       .batch_interval = stream_info.common_info.batch_interval,
       .batch_size = stream_info.common_info.batch_size,
+      .transformation_query = stream_info.common_info.transformation_query,
       .public_configs = std::move(stream_info.configs),
       .private_configs = std::move(stream_info.credentials),
   };
@@ -35,7 +36,8 @@ KafkaStream::StreamInfo KafkaStream::Info(std::string transformation_name) const
   const auto &info = consumer_->Info();
   return {{.batch_interval = info.batch_interval,
            .batch_size = info.batch_size,
-           .transformation_name = std::move(transformation_name)},
+           .transformation_name = std::move(transformation_name),
+           .transformation_query = info.transformation_query},
           .topics = info.topics,
           .consumer_group = info.consumer_group,
           .bootstrap_servers = info.bootstrap_servers,
@@ -103,7 +105,8 @@ PulsarStream::StreamInfo PulsarStream::Info(std::string transformation_name) con
   const auto &info = consumer_->Info();
   return {{.batch_interval = info.batch_interval,
            .batch_size = info.batch_size,
-           .transformation_name = std::move(transformation_name)},
+           .transformation_name = std::move(transformation_name),
+           .transformation_query = info.transformation_query},
           .topics = info.topics,
           .service_url = info.service_url};
 }

src/query/stream/streams.cpp

@@ -82,9 +82,10 @@ std::pair<TypedValue /*query*/, TypedValue /*parameters*/> ExtractTransformation
 }
 
 template <typename TMessage>
-void CallCustomTransformation(const std::string &transformation_name, const std::vector<TMessage> &messages,
+void CallCustomTransformation(const std::string &transformation_name, const std::string &transformation_query,
-                              mgp_result &result, storage::Storage::Accessor &storage_accessor,
+                              const std::vector<TMessage> &messages, mgp_result &result,
-                              utils::MemoryResource &memory_resource, const std::string &stream_name) {
+                              storage::Storage::Accessor &storage_accessor, utils::MemoryResource &memory_resource,
+                              const std::string &stream_name) {
   DbAccessor db_accessor{&storage_accessor};
   {
     auto maybe_transformation =
@@ -97,7 +98,7 @@ void CallCustomTransformation(const std::string &transformation_name, const std:
     mgp_messages mgp_messages{mgp_messages::storage_type{&memory_resource}};
     std::transform(messages.begin(), messages.end(), std::back_inserter(mgp_messages.messages),
                    [](const TMessage &message) { return mgp_message{message}; });
-    mgp_graph graph{&db_accessor, storage::View::OLD, nullptr};
+    mgp_trans_context ctx{&db_accessor, storage::View::OLD, transformation_query};
     mgp_memory memory{&memory_resource};
     result.rows.clear();
     result.error_msg.reset();
@@ -108,7 +109,7 @@ void CallCustomTransformation(const std::string &transformation_name, const std:
     MG_ASSERT(result.signature->contains(params_param_name));
 
     spdlog::trace("Calling transformation in stream '{}'", stream_name);
-    trans.cb(&mgp_messages, &graph, &result, &memory);
+    trans.cb(&mgp_messages, &ctx, &result, &memory);
   }
   if (result.error_msg.has_value()) {
     throw StreamsException(result.error_msg->c_str());
@@ -483,7 +484,8 @@ Streams::StreamsMap::iterator Streams::CreateConsumer(StreamsMap &map, const std
   auto *memory_resource = utils::NewDeleteResource();
 
   auto consumer_function = [interpreter_context = interpreter_context_, memory_resource, stream_name,
-                            transformation_name = stream_info.common_info.transformation_name, owner = owner,
+                            transformation_name = stream_info.common_info.transformation_name,
+                            transformation_query = stream_info.common_info.transformation_query, owner = owner,
                             interpreter = std::make_shared<Interpreter>(interpreter_context_),
                             result = mgp_result{nullptr, memory_resource},
                             total_retries = interpreter_context_->config.stream_transaction_conflict_retries,
@@ -496,7 +498,8 @@ Streams::StreamsMap::iterator Streams::CreateConsumer(StreamsMap &map, const std
         [interpreter_context, interpreter]() { interpreter_context->interpreters->erase(interpreter.get()); }};
 
     memgraph::metrics::IncrementCounter(memgraph::metrics::MessagesConsumed, messages.size());
-    CallCustomTransformation(transformation_name, messages, result, *accessor, *memory_resource, stream_name);
+    CallCustomTransformation(transformation_name, transformation_query, messages, result, *accessor, *memory_resource,
+                             stream_name);
 
     DiscardValueResultStream stream;
 
@@ -733,6 +736,7 @@ TransformationResult Streams::Check(const std::string &stream_name, std::optiona
         // that
         const auto locked_stream_source = stream_data.stream_source->ReadLock();
         const auto transformation_name = stream_data.transformation_name;
+        const auto transformation_query = "";
         locked_streams.reset();
 
         auto *memory_resource = utils::NewDeleteResource();
@@ -740,10 +744,12 @@ TransformationResult Streams::Check(const std::string &stream_name, std::optiona
         TransformationResult test_result;
 
         auto consumer_function = [interpreter_context = interpreter_context_, memory_resource, &stream_name,
-                                  &transformation_name = transformation_name, &result,
+                                  &transformation_name = transformation_name,
+                                  transformation_query = transformation_query, &result,
                                   &test_result]<typename T>(const std::vector<T> &messages) mutable {
           auto accessor = interpreter_context->db->Access();
-          CallCustomTransformation(transformation_name, messages, result, *accessor, *memory_resource, stream_name);
+          CallCustomTransformation(transformation_name, transformation_query, messages, result, *accessor,
+                                   *memory_resource, stream_name);
 
           auto result_row = std::vector<TypedValue>();
           result_row.reserve(kCheckStreamResultSize);

tests/e2e/streams/transformations/c_transformations.cpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -12,7 +12,7 @@
 #include "mg_procedure.h"
 
 extern "C" int mgp_init_module(mgp_module *module, mgp_memory *memory) {
-  static const auto no_op_cb = [](mgp_messages *msg, mgp_graph *graph, mgp_result *result, mgp_memory *memory) {};
+  static const auto no_op_cb = [](mgp_messages *msg, mgp_trans_context *ctx, mgp_result *result, mgp_memory *memory) {};
 
   if (mgp_error::MGP_ERROR_NO_ERROR != mgp_module_add_transformation(module, "empty_transformation", no_op_cb)) {
     return 1;

tests/unit/mgp_trans_c_api.cpp

@@ -1,4 +1,4 @@
-// Copyright 2022 Memgraph Ltd.
+// Copyright 2023 Memgraph Ltd.
 //
 // Use of this software is governed by the Business Source License
 // included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
@@ -16,7 +16,8 @@
 #include "test_utils.hpp"
 
 TEST(MgpTransTest, TestMgpTransApi) {
-  static constexpr auto no_op_cb = [](mgp_messages *msg, mgp_graph *graph, mgp_result *result, mgp_memory *memory) {};
+  static constexpr auto no_op_cb = [](mgp_messages *msg, mgp_trans_context *ctx, mgp_result *result,
+                                      mgp_memory *memory) {};
   mgp_module module(memgraph::utils::NewDeleteResource());
   // If this is false, then mgp_module_add_transformation()
   // correctly calls IsValidIdentifier(). We don't need to test