[fix](be) Validate aggregate state parameters and preserve nullable layouts

be/src/exprs/aggregate/aggregate_function_collect.h

@@ -28,6 +28,7 @@
 #include <string>
 #include <type_traits>
 
+#include "common/exception.h"
 #include "core/assert_cast.h"
 #include "core/column/column.h"
 #include "core/column/column_array.h"
@@ -110,7 +111,10 @@ struct AggregateFunctionCollectSetData {
         }
     }
 
-    void reset() { data_set.clear(); }
+    void reset() {
+        data_set.clear();
+        max_size = -1;
+    }
 };
 
 template <PrimitiveType T, bool HasLimit>
@@ -178,7 +182,10 @@ struct AggregateFunctionCollectSetData<T, HasLimit> {
         }
     }
 
-    void reset() { data_set.clear(); }
+    void reset() {
+        data_set.clear();
+        max_size = -1;
+    }
 };
 
 template <PrimitiveType T, bool HasLimit>
@@ -230,7 +237,10 @@ struct AggregateFunctionCollectListData {
         read_var_int(max_size, buf);
     }
 
-    void reset() { data.clear(); }
+    void reset() {
+        data.clear();
+        max_size = -1;
+    }
 
     void insert_result_into(IColumn& to) const {
         auto& vec = assert_cast<ColVecType&, TypeCheckOnRelease::DISABLE>(to).get_data();
@@ -296,7 +306,10 @@ struct AggregateFunctionCollectListData<T, HasLimit> {
         read_var_int(max_size, buf);
     }
 
-    void reset() { data->clear(); }
+    void reset() {
+        data->clear();
+        max_size = -1;
+    }
 
     void insert_result_into(IColumn& to) const {
         auto& to_str = assert_cast<ColVecType&, TypeCheckOnRelease::DISABLE>(to);
@@ -387,7 +400,10 @@ struct AggregateFunctionCollectListData<T, HasLimit> {
         read_var_int(max_size, buf);
     }
 
-    void reset() { column_data->clear(); }
+    void reset() {
+        column_data->clear();
+        max_size = -1;
+    }
 
     void insert_result_into(IColumn& to) const { to.insert_range_from(*column_data, 0, size()); }
 };
@@ -425,8 +441,7 @@ class AggregateFunctionCollect final
         if constexpr (HasLimit) {
             if (data.max_size == -1) {
                 data.max_size =
-                        (UInt64)assert_cast<const ColumnInt32*, TypeCheckOnRelease::DISABLE>(
-                                columns[1])
+                        assert_cast<const ColumnInt32*, TypeCheckOnRelease::DISABLE>(columns[1])
                                 ->get_element(row_num);
             }
             if (data.size() >= data.max_size) {
@@ -444,6 +459,18 @@ class AggregateFunctionCollect final
                Arena& arena) const override {
         auto& data = this->data(place);
         const auto& rhs_data = this->data(rhs);
+        if constexpr (HasLimit) {
+            if (rhs_data.size() == 0) {
+                return;
+            }
+            if (data.size() == 0) {
+                data.max_size = rhs_data.max_size;
+            } else if (UNLIKELY(data.max_size != rhs_data.max_size)) {
+                throw Exception(ErrorCode::INVALID_ARGUMENT,
+                                "{} aggregate states have incompatible limits: {} vs {}",
+                                get_name(), data.max_size, rhs_data.max_size);
+            }
+        }
         if constexpr (ENABLE_ARENA) {
             data.merge(rhs_data, arena);
         } else {

be/src/exprs/aggregate/aggregate_function_ema.h

@@ -21,8 +21,10 @@
 #pragma once
 
 #include <cmath>
+#include <limits>
 #include <memory>
 
+#include "common/exception.h"
 #include "core/assert_cast.h"
 #include "core/column/column_vector.h"
 #include "core/data_type/data_type_number.h"
@@ -56,18 +58,24 @@ class IColumn;
  *   - value:      numeric column to average
  *   - timeunit:   numeric time index (not raw timestamp; use intDiv if needed)
  * Returns DOUBLE.
+ *
+ * A zero half_decay returns 0 but remains an initialized configuration. Only
+ * fresh/reset states are identities; all initialized states must have matching
+ * half decays when merged.
  */
 struct ExponentialMovingAverageData {
     double value = 0.0;
     double time = 0.0;
     double half_decay = 0.0;
+    bool initialized = false;
 
     static double scale(double time_passed, double hd) { return std::exp2(-time_passed / hd); }
 
     static double sum_weights(double hd) { return 1.0 / (1.0 - std::exp2(-1.0 / hd)); }
 
     void add(double new_value, double current_time, double hd) {
         half_decay = hd;
+        initialized = true;
         ExponentialMovingAverageData other;
         other.value = new_value;
         other.time = current_time;
@@ -86,37 +94,60 @@ struct ExponentialMovingAverageData {
     }
 
     void merge(const ExponentialMovingAverageData& rhs) {
-        double hd = half_decay != 0.0 ? half_decay : rhs.half_decay;
-        if (hd == 0.0) {
+        if (!rhs.initialized) {
             return;
         }
-        half_decay = hd;
-        merge_point(rhs, hd);
+        if (!initialized) {
+            *this = rhs;
+            return;
+        }
+        if (UNLIKELY(half_decay != rhs.half_decay)) {
+            throw Exception(
+                    ErrorCode::INVALID_ARGUMENT,
+                    "exponential_moving_average aggregate states have incompatible half decay");
+        }
+        merge_point(rhs, half_decay);
     }
 
     double get() const {
+        check_half_decay();
         if (half_decay == 0.0) {
             return 0.0;
         }
         return value / sum_weights(half_decay);
     }
 
     void write(BufferWritable& buf) const {
+        check_half_decay();
         buf.write_binary(value);
         buf.write_binary(time);
-        buf.write_binary(half_decay);
+        // NaN is rejected for configured states, so it can encode initialization without
+        // adding a field or colliding with the valid zero half-decay configuration.
+        buf.write_binary(initialized ? half_decay : std::numeric_limits<double>::quiet_NaN());
     }
 
     void read(BufferReadable& buf) {
         buf.read_binary(value);
         buf.read_binary(time);
         buf.read_binary(half_decay);
+        initialized = !std::isnan(half_decay);
+        if (!initialized) {
+            half_decay = 0.0;
+        }
+    }
+
+    void check_half_decay() const {
+        if (UNLIKELY(std::isnan(half_decay))) {
+            throw Exception(ErrorCode::INVALID_ARGUMENT,
+                            "exponential_moving_average half decay must not be NaN");
+        }
     }
 
     void reset() {
         value = 0.0;
         time = 0.0;
         half_decay = 0.0;
+        initialized = false;
     }
 };
 

be/src/exprs/aggregate/aggregate_function_group_concat.h

@@ -22,6 +22,7 @@
 #include <memory>
 #include <string>
 
+#include "common/exception.h"
 #include "core/assert_cast.h"
 #include "core/column/column_string.h"
 #include "core/data_type/data_type_string.h"
@@ -73,6 +74,10 @@ struct AggregateFunctionGroupConcatData {
             separator = rhs.separator;
             data.assign(rhs.data);
         } else {
+            if (UNLIKELY(separator != rhs.separator)) {
+                throw Exception(ErrorCode::INVALID_ARGUMENT,
+                                "group_concat aggregate states have incompatible separators");
+            }
             auto offset = data.size();
 
             auto delta_size = separator.size() + rhs.data.size();

be/src/exprs/aggregate/aggregate_function_histogram.h

@@ -50,7 +50,10 @@ struct AggregateFunctionHistogramData {
 
     void set_parameters(size_t input_max_num_buckets) { max_num_buckets = input_max_num_buckets; }
 
-    void reset() { ordered_map.clear(); }
+    void reset() {
+        ordered_map.clear();
+        max_num_buckets = BUCKET_NUM_INIT_VALUE;
+    }
 
     void add(const StringRef& value, const UInt64& number = 1) {
         std::string data = value.to_string();
@@ -78,7 +81,13 @@ struct AggregateFunctionHistogramData {
             return;
         }
 
-        max_num_buckets = rhs.max_num_buckets;
+        if (!max_num_buckets) {
+            max_num_buckets = rhs.max_num_buckets;
+        } else if (UNLIKELY(max_num_buckets != rhs.max_num_buckets)) {
+            throw Exception(ErrorCode::INVALID_ARGUMENT,
+                            "histogram aggregate states have incompatible bucket counts: {} vs {}",
+                            max_num_buckets, rhs.max_num_buckets);
+        }
 
         for (auto rhs_it : rhs.ordered_map) {
             auto lhs_it = ordered_map.find(rhs_it.first);

be/src/exprs/aggregate/aggregate_function_orthogonal_bitmap.h

@@ -150,6 +150,12 @@ struct AggIntersectCount : public AggOrthBitmapBaseData<T> {
         if (rhs.first_init) {
             return;
         }
+        if (!AggOrthBitmapBaseData<T>::first_init) {
+            if (UNLIKELY(!AggOrthBitmapBaseData<T>::bitmap.has_same_keys(rhs.bitmap))) {
+                throw Exception(ErrorCode::INVALID_ARGUMENT,
+                                "intersect_count aggregate states have incompatible filter values");
+            }
+        }
         AggOrthBitmapBaseData<T>::bitmap.merge(rhs.bitmap);
         AggOrthBitmapBaseData<T>::first_init = false;
     }