YQL-19424: Force BlockExpandChunked process stream input

commit_hash:183c6a5ff0706ee5c612ff07806a656f19c5bd30

4 files changed, 17 insertions, 10 deletions

diff --git a/yql/essentials/minikql/comp_nodes/ut/mkql_block_compress_ut.cpp b/yql/essentials/minikql/comp_nodes/ut/mkql_block_compress_ut.cpp
index 2b8fd3ffd0c..d32dc80aff1 100644
--- a/yql/essentials/minikql/comp_nodes/ut/mkql_block_compress_ut.cpp
+++ b/yql/essentials/minikql/comp_nodes/ut/mkql_block_compress_ut.cpp
@@ -81,10 +81,10 @@ void DoNestedTuplesCompressTest() {
     node = pb.ExpandMap(node, [&](TRuntimeNode item) -> TRuntimeNode::TList {
         return {pb.Nth(item, 0U), pb.Nth(item, 1U), pb.Nth(item, 2U)};
     });
-    node = pb.ToFlow(pb.WideToBlocks(pb.FromFlow(node)));
+    node = pb.WideToBlocks(pb.FromFlow(node));
 
     node = pb.BlockExpandChunked(node);
-    node = pb.WideSkipBlocks(pb.FromFlow(node), pb.template NewDataLiteral<ui64>(19));
+    node = pb.WideSkipBlocks(node, pb.template NewDataLiteral<ui64>(19));
     node = pb.BlockCompress(node, 2);
     node = pb.ToFlow(pb.WideFromBlocks(node));
 
diff --git a/yql/essentials/minikql/comp_nodes/ut/mkql_blocks_ut.cpp b/yql/essentials/minikql/comp_nodes/ut/mkql_blocks_ut.cpp
index e9014f0335b..7fb6bae3a10 100644
--- a/yql/essentials/minikql/comp_nodes/ut/mkql_blocks_ut.cpp
+++ b/yql/essentials/minikql/comp_nodes/ut/mkql_blocks_ut.cpp
@@ -331,15 +331,15 @@ void TestChunked(bool withBlockExpand) {
     node = pb.ExpandMap(node, [&](TRuntimeNode item) -> TRuntimeNode::TList {
         return {pb.Nth(item, 0U), pb.Nth(item, 1U), pb.Nth(item, 2U), pb.Nth(item, 3U)};
     });
-    node = pb.ToFlow(pb.WideToBlocks(pb.FromFlow(node)));
+    node = pb.WideToBlocks(pb.FromFlow(node));
     if (withBlockExpand) {
         node = pb.BlockExpandChunked(node);
         // WideTakeBlocks won't work on chunked blocks
-        node = pb.WideTakeBlocks(pb.FromFlow(node), pb.NewDataLiteral<ui64>(19));
+        node = pb.WideTakeBlocks(node, pb.NewDataLiteral<ui64>(19));
         node = pb.ToFlow(pb.WideFromBlocks(node));
     } else {
         // WideFromBlocks should support chunked blocks
-        node = pb.ToFlow(pb.WideFromBlocks(pb.FromFlow(node)));
+        node = pb.ToFlow(pb.WideFromBlocks(node));
         node = pb.Take(node, pb.NewDataLiteral<ui64>(19));
     }
     node = pb.NarrowMap(node, [&](TRuntimeNode::TList items) -> TRuntimeNode {
diff --git a/yql/essentials/minikql/mkql_program_builder.cpp b/yql/essentials/minikql/mkql_program_builder.cpp
index 92229a53cc2..e29bbf108a0 100644
--- a/yql/essentials/minikql/mkql_program_builder.cpp
+++ b/yql/essentials/minikql/mkql_program_builder.cpp
@@ -1646,12 +1646,19 @@ TRuntimeNode TProgramBuilder::BlockCompress(TRuntimeNode stream, ui32 bitmapInde
 }
 
 TRuntimeNode TProgramBuilder::BlockExpandChunked(TRuntimeNode comp) {
-    if (comp.GetStaticType()->IsStream()) {
-        ValidateBlockStreamType(comp.GetStaticType());
+    const auto inputType = comp.GetStaticType();
+    if constexpr (RuntimeVersion < 74U) {
+        if (inputType->IsStream()) {
+            ValidateBlockStreamType(inputType);
+        } else {
+            ValidateBlockFlowType(inputType);
+        }
     } else {
-        ValidateBlockFlowType(comp.GetStaticType());
+        MKQL_ENSURE(inputType->IsStream(), "Expected stream as input type");
+        ValidateBlockStreamType(inputType);
     }
-    TCallableBuilder callableBuilder(Env_, __func__, comp.GetStaticType());
+
+    TCallableBuilder callableBuilder(Env_, __func__, inputType);
     callableBuilder.Add(comp);
     return TRuntimeNode(callableBuilder.Build(), false);
 }
diff --git a/yql/essentials/minikql/mkql_runtime_version.h b/yql/essentials/minikql/mkql_runtime_version.h
index 2ccb5d4b3a8..63f5edfff4e 100644
--- a/yql/essentials/minikql/mkql_runtime_version.h
+++ b/yql/essentials/minikql/mkql_runtime_version.h
@@ -27,7 +27,7 @@ namespace NKikimr::NMiniKQL {
 // 1. Bump this version every time incompatible runtime nodes are introduced.
 // 2. Make sure you provide runtime node generation for previous runtime versions.
 #ifndef MKQL_RUNTIME_VERSION
-    #define MKQL_RUNTIME_VERSION 73U
+    #define MKQL_RUNTIME_VERSION 74U
 #endif
 
 class TRuntimeVersion {