[feature](variable) enables views, materialized views, generated columns, and alias functions to persist session variables (apache#58031)

### What problem does this PR solve?

Problem Summary:
When creating views, generated columns, materialized views, or alias
functions, session variables that affect query results (e.g.,
`enable_decimal256`) are not persisted. This causes inconsistent query
results when session variables differ between creation and query time.

This PR persists session variables marked with `affectQueryResult()`
annotation for:
- Views
- Generated Columns  
- Materialized Views
- Alias Functions

When querying these objects, the system automatically uses the persisted
session variables from creation time, ensuring consistent results.

#### Key Changes

- Added `sessionVariables` field to `View`, `Column`, `MTMV`, and
`AliasFunction` classes
- Created `AutoCloseSessionVariable` utility to temporarily apply
persisted variables during query processing
- Modified view/column/MV/function parsing logic to use persisted
session variables
- Added `SessionVarGuardExpr` to protect expressions that depend on
session variables

---------


Co-authored-by: jacktengg <[email protected]>

Author: feiniaofeiafei

be/src/olap/memtable.cpp

@@ -114,13 +114,13 @@ void MemTable::_init_agg_functions(const vectorized::Block* block) {
             // the aggregate function manually.
             if (_skip_bitmap_col_idx != cid) {
                 function = vectorized::AggregateFunctionSimpleFactory::instance().get(
-                        "replace_load", {block->get_data_type(cid)},
+                        "replace_load", {block->get_data_type(cid)}, block->get_data_type(cid),
                         block->get_data_type(cid)->is_nullable(),
                         BeExecVersionManager::get_newest_version());
             } else {
                 function = vectorized::AggregateFunctionSimpleFactory::instance().get(
-                        "bitmap_intersect", {block->get_data_type(cid)}, false,
-                        BeExecVersionManager::get_newest_version());
+                        "bitmap_intersect", {block->get_data_type(cid)}, block->get_data_type(cid),
+                        false, BeExecVersionManager::get_newest_version());
             }
         } else {
             function = _tablet_schema->column(cid).get_aggregate_function(

be/src/olap/tablet_schema.cpp

@@ -776,7 +776,8 @@ vectorized::AggregateFunctionPtr TabletColumn::get_aggregate_function(
         std::transform(agg_name.begin(), agg_name.end(), agg_name.begin(),
                        [](unsigned char c) { return std::tolower(c); });
         function = vectorized::AggregateFunctionSimpleFactory::instance().get(
-                agg_name, {type}, type->is_nullable(), BeExecVersionManager::get_newest_version());
+                agg_name, {type}, type, type->is_nullable(),
+                BeExecVersionManager::get_newest_version());
         if (!function) {
             LOG(WARNING) << "get column aggregate function failed, aggregation_name=" << origin_name
                          << ", column_type=" << type->get_name();

be/src/pipeline/exec/analytic_sink_operator.h

@@ -186,9 +186,15 @@ class AnalyticSinkOperatorX final : public DataSinkOperatorX<AnalyticSinkLocalSt
 #ifdef BE_TEST
     AnalyticSinkOperatorX(ObjectPool* pool)
             : _pool(pool),
+              _intermediate_tuple_id(0),
+              _output_tuple_id(0),
               _buffered_tuple_id(0),
               _is_colocate(false),
-              _require_bucket_distribution(false) {}
+              _require_bucket_distribution(false),
+              _has_window(false),
+              _has_range_window(false),
+              _has_window_start(false),
+              _has_window_end(false) {}
 #endif
 
     Status init(const TDataSink& tsink) override {

be/src/runtime/primitive_type.h

@@ -206,10 +206,6 @@ class RuntimeState {
         return _query_options.__isset.enable_insert_strict && _query_options.enable_insert_strict;
     }
 
-    bool enable_decimal256() const {
-        return _query_options.__isset.enable_decimal256 && _query_options.enable_decimal256;
-    }
-
     bool enable_common_expr_pushdown() const {
         return _query_options.__isset.enable_common_expr_pushdown &&
                _query_options.enable_common_expr_pushdown;

be/src/runtime/runtime_state.h

@@ -46,7 +46,6 @@ class IColumn;
 class IDataType;
 
 struct AggregateFunctionAttr {
-    bool enable_decimal256 {false};
     bool is_window_function {false};
     bool is_foreach {false};
     std::vector<std::string> column_names;

be/src/vec/aggregate_functions/aggregate_function.h

@@ -26,7 +26,7 @@ QueryContext* AggregateFunctionAIAggData::_ctx = nullptr;
 void register_aggregate_function_ai_agg(AggregateFunctionSimpleFactory& factory) {
     factory.register_function_both("ai_agg",
                                    [](const std::string& name, const DataTypes& argument_types,
-                                      const bool result_is_nullable,
+                                      const DataTypePtr& result_type, const bool result_is_nullable,
                                       const AggregateFunctionAttr& attr) -> AggregateFunctionPtr {
                                        return creator_without_type::create<AggregateFunctionAIAgg>(
                                                argument_types, result_is_nullable, attr);

be/src/vec/aggregate_functions/aggregate_function_ai_agg.cpp

@@ -25,8 +25,8 @@ namespace doris::vectorized {
 #include "common/compile_check_begin.h"
 
 AggregateFunctionPtr create_aggregate_function_approx_count_distinct(
-        const std::string& name, const DataTypes& argument_types, const bool result_is_nullable,
-        const AggregateFunctionAttr& attr) {
+        const std::string& name, const DataTypes& argument_types, const DataTypePtr& result_type,
+        const bool result_is_nullable, const AggregateFunctionAttr& attr) {
     return creator_with_type_list<
             TYPE_BOOLEAN, TYPE_TINYINT, TYPE_SMALLINT, TYPE_INT, TYPE_BIGINT, TYPE_LARGEINT,
             TYPE_FLOAT, TYPE_DOUBLE, TYPE_DECIMAL32, TYPE_DECIMAL64, TYPE_DECIMAL128I,

be/src/vec/aggregate_functions/aggregate_function_approx_count_distinct.cpp

@@ -42,6 +42,7 @@ AggregateFunctionPtr do_create_agg_function_collect(const DataTypes& argument_ty
 
 AggregateFunctionPtr create_aggregate_function_array_agg(const std::string& name,
                                                          const DataTypes& argument_types,
+                                                         const DataTypePtr& result_type,
                                                          const bool result_is_nullable,
                                                          const AggregateFunctionAttr& attr) {
     AggregateFunctionPtr agg_fn;

be/src/vec/aggregate_functions/aggregate_function_array_agg.cpp

@@ -28,41 +28,43 @@
 namespace doris::vectorized {
 #include "common/compile_check_begin.h"
 
+// TODO: use result type got from FE plan
 template <PrimitiveType T>
 struct Avg {
-    using FieldType = typename PrimitiveTypeTraits<T>::AvgNearestFieldType;
+    static constexpr PrimitiveType ResultPType = T == TYPE_DECIMALV2 ? T : TYPE_DOUBLE;
     using Function = AggregateFunctionAvg<
-            T, AggregateFunctionAvgData<PrimitiveTypeTraits<T>::AvgNearestPrimitiveType>>;
+            T, ResultPType,
+            AggregateFunctionAvgData<PrimitiveTypeTraits<T>::AvgNearestPrimitiveType>>;
 };
 
 template <PrimitiveType T>
 using AggregateFuncAvg = typename Avg<T>::Function;
 
-template <PrimitiveType T>
-struct AvgDecimal256 {
-    using FieldType = typename PrimitiveTypeTraits<T>::AvgNearestFieldType256;
-    using Function = AggregateFunctionAvg<
-            T, AggregateFunctionAvgData<PrimitiveTypeTraits<T>::AvgNearestPrimitiveType256>>;
+// use result type got from FE plan
+template <PrimitiveType InputType, PrimitiveType ResultType>
+struct AvgDecimalV3 {
+    using Function =
+            AggregateFunctionAvg<InputType, ResultType, AggregateFunctionAvgData<ResultType>>;
 };
 
-template <PrimitiveType T>
-using AggregateFuncAvgDecimal256 = typename AvgDecimal256<T>::Function;
+template <PrimitiveType InputType, PrimitiveType ResultType>
+using AggregateFuncAvgDecimalV3 = typename AvgDecimalV3<InputType, ResultType>::Function;
 
 void register_aggregate_function_avg(AggregateFunctionSimpleFactory& factory) {
     AggregateFunctionCreator creator = [&](const std::string& name, const DataTypes& types,
+                                           const DataTypePtr& result_type,
                                            const bool result_is_nullable,
                                            const AggregateFunctionAttr& attr) {
-        if (attr.enable_decimal256 && is_decimal(types[0]->get_primitive_type())) {
-            return creator_with_type_list<
-                    TYPE_DECIMAL32, TYPE_DECIMAL64, TYPE_DECIMAL128I,
-                    TYPE_DECIMAL256>::creator<AggregateFuncAvgDecimal256>(name, types,
-                                                                          result_is_nullable, attr);
+        if (is_decimalv3(types[0]->get_primitive_type())) {
+            return creator_with_type_list<TYPE_DECIMAL32, TYPE_DECIMAL64, TYPE_DECIMAL128I,
+                                          TYPE_DECIMAL256>::
+                    creator_with_result_type<AggregateFuncAvgDecimalV3>(name, types, result_type,
+                                                                        result_is_nullable, attr);
         } else {
             return creator_with_type_list<
                     TYPE_TINYINT, TYPE_SMALLINT, TYPE_INT, TYPE_BIGINT, TYPE_LARGEINT, TYPE_DOUBLE,
-                    TYPE_DECIMAL32, TYPE_DECIMAL64, TYPE_DECIMAL128I,
-                    TYPE_DECIMALV2>::creator<AggregateFuncAvg>(name, types, result_is_nullable,
-                                                               attr);
+                    TYPE_DECIMALV2>::creator<AggregateFuncAvg>(name, types, result_type,
+                                                               result_is_nullable, attr);
         }
     };
     factory.register_function_both("avg", creator);