[Native] - reduce_agg seems to be producing wrong results against c++ workers

[mknegi]

Expected Behavior

assertQuery("SELECT id, reduce_agg(value, 's', (a, b) -> concat(a, b, 's'), (a, b) -> concat(a, b, 's')) FROM ( VALUES (1, '2'), (1, '3'), (1, '4'), (2, '20'), (2, '30'), (2, '40') ) AS t(id, value) GROUP BY id",
                "values (1, 's2s3s4s'), (2, 's20s30s40s')");       

Current Behavior

presto:tpch> SELECT id, reduce_agg(value, 's', (a, b) -> concat(a, b, 's'), (a, b) -> concat(a, b, 's')) FROM ( VALUES (1, '2'), (1, '3'), (1, '4'), (2, '20'), (2, '30'), (2, '40') ) AS t(id, value) GROUP BY id;
 id |     _col1
----+----------------
  1 | s2ss3sss4ss
  2 | s20ss30sss40ss
(2 rows)

Query Plan:
presto:tpch> explain SELECT id, 's' || reduce_agg(value, '', (a, b) -> concat(a, b, 's'), (a, b) -> concat(a, b, 's')) FROM ( VALUES (1, '2'), (1, '3'), (1, '4'), (2, '20'), (2, '30'), (2, '40') ) AS t(id, value) GROUP BY id;
                                                                                                        Query Plan              >
-------------------------------------------------------------------------------------------------------------------------------->
 - Output[PlanNodeId 11][id, _col1] => [field:integer, concat:varchar]                                                          >
         id := field (1:16)                                                                                                     >
         _col1 := concat (1:24)                                                                                                 >
     - Project[PlanNodeId 7][projectLocality = LOCAL] => [field:integer, concat:varchar]                                        >
             concat := concat(VARCHAR's', reduce_agg) (1:24)                                                                    >
         - Aggregate(FINAL)[field][PlanNodeId 6] => [field:integer, reduce_agg:varchar]                                         >
                 reduce_agg := "presto.default.reduce_agg"((reduce_agg_18,(expr, expr_0) -> concat(expr, CAST(expr_0 AS varchar)>
             - LocalExchange[PlanNodeId 379][HASH] (field) => [field:integer, reduce_agg_18:varchar]                            >
                 - Aggregate(PARTIAL)[field][PlanNodeId 377] => [field:integer, reduce_agg_18:varchar]                          >
                         reduce_agg_18 := "presto.default.reduce_agg"((field_3,expr_10,(expr, expr_0) -> concat(expr, CAST(expr_>
                     - Project[PlanNodeId 229][projectLocality = LOCAL] => [field:integer, field_3:varchar(2), expr_10:varchar] >
                             Estimates: {source: CostBasedSourceInfo, rows: 6 (360B), cpu: 1,050.00, memory: 0.00, network: 0.00>
                             expr_10 := VARCHAR''                                                                               >
                         - LocalExchange[PlanNodeId 365][ROUND_ROBIN] () => [field:integer, field_3:varchar(2)]                 >
                                 Estimates: {source: CostBasedSourceInfo, rows: 6 (360B), cpu: 360.00, memory: 0.00, network: 0.>
                             - Values[PlanNodeId 0] => [field:integer, field_3:varchar(2)]                                      >
                                     Estimates: {source: CostBasedSourceInfo, rows: 6 (360B), cpu: 0.00, memory: 0.00, network: >
                                     (INTEGER'1', VARCHAR'2')                                                                   >
                                     (INTEGER'1', VARCHAR'3')                                                                   >
                                     (INTEGER'1', VARCHAR'4')                                                                   >
                                     (INTEGER'2', VARCHAR'20')                                                                  >
                                     (INTEGER'2', VARCHAR'30')                                                                  >
                                     (INTEGER'2', VARCHAR'40')                                                                  >
                                                                                                                                >
(1 row)