Introduce a flag to control type annotation for WideFromBlocks

commit_hash:e0e84d14ea42f099c639f993508e18f7df307bb4

5 files changed, 117 insertions, 0 deletions

diff --git a/yql/essentials/core/peephole_opt/yql_opt_peephole_physical.cpp b/yql/essentials/core/peephole_opt/yql_opt_peephole_physical.cpp
index 9efc34892d..07e5880c23 100644
--- a/yql/essentials/core/peephole_opt/yql_opt_peephole_physical.cpp
+++ b/yql/essentials/core/peephole_opt/yql_opt_peephole_physical.cpp
@@ -128,6 +128,15 @@ TExprNode::TPtr RebuildArgumentsOnlyLambdaForBlocks(const TExprNode& lambda, TEx
 
 TExprNode::TPtr OptimizeWideToBlocks(const TExprNode::TPtr& node, TExprContext& ctx, TTypeAnnotationContext& types) {
     Y_UNUSED(types);
+
+    // Static assert to ensure backward compatible change: if the
+    // constant below is true, both input and output types of
+    // WideFromBlocks callable have to be WideStream; otherwise,
+    // both input and output types have to be WideFlow.
+    // FIXME: When all spots using WideFromBlocks are adjusted
+    // to work with WideStream, drop the assertion below.
+    static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
     if (node->Head().IsCallable("WideFromBlocks")) {
         YQL_CLOG(DEBUG, CorePeepHole) << "Drop " << node->Content() << " over " << node->Head().Content();
         return ctx.NewCallable(node->Pos(), "ReplicateScalars", { node->Head().HeadPtr() });
@@ -148,6 +157,15 @@ TExprNode::TPtr OptimizeWideToBlocks(const TExprNode::TPtr& node, TExprContext&
 
 TExprNode::TPtr OptimizeWideFromBlocks(const TExprNode::TPtr& node, TExprContext& ctx, TTypeAnnotationContext& types) {
     Y_UNUSED(types);
+
+    // Static assert to ensure backward compatible change: if the
+    // constant below is true, both input and output types of
+    // WideFromBlocks callable have to be WideStream; otherwise,
+    // both input and output types have to be WideFlow.
+    // FIXME: When all spots using WideFromBlocks are adjusted
+    // to work with WideStream, drop the assertion below.
+    static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
     if (node->Head().IsCallable("WideToBlocks")) {
         YQL_CLOG(DEBUG, CorePeepHole) << "Drop " << node->Content() << " over " << node->Head().Content();
         return node->Head().HeadPtr();
@@ -6079,6 +6097,15 @@ TExprNode::TPtr OptimizeWideMapBlocks(const TExprNode::TPtr& node, TExprContext&
     if (node->Head().IsCallable("WideFromBlocks")) {
         if (auto newLambda = RebuildArgumentsOnlyLambdaForBlocks(*lambda, ctx, types)) {
             YQL_CLOG(DEBUG, CorePeepHole) << "Swap " << node->Head().Content() << " with " << node->Content();
+
+            // Static assert to ensure backward compatible change: if the
+            // constant below is true, both input and output types of
+            // WideFromBlocks callable have to be WideStream; otherwise,
+            // both input and output types have to be WideFlow.
+            // FIXME: When all spots using WideFromBlocks are adjusted
+            // to work with WideStream, drop the assertion below.
+            static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
             return ctx.Builder(node->Pos())
                 .Callable("WideFromBlocks")
                     .Callable(0, "WideMap")
@@ -6111,6 +6138,14 @@ TExprNode::TPtr OptimizeWideMapBlocks(const TExprNode::TPtr& node, TExprContext&
     YQL_CLOG(DEBUG, CorePeepHole) << "Convert " << node->Content() << " to blocks, extra nodes: " << newNodes
                                   << ", extra columns: " << rewritePositions.size();
 
+    // Static assert to ensure backward compatible change: if the
+    // constant below is true, both input and output types of
+    // WideFromBlocks callable have to be WideStream; otherwise,
+    // both input and output types have to be WideFlow.
+    // FIXME: When all spots using WideFromBlocks are adjusted
+    // to work with WideStream, drop the assertion below.
+    static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
     auto ret = ctx.Builder(node->Pos())
         .Callable("WideFromBlocks")
             .Callable(0, "WideMap")
@@ -6178,6 +6213,15 @@ TExprNode::TPtr OptimizeWideFilterBlocks(const TExprNode::TPtr& node, TExprConte
 
         YQL_CLOG(DEBUG, CorePeepHole) << "Convert " << node->Content() << " to blocks, extra nodes: " << newNodes
                                       << ", extra columns: " << rewritePositions.size();
+
+        // Static assert to ensure backward compatible change: if the
+        // constant below is true, both input and output types of
+        // WideFromBlocks callable have to be WideStream; otherwise,
+        // both input and output types have to be WideFlow.
+        // FIXME: When all spots using WideFromBlocks are adjusted
+        // to work with WideStream, drop the assertion below.
+        static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
         return ctx.Builder(node->Pos())
             .Callable("WideFromBlocks")
                 .Add(0, result)
@@ -6189,6 +6233,14 @@ TExprNode::TPtr OptimizeWideFilterBlocks(const TExprNode::TPtr& node, TExprConte
         return node;
     }
 
+    // Static assert to ensure backward compatible change: if the
+    // constant below is true, both input and output types of
+    // WideFromBlocks callable have to be WideStream; otherwise,
+    // both input and output types have to be WideFlow.
+    // FIXME: When all spots using WideFromBlocks are adjusted
+    // to work with WideStream, drop the assertion below.
+    static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
     auto filtered = ctx.Builder(node->Pos())
         .Callable("WideFilter")
             .Callable(0, "WideFromBlocks")
@@ -6261,6 +6313,15 @@ TExprNode::TPtr OptimizeSkipTakeToBlocks(const TExprNode::TPtr& node, TExprConte
 
     TStringBuf newName = node->Content() == "Skip" ? "WideSkipBlocks" : "WideTakeBlocks";
     YQL_CLOG(DEBUG, CorePeepHole) << "Convert " << node->Content() << " to " << newName;
+
+    // Static assert to ensure backward compatible change: if the
+    // constant below is true, both input and output types of
+    // WideFromBlocks callable have to be WideStream; otherwise,
+    // both input and output types have to be WideFlow.
+    // FIXME: When all spots using WideFromBlocks are adjusted
+    // to work with WideStream, drop the assertion below.
+    static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
     return ctx.Builder(node->Pos())
         .Callable("WideFromBlocks")
             .Callable(0, newName)
@@ -6308,6 +6369,14 @@ TExprNode::TPtr OptimizeTopOrSortBlocks(const TExprNode::TPtr& node, TExprContex
     auto children = node->ChildrenList();
     children[0] = ctx.NewCallable(node->Pos(), "WideToBlocks", { children[0] });
 
+    // Static assert to ensure backward compatible change: if the
+    // constant below is true, both input and output types of
+    // WideFromBlocks callable have to be WideStream; otherwise,
+    // both input and output types have to be WideFlow.
+    // FIXME: When all spots using WideFromBlocks are adjusted
+    // to work with WideStream, drop the assertion below.
+    static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
     return ctx.Builder(node->Pos())
         .Callable("WideFromBlocks")
             .Add(0, ctx.NewCallable(node->Pos(), newName, std::move(children)))
@@ -6532,6 +6601,15 @@ TExprNode::TPtr OptimizeWideMaps(const TExprNode::TPtr& node, TExprContext& ctx)
             }
         } else if (input.IsCallable({"WideFromBlocks", "WideTakeBlocks", "WideSkipBlocks", "BlockExpandChunked"})) {
             YQL_CLOG(DEBUG, CorePeepHole) << node->Content() << " over " << input.Content() << " with " << unused.size() << " unused fields.";
+
+            // Static assert to ensure backward compatible change: if the
+            // constant below is true, both input and output types of
+            // WideFromBlocks callable have to be WideStream; otherwise,
+            // both input and output types have to be WideFlow.
+            // FIXME: When all spots using WideFromBlocks are adjusted
+            // to work with WideStream, drop the assertion below.
+            static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
             return ctx.Builder(node->Pos())
                 .Callable(node->Content())
                     .Add(0, ctx.ChangeChild(input, 0U, MakeWideMapForDropUnused(input.HeadPtr(), unused, ctx)))
diff --git a/yql/essentials/core/type_ann/type_ann_blocks.cpp b/yql/essentials/core/type_ann/type_ann_blocks.cpp
index 8dcfe1a180..4f3a092dde 100644
--- a/yql/essentials/core/type_ann/type_ann_blocks.cpp
+++ b/yql/essentials/core/type_ann/type_ann_blocks.cpp
@@ -966,6 +966,15 @@ IGraphTransformer::TStatus WideToBlocksWrapper(const TExprNode::TPtr& input, TEx
 
 IGraphTransformer::TStatus WideFromBlocksWrapper(const TExprNode::TPtr& input, TExprNode::TPtr& output, TContext& ctx) {
     Y_UNUSED(output);
+
+    // Static assert to ensure backward compatible change: if the
+    // constant below is true, both input and output types of
+    // WideFromBlocks callable have to be WideStream; otherwise,
+    // both input and output types have to be WideFlow.
+    // FIXME: When all spots using WideFromBlocks are adjusted
+    // to work with WideStream, drop the assertion below.
+    static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
     if (!EnsureArgsCount(*input, 1U, ctx.Expr)) {
         return IGraphTransformer::TStatus::Error;
     }
diff --git a/yql/essentials/core/yql_aggregate_expander.cpp b/yql/essentials/core/yql_aggregate_expander.cpp
index 4e81c16d0e..a9b5a16162 100644
--- a/yql/essentials/core/yql_aggregate_expander.cpp
+++ b/yql/essentials/core/yql_aggregate_expander.cpp
@@ -708,6 +708,15 @@ TExprNode::TPtr TAggregateExpander::TryGenerateBlockCombineAllOrHashed() {
 
     TExprNode::TPtr aggWideFlow;
     if (hashed) {
+
+        // Static assert to ensure backward compatible change: if the
+        // constant below is true, both input and output types of
+        // WideFromBlocks callable have to be WideStream; otherwise,
+        // both input and output types have to be WideFlow.
+        // FIXME: When all spots using WideFromBlocks are adjusted
+        // to work with WideStream, drop the assertion below.
+        static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
         aggWideFlow = Ctx.Builder(Node->Pos())
             .Callable("WideFromBlocks")
                 .Callable(0, "ToFlow")
@@ -2930,6 +2939,14 @@ TExprNode::TPtr TAggregateExpander::TryGenerateBlockMergeFinalizeHashed() {
             .Build();
     }
 
+    // Static assert to ensure backward compatible change: if the
+    // constant below is true, both input and output types of
+    // WideFromBlocks callable have to be WideStream; otherwise,
+    // both input and output types have to be WideFlow.
+    // FIXME: When all spots using WideFromBlocks are adjusted
+    // to work with WideStream, drop the assertion below.
+    static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
     auto aggWideFlow = Ctx.NewCallable(Node->Pos(), "WideFromBlocks", { aggBlocks });
     auto finalFlow = MakeNarrowMap(Node->Pos(), outputColumns, aggWideFlow, Ctx);
     auto root = Ctx.NewCallable(Node->Pos(), "FromFlow", { finalFlow });
diff --git a/yql/essentials/core/yql_expr_type_annotation.h b/yql/essentials/core/yql_expr_type_annotation.h
index 94d7f0dd46..a8def26cee 100644
--- a/yql/essentials/core/yql_expr_type_annotation.h
+++ b/yql/essentials/core/yql_expr_type_annotation.h
@@ -353,4 +353,8 @@ TStringBuf NormalizeCallableName(TStringBuf name);
 
 void CheckExpectedTypeAndColumnOrder(const TExprNode& node, TExprContext& ctx, TTypeAnnotationContext& typesCtx);
 
+namespace NBlockStreamIO {
+    constexpr bool WideFromBlocks = false;
+} // namespace NBlockStreamIO
+
 }
diff --git a/yql/essentials/public/purecalc/common/transformations/utils.cpp b/yql/essentials/public/purecalc/common/transformations/utils.cpp
index 4e2da41835..6f63be0208 100644
--- a/yql/essentials/public/purecalc/common/transformations/utils.cpp
+++ b/yql/essentials/public/purecalc/common/transformations/utils.cpp
@@ -13,6 +13,15 @@ TExprNode::TPtr NYql::NPureCalc::NodeFromBlocks(
 ) {
     const auto items = structType->GetItems();
     Y_ENSURE(items.size() > 0);
+
+    // Static assert to ensure backward compatible change: if the
+    // constant below is true, both input and output types of
+    // WideFromBlocks callable have to be WideStream; otherwise,
+    // both input and output types have to be WideFlow.
+    // FIXME: When all spots using WideFromBlocks are adjusted
+    // to work with WideStream, drop the assertion below.
+    static_assert(!NYql::NBlockStreamIO::WideFromBlocks);
+
     return ctx.Builder(pos)
         .Lambda()
             .Param("stream")