Make DQ use only WideStream for WideToBlocks (#14499)

Author: igormunkin

ydb/library/yql/dq/opt/dq_opt_build.cpp

@@ -789,21 +789,6 @@ bool CanRebuildForWideBlockChannelOutput(bool forceBlocks, const TDqOutput& outp
 
 TDqPhyStage RebuildStageOutputAsWideBlock(const TDqPhyStage& stage, TExprContext& ctx)
 {
-    if constexpr (!NYql::NBlockStreamIO::WideToBlocks) {
-        return Build<TDqPhyStage>(ctx, stage.Pos())
-            .InitFrom(stage)
-            .Program()
-                .Args(stage.Program().Args())
-                .Body<TCoFromFlow>()
-                    .Input<TCoWideToBlocks>()
-                        .Input<TCoToFlow>()
-                            .Input(stage.Program().Body())
-                        .Build()
-                    .Build()
-                .Build()
-            .Build()
-            .Done();
-    }
     return Build<TDqPhyStage>(ctx, stage.Pos())
         .InitFrom(stage)
         .Program()

ydb/library/yql/dq/opt/dq_opt_peephole.cpp

@@ -797,39 +797,6 @@ NNodes::TExprBase DqPeepholeRewriteLength(const NNodes::TExprBase& node, TExprCo
 
     auto dqPhyLength = node.Cast<TDqPhyLength>();
     if (typesCtx.IsBlockEngineEnabled()) {
-        if constexpr (!NYql::NBlockStreamIO::WideToBlocks) {
-            return NNodes::TExprBase(ctx.Builder(node.Pos())
-                .Callable("NarrowMap")
-                    .Callable(0, "ToFlow")
-                        .Callable(0, "BlockCombineAll")
-                            .Callable(0, "FromFlow")
-                                .Callable(0, "WideToBlocks")
-                                    .Add(0, MakeExpandMap(node.Pos(), {}, dqPhyLength.Input().Ptr(), ctx))
-                                .Seal()
-                            .Seal()
-                            .Callable(1, "Void")
-                            .Seal()
-                            .List(2)
-                                .List(0)
-                                    .Callable(0, "AggBlockApply")
-                                        .Atom(0, "count_all")
-                                    .Seal()
-                                .Seal()
-                            .Seal()
-                        .Seal()
-                    .Seal()
-                    .Lambda(1)
-                        .Param("value")
-                        .Callable("AsStruct")
-                            .List(0)
-                                .Atom(0, dqPhyLength.Name())
-                                .Arg(1, "value")
-                            .Seal()
-                        .Seal()
-                    .Seal()
-                .Seal()
-                .Build());
-        }
         return NNodes::TExprBase(ctx.Builder(node.Pos())
             .Callable("NarrowMap")
                 .Callable(0, "ToFlow")

ydb/library/yql/providers/dq/provider/exec/yql_dq_exectransformer.cpp

@@ -335,16 +335,9 @@ TExprNode::TPtr DqMarkBlockStage(const TDqStatePtr& state, const TPublicIds::TPt
             return false;
         }
 
-        if constexpr (!NYql::NBlockStreamIO::WideToBlocks) {
-            if (node->IsCallable("WideToBlocks") && node->Head().IsCallable("ToFlow") && node->Head().Head().IsArgument()) {
-                // scalar channel as input
-                return false;
-            }
-        } else {
-            if (node->IsCallable("WideToBlocks") && node->Head().IsArgument()) {
-                // scalar channel as input
-                return false;
-            }
+        if (node->IsCallable("WideToBlocks") && node->Head().IsArgument()) {
+            // scalar channel as input
+            return false;
         }
 
         const TTypeAnnotationNode* nodeType = node->GetTypeAnn();