Add combine step for hopping aggregate

commit_hash:73a530e711a4f8aebf9b4a6493da6ba523786389

3 files changed, 45 insertions, 7 deletions

diff --git a/yt/yql/providers/ytflow/expr_nodes/yql_ytflow_expr_nodes.json b/yt/yql/providers/ytflow/expr_nodes/yql_ytflow_expr_nodes.json
index 90c1dc8d7cd..40da0493503 100644
--- a/yt/yql/providers/ytflow/expr_nodes/yql_ytflow_expr_nodes.json
+++ b/yt/yql/providers/ytflow/expr_nodes/yql_ytflow_expr_nodes.json
@@ -135,13 +135,10 @@
                 {"Index": 4, "Name": "Keys", "Type": "TCoAtomList"},
                 {"Index": 5, "Name": "UpdateStateLambda", "Type": "TCoLambda"},
                 {"Index": 6, "Name": "PostprocessLambda", "Type": "TCoLambda"},
-                {"Index": 7, "Name": "Column", "Type": "TCoAtom"},
-                {"Index": 8, "Name": "Hop", "Type": "TCoAtom"},
-                {"Index": 9, "Name": "Interval", "Type": "TCoAtom"},
-                {"Index": 10, "Name": "Delay", "Type": "TCoAtom"},
-                {"Index": 11, "Name": "ItemType", "Type": "TExprBase"},
-                {"Index": 12, "Name": "InitLambda", "Type": "TCoLambda"},
-                {"Index": 13, "Name": "SaveLambda", "Type": "TCoLambda"}
+                {"Index": 7, "Name": "Hop", "Type": "TCoAtom"},
+                {"Index": 8, "Name": "Interval", "Type": "TCoAtom"},
+                {"Index": 9, "Name": "Delay", "Type": "TCoAtom"},
+                {"Index": 10, "Name": "SavedStateType", "Type": "TExprBase"}
             ]
         },
         {
@@ -169,6 +166,14 @@
             ]
         },
         {
+            "Name": "TYtflowChunkedForwardList",
+            "Base": "TCallable",
+            "Match": {"Type": "Callable", "Name": "YtflowChunkedForwardList"},
+            "Children": [
+                {"Index": 0, "Name": "Stream", "Type": "TExprBase"}
+            ]
+        },
+        {
             "Name": "TYtflowOutput",
             "Base": "TCallable",
             "Match": {"Type": "Callable", "Name": "YtflowOutput"},
diff --git a/yt/yql/tests/sql/suites/ytflow/hopping_aggregate_over_map.cfg b/yt/yql/tests/sql/suites/ytflow/hopping_aggregate_over_map.cfg
new file mode 100644
index 00000000000..07bbcd82943
--- /dev/null
+++ b/yt/yql/tests/sql/suites/ytflow/hopping_aggregate_over_map.cfg
@@ -0,0 +1,5 @@
+in Input input_hopping_aggregate.txt
+out Output output_hopping_aggregate.txt
+providers ytflow
+udf datetime2_udf
+langver 2025.04
diff --git a/yt/yql/tests/sql/suites/ytflow/hopping_aggregate_over_map.yql b/yt/yql/tests/sql/suites/ytflow/hopping_aggregate_over_map.yql
new file mode 100644
index 00000000000..88390e96619
--- /dev/null
+++ b/yt/yql/tests/sql/suites/ytflow/hopping_aggregate_over_map.yql
@@ -0,0 +1,28 @@
+/* sort outputs */
+
+use plato;
+
+pragma Engine = "ytflow";
+
+pragma Ytflow.Cluster = "plato";
+pragma Ytflow.PipelineDirectory = "pipelines";
+pragma Ytflow.PipelineName = "test";
+
+$stream = select
+    *
+from Input
+where value > 0;
+
+insert into Output
+select 
+    HOP_START() as window_start,
+    HOP_END() as window_end,
+    key,
+    sum(value) as sum_values,
+    sum_if(value, value > 10) as sum_if_values,
+    count(*) as count_values,
+    count_if(value > 10) as count_if_values
+from $stream
+group by
+    key,
+    HOP(Datetime::FromSeconds(Unwrap(ts)), "PT5S", "PT5S", "PT10S");