Add parsing without quotes

include/_mgp.hpp

@@ -816,6 +816,12 @@ 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) {

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

@@ -1628,6 +1628,19 @@ class Messages {
   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
@@ -4446,6 +4459,10 @@ inline Messages::Iterator &Messages::Iterator::operator++() {
 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,

query_modules/CMakeLists.txt

@@ -6,6 +6,8 @@ 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.
@@ -58,6 +60,16 @@ install(PROGRAMS $<TARGET_FILE:example_cpp_transformation>
 # 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

@@ -27,7 +27,7 @@ std::string EscapeString(std::string s) {
   return s;
 }
 
-void Transformation(struct mgp_messages *messages, mgp_graph *graph, mgp_result *result, mgp_memory *memory) {
+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 {

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) {
-        CallPythonTransformation(py_cb, msgs, graph, result, memory);
+      [py_cb](mgp_messages *msgs, mgp_trans_context *ctx, mgp_result *result, mgp_memory *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,
-                              mgp_result &result, storage::Storage::Accessor &storage_accessor,
-                              utils::MemoryResource &memory_resource, const std::string &stream_name) {
+void CallCustomTransformation(const std::string &transformation_name, const std::string &transformation_query,
+                              const std::vector<TMessage> &messages, mgp_result &result,
+                              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,11 +12,11 @@
 #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;
   }
 
   return 0;
-}
+}

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