From 8470bc15da9970db212fde857b6dd48d2b1d9545 Mon Sep 17 00:00:00 2001
From: Piotr Rudnicki <prn@touk.pl>
Date: Thu, 26 Sep 2024 16:22:55 +0200
Subject: [PATCH] add expand method

---
 .../flink/util/transformer/aggregate/aggregates.scala       | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/engine/flink/components/base-unbounded/src/main/scala/pl/touk/nussknacker/engine/flink/util/transformer/aggregate/aggregates.scala b/engine/flink/components/base-unbounded/src/main/scala/pl/touk/nussknacker/engine/flink/util/transformer/aggregate/aggregates.scala
index bdb37f86bee..41a93c43d2f 100644
--- a/engine/flink/components/base-unbounded/src/main/scala/pl/touk/nussknacker/engine/flink/util/transformer/aggregate/aggregates.scala
+++ b/engine/flink/components/base-unbounded/src/main/scala/pl/touk/nussknacker/engine/flink/util/transformer/aggregate/aggregates.scala
@@ -346,6 +346,12 @@ object aggregates {
       _.computeStoredType(_)
     )
 
+    def expand(newField: (String, Aggregator)): MapAggregator = {
+      val expandedMap: Map[String, Aggregator] = (scalaFields + newField)
+
+      new MapAggregator(expandedMap.asJava)
+    }
+
     private def computeTypeByFields(
         input: TypingResult,
         objType: TypedClass,