[BugFix] Fix chunk accumulator miss assign _max_size (#52579)

``` (java.sql.SQLException)(conn=143)Intermediate chunk size mustnot be greater than 1024,actually 4095 after 0-th operator spillable_org .mariadb.jdbc.export.ExceptionFactory.createException():299 org.mariadb.jdbc.export.ExceptionFactory.create():370 org .mariadb.jdbc.message.ClientMessage.readPacket():137org.mariadb.jdbc.client.impl.StandardClient.readPacket():840org .mariadb.jdbc.client.impl.StandardClient.readResults():779org.mariadb.jdbc.client.impl.StandardClient.readResponse():698org. mariadb.jdbc.client.impl.StandardClient.execute():641 org .mariadb.jdbc.client.impl.MultiPrimaryClient.execute():347 org.mariadb.jdbc.Statement.executelnternal():935 org.mariadb.jdbc.Statement.executeUpdate():917 ``` Signed-off-by: stdpain <[email protected]>
StarRocks · Nov 4, 2024 · 19c386f · 19c386f
1 parent 3966363
commit 19c386f
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 0 deletions.
diff --git a/be/src/exec/pipeline/aggregate/spillable_aggregate_blocking_source_operator.cpp b/be/src/exec/pipeline/aggregate/spillable_aggregate_blocking_source_operator.cpp
@@ -24,6 +24,7 @@ Status SpillableAggregateBlockingSourceOperator::prepare(RuntimeState* state) {
     RETURN_IF_ERROR(AggregateBlockingSourceOperator::prepare(state));
     RETURN_IF_ERROR(_stream_aggregator->prepare(state, state->obj_pool(), _unique_metrics.get()));
     RETURN_IF_ERROR(_stream_aggregator->open(state));
+    _accumulator.set_max_size(state->chunk_size());
     return Status::OK();
 }
 

diff --git a/be/src/exec/pipeline/aggregate/spillable_aggregate_distinct_blocking_operator.cpp b/be/src/exec/pipeline/aggregate/spillable_aggregate_distinct_blocking_operator.cpp
@@ -182,6 +182,7 @@ Status SpillableAggregateDistinctBlockingSourceOperator::prepare(RuntimeState* s
     RETURN_IF_ERROR(AggregateDistinctBlockingSourceOperator::prepare(state));
     RETURN_IF_ERROR(_stream_aggregator->prepare(state, state->obj_pool(), _unique_metrics.get()));
     RETURN_IF_ERROR(_stream_aggregator->open(state));
+    _accumulator.set_max_size(state->chunk_size());
     return Status::OK();
 }
 

diff --git a/be/src/exec/pipeline/hash_partition_context.cpp b/be/src/exec/pipeline/hash_partition_context.cpp
@@ -37,6 +37,7 @@ Status HashPartitionContext::prepare(RuntimeState* state, RuntimeProfile* profil
         _has_nullable_key = _has_nullable_key || _partition_types[i].is_nullable;
     }
 
+    _acc.set_max_size(state->chunk_size());
     _chunks_partitioner = std::make_unique<ChunksPartitioner>(_has_nullable_key, _partition_exprs, _partition_types);
     return _chunks_partitioner->prepare(state, profile);
 }

diff --git a/test/sql/test_spill/R/test_spill_aggregate b/test/sql/test_spill/R/test_spill_aggregate
@@ -207,4 +207,15 @@ select distinct c0, c1 from t5 order by 1, 2 limit 1;
 select count(*) from (select distinct c0, c1 from t5 limit 10) tb;
 -- result:
 10
+-- !result
+set chunk_size = 1024;
+-- result:
+-- !result
+select max(s1), avg(s2) from (select sum(c1) s1, sum(c0) s2 from t5 group by c0) tb;
+-- result:
+649999	325000.5
+-- !result
+select count(*) from (select distinct c0, c1 from t5) tb;
+-- result:
+650000
 -- !result
diff --git a/test/sql/test_spill/T/test_spill_aggregate b/test/sql/test_spill/T/test_spill_aggregate
@@ -98,3 +98,7 @@ select count(s1), count(s2) from (select sum(c1) s1, sum(c0) s2, c0 from t5 grou
 select count(s1), count(s2) from (select sum(c1) s1, sum(c0) s2, c0, c1 from t5 group by c0, c1 limit 10) tb;
 select distinct c0, c1 from t5 order by 1, 2 limit 1;
 select count(*) from (select distinct c0, c1 from t5 limit 10) tb;
+
+set chunk_size = 1024;
+select max(s1), avg(s2) from (select sum(c1) s1, sum(c0) s2 from t5 group by c0) tb;
+select count(*) from (select distinct c0, c1 from t5) tb;