[CH] Logical error: 'Invalid number of columns in chunk pushed to OutputPort.

[loneylee]

Describe the bug
CHHashAggregate output is error when child is instanceOf InputAdapter.

Run sql

SELECT LTRIM(o_orderpriority, '-') ltm, count(*) AS order_count
FROM orders
WHERE o_orderdate >= date '1993-07-01'
  AND o_orderdate < date '1993-07-01' + interval 3 month
  AND EXISTS ( SELECT * FROM lineitem WHERE l_orderkey = o_orderkey
  AND l_commitdate
    < l_receiptdate)
GROUP BY ltm
ORDER BY ltm;

Plan is

DeserializeToObject createexternalrow(ltm#872.toString, order_count#873L, StructField(ltm,StringType,true), StructField(order_count,LongType,false)), obj#881: org.apache.spark.sql.Row
+- CHNativeColumnarToRow
   +- SortExecTransformer [ltm#872 ASC NULLS FIRST], true, 0
      +- CoalesceBatches
         +- ColumnarExchange rangepartitioning(ltm#872 ASC NULLS FIRST, 5), ENSURE_REQUIREMENTS, [plan_id=891], [id=#891], [OUTPUT] List(ltm:StringType, order_count:LongType), [OUTPUT] List(ltm:StringType, order_count:LongType)
            +- HashAggregateTransformer(keys=[_groupingexpression#882], functions=[count(1)], output=[ltm#872, order_count#873L])
               +- CoalesceBatches
                  +- ColumnarExchange hashpartitioning(_groupingexpression#882, 5), ENSURE_REQUIREMENTS, [plan_id=888], [id=#888], [OUTPUT] Vector(_groupingexpression:StringType, count:LongType), [OUTPUT] Vector(_groupingexpression:StringType, count:LongType)
                     +- HashAggregateTransformer(keys=[_groupingexpression#882], functions=[partial_count(1)], output=[_groupingexpression#882, count#880L])
                        +- RowToCHNativeColumnar
                           +- Project [ltrim(-, Some(o_orderpriority#844)) AS _groupingexpression#882]
                              +- CHNativeColumnarToRow
                                 +- CHBroadcastHashJoinExecTransformer [o_orderkey#839L], [l_orderkey#848L], LeftSemi, BuildRight, false
                                    :- ProjectExecTransformer [o_orderkey#839L, o_orderpriority#844]
                                    :  +- FilterExecTransformer ((isnotnull(o_orderdate#843) AND (o_orderdate#843 >= 1993-07-01)) AND (o_orderdate#843 < 1993-10-01))
                                    :     +- NativeFileScan parquet default.orders[o_orderkey#839L,o_orderdate#843,o_orderpriority#844] Batched: true, DataFilters: [isnotnull(o_orderdate#843), (o_orderdate#843 >= 1993-07-01), (o_orderdate#843 < 1993-10-01)], Format: Parquet, Location: InMemoryFileIndex(1 paths)[file:/home/admin123/Documents/work/code/Gluten-Dev/backends-clickhouse..., PartitionFilters: [], PushedFilters: [IsNotNull(o_orderdate), GreaterThanOrEqual(o_orderdate,1993-07-01), LessThan(o_orderdate,1993-10..., ReadSchema: struct<o_orderkey:bigint,o_orderdate:date,o_orderpriority:string>
                                    +- ColumnarBroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, true]),false), [plan_id=861]
                                       +- ProjectExecTransformer [l_orderkey#848L]
                                          +- FilterExecTransformer ((isnotnull(l_commitdate#859) AND isnotnull(l_receiptdate#860)) AND (l_commitdate#859 < l_receiptdate#860))
                                             +- NativeFileScan parquet default.lineitem[l_orderkey#848L,l_commitdate#859,l_receiptdate#860] Batched: true, DataFilters:

With error:

Logical error: 'Invalid number of columns in chunk pushed to OutputPort. Expected 2, found 1
Header: _groupingexpression#882 Nullable(String) Nullable(size = 0, String(size = 0), UInt8(size = 0)), count#875 Int64 Int64(size = 0)
Chunk:  Nullable(size = 3747, String(size = 3747), UInt8(size = 3747))