WideCombiner<SkipYields=true> (1st stage) ignores MemLimit for DISTINCT aggregations

[Hor911]

TPC-H Q4 scale 1000

$border = Date("1994-03-01");

$o = (
select
    o_orderpriority,
    o_orderkey
from
    orders
where
    o_orderdate >= $border
    and o_orderdate < DateTime::MakeDate(DateTime::ShiftMonths($border, 3))
);

$l = (
select
    distinct l_orderkey
from
    lineitem
where
    l_commitdate < l_receiptdate
);

select
    o.o_orderpriority as o_orderpriority,
    count(*) as order_count
from
    $o as o
join
    $l as l
on
    o.o_orderkey = l.l_orderkey
group by
    o.o_orderpriority
order by
    o_orderpriority;

Notice SELECT DISTINCT above

Default run causes over-expected memory consumption in stage #1

Here is AST for stage

(let $30 (DqPhyStage '((DqSource $22 $26)) (lambda '($75) (block '(
  (let $76 (DqSourceWideBlockWrap $75 $22 $25 $16))
  (let $77 (lambda '($78 $79 $80 $81) (block '(
    (let $82 (BlockFunc '"Less" $18 $78 $80))
    (return $79 $82 $81)
  ))))
  (return (FromFlow (WideCombiner (WideFromBlocks (BlockCompress (WideMap $76 $77) '1)) '-1073741824 $27 (lambda '($84 $85) $84) (lambda '($86 $87 $88) $88) (lambda '($89 $90) $90))))
))) $29))

Notice negative MemLimit -1073741824 (it causes SkipYields=true later on)

Let's force positive MemLimit

diff --git a/ydb/library/yql/core/peephole_opt/yql_opt_peephole_physical.cpp b/ydb/library/yql/core/peephole_opt/yql_opt_peephole_physical.cpp
index e88e436f12..f8cccaf8f2 100644
--- a/ydb/library/yql/core/peephole_opt/yql_opt_peephole_physical.cpp
+++ b/ydb/library/yql/core/peephole_opt/yql_opt_peephole_physical.cpp
@@ -3326,7 +3326,7 @@ TExprNode::TPtr ExpandCombineByKey(const TExprNode::TPtr& node, TExprContext& ct
         .UpdateHandler(combine.UpdateHandlerLambda())
         .FinishHandler(combine.FinishHandlerLambda())
         .MemLimit()
-            .Value("0")
+            .Value("1073741824")
             .Build()
         .Done()
         .Ptr();

Now, memory consumption is significantly lower

The difference is in positive MemLimit in AST

(let $30 (DqPhyStage '((DqSource $22 $26)) (lambda '($75) (block '(
  (let $76 (DqSourceWideBlockWrap $75 $22 $25 $16))
  (let $77 (lambda '($78 $79 $80 $81) (block '(
    (let $82 (BlockFunc '"Less" $18 $78 $80))
    (return $79 $82 $81)
  ))))
  (return (FromFlow (WideCombiner (WideFromBlocks (BlockCompress (WideMap $76 $77) '1)) '"1073741824" $27 (lambda '($84 $85) $84) (lambda '($86 $87 $88) $88) (lambda '($89 $90) $90))))
))) $29))

The problem is observed for DISTINCT aggregation only

Other aggregations like GROUP BY work nicely (with low memory consumption)