diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java
index 532a49b0ebf96a..4b00ca94589e49 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Column.java
@@ -57,6 +57,7 @@ public class Column implements Writable {
     public static final String DELETE_SIGN = "__DORIS_DELETE_SIGN__";
     public static final String SEQUENCE_COL = "__DORIS_SEQUENCE_COL__";
     private static final String COLUMN_ARRAY_CHILDREN = "item";
+    private static final String COLUMN_STRUCT_CHILDREN = "field";
     public static final int COLUMN_UNIQUE_ID_INIT_VALUE = -1;
 
     @SerializedName(value = "name")
@@ -185,6 +186,13 @@ public void createChildrenColumn(Type type, Column column) {
             Column c = new Column(COLUMN_ARRAY_CHILDREN, ((ArrayType) type).getItemType());
             c.setIsAllowNull(((ArrayType) type).getContainsNull());
             column.addChildrenColumn(c);
+        } else if (type.isStructType()) {
+            ArrayList<StructField> fileds = ((StructType) type).getFields();
+            for (StructField field : fileds) {
+                Column c = new Column(COLUMN_STRUCT_CHILDREN, field.getType());
+                // c.setIsAllowNull(field.getContainsNull());
+                column.addChildrenColumn(c);
+            }
         }
     }
 
@@ -395,34 +403,41 @@ public TColumn toThrift() {
         return tColumn;
     }
 
+    private void setChildrenTColumn(Column children, TColumn tColumn) {
+        TColumn childrenTColumn = new TColumn();
+        childrenTColumn.setColumnName(children.name);
+
+        TColumnType childrenTColumnType = new TColumnType();
+        childrenTColumnType.setType(children.getDataType().toThrift());
+        childrenTColumnType.setLen(children.getStrLen());
+        childrenTColumnType.setPrecision(children.getPrecision());
+        childrenTColumnType.setScale(children.getScale());
+        childrenTColumnType.setIndexLen(children.getOlapColumnIndexSize());
+
+        childrenTColumn.setColumnType(childrenTColumnType);
+        childrenTColumn.setIsAllowNull(children.isAllowNull());
+        // TODO: If we don't set the aggregate type for children, the type will be
+        //  considered as TAggregationType::SUM after deserializing in BE.
+        //  For now, we make children inherit the aggregate type from their parent.
+        if (tColumn.getAggregationType() != null) {
+            childrenTColumn.setAggregationType(tColumn.getAggregationType());
+        }
+
+        tColumn.children_column.add(childrenTColumn);
+        toChildrenThrift(children, childrenTColumn);
+    }
+
     private void toChildrenThrift(Column column, TColumn tColumn) {
         if (column.type.isArrayType()) {
             Column children = column.getChildren().get(0);
-
-            TColumn childrenTColumn = new TColumn();
-            childrenTColumn.setColumnName(children.name);
-
-            TColumnType childrenTColumnType = new TColumnType();
-            childrenTColumnType.setType(children.getDataType().toThrift());
-            childrenTColumnType.setType(children.getDataType().toThrift());
-            childrenTColumnType.setLen(children.getStrLen());
-            childrenTColumnType.setPrecision(children.getPrecision());
-            childrenTColumnType.setScale(children.getScale());
-
-            childrenTColumnType.setIndexLen(children.getOlapColumnIndexSize());
-            childrenTColumn.setColumnType(childrenTColumnType);
-            childrenTColumn.setIsAllowNull(children.isAllowNull());
-            // TODO: If we don't set the aggregate type for children, the type will be
-            //  considered as TAggregationType::SUM after deserializing in BE.
-            //  For now, we make children inherit the aggregate type from their parent.
-            if (tColumn.getAggregationType() != null) {
-                childrenTColumn.setAggregationType(tColumn.getAggregationType());
-            }
-
             tColumn.setChildrenColumn(new ArrayList<>());
-            tColumn.children_column.add(childrenTColumn);
-
-            toChildrenThrift(children, childrenTColumn);
+            setChildrenTColumn(children, tColumn);
+        } else if (column.type.isStructType()) {
+            List<Column> childrenColumns = column.getChildren();
+            tColumn.setChildrenColumn(new ArrayList<>());
+            for (Column children : childrenColumns) {
+                setChildrenTColumn(children, tColumn);
+            }
         }
     }
 
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/StructType.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/StructType.java
index e3ffa3a521892a..d5c74b9958554a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/StructType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/StructType.java
@@ -93,6 +93,11 @@ public void clearFields() {
         fieldMap.clear();
     }
 
+    @Override
+    public PrimitiveType getPrimitiveType() {
+        return PrimitiveType.STRUCT;
+    }
+
     @Override
     public boolean equals(Object other) {
         if (!(other instanceof StructType)) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Type.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Type.java
index 74cab468088cd6..1aa81c7f222547 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Type.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Type.java
@@ -367,7 +367,7 @@ public boolean isComplexType() {
     }
 
     public boolean isCollectionType() {
-        return isMapType() || isArrayType() || isMultiRowType();
+        return isMapType() || isArrayType() || isMultiRowType() || isStructType();
     }
 
     public boolean isMapType() {
@@ -492,6 +492,8 @@ public static boolean canCastTo(Type sourceType, Type targetType) {
                 && !sourceType.isNull()) {
             // TODO: current not support cast any non-array type(except for null) to nested array type.
             return false;
+        } else if (targetType.isStructType() && sourceType.isStringType()) {
+            return true;
         }
         return sourceType.isNull() || sourceType.getPrimitiveType().isCharFamily();
     }