diff --git a/be/src/olap/field.h b/be/src/olap/field.h
index c88dcbdfed..8a23fec04b 100644
--- a/be/src/olap/field.h
+++ b/be/src/olap/field.h
@@ -306,7 +306,7 @@ public:
 
 class CharField : public Field {
 public:
-    explicit CharField() : Field() {}
+    explicit CharField() = default;
     explicit CharField(const TabletColumn& column) : Field(column) {}
 
     size_t get_variable_len() const override { return _length; }
@@ -367,7 +367,7 @@ public:
 
 class VarcharField : public Field {
 public:
-    explicit VarcharField() : Field() {}
+    explicit VarcharField() = default;
     explicit VarcharField(const TabletColumn& column) : Field(column) {}
 
     size_t get_variable_len() const override { return _length - OLAP_VARCHAR_MAX_BYTES; }
@@ -428,7 +428,7 @@ public:
 };
 class StringField : public Field {
 public:
-    explicit StringField() : Field() {}
+    explicit StringField() = default;
     explicit StringField(const TabletColumn& column) : Field(column) {}
 
     // minus OLAP_VARCHAR_MAX_BYTES here just for being compatible with old storage format
@@ -480,7 +480,7 @@ public:
 
 class BitmapAggField : public Field {
 public:
-    explicit BitmapAggField() : Field() {}
+    explicit BitmapAggField() = default;
     explicit BitmapAggField(const TabletColumn& column) : Field(column) {}
 
     char* allocate_memory(char* cell_ptr, char* variable_ptr) const override {
@@ -498,7 +498,7 @@ public:
 
 class QuantileStateAggField : public Field {
 public:
-    explicit QuantileStateAggField() : Field() {}
+    explicit QuantileStateAggField() = default;
     explicit QuantileStateAggField(const TabletColumn& column) : Field(column) {}
 
     char* allocate_memory(char* cell_ptr, char* variable_ptr) const override {
@@ -514,9 +514,27 @@ public:
     }
 };
 
+class AggStateField : public Field {
+public:
+    explicit AggStateField() = default;
+    explicit AggStateField(const TabletColumn& column) : Field(column) {}
+
+    char* allocate_memory(char* cell_ptr, char* variable_ptr) const override {
+        auto slice = (Slice*)cell_ptr;
+        slice->data = nullptr;
+        return variable_ptr;
+    }
+
+    AggStateField* clone() const override {
+        auto* local = new AggStateField();
+        Field::clone(local);
+        return local;
+    }
+};
+
 class HllAggField : public Field {
 public:
-    explicit HllAggField() : Field() {}
+    explicit HllAggField() = default;
     explicit HllAggField(const TabletColumn& column) : Field(column) {}
 
     char* allocate_memory(char* cell_ptr, char* variable_ptr) const override {
@@ -541,6 +559,7 @@ public:
             case FieldType::OLAP_FIELD_TYPE_CHAR:
                 return new CharField(column);
             case FieldType::OLAP_FIELD_TYPE_VARCHAR:
+            case FieldType::OLAP_FIELD_TYPE_AGG_STATE:
                 return new VarcharField(column);
             case FieldType::OLAP_FIELD_TYPE_STRING:
                 return new StringField(column);
@@ -588,12 +607,12 @@ public:
 
         // for value column
         switch (column.aggregation()) {
-        case OLAP_FIELD_AGGREGATION_NONE:
-        case OLAP_FIELD_AGGREGATION_SUM:
-        case OLAP_FIELD_AGGREGATION_MIN:
-        case OLAP_FIELD_AGGREGATION_MAX:
-        case OLAP_FIELD_AGGREGATION_REPLACE:
-        case OLAP_FIELD_AGGREGATION_REPLACE_IF_NOT_NULL:
+        case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE:
+        case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_SUM:
+        case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_MIN:
+        case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_MAX:
+        case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE:
+        case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE_IF_NOT_NULL:
             switch (column.type()) {
             case FieldType::OLAP_FIELD_TYPE_CHAR:
                 return new CharField(column);
@@ -642,22 +661,23 @@ public:
             default:
                 return new Field(column);
             }
-        case OLAP_FIELD_AGGREGATION_HLL_UNION:
+        case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_HLL_UNION:
             return new HllAggField(column);
-        case OLAP_FIELD_AGGREGATION_BITMAP_UNION:
+        case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_BITMAP_UNION:
             return new BitmapAggField(column);
-        case OLAP_FIELD_AGGREGATION_QUANTILE_UNION:
+        case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_QUANTILE_UNION:
             return new QuantileStateAggField(column);
-        case OLAP_FIELD_AGGREGATION_UNKNOWN:
-            LOG(WARNING) << "WOW! value column agg type is unknown";
+        case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_GENERIC:
+            return new AggStateField(column);
+        case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_UNKNOWN:
+            CHECK(false) << ", value column no agg type";
             return nullptr;
         }
-        LOG(WARNING) << "WOW! value column no agg type";
         return nullptr;
     }
 
     static Field* create_by_type(const FieldType& type) {
-        TabletColumn column(OLAP_FIELD_AGGREGATION_NONE, type);
+        TabletColumn column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, type);
         return create(column);
     }
 };
diff --git a/be/src/olap/memtable.cpp b/be/src/olap/memtable.cpp
index b09fe144f6..e87bc847e0 100644
--- a/be/src/olap/memtable.cpp
+++ b/be/src/olap/memtable.cpp
@@ -118,8 +118,8 @@ void MemTable::_init_agg_functions(const vectorized::Block* block) {
                     "replace_load", {block->get_data_type(cid)},
                     block->get_data_type(cid)->is_nullable());
         } else {
-            function = _tablet_schema->column(cid).get_aggregate_function(
-                    {block->get_data_type(cid)}, vectorized::AGG_LOAD_SUFFIX);
+            function =
+                    _tablet_schema->column(cid).get_aggregate_function(vectorized::AGG_LOAD_SUFFIX);
         }
 
         DCHECK(function != nullptr);
diff --git a/be/src/olap/olap_common.h b/be/src/olap/olap_common.h
index 4f9351d5e9..9ab2a26962 100644
--- a/be/src/olap/olap_common.h
+++ b/be/src/olap/olap_common.h
@@ -149,7 +149,7 @@ enum class FieldType {
 // Note that in practice, not all types can use all the following aggregation methods
 // For example, it is meaningless to use SUM for the string type (but it will not cause the program to crash)
 // The implementation of the Field class does not perform such checks, and should be constrained when creating the table
-enum FieldAggregationMethod {
+enum class FieldAggregationMethod {
     OLAP_FIELD_AGGREGATION_NONE = 0,
     OLAP_FIELD_AGGREGATION_SUM = 1,
     OLAP_FIELD_AGGREGATION_MIN = 2,
@@ -160,7 +160,8 @@ enum FieldAggregationMethod {
     OLAP_FIELD_AGGREGATION_BITMAP_UNION = 7,
     // Replace if and only if added value is not null
     OLAP_FIELD_AGGREGATION_REPLACE_IF_NOT_NULL = 8,
-    OLAP_FIELD_AGGREGATION_QUANTILE_UNION = 9
+    OLAP_FIELD_AGGREGATION_QUANTILE_UNION = 9,
+    OLAP_FIELD_AGGREGATION_GENERIC = 10
 };
 
 enum class PushType {
diff --git a/be/src/olap/reader.h b/be/src/olap/reader.h
index 394ca247b8..d03ff346e1 100644
--- a/be/src/olap/reader.h
+++ b/be/src/olap/reader.h
@@ -68,8 +68,7 @@ class VExprContext;
 // So we should compare the common prefix columns of lhs and rhs.
 //
 // NOTE: if you are not sure if you can use it, please don't use this function.
-template <typename LhsRowType, typename RhsRowType>
-int compare_row_key(const LhsRowType& lhs, const RhsRowType& rhs) {
+inline int compare_row_key(const RowCursor& lhs, const RowCursor& rhs) {
     auto cmp_cids = std::min(lhs.schema()->num_column_ids(), rhs.schema()->num_column_ids());
     for (uint32_t cid = 0; cid < cmp_cids; ++cid) {
         auto res = lhs.schema()->column(cid)->compare_cell(lhs.cell(cid), rhs.cell(cid));
diff --git a/be/src/olap/rowset/segment_v2/column_reader.cpp b/be/src/olap/rowset/segment_v2/column_reader.cpp
index c4219916c9..9a8a434a0d 100644
--- a/be/src/olap/rowset/segment_v2/column_reader.cpp
+++ b/be/src/olap/rowset/segment_v2/column_reader.cpp
@@ -31,6 +31,7 @@
 #include "olap/column_predicate.h"
 #include "olap/decimal12.h"
 #include "olap/inverted_index_parser.h"
+#include "olap/olap_common.h"
 #include "olap/rowset/segment_v2/binary_dict_page.h" // for BinaryDictPageDecoder
 #include "olap/rowset/segment_v2/binary_plain_page.h"
 #include "olap/rowset/segment_v2/bitmap_index_reader.h"
@@ -1287,7 +1288,8 @@ void DefaultValueColumnIterator::insert_default_data(const TypeInfo* type_info,
     case FieldType::OLAP_FIELD_TYPE_STRING:
     case FieldType::OLAP_FIELD_TYPE_VARCHAR:
     case FieldType::OLAP_FIELD_TYPE_CHAR:
-    case FieldType::OLAP_FIELD_TYPE_JSONB: {
+    case FieldType::OLAP_FIELD_TYPE_JSONB:
+    case FieldType::OLAP_FIELD_TYPE_AGG_STATE: {
         char* data_ptr = ((Slice*)mem_value)->data;
         size_t data_len = ((Slice*)mem_value)->size;
         dst->insert_many_data(data_ptr, data_len, n);
diff --git a/be/src/olap/rowset/segment_v2/column_writer.cpp b/be/src/olap/rowset/segment_v2/column_writer.cpp
index 0ee3608f84..3468a94dcc 100644
--- a/be/src/olap/rowset/segment_v2/column_writer.cpp
+++ b/be/src/olap/rowset/segment_v2/column_writer.cpp
@@ -155,9 +155,10 @@ Status ColumnWriter::create(const ColumnWriterOptions& opts, const TabletColumn*
                 null_options.need_bloom_filter = false;
                 null_options.need_bitmap_index = false;
 
-                TabletColumn null_column = TabletColumn(
-                        OLAP_FIELD_AGGREGATION_NONE, null_type, null_options.meta->is_nullable(),
-                        null_options.meta->unique_id(), null_options.meta->length());
+                TabletColumn null_column =
+                        TabletColumn(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, null_type,
+                                     null_options.meta->is_nullable(),
+                                     null_options.meta->unique_id(), null_options.meta->length());
                 null_column.set_name("nullable");
                 null_column.set_index_length(-1); // no short key index
                 std::unique_ptr<Field> null_field(FieldFactory::create(null_column));
@@ -212,9 +213,10 @@ Status ColumnWriter::create(const ColumnWriterOptions& opts, const TabletColumn*
             length_options.need_bloom_filter = false;
             length_options.need_bitmap_index = false;
 
-            TabletColumn length_column = TabletColumn(
-                    OLAP_FIELD_AGGREGATION_NONE, length_type, length_options.meta->is_nullable(),
-                    length_options.meta->unique_id(), length_options.meta->length());
+            TabletColumn length_column =
+                    TabletColumn(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, length_type,
+                                 length_options.meta->is_nullable(),
+                                 length_options.meta->unique_id(), length_options.meta->length());
             length_column.set_name("length");
             length_column.set_index_length(-1); // no short key index
             std::unique_ptr<Field> bigint_field(FieldFactory::create(length_column));
@@ -240,9 +242,10 @@ Status ColumnWriter::create(const ColumnWriterOptions& opts, const TabletColumn*
                 null_options.need_bloom_filter = false;
                 null_options.need_bitmap_index = false;
 
-                TabletColumn null_column = TabletColumn(
-                        OLAP_FIELD_AGGREGATION_NONE, null_type, length_options.meta->is_nullable(),
-                        null_options.meta->unique_id(), null_options.meta->length());
+                TabletColumn null_column =
+                        TabletColumn(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, null_type,
+                                     length_options.meta->is_nullable(),
+                                     null_options.meta->unique_id(), null_options.meta->length());
                 null_column.set_name("nullable");
                 null_column.set_index_length(-1); // no short key index
                 std::unique_ptr<Field> null_field(FieldFactory::create(null_column));
@@ -303,9 +306,10 @@ Status ColumnWriter::create(const ColumnWriterOptions& opts, const TabletColumn*
             length_options.need_bloom_filter = false;
             length_options.need_bitmap_index = false;
 
-            TabletColumn length_column = TabletColumn(
-                    OLAP_FIELD_AGGREGATION_NONE, length_type, length_options.meta->is_nullable(),
-                    length_options.meta->unique_id(), length_options.meta->length());
+            TabletColumn length_column =
+                    TabletColumn(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, length_type,
+                                 length_options.meta->is_nullable(),
+                                 length_options.meta->unique_id(), length_options.meta->length());
             length_column.set_name("length");
             length_column.set_index_length(-1); // no short key index
             std::unique_ptr<Field> bigint_field(FieldFactory::create(length_column));
@@ -330,9 +334,9 @@ Status ColumnWriter::create(const ColumnWriterOptions& opts, const TabletColumn*
                 null_options.need_bloom_filter = false;
                 null_options.need_bitmap_index = false;
 
-                TabletColumn null_column =
-                        TabletColumn(OLAP_FIELD_AGGREGATION_NONE, null_type, false,
-                                     null_options.meta->unique_id(), null_options.meta->length());
+                TabletColumn null_column = TabletColumn(
+                        FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, null_type, false,
+                        null_options.meta->unique_id(), null_options.meta->length());
                 null_column.set_name("nullable");
                 null_column.set_index_length(-1); // no short key index
                 std::unique_ptr<Field> null_field(FieldFactory::create(null_column));
diff --git a/be/src/olap/rowset/segment_v2/encoding_info.cpp b/be/src/olap/rowset/segment_v2/encoding_info.cpp
index c500d7e902..573ea92532 100644
--- a/be/src/olap/rowset/segment_v2/encoding_info.cpp
+++ b/be/src/olap/rowset/segment_v2/encoding_info.cpp
@@ -326,6 +326,8 @@ EncodingInfoResolver::EncodingInfoResolver() {
     _add_map<FieldType::OLAP_FIELD_TYPE_OBJECT, PLAIN_ENCODING>();
 
     _add_map<FieldType::OLAP_FIELD_TYPE_QUANTILE_STATE, PLAIN_ENCODING>();
+
+    _add_map<FieldType::OLAP_FIELD_TYPE_AGG_STATE, PLAIN_ENCODING>();
 }
 
 EncodingInfoResolver::~EncodingInfoResolver() {
diff --git a/be/src/olap/rowset/segment_v2/segment_writer.cpp b/be/src/olap/rowset/segment_v2/segment_writer.cpp
index 06cadcf1af..466fe5ef53 100644
--- a/be/src/olap/rowset/segment_v2/segment_writer.cpp
+++ b/be/src/olap/rowset/segment_v2/segment_writer.cpp
@@ -203,6 +203,15 @@ Status SegmentWriter::init(const std::vector<uint32_t>& col_ids, bool has_key,
                 return Status::NotSupported("Do not support bitmap index for jsonb type");
             }
         }
+        if (column.type() == FieldType::OLAP_FIELD_TYPE_AGG_STATE) {
+            opts.need_zone_map = false;
+            if (opts.need_bloom_filter) {
+                return Status::NotSupported("Do not support bloom filter for agg_state type");
+            }
+            if (opts.need_bitmap_index) {
+                return Status::NotSupported("Do not support bitmap index for agg_state type");
+            }
+        }
         if (column.type() == FieldType::OLAP_FIELD_TYPE_MAP) {
             opts.need_zone_map = false;
             if (opts.need_bloom_filter) {
diff --git a/be/src/olap/schema_change.cpp b/be/src/olap/schema_change.cpp
index 3ceebd22d3..34b75436d3 100644
--- a/be/src/olap/schema_change.cpp
+++ b/be/src/olap/schema_change.cpp
@@ -121,7 +121,6 @@ public:
                 try {
                     vectorized::AggregateFunctionPtr function =
                             tablet_schema->column(i).get_aggregate_function(
-                                    {finalized_block.get_data_type(i)},
                                     vectorized::AGG_LOAD_SUFFIX);
                     agg_functions.push_back(function);
                     // create aggregate data
diff --git a/be/src/olap/tablet_meta.cpp b/be/src/olap/tablet_meta.cpp
index 1294720fa2..74a48b07c8 100644
--- a/be/src/olap/tablet_meta.cpp
+++ b/be/src/olap/tablet_meta.cpp
@@ -303,9 +303,13 @@ void TabletMeta::init_column_from_tcolumn(uint32_t unique_id, const TColumn& tco
     }
     if (!tcolumn.is_key) {
         column->set_is_key(false);
-        string aggregation_type;
-        EnumToString(TAggregationType, tcolumn.aggregation_type, aggregation_type);
-        column->set_aggregation(aggregation_type);
+        if (tcolumn.__isset.aggregation) {
+            column->set_aggregation(tcolumn.aggregation);
+        } else {
+            string aggregation_type;
+            EnumToString(TAggregationType, tcolumn.aggregation_type, aggregation_type);
+            column->set_aggregation(aggregation_type);
+        }
     } else {
         column->set_is_key(true);
         column->set_aggregation("NONE");
@@ -317,19 +321,9 @@ void TabletMeta::init_column_from_tcolumn(uint32_t unique_id, const TColumn& tco
     if (tcolumn.__isset.is_bloom_filter_column) {
         column->set_is_bf_column(tcolumn.is_bloom_filter_column);
     }
-    if (tcolumn.column_type.type == TPrimitiveType::STRUCT) {
-        for (size_t i = 0; i < tcolumn.children_column.size(); i++) {
-            ColumnPB* children_column = column->add_children_columns();
-            init_column_from_tcolumn(i, tcolumn.children_column[i], children_column);
-        }
-    } else if (tcolumn.column_type.type == TPrimitiveType::ARRAY) {
+    for (size_t i = 0; i < tcolumn.children_column.size(); i++) {
         ColumnPB* children_column = column->add_children_columns();
-        init_column_from_tcolumn(0, tcolumn.children_column[0], children_column);
-    } else if (tcolumn.column_type.type == TPrimitiveType::MAP) {
-        ColumnPB* key_column = column->add_children_columns();
-        init_column_from_tcolumn(0, tcolumn.children_column[0], key_column);
-        ColumnPB* val_column = column->add_children_columns();
-        init_column_from_tcolumn(0, tcolumn.children_column[1], val_column);
+        init_column_from_tcolumn(i, tcolumn.children_column[i], children_column);
     }
 }
 
diff --git a/be/src/olap/tablet_schema.cpp b/be/src/olap/tablet_schema.cpp
index 7c288f30ff..7fc3e63c15 100644
--- a/be/src/olap/tablet_schema.cpp
+++ b/be/src/olap/tablet_schema.cpp
@@ -114,6 +114,8 @@ FieldType TabletColumn::get_field_type_by_string(const std::string& type_str) {
         type = FieldType::OLAP_FIELD_TYPE_ARRAY;
     } else if (0 == upper_type_str.compare("QUANTILE_STATE")) {
         type = FieldType::OLAP_FIELD_TYPE_QUANTILE_STATE;
+    } else if (0 == upper_type_str.compare("AGG_STATE")) {
+        type = FieldType::OLAP_FIELD_TYPE_AGG_STATE;
     } else {
         LOG(WARNING) << "invalid type string. [type='" << type_str << "']";
         type = FieldType::OLAP_FIELD_TYPE_UNKNOWN;
@@ -129,26 +131,27 @@ FieldAggregationMethod TabletColumn::get_aggregation_type_by_string(const std::s
     FieldAggregationMethod aggregation_type;
 
     if (0 == upper_str.compare("NONE")) {
-        aggregation_type = OLAP_FIELD_AGGREGATION_NONE;
+        aggregation_type = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE;
     } else if (0 == upper_str.compare("SUM")) {
-        aggregation_type = OLAP_FIELD_AGGREGATION_SUM;
+        aggregation_type = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_SUM;
     } else if (0 == upper_str.compare("MIN")) {
-        aggregation_type = OLAP_FIELD_AGGREGATION_MIN;
+        aggregation_type = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_MIN;
     } else if (0 == upper_str.compare("MAX")) {
-        aggregation_type = OLAP_FIELD_AGGREGATION_MAX;
+        aggregation_type = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_MAX;
     } else if (0 == upper_str.compare("REPLACE")) {
-        aggregation_type = OLAP_FIELD_AGGREGATION_REPLACE;
+        aggregation_type = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE;
     } else if (0 == upper_str.compare("REPLACE_IF_NOT_NULL")) {
-        aggregation_type = OLAP_FIELD_AGGREGATION_REPLACE_IF_NOT_NULL;
+        aggregation_type = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE_IF_NOT_NULL;
     } else if (0 == upper_str.compare("HLL_UNION")) {
-        aggregation_type = OLAP_FIELD_AGGREGATION_HLL_UNION;
+        aggregation_type = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_HLL_UNION;
     } else if (0 == upper_str.compare("BITMAP_UNION")) {
-        aggregation_type = OLAP_FIELD_AGGREGATION_BITMAP_UNION;
+        aggregation_type = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_BITMAP_UNION;
     } else if (0 == upper_str.compare("QUANTILE_UNION")) {
-        aggregation_type = OLAP_FIELD_AGGREGATION_QUANTILE_UNION;
+        aggregation_type = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_QUANTILE_UNION;
+    } else if (!upper_str.empty()) {
+        aggregation_type = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_GENERIC;
     } else {
-        LOG(WARNING) << "invalid aggregation type string. [aggregation='" << str << "']";
-        aggregation_type = OLAP_FIELD_AGGREGATION_UNKNOWN;
+        aggregation_type = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_UNKNOWN;
     }
 
     return aggregation_type;
@@ -247,7 +250,8 @@ std::string TabletColumn::get_string_by_field_type(FieldType type) {
         return "OBJECT";
     case FieldType::OLAP_FIELD_TYPE_QUANTILE_STATE:
         return "QUANTILE_STATE";
-
+    case FieldType::OLAP_FIELD_TYPE_AGG_STATE:
+        return "AGG_STATE";
     default:
         return "UNKNOWN";
     }
@@ -255,31 +259,31 @@ std::string TabletColumn::get_string_by_field_type(FieldType type) {
 
 std::string TabletColumn::get_string_by_aggregation_type(FieldAggregationMethod type) {
     switch (type) {
-    case OLAP_FIELD_AGGREGATION_NONE:
+    case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE:
         return "NONE";
 
-    case OLAP_FIELD_AGGREGATION_SUM:
+    case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_SUM:
         return "SUM";
 
-    case OLAP_FIELD_AGGREGATION_MIN:
+    case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_MIN:
         return "MIN";
 
-    case OLAP_FIELD_AGGREGATION_MAX:
+    case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_MAX:
         return "MAX";
 
-    case OLAP_FIELD_AGGREGATION_REPLACE:
+    case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE:
         return "REPLACE";
 
-    case OLAP_FIELD_AGGREGATION_REPLACE_IF_NOT_NULL:
+    case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_REPLACE_IF_NOT_NULL:
         return "REPLACE_IF_NOT_NULL";
 
-    case OLAP_FIELD_AGGREGATION_HLL_UNION:
+    case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_HLL_UNION:
         return "HLL_UNION";
 
-    case OLAP_FIELD_AGGREGATION_BITMAP_UNION:
+    case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_BITMAP_UNION:
         return "BITMAP_UNION";
 
-    case OLAP_FIELD_AGGREGATION_QUANTILE_UNION:
+    case FieldAggregationMethod::OLAP_FIELD_AGGREGATION_QUANTILE_UNION:
         return "QUANTILE_UNION";
 
     default:
@@ -319,6 +323,7 @@ uint32_t TabletColumn::get_field_length_by_type(TPrimitiveType::type type, uint3
         return string_length;
     case TPrimitiveType::VARCHAR:
     case TPrimitiveType::HLL:
+    case TPrimitiveType::AGG_STATE:
         return string_length + sizeof(OLAP_VARCHAR_MAX_LENGTH);
     case TPrimitiveType::STRING:
         return string_length + sizeof(OLAP_STRING_MAX_LENGTH);
@@ -346,7 +351,7 @@ uint32_t TabletColumn::get_field_length_by_type(TPrimitiveType::type type, uint3
     }
 }
 
-TabletColumn::TabletColumn() : _aggregation(OLAP_FIELD_AGGREGATION_NONE) {}
+TabletColumn::TabletColumn() : _aggregation(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE) {}
 
 TabletColumn::TabletColumn(FieldAggregationMethod agg, FieldType type) {
     _aggregation = agg;
@@ -418,30 +423,21 @@ void TabletColumn::init_from_pb(const ColumnPB& column) {
     }
     if (column.has_aggregation()) {
         _aggregation = get_aggregation_type_by_string(column.aggregation());
+        _aggregation_name = column.aggregation();
     }
     if (column.has_visible()) {
         _visible = column.visible();
     }
-    if (_type == FieldType::OLAP_FIELD_TYPE_STRUCT) {
-        for (size_t i = 0; i < column.children_columns_size(); i++) {
-            TabletColumn child_column;
-            child_column.init_from_pb(column.children_columns(i));
-            add_sub_column(child_column);
-        }
-    } else if (_type == FieldType::OLAP_FIELD_TYPE_ARRAY) {
-        DCHECK(column.children_columns_size() == 1) << "ARRAY type has more than 1 children types.";
-        TabletColumn child_column;
-        child_column.init_from_pb(column.children_columns(0));
-        add_sub_column(child_column);
+    if (_type == FieldType::OLAP_FIELD_TYPE_ARRAY) {
+        CHECK(column.children_columns_size() == 1) << "ARRAY type has more than 1 children types.";
     }
     if (_type == FieldType::OLAP_FIELD_TYPE_MAP) {
-        DCHECK(column.children_columns_size() == 2) << "MAP type has more than 2 children types.";
-        TabletColumn key_column;
-        TabletColumn value_column;
-        key_column.init_from_pb(column.children_columns(0));
-        value_column.init_from_pb(column.children_columns(1));
-        add_sub_column(key_column);
-        add_sub_column(value_column);
+        CHECK(column.children_columns_size() == 2) << "MAP type has more than 2 children types.";
+    }
+    for (size_t i = 0; i < column.children_columns_size(); i++) {
+        TabletColumn child_column;
+        child_column.init_from_pb(column.children_columns(i));
+        add_sub_column(child_column);
     }
 }
 
@@ -463,28 +459,24 @@ void TabletColumn::to_schema_pb(ColumnPB* column) const {
     if (_is_bf_column) {
         column->set_is_bf_column(_is_bf_column);
     }
-    column->set_aggregation(get_string_by_aggregation_type(_aggregation));
+    if (!_aggregation_name.empty()) {
+        column->set_aggregation(_aggregation_name);
+    }
     if (_has_bitmap_index) {
         column->set_has_bitmap_index(_has_bitmap_index);
     }
     column->set_visible(_visible);
 
-    if (_type == FieldType::OLAP_FIELD_TYPE_STRUCT) {
-        for (size_t i = 0; i < _sub_columns.size(); i++) {
-            ColumnPB* child = column->add_children_columns();
-            _sub_columns[i].to_schema_pb(child);
-        }
-    } else if (_type == FieldType::OLAP_FIELD_TYPE_ARRAY) {
-        DCHECK(_sub_columns.size() == 1) << "ARRAY type has more than 1 children types.";
-        ColumnPB* child = column->add_children_columns();
-        _sub_columns[0].to_schema_pb(child);
+    if (_type == FieldType::OLAP_FIELD_TYPE_ARRAY) {
+        CHECK(_sub_columns.size() == 1) << "ARRAY type has more than 1 children types.";
     }
     if (_type == FieldType::OLAP_FIELD_TYPE_MAP) {
-        DCHECK(_sub_columns.size() == 2) << "MAP type has more than 2 children types.";
-        ColumnPB* child_key = column->add_children_columns();
-        _sub_columns[0].to_schema_pb(child_key);
-        ColumnPB* child_val = column->add_children_columns();
-        _sub_columns[1].to_schema_pb(child_val);
+        CHECK(_sub_columns.size() == 2) << "MAP type has more than 2 children types.";
+    }
+
+    for (size_t i = 0; i < _sub_columns.size(); i++) {
+        ColumnPB* child = column->add_children_columns();
+        _sub_columns[i].to_schema_pb(child);
     }
 }
 
@@ -498,14 +490,30 @@ bool TabletColumn::is_row_store_column() const {
     return _col_name == BeConsts::ROW_STORE_COL;
 }
 
-vectorized::AggregateFunctionPtr TabletColumn::get_aggregate_function(
-        vectorized::DataTypes argument_types, std::string suffix) const {
+vectorized::AggregateFunctionPtr TabletColumn::get_aggregate_function_merge() const {
+    vectorized::DataTypes argument_types;
+    for (auto col : _sub_columns) {
+        argument_types.push_back(vectorized::DataTypeFactory::instance().create_data_type(col));
+    }
+    auto function = vectorized::AggregateFunctionSimpleFactory::instance().get(
+            _aggregation_name, argument_types, false);
+    return function;
+}
+
+vectorized::AggregateFunctionPtr TabletColumn::get_aggregate_function(std::string suffix) const {
+    std::string origin_name = TabletColumn::get_string_by_aggregation_type(_aggregation);
+    auto type = vectorized::DataTypeFactory::instance().create_data_type(*this);
+
     std::string agg_name = TabletColumn::get_string_by_aggregation_type(_aggregation) + suffix;
     std::transform(agg_name.begin(), agg_name.end(), agg_name.begin(),
                    [](unsigned char c) { return std::tolower(c); });
 
-    return vectorized::AggregateFunctionSimpleFactory::instance().get(
-            agg_name, argument_types, argument_types.back()->is_nullable());
+    auto function = vectorized::AggregateFunctionSimpleFactory::instance().get(agg_name, {type},
+                                                                               type->is_nullable());
+    if (function) {
+        return function;
+    }
+    return get_aggregate_function_merge();
 }
 
 void TabletIndex::init_from_thrift(const TOlapTableIndex& index,
@@ -550,7 +558,7 @@ void TabletIndex::init_from_thrift(const TOlapTableIndex& index,
                                    const std::vector<int32_t>& column_uids) {
     _index_id = index.index_id;
     _index_name = index.index_name;
-    _col_unique_ids = std::move(column_uids);
+    _col_unique_ids = column_uids;
 
     switch (index.index_type) {
     case TIndexType::BITMAP:
diff --git a/be/src/olap/tablet_schema.h b/be/src/olap/tablet_schema.h
index 0249a87ef2..2ee7122a6b 100644
--- a/be/src/olap/tablet_schema.h
+++ b/be/src/olap/tablet_schema.h
@@ -77,7 +77,8 @@ public:
                _type == FieldType::OLAP_FIELD_TYPE_STRING ||
                _type == FieldType::OLAP_FIELD_TYPE_HLL ||
                _type == FieldType::OLAP_FIELD_TYPE_OBJECT ||
-               _type == FieldType::OLAP_FIELD_TYPE_QUANTILE_STATE;
+               _type == FieldType::OLAP_FIELD_TYPE_QUANTILE_STATE ||
+               _type == FieldType::OLAP_FIELD_TYPE_AGG_STATE;
     }
     bool has_default_value() const { return _has_default_value; }
     std::string default_value() const { return _default_value; }
@@ -88,8 +89,8 @@ public:
     void set_is_nullable(bool is_nullable) { _is_nullable = is_nullable; }
     void set_has_default_value(bool has) { _has_default_value = has; }
     FieldAggregationMethod aggregation() const { return _aggregation; }
-    vectorized::AggregateFunctionPtr get_aggregate_function(vectorized::DataTypes argument_types,
-                                                            std::string suffix) const;
+    vectorized::AggregateFunctionPtr get_aggregate_function_merge() const;
+    vectorized::AggregateFunctionPtr get_aggregate_function(std::string suffix) const;
     int precision() const { return _precision; }
     int frac() const { return _frac; }
     inline bool visible() const { return _visible; }
@@ -120,6 +121,7 @@ private:
     FieldType _type;
     bool _is_key = false;
     FieldAggregationMethod _aggregation;
+    std::string _aggregation_name;
     bool _is_nullable = false;
 
     bool _has_default_value = false;
@@ -299,7 +301,6 @@ private:
     friend bool operator==(const TabletSchema& a, const TabletSchema& b);
     friend bool operator!=(const TabletSchema& a, const TabletSchema& b);
 
-private:
     KeysType _keys_type = DUP_KEYS;
     SortType _sort_type = SortType::LEXICAL;
     size_t _sort_col_num = 0;
diff --git a/be/src/olap/types.cpp b/be/src/olap/types.cpp
index 868924cf46..e179e59b18 100644
--- a/be/src/olap/types.cpp
+++ b/be/src/olap/types.cpp
@@ -51,6 +51,7 @@ bool is_olap_string_type(FieldType field_type) {
     case FieldType::OLAP_FIELD_TYPE_OBJECT:
     case FieldType::OLAP_FIELD_TYPE_STRING:
     case FieldType::OLAP_FIELD_TYPE_JSONB:
+    case FieldType::OLAP_FIELD_TYPE_AGG_STATE:
         return true;
     default:
         return false;
@@ -95,7 +96,8 @@ const TypeInfo* get_scalar_type_info(FieldType field_type) {
             get_scalar_type_info<FieldType::OLAP_FIELD_TYPE_DECIMAL64>(),
             get_scalar_type_info<FieldType::OLAP_FIELD_TYPE_DECIMAL128I>(),
             get_scalar_type_info<FieldType::OLAP_FIELD_TYPE_JSONB>(),
-    };
+            nullptr,
+            get_scalar_type_info<FieldType::OLAP_FIELD_TYPE_AGG_STATE>()};
     return field_type_array[int(field_type)];
 }
 
@@ -166,7 +168,8 @@ const TypeInfo* get_array_type_info(FieldType leaf_type, int32_t iterations) {
             INIT_ARRAY_TYPE_INFO_LIST(FieldType::OLAP_FIELD_TYPE_DECIMAL64),
             INIT_ARRAY_TYPE_INFO_LIST(FieldType::OLAP_FIELD_TYPE_DECIMAL128I),
             INIT_ARRAY_TYPE_INFO_LIST(FieldType::OLAP_FIELD_TYPE_JSONB),
-    };
+            {nullptr, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr, nullptr},
+            INIT_ARRAY_TYPE_INFO_LIST(FieldType::OLAP_FIELD_TYPE_AGG_STATE)};
     return array_type_Info_arr[int(leaf_type)][iterations];
 }
 
diff --git a/be/src/olap/types.h b/be/src/olap/types.h
index a079d96b20..1cdf6e995d 100644
--- a/be/src/olap/types.h
+++ b/be/src/olap/types.h
@@ -739,6 +739,12 @@ template <>
 struct CppTypeTraits<FieldType::OLAP_FIELD_TYPE_QUANTILE_STATE> {
     using CppType = Slice;
 };
+
+template <>
+struct CppTypeTraits<FieldType::OLAP_FIELD_TYPE_AGG_STATE> {
+    using CppType = Slice;
+};
+
 template <>
 struct CppTypeTraits<FieldType::OLAP_FIELD_TYPE_STRUCT> {
     using CppType = StructValue;
@@ -1410,6 +1416,10 @@ template <>
 struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_QUANTILE_STATE>
         : public FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_VARCHAR> {};
 
+template <>
+struct FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_AGG_STATE>
+        : public FieldTypeTraits<FieldType::OLAP_FIELD_TYPE_VARCHAR> {};
+
 // Instantiate this template to get static access to the type traits.
 template <FieldType field_type>
 struct TypeTraits : public FieldTypeTraits<field_type> {
diff --git a/be/src/runtime/primitive_type.h b/be/src/runtime/primitive_type.h
index dfc121c330..2caad79546 100644
--- a/be/src/runtime/primitive_type.h
+++ b/be/src/runtime/primitive_type.h
@@ -108,7 +108,6 @@ bool is_type_compatible(PrimitiveType lhs, PrimitiveType rhs);
 TExprOpcode::type to_in_opcode(PrimitiveType t);
 PrimitiveType thrift_to_type(TPrimitiveType::type ttype);
 TPrimitiveType::type to_thrift(PrimitiveType ptype);
-TColumnType to_tcolumn_type_thrift(TPrimitiveType::type ttype);
 std::string type_to_string(PrimitiveType t);
 std::string type_to_odbc_string(PrimitiveType t);
 TTypeDesc gen_type_desc(const TPrimitiveType::type val);
diff --git a/be/src/vec/data_types/data_type_agg_state.h b/be/src/vec/data_types/data_type_agg_state.h
index 468d10b36f..27d0d2c0e1 100644
--- a/be/src/vec/data_types/data_type_agg_state.h
+++ b/be/src/vec/data_types/data_type_agg_state.h
@@ -21,12 +21,14 @@
 #pragma once
 
 #include <gen_cpp/Types_types.h>
+#include <gen_cpp/data.pb.h>
 #include <stddef.h>
 #include <stdint.h>
 
 #include <memory>
 #include <string>
 
+#include "vec/data_types/data_type.h"
 #include "vec/data_types/data_type_string.h"
 
 namespace doris {
@@ -54,6 +56,13 @@ public:
 
     void add_sub_type(DataTypePtr type) { sub_types.push_back(type); }
 
+    void to_pb_column_meta(PColumnMeta* col_meta) const override {
+        IDataType::to_pb_column_meta(col_meta);
+        for (auto type : sub_types) {
+            type->to_pb_column_meta(col_meta->add_children());
+        }
+    }
+
 private:
     DataTypes sub_types;
 };
diff --git a/be/src/vec/data_types/data_type_factory.cpp b/be/src/vec/data_types/data_type_factory.cpp
index 23d7c37b4b..438e266658 100644
--- a/be/src/vec/data_types/data_type_factory.cpp
+++ b/be/src/vec/data_types/data_type_factory.cpp
@@ -544,6 +544,13 @@ DataTypePtr DataTypeFactory::create_data_type(const PColumnMeta& pcolumn) {
         nested = std::make_shared<DataTypeTime>();
         break;
     }
+    case PGenericType::AGG_STATE: {
+        nested = std::make_shared<DataTypeAggState>();
+        for (auto child : pcolumn.children()) {
+            ((DataTypeAggState*)nested.get())->add_sub_type(create_data_type(child));
+        }
+        break;
+    }
     default: {
         LOG(FATAL) << fmt::format("Unknown data type: {}", pcolumn.type());
         return nullptr;
diff --git a/be/src/vec/exprs/vexpr.cpp b/be/src/vec/exprs/vexpr.cpp
index ea07a3359f..df3705c53c 100644
--- a/be/src/vec/exprs/vexpr.cpp
+++ b/be/src/vec/exprs/vexpr.cpp
@@ -78,16 +78,7 @@ VExpr::VExpr(const doris::TExprNode& node)
     _data_type = DataTypeFactory::instance().create_data_type(_type, is_nullable);
 }
 
-VExpr::VExpr(const VExpr& vexpr)
-        : _node_type(vexpr._node_type),
-          _opcode(vexpr._opcode),
-          _type(vexpr._type),
-          _data_type(vexpr._data_type),
-          _children(vexpr._children),
-          _fn(vexpr._fn),
-          _fn_context_index(vexpr._fn_context_index),
-          _constant_col(vexpr._constant_col),
-          _prepared(vexpr._prepared) {}
+VExpr::VExpr(const VExpr& vexpr) = default;
 
 VExpr::VExpr(const TypeDescriptor& type, bool is_slotref, bool is_nullable)
         : _opcode(TExprOpcode::INVALID_OPCODE),
diff --git a/be/src/vec/functions/function.cpp b/be/src/vec/functions/function.cpp
index 101365c2a8..c0eced8691 100644
--- a/be/src/vec/functions/function.cpp
+++ b/be/src/vec/functions/function.cpp
@@ -109,8 +109,12 @@ NullPresence get_null_presence(const Block& block, const ColumnNumbers& args) {
     for (const auto& arg : args) {
         const auto& elem = block.get_by_position(arg);
 
-        if (!res.has_nullable) res.has_nullable = elem.type->is_nullable();
-        if (!res.has_null_constant) res.has_null_constant = elem.type->only_null();
+        if (!res.has_nullable) {
+            res.has_nullable = elem.type->is_nullable();
+        }
+        if (!res.has_null_constant) {
+            res.has_null_constant = elem.type->only_null();
+        }
     }
 
     return res;
@@ -120,8 +124,12 @@ NullPresence get_null_presence(const Block& block, const ColumnNumbers& args) {
     NullPresence res;
 
     for (const auto& elem : args) {
-        if (!res.has_nullable) res.has_nullable = elem.type->is_nullable();
-        if (!res.has_null_constant) res.has_null_constant = elem.type->only_null();
+        if (!res.has_nullable) {
+            res.has_nullable = elem.type->is_nullable();
+        }
+        if (!res.has_null_constant) {
+            res.has_null_constant = elem.type->only_null();
+        }
     }
 
     return res;
@@ -266,27 +274,14 @@ Status PreparedFunctionImpl::execute_without_low_cardinality_columns(
 Status PreparedFunctionImpl::execute(FunctionContext* context, Block& block,
                                      const ColumnNumbers& args, size_t result,
                                      size_t input_rows_count, bool dry_run) {
-    //    if (use_default_implementation_for_low_cardinality_columns()) {
-    //        auto& res = block.safe_get_by_position(result);
-    //        Block block_without_low_cardinality = block.clone_without_columns();
-    //
-    //        for (auto arg : args)
-    //            block_without_low_cardinality.safe_get_by_position(arg).column =
-    //                    block.safe_get_by_position(arg).column;
-    //
-    //        {
-    //            RETURN_IF_ERROR(execute_without_low_cardinality_columns(
-    //                    context, block_without_low_cardinality, args, result, input_rows_count,
-    //                    dry_run));
-    //            res.column = block_without_low_cardinality.safe_get_by_position(result).column;
-    //        }
-    //    } else
     return execute_without_low_cardinality_columns(context, block, args, result, input_rows_count,
                                                    dry_run);
 }
 
 void FunctionBuilderImpl::check_number_of_arguments(size_t number_of_arguments) const {
-    if (is_variadic()) return;
+    if (is_variadic()) {
+        return;
+    }
 
     size_t expected_number_of_arguments = get_number_of_arguments();
 
@@ -325,8 +320,9 @@ DataTypePtr FunctionBuilderImpl::get_return_type(const ColumnsWithTypeAndName& a
 
         for (ColumnWithTypeAndName& arg : args_without_low_cardinality) {
             bool is_const = arg.column && is_column_const(*arg.column);
-            if (is_const)
+            if (is_const) {
                 arg.column = assert_cast<const ColumnConst&>(*arg.column).remove_low_cardinality();
+            }
         }
 
         auto type_without_low_cardinality =
diff --git a/be/src/vec/olap/block_reader.cpp b/be/src/vec/olap/block_reader.cpp
index 5ae2ae5e3e..c7f4ebbc25 100644
--- a/be/src/vec/olap/block_reader.cpp
+++ b/be/src/vec/olap/block_reader.cpp
@@ -147,11 +147,11 @@ void BlockReader::_init_agg_state(const ReaderParams& read_params) {
 
     auto& tablet_schema = *_tablet_schema;
     for (auto idx : _agg_columns_idx) {
+        auto column = tablet_schema.column(
+                read_params.origin_return_columns->at(_return_columns_loc[idx]));
         AggregateFunctionPtr function =
-                tablet_schema
-                        .column(read_params.origin_return_columns->at(_return_columns_loc[idx]))
-                        .get_aggregate_function({_next_row.block->get_data_type(idx)},
-                                                vectorized::AGG_READER_SUFFIX);
+                column.get_aggregate_function(vectorized::AGG_READER_SUFFIX);
+
         DCHECK(function != nullptr);
         _agg_functions.push_back(function);
         // create aggregate data
diff --git a/be/src/vec/olap/block_reader.h b/be/src/vec/olap/block_reader.h
index db6dacfdc7..bf9e20def4 100644
--- a/be/src/vec/olap/block_reader.h
+++ b/be/src/vec/olap/block_reader.h
@@ -122,6 +122,8 @@ private:
     ColumnPtr _delete_filter_column;
 
     bool _is_rowsets_overlapping = true;
+
+    Arena _arena;
 };
 
 } // namespace vectorized
diff --git a/be/src/vec/olap/olap_data_convertor.cpp b/be/src/vec/olap/olap_data_convertor.cpp
index ab1cd1ed6e..3070975a3c 100644
--- a/be/src/vec/olap/olap_data_convertor.cpp
+++ b/be/src/vec/olap/olap_data_convertor.cpp
@@ -75,6 +75,9 @@ OlapBlockDataConvertor::create_olap_column_data_convertor(const TabletColumn& co
     case FieldType::OLAP_FIELD_TYPE_QUANTILE_STATE: {
         return std::make_unique<OlapColumnDataConvertorQuantileState>();
     }
+    case FieldType::OLAP_FIELD_TYPE_AGG_STATE: {
+        return std::make_unique<OlapColumnDataConvertorVarChar>(false);
+    }
     case FieldType::OLAP_FIELD_TYPE_HLL: {
         return std::make_unique<OlapColumnDataConvertorHLL>();
     }
diff --git a/be/src/vec/olap/vertical_block_reader.cpp b/be/src/vec/olap/vertical_block_reader.cpp
index d9f4fb3256..a01726f50a 100644
--- a/be/src/vec/olap/vertical_block_reader.cpp
+++ b/be/src/vec/olap/vertical_block_reader.cpp
@@ -162,8 +162,7 @@ void VerticalBlockReader::_init_agg_state(const ReaderParams& read_params) {
     for (size_t idx = 0; idx < _return_columns.size(); ++idx) {
         AggregateFunctionPtr function =
                 tablet_schema.column(_return_columns.at(idx))
-                        .get_aggregate_function({_next_row.block->get_data_type(idx)},
-                                                vectorized::AGG_READER_SUFFIX);
+                        .get_aggregate_function(vectorized::AGG_READER_SUFFIX);
         DCHECK(function != nullptr);
         _agg_functions.push_back(function);
         // create aggregate data
diff --git a/be/src/vec/sink/vmysql_result_writer.cpp b/be/src/vec/sink/vmysql_result_writer.cpp
index 086e28aec0..8896c48e86 100644
--- a/be/src/vec/sink/vmysql_result_writer.cpp
+++ b/be/src/vec/sink/vmysql_result_writer.cpp
@@ -34,6 +34,7 @@
 #include "olap/hll.h"
 #include "runtime/buffer_control_block.h"
 #include "runtime/decimalv2_value.h"
+#include "runtime/define_primitive_type.h"
 #include "runtime/large_int_value.h"
 #include "runtime/primitive_type.h"
 #include "runtime/runtime_state.h"
@@ -709,7 +710,8 @@ Status VMysqlResultWriter<is_binary_format>::append_block(Block& input_block) {
         }
         case TYPE_STRING:
         case TYPE_CHAR:
-        case TYPE_VARCHAR: {
+        case TYPE_VARCHAR:
+        case TYPE_AGG_STATE: {
             if (type_ptr->is_nullable()) {
                 status = _add_one_column<PrimitiveType::TYPE_VARCHAR, true>(column_ptr, result,
                                                                             rows_buffer);
diff --git a/be/test/olap/delta_writer_test.cpp b/be/test/olap/delta_writer_test.cpp
index 19efddd4b8..138b9fcbec 100644
--- a/be/test/olap/delta_writer_test.cpp
+++ b/be/test/olap/delta_writer_test.cpp
@@ -330,28 +330,72 @@ static TDescriptorTable create_descriptor_tablet() {
     tuple_builder.add_slot(
             TSlotDescriptorBuilder().type(TYPE_DATEV2).column_name("k11").column_pos(10).build());
 
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().type(TYPE_TINYINT).column_name("v1").column_pos(11).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().type(TYPE_SMALLINT).column_name("v2").column_pos(12).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().type(TYPE_INT).column_name("v3").column_pos(13).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().type(TYPE_BIGINT).column_name("v4").column_pos(14).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().type(TYPE_LARGEINT).column_name("v5").column_pos(15).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().type(TYPE_DATE).column_name("v6").column_pos(16).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().type(TYPE_DATETIME).column_name("v7").column_pos(17).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().string_type(4).column_name("v8").column_pos(18).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().string_type(65).column_name("v9").column_pos(19).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().decimal_type(6, 3).column_name("v10").column_pos(20).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().type(TYPE_DATEV2).column_name("v11").column_pos(21).build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .type(TYPE_TINYINT)
+                                   .column_name("v1")
+                                   .column_pos(11)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .type(TYPE_SMALLINT)
+                                   .column_name("v2")
+                                   .column_pos(12)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .type(TYPE_INT)
+                                   .column_name("v3")
+                                   .column_pos(13)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .type(TYPE_BIGINT)
+                                   .column_name("v4")
+                                   .column_pos(14)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .type(TYPE_LARGEINT)
+                                   .column_name("v5")
+                                   .column_pos(15)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .type(TYPE_DATE)
+                                   .column_name("v6")
+                                   .column_pos(16)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .type(TYPE_DATETIME)
+                                   .column_name("v7")
+                                   .column_pos(17)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .string_type(4)
+                                   .column_name("v8")
+                                   .column_pos(18)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .string_type(65)
+                                   .column_name("v9")
+                                   .column_pos(19)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .decimal_type(6, 3)
+                                   .column_name("v10")
+                                   .column_pos(20)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .type(TYPE_DATEV2)
+                                   .column_name("v11")
+                                   .column_pos(21)
+                                   .nullable(false)
+                                   .build());
     tuple_builder.build(&dtb);
 
     return dtb.desc_tbl();
@@ -365,14 +409,23 @@ static TDescriptorTable create_descriptor_tablet_with_sequence_col() {
             TSlotDescriptorBuilder().type(TYPE_TINYINT).column_name("k1").column_pos(0).build());
     tuple_builder.add_slot(
             TSlotDescriptorBuilder().type(TYPE_SMALLINT).column_name("k2").column_pos(1).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().type(TYPE_DATETIME).column_name("v1").column_pos(2).build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().type(TYPE_DATEV2).column_name("v2").column_pos(3).build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .type(TYPE_DATETIME)
+                                   .column_name("v1")
+                                   .column_pos(2)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .type(TYPE_DATEV2)
+                                   .column_name("v2")
+                                   .column_pos(3)
+                                   .nullable(false)
+                                   .build());
     tuple_builder.add_slot(TSlotDescriptorBuilder()
                                    .type(TYPE_INT)
                                    .column_name(SEQUENCE_COL)
                                    .column_pos(4)
+                                   .nullable(false)
                                    .build());
     tuple_builder.build(&dtb);
 
diff --git a/be/test/olap/storage_types_test.cpp b/be/test/olap/storage_types_test.cpp
index d96c1554fe..ab7b80a18f 100644
--- a/be/test/olap/storage_types_test.cpp
+++ b/be/test/olap/storage_types_test.cpp
@@ -150,13 +150,15 @@ TEST(TypesTest, copy_and_equal) {
 
 template <FieldType item_type>
 void common_test_array(CollectionValue src_val) {
-    TabletColumn list_column(OLAP_FIELD_AGGREGATION_NONE, FieldType::OLAP_FIELD_TYPE_ARRAY);
+    TabletColumn list_column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE,
+                             FieldType::OLAP_FIELD_TYPE_ARRAY);
     int32 item_length = 0;
     if (item_type == FieldType::OLAP_FIELD_TYPE_CHAR ||
         item_type == FieldType::OLAP_FIELD_TYPE_VARCHAR) {
         item_length = 10;
     }
-    TabletColumn item_column(OLAP_FIELD_AGGREGATION_NONE, item_type, true, 0, item_length);
+    TabletColumn item_column(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE, item_type, true,
+                             0, item_length);
     list_column.add_sub_column(item_column);
 
     auto array_type = get_type_info(&list_column);
diff --git a/be/test/olap/tablet_cooldown_test.cpp b/be/test/olap/tablet_cooldown_test.cpp
index 320f741d2c..7472ac8670 100644
--- a/be/test/olap/tablet_cooldown_test.cpp
+++ b/be/test/olap/tablet_cooldown_test.cpp
@@ -319,9 +319,14 @@ static TDescriptorTable create_descriptor_tablet_with_sequence_col() {
                                    .type(TYPE_INT)
                                    .column_name(SEQUENCE_COL)
                                    .column_pos(2)
+                                   .nullable(false)
+                                   .build());
+    tuple_builder.add_slot(TSlotDescriptorBuilder()
+                                   .type(TYPE_DATETIME)
+                                   .column_name("v1")
+                                   .column_pos(3)
+                                   .nullable(false)
                                    .build());
-    tuple_builder.add_slot(
-            TSlotDescriptorBuilder().type(TYPE_DATETIME).column_name("v1").column_pos(3).build());
     tuple_builder.build(&desc_tbl_builder);
 
     return desc_tbl_builder.desc_tbl();
diff --git a/be/test/olap/tablet_schema_helper.h b/be/test/olap/tablet_schema_helper.h
index b93cce162e..547882a18c 100644
--- a/be/test/olap/tablet_schema_helper.h
+++ b/be/test/olap/tablet_schema_helper.h
@@ -33,10 +33,11 @@ class Arena;
 TabletColumn create_int_key(int32_t id, bool is_nullable = true, bool is_bf_column = false,
                             bool has_bitmap_index = false);
 
-TabletColumn create_int_value(int32_t id,
-                              FieldAggregationMethod agg_method = OLAP_FIELD_AGGREGATION_SUM,
-                              bool is_nullable = true, const std::string default_value = "",
-                              bool is_bf_column = false, bool has_bitmap_index = false);
+TabletColumn create_int_value(
+        int32_t id,
+        FieldAggregationMethod agg_method = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_SUM,
+        bool is_nullable = true, const std::string default_value = "", bool is_bf_column = false,
+        bool has_bitmap_index = false);
 
 TabletColumn create_char_key(int32_t id, bool is_nullable = true);
 
@@ -49,7 +50,7 @@ TabletColumn create_with_default_value(std::string default_value) {
     TabletColumn column;
     column._type = type;
     column._is_nullable = true;
-    column._aggregation = OLAP_FIELD_AGGREGATION_NONE;
+    column._aggregation = FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE;
     column._has_default_value = true;
     column._default_value = default_value;
     column._length = 4;
diff --git a/be/test/vec/exec/vgeneric_iterators_test.cpp b/be/test/vec/exec/vgeneric_iterators_test.cpp
index 7f2c8d2574..cdc0f6faa7 100644
--- a/be/test/vec/exec/vgeneric_iterators_test.cpp
+++ b/be/test/vec/exec/vgeneric_iterators_test.cpp
@@ -45,12 +45,14 @@ public:
 
 Schema create_schema() {
     std::vector<TabletColumn> col_schemas;
-    col_schemas.emplace_back(OLAP_FIELD_AGGREGATION_NONE, FieldType::OLAP_FIELD_TYPE_SMALLINT,
-                             true);
+    col_schemas.emplace_back(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE,
+                             FieldType::OLAP_FIELD_TYPE_SMALLINT, true);
     // c2: int
-    col_schemas.emplace_back(OLAP_FIELD_AGGREGATION_NONE, FieldType::OLAP_FIELD_TYPE_INT, true);
+    col_schemas.emplace_back(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_NONE,
+                             FieldType::OLAP_FIELD_TYPE_INT, true);
     // c3: big int
-    col_schemas.emplace_back(OLAP_FIELD_AGGREGATION_SUM, FieldType::OLAP_FIELD_TYPE_BIGINT, true);
+    col_schemas.emplace_back(FieldAggregationMethod::OLAP_FIELD_AGGREGATION_SUM,
+                             FieldType::OLAP_FIELD_TYPE_BIGINT, true);
 
     Schema schema(col_schemas, 2);
     return schema;