From dab32360f933945da923ade0f4b35b379bcf7f57 Mon Sep 17 00:00:00 2001
From: philo <feilong.he@intel.com>
Date: Thu, 25 Aug 2022 20:32:41 +0800
Subject: [PATCH] Correct the setting for length_

---
 .../codegen/arrow_compute/ext/actions_impl.cc | 276 ++++++++++++------
 1 file changed, 183 insertions(+), 93 deletions(-)

diff --git a/native-sql-engine/cpp/src/codegen/arrow_compute/ext/actions_impl.cc b/native-sql-engine/cpp/src/codegen/arrow_compute/ext/actions_impl.cc
index d644574b5..9dac96016 100644
--- a/native-sql-engine/cpp/src/codegen/arrow_compute/ext/actions_impl.cc
+++ b/native-sql-engine/cpp/src/codegen/arrow_compute/ext/actions_impl.cc
@@ -150,6 +150,10 @@ class UniqueAction : public ActionBase {
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = std::make_shared<ArrayType>(in_list[0]);
     row_id_ = 0;
@@ -198,13 +202,12 @@ class UniqueAction : public ActionBase {
     cache_validity_.resize(max_group_size, false);
     null_flag_.resize(max_group_size, false);
     cache_.resize(max_group_size + 1);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     if (cache_validity_[dest_group_id] == false) {
       cache_validity_[dest_group_id] = true;
@@ -215,7 +218,7 @@ class UniqueAction : public ActionBase {
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     if (cache_validity_[dest_group_id] == false) {
       cache_validity_[dest_group_id] = true;
@@ -308,6 +311,10 @@ class CountAction : public ActionBase {
     if (cache_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_list_ = in_list;
     row_id = 0;
@@ -356,7 +363,6 @@ class CountAction : public ActionBase {
       max_group_size = target_group_size * 2;
     }
     cache_.resize(max_group_size, 0);
-    length_ = cache_.size();
     return arrow::Status::OK();
   }
 
@@ -387,7 +393,7 @@ class CountAction : public ActionBase {
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
 
     cache_[dest_group_id] += 1;
@@ -396,7 +402,7 @@ class CountAction : public ActionBase {
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -471,6 +477,10 @@ class CountDistinctAction : public ActionBase {
     if (cache_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     // prepare evaluate lambda
     *on_valid = [this](int dest_group_id) {
@@ -490,7 +500,6 @@ class CountDistinctAction : public ActionBase {
       max_group_size = target_group_size * 2;
     }
     cache_.resize(max_group_size, 0);
-    length_ = cache_.size();
     return arrow::Status::OK();
   }
 
@@ -540,7 +549,7 @@ class CountDistinctAction : public ActionBase {
 
   arrow::Status Evaluate(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_[dest_group_id] += 1;
     return arrow::Status::OK();
@@ -548,7 +557,7 @@ class CountDistinctAction : public ActionBase {
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -622,6 +631,10 @@ class CountLiteralAction : public ActionBase {
     if (cache_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     // prepare evaluate lambda
     *on_valid = [this](int dest_group_id) {
@@ -641,7 +654,6 @@ class CountLiteralAction : public ActionBase {
       max_group_size = target_group_size * 2;
     }
     cache_.resize(max_group_size, 0);
-    length_ = cache_.size();
     return arrow::Status::OK();
   }
 
@@ -661,7 +673,7 @@ class CountLiteralAction : public ActionBase {
 
   arrow::Status Evaluate(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_[dest_group_id] += 1;
     return arrow::Status::OK();
@@ -669,7 +681,7 @@ class CountLiteralAction : public ActionBase {
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -746,6 +758,10 @@ class MinAction<DataType, CType, precompile::enable_if_number<DataType>>
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
     GetFunction<CType>(in_list, max_group_id, on_valid, on_null);
     return arrow::Status::OK();
   }
@@ -759,7 +775,6 @@ class MinAction<DataType, CType, precompile::enable_if_number<DataType>>
     }
     cache_validity_.resize(max_group_size, false);
     cache_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -775,7 +790,7 @@ class MinAction<DataType, CType, precompile::enable_if_number<DataType>>
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     GetMinResultWithGroupBy<CType>(dest_group_id, data);
     return arrow::Status::OK();
@@ -783,7 +798,7 @@ class MinAction<DataType, CType, precompile::enable_if_number<DataType>>
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -1027,6 +1042,10 @@ class MinAction<DataType, CType, precompile::enable_if_decimal<DataType>>
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = std::make_shared<ArrayType>(in_list[0]);
     in_null_count_ = in_->null_count();
@@ -1063,7 +1082,6 @@ class MinAction<DataType, CType, precompile::enable_if_decimal<DataType>>
     }
     cache_validity_.resize(max_group_size, false);
     cache_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -1084,7 +1102,7 @@ class MinAction<DataType, CType, precompile::enable_if_decimal<DataType>>
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     if (!cache_validity_[dest_group_id]) {
       cache_[dest_group_id] = *(CType*)data;
@@ -1098,7 +1116,7 @@ class MinAction<DataType, CType, precompile::enable_if_decimal<DataType>>
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -1183,6 +1201,10 @@ class MinAction<DataType, CType, precompile::enable_if_string_like<DataType>>
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = std::make_shared<ArrayType>(in_list[0]);
     in_null_count_ = in_->null_count();
@@ -1219,7 +1241,6 @@ class MinAction<DataType, CType, precompile::enable_if_string_like<DataType>>
     }
     cache_validity_.resize(max_group_size, false);
     cache_.resize(max_group_size, "");
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -1249,7 +1270,7 @@ class MinAction<DataType, CType, precompile::enable_if_string_like<DataType>>
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     if (!cache_validity_[dest_group_id]) {
       cache_[dest_group_id] = *(CType*)data;
@@ -1263,7 +1284,7 @@ class MinAction<DataType, CType, precompile::enable_if_string_like<DataType>>
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -1353,6 +1374,10 @@ class MaxAction<DataType, CType, precompile::enable_if_number<DataType>>
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
     GetFunction<CType>(in_list, max_group_id, on_valid, on_null);
     return arrow::Status::OK();
   }
@@ -1366,7 +1391,6 @@ class MaxAction<DataType, CType, precompile::enable_if_number<DataType>>
     }
     cache_validity_.resize(max_group_size, false);
     cache_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -1382,7 +1406,7 @@ class MaxAction<DataType, CType, precompile::enable_if_number<DataType>>
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     GetMaxResultWithGroupBy<CType>(dest_group_id, data);
     return arrow::Status::OK();
@@ -1390,7 +1414,7 @@ class MaxAction<DataType, CType, precompile::enable_if_number<DataType>>
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -1629,6 +1653,10 @@ class MaxAction<DataType, CType, precompile::enable_if_decimal<DataType>>
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = std::make_shared<ArrayType>(in_list[0]);
     in_null_count_ = in_->null_count();
@@ -1665,7 +1693,6 @@ class MaxAction<DataType, CType, precompile::enable_if_decimal<DataType>>
     }
     cache_validity_.resize(max_group_size, false);
     cache_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -1685,7 +1712,7 @@ class MaxAction<DataType, CType, precompile::enable_if_decimal<DataType>>
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     if (!cache_validity_[dest_group_id]) {
       cache_[dest_group_id] = *(CType*)data;
@@ -1699,7 +1726,7 @@ class MaxAction<DataType, CType, precompile::enable_if_decimal<DataType>>
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -1784,6 +1811,10 @@ class MaxAction<DataType, CType, precompile::enable_if_string_like<DataType>>
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = std::make_shared<ArrayType>(in_list[0]);
     in_null_count_ = in_->null_count();
@@ -1821,7 +1852,6 @@ class MaxAction<DataType, CType, precompile::enable_if_string_like<DataType>>
     }
     cache_validity_.resize(max_group_size, false);
     cache_.resize(max_group_size, "");
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -1851,7 +1881,7 @@ class MaxAction<DataType, CType, precompile::enable_if_string_like<DataType>>
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     if (!cache_validity_[dest_group_id]) {
       cache_[dest_group_id] = *(CType*)data;
@@ -1865,7 +1895,7 @@ class MaxAction<DataType, CType, precompile::enable_if_string_like<DataType>>
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -1956,6 +1986,10 @@ class SumAction<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = in_list[0];
     in_null_count_ = in_->null_count();
@@ -1998,7 +2032,6 @@ class SumAction<DataType, CType, ResDataType, ResCType,
     }
     cache_validity_.resize(max_group_size, false);
     cache_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -2022,7 +2055,7 @@ class SumAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_validity_[dest_group_id] = true;
     cache_[dest_group_id] += *(CType*)data;
@@ -2031,7 +2064,7 @@ class SumAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -2116,6 +2149,10 @@ class SumAction<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = std::make_shared<ArrayType>(in_list[0]);
     in_null_count_ = in_->null_count();
@@ -2156,7 +2193,6 @@ class SumAction<DataType, CType, ResDataType, ResCType,
     }
     cache_validity_.resize(max_group_size, false);
     cache_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -2176,7 +2212,7 @@ class SumAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_validity_[dest_group_id] = true;
     cache_[dest_group_id] += *(CType*)data;
@@ -2185,7 +2221,7 @@ class SumAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -2282,6 +2318,10 @@ class SumActionPartial<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = in_list[0];
     in_null_count_ = in_->null_count();
@@ -2325,7 +2365,6 @@ class SumActionPartial<DataType, CType, ResDataType, ResCType,
     }
     cache_validity_.resize(max_group_size, false);
     cache_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -2348,7 +2387,7 @@ class SumActionPartial<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_validity_[dest_group_id] = true;
     cache_[dest_group_id] += *(CType*)data;
@@ -2357,7 +2396,7 @@ class SumActionPartial<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -2447,6 +2486,10 @@ class SumActionPartial<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = std::make_shared<ArrayType>(in_list[0]);
     in_null_count_ = in_->null_count();
@@ -2489,7 +2532,6 @@ class SumActionPartial<DataType, CType, ResDataType, ResCType,
     }
     cache_validity_.resize(max_group_size, false);
     cache_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -2509,7 +2551,7 @@ class SumActionPartial<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_validity_[dest_group_id] = true;
     cache_[dest_group_id] += *(CType*)data;
@@ -2518,7 +2560,7 @@ class SumActionPartial<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -2617,6 +2659,10 @@ class AvgAction<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = in_list[0];
     // prepare evaluate lambda
@@ -2659,7 +2705,6 @@ class AvgAction<DataType, CType, ResDataType, ResCType,
     cache_validity_.resize(max_group_size, false);
     cache_sum_.resize(max_group_size, 0);
     cache_count_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -2688,7 +2733,7 @@ class AvgAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_validity_[dest_group_id] = true;
     cache_sum_[dest_group_id] += *(CType*)data;
@@ -2698,7 +2743,7 @@ class AvgAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -2786,6 +2831,10 @@ class AvgAction<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = in_list[0];
     // prepare evaluate lambda
@@ -2828,7 +2877,6 @@ class AvgAction<DataType, CType, ResDataType, ResCType,
     cache_validity_.resize(max_group_size, false);
     cache_sum_.resize(max_group_size, 0);
     cache_count_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -2857,7 +2905,7 @@ class AvgAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_validity_[dest_group_id] = true;
     cache_sum_[dest_group_id] += *(CType*)data;
@@ -2867,7 +2915,7 @@ class AvgAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -2971,6 +3019,10 @@ class SumCountAction<DataType, CType, ResDataType, ResCType,
     if (cache_sum_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = in_list[0];
     in_null_count_ = in_->null_count();
@@ -3015,7 +3067,6 @@ class SumCountAction<DataType, CType, ResDataType, ResCType,
     cache_validity_.resize(max_group_size, false);
     cache_sum_.resize(max_group_size, 0);
     cache_count_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -3043,7 +3094,7 @@ class SumCountAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_sum_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_sum_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_sum_[dest_group_id] += *(CType*)data;
     cache_count_[dest_group_id] += 1;
@@ -3053,7 +3104,7 @@ class SumCountAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_sum_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_sum_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -3158,6 +3209,10 @@ class SumCountAction<DataType, CType, ResDataType, ResCType,
     if (cache_sum_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = std::make_shared<ArrayType>(in_list[0]);
     in_null_count_ = in_->null_count();
@@ -3202,7 +3257,6 @@ class SumCountAction<DataType, CType, ResDataType, ResCType,
     cache_validity_.resize(max_group_size, false);
     cache_sum_.resize(max_group_size, 0);
     cache_count_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -3223,7 +3277,7 @@ class SumCountAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_sum_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_sum_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_sum_[dest_group_id] += *(CType*)data;
     cache_count_[dest_group_id] += 1;
@@ -3233,7 +3287,7 @@ class SumCountAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_sum_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_sum_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -3341,6 +3395,10 @@ class SumCountMergeAction<DataType, CType, ResDataType, ResCType,
     if (cache_sum_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_sum_ = in_list[0];
     in_count_ = in_list[1];
@@ -3377,7 +3435,6 @@ class SumCountMergeAction<DataType, CType, ResDataType, ResCType,
     cache_validity_.resize(max_group_size, false);
     cache_sum_.resize(max_group_size, 0);
     cache_count_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -3404,7 +3461,7 @@ class SumCountMergeAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data, void* data2) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_sum_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_sum_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_sum_[dest_group_id] += *(CType*)data;
     cache_count_[dest_group_id] += *(int64_t*)data2;
@@ -3414,7 +3471,7 @@ class SumCountMergeAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_sum_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_sum_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -3520,6 +3577,10 @@ class SumCountMergeAction<DataType, CType, ResDataType, ResCType,
     if (cache_sum_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_sum_ = std::make_shared<ArrayType>(in_list[0]);
     in_count_ = std::make_shared<precompile::Int64Array>(in_list[1]);
@@ -3554,7 +3615,6 @@ class SumCountMergeAction<DataType, CType, ResDataType, ResCType,
     cache_validity_.resize(max_group_size, false);
     cache_sum_.resize(max_group_size, 0);
     cache_count_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -3576,7 +3636,7 @@ class SumCountMergeAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data, void* data2) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_sum_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_sum_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_sum_[dest_group_id] += *(CType*)data;
     cache_count_[dest_group_id] += *(int64_t*)data2;
@@ -3586,7 +3646,7 @@ class SumCountMergeAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_sum_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_sum_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -3688,6 +3748,10 @@ class AvgByCountAction<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_sum_ = in_list[0];
     in_count_ = in_list[1];
@@ -3723,7 +3787,6 @@ class AvgByCountAction<DataType, CType, ResDataType, ResCType,
     cache_validity_.resize(max_group_size, false);
     cache_sum_.resize(max_group_size, 0);
     cache_count_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -3751,7 +3814,7 @@ class AvgByCountAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data, void* data2) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_sum_[dest_group_id] += *(CType*)data;
     cache_count_[dest_group_id] += *(int64_t*)data2;
@@ -3760,7 +3823,7 @@ class AvgByCountAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -3866,6 +3929,10 @@ class AvgByCountAction<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_sum_ = std::make_shared<ArrayType>(in_list[0]);
     in_count_ = std::make_shared<precompile::Int64Array>(in_list[1]);
@@ -3899,7 +3966,6 @@ class AvgByCountAction<DataType, CType, ResDataType, ResCType,
     cache_validity_.resize(max_group_size, false);
     cache_sum_.resize(max_group_size, 0);
     cache_count_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -3921,7 +3987,7 @@ class AvgByCountAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data, void* data2) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_sum_[dest_group_id] += *(CType*)data;
     cache_count_[dest_group_id] += *(int64_t*)data2;
@@ -3930,7 +3996,7 @@ class AvgByCountAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -4059,6 +4125,10 @@ class StddevSampPartialAction<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = std::make_shared<ArrayType>(in_list[0]);
     in_null_count_ = in_->null_count();
@@ -4100,7 +4170,6 @@ class StddevSampPartialAction<DataType, CType, ResDataType, ResCType,
     cache_sum_.resize(max_group_size, 0);
     cache_count_.resize(max_group_size, 0);
     cache_m2_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -4131,7 +4200,7 @@ class StddevSampPartialAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_validity_[dest_group_id] = true;
     auto pre_avg = cache_sum_[dest_group_id] * 1.0 /
@@ -4146,7 +4215,7 @@ class StddevSampPartialAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -4290,6 +4359,10 @@ class StddevSampPartialAction<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = std::make_shared<ArrayType>(in_list[0]);
     in_null_count_ = in_->null_count();
@@ -4331,7 +4404,6 @@ class StddevSampPartialAction<DataType, CType, ResDataType, ResCType,
     cache_sum_.resize(max_group_size, 0);
     cache_count_.resize(max_group_size, 0);
     cache_m2_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -4362,7 +4434,7 @@ class StddevSampPartialAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_validity_[dest_group_id] = true;
     auto double_data = (*(CType*)data).ToDouble(scale);
@@ -4378,7 +4450,7 @@ class StddevSampPartialAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -4514,6 +4586,11 @@ class StddevSampFinalAction<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
+
     in_count_ = std::make_shared<ArrayType>(in_list[0]);
     in_avg_ = std::make_shared<ArrayType>(in_list[1]);
     in_m2_ = std::make_shared<ArrayType>(in_list[2]);
@@ -4556,7 +4633,6 @@ class StddevSampFinalAction<DataType, CType, ResDataType, ResCType,
     cache_count_.resize(max_group_size, 0);
     cache_avg_.resize(max_group_size, 0);
     cache_m2_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -4582,7 +4658,7 @@ class StddevSampFinalAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data, void* data2, void* data3) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_validity_[dest_group_id] = true;
     auto pre_avg = cache_avg_[dest_group_id];
@@ -4598,7 +4674,7 @@ class StddevSampFinalAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -4718,6 +4794,10 @@ class StddevSampFinalAction<DataType, CType, ResDataType, ResCType,
     if (cache_validity_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_count_ = std::make_shared<CountArrayType>(in_list[0]);
     in_avg_ = std::make_shared<ArrayType>(in_list[1]);
@@ -4766,7 +4846,6 @@ class StddevSampFinalAction<DataType, CType, ResDataType, ResCType,
     cache_count_.resize(max_group_size, 0);
     cache_avg_.resize(max_group_size, 0);
     cache_m2_.resize(max_group_size, 0);
-    length_ = cache_validity_.size();
     return arrow::Status::OK();
   }
 
@@ -4792,7 +4871,7 @@ class StddevSampFinalAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status Evaluate(int dest_group_id, void* data, void* data2, void* data3) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     cache_validity_[dest_group_id] = true;
     auto pre_avg = cache_avg_[dest_group_id];
@@ -4813,7 +4892,7 @@ class StddevSampFinalAction<DataType, CType, ResDataType, ResCType,
 
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_validity_.size() <= target_group_size) GrowByFactor(target_group_size);
+    if (cache_validity_.size() < target_group_size) GrowByFactor(target_group_size);
     if (length_ < target_group_size) length_ = target_group_size;
     return arrow::Status::OK();
   }
@@ -4954,6 +5033,10 @@ class FirstPartialAction<DataType, CType, ResDataType, ResCType,
     if (cache_first_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = in_list[0];
     in_null_count_ = in_->null_count();
@@ -5000,7 +5083,6 @@ class FirstPartialAction<DataType, CType, ResDataType, ResCType,
     cache_first_.resize(max_group_size);
     cache_value_set_.resize(max_group_size, false);
     cache_null_flag_.resize(max_group_size, false);
-    length_ = cache_first_.size();
     return arrow::Status::OK();
   }
 
@@ -5041,7 +5123,7 @@ class FirstPartialAction<DataType, CType, ResDataType, ResCType,
   // For codegen.
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_first_.size() <= target_group_size) {
+    if (cache_first_.size() < target_group_size) {
       GrowByFactor(target_group_size);
     }
     if (length_ < target_group_size) {
@@ -5059,7 +5141,7 @@ class FirstPartialAction<DataType, CType, ResDataType, ResCType,
   // For codegen.
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_first_.size() <= target_group_size) {
+    if (cache_first_.size() < target_group_size) {
       GrowByFactor(target_group_size);
     }
     if (length_ < target_group_size) {
@@ -5191,6 +5273,10 @@ class FirstPartialAction<DataType, CType, ResDataType, ResCType,
     if (cache_first_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     in_ = std::make_shared<ArrayType>(in_list[0]);
     in_null_count_ = in_->null_count();
@@ -5236,7 +5322,6 @@ class FirstPartialAction<DataType, CType, ResDataType, ResCType,
     cache_first_.resize(max_group_size);
     cache_value_set_.resize(max_group_size, false);
     cache_null_flag_.resize(max_group_size, false);
-    length_ = cache_first_.size();
     return arrow::Status::OK();
   }
 
@@ -5277,7 +5362,7 @@ class FirstPartialAction<DataType, CType, ResDataType, ResCType,
   // For codegen.
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_first_.size() <= target_group_size) {
+    if (cache_first_.size() < target_group_size) {
       GrowByFactor(target_group_size);
     }
     if (length_ < target_group_size) {
@@ -5295,7 +5380,7 @@ class FirstPartialAction<DataType, CType, ResDataType, ResCType,
   // For codegen.
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_first_.size() <= target_group_size) {
+    if (cache_first_.size() < target_group_size) {
       GrowByFactor(target_group_size);
     }
     if (length_ < target_group_size) {
@@ -5425,6 +5510,10 @@ class FirstFinalAction<DataType, CType, ResDataType, ResCType,
     if (cache_first_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
 
     // The input for first value.
     in_ = in_list[0];
@@ -5475,7 +5564,6 @@ class FirstFinalAction<DataType, CType, ResDataType, ResCType,
     cache_first_.resize(max_group_size);
     cache_value_set_.resize(max_group_size, false);
     cache_null_flag_.resize(max_group_size, false);
-    length_ = cache_first_.size();
     return arrow::Status::OK();
   }
 
@@ -5524,7 +5612,7 @@ class FirstFinalAction<DataType, CType, ResDataType, ResCType,
   arrow::Status Evaluate(int dest_group_id, void* data1, void* data2) {
     auto target_group_size = dest_group_id + 1;
     // cache_first_ size should be always as same as others (cache_value_set_, etc).
-    if (cache_first_.size() <= target_group_size) {
+    if (cache_first_.size() < target_group_size) {
       GrowByFactor(target_group_size);
     }
     if (length_ < target_group_size) {
@@ -5553,7 +5641,7 @@ class FirstFinalAction<DataType, CType, ResDataType, ResCType,
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
     // cache_first_ size should be always as same as others, like cache_value_set_, etc.
-    if (cache_first_.size() <= target_group_size) {
+    if (cache_first_.size() < target_group_size) {
       GrowByFactor(target_group_size);
     }
     if (length_ < target_group_size) {
@@ -5579,7 +5667,7 @@ class FirstFinalAction<DataType, CType, ResDataType, ResCType,
   // Just for codegen.
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_first_.size() <= target_group_size) {
+    if (cache_first_.size() < target_group_size) {
       GrowByFactor(target_group_size);
     }
     if (length_ < target_group_size) {
@@ -5683,7 +5771,10 @@ class FirstFinalAction<DataType, CType, ResDataType, ResCType,
     if (cache_first_.size() <= max_group_id) {
       GrowByFactor(max_group_id + 1);
     }
-
+    // The actual size will be used to keep result for each group.
+    if (length_ < max_group_id + 1) {
+      length_ = max_group_id + 1;
+    }
     // The input for first value.
     in_ = std::make_shared<ArrayType>(in_list[0]);
     auto value_set_array = in_list[1];
@@ -5730,7 +5821,6 @@ class FirstFinalAction<DataType, CType, ResDataType, ResCType,
     cache_first_.resize(max_group_size);
     cache_value_set_.resize(max_group_size, false);
     cache_null_flag_.resize(max_group_size, false);
-    length_ = cache_first_.size();
     return arrow::Status::OK();
   }
 
@@ -5776,7 +5866,7 @@ class FirstFinalAction<DataType, CType, ResDataType, ResCType,
   arrow::Status Evaluate(int dest_group_id, void* data1, void* data2) {
     auto target_group_size = dest_group_id + 1;
     // cache_first_ size should be always as same as others (cache_value_set_, etc).
-    if (cache_first_.size() <= target_group_size) {
+    if (cache_first_.size() < target_group_size) {
       GrowByFactor(target_group_size);
     }
     if (length_ < target_group_size) {
@@ -5805,7 +5895,7 @@ class FirstFinalAction<DataType, CType, ResDataType, ResCType,
   arrow::Status Evaluate(int dest_group_id, void* data) {
     auto target_group_size = dest_group_id + 1;
     // cache_first_ size should be always as same as others, like cache_value_set_, etc.
-    if (cache_first_.size() <= target_group_size) {
+    if (cache_first_.size() < target_group_size) {
       GrowByFactor(target_group_size);
     }
     if (length_ < target_group_size) {
@@ -5831,7 +5921,7 @@ class FirstFinalAction<DataType, CType, ResDataType, ResCType,
   // Just for codegen.
   arrow::Status EvaluateNull(int dest_group_id) {
     auto target_group_size = dest_group_id + 1;
-    if (cache_first_.size() <= target_group_size) {
+    if (cache_first_.size() < target_group_size) {
       GrowByFactor(target_group_size);
     }
     if (length_ < target_group_size) {