diff --git a/fe/src/com/baidu/palo/alter/SchemaChangeHandler.java b/fe/src/com/baidu/palo/alter/SchemaChangeHandler.java
index 84269f22de6c7c..af3ec037dd0094 100644
--- a/fe/src/com/baidu/palo/alter/SchemaChangeHandler.java
+++ b/fe/src/com/baidu/palo/alter/SchemaChangeHandler.java
@@ -515,8 +515,7 @@ private void addColumnInternal(OlapTable olapTable, Column newColumn, ColumnPosi
             }
         }
 
-
-        // hll must be userd in agg_keys
+        // hll must be used in agg_keys
         if (newColumn.getType().isHllType() && KeysType.AGG_KEYS != olapTable.getKeysType()) {
             throw new DdlException("HLL must be used in AGG_KEYS");
         }
@@ -648,6 +647,8 @@ private void checkAndAddColumn(List<Column> modIndexSchema, Column newColumn, Co
                 modIndexSchema.add(newColumn);
             }
         }
+
+        checkRowLength(modIndexSchema);
     }
 
     private void checkKeyModificationIfInRandomDistributedTable(OlapTable olapTable) throws DdlException {
@@ -660,6 +661,18 @@ private void checkKeyModificationIfInRandomDistributedTable(OlapTable olapTable)
         }
     }
 
+    private void checkRowLength(List<Column> modIndexSchema) throws DdlException {
+        int rowLengthBytes = 0;
+        for (Column column : modIndexSchema) {
+            rowLengthBytes += column.getColumnType().getMemlayoutBytes();
+        }
+
+        if (rowLengthBytes > Config.max_layout_length_per_row) {
+            throw new DdlException("The size of a row (" + rowLengthBytes + ") exceed the maximal row size: "
+                    + Config.max_layout_length_per_row);
+        }
+    }
+
     private void createJob(long dbId, OlapTable olapTable, Map<Long, LinkedList<Column>> indexSchemaMap,
                            Map<String, String> propertyMap) throws DdlException {
         if (olapTable.getState() == OlapTableState.ROLLUP) {
diff --git a/fe/src/com/baidu/palo/analysis/CreateTableStmt.java b/fe/src/com/baidu/palo/analysis/CreateTableStmt.java
index d5a15fc15bac7d..0694110947c387 100644
--- a/fe/src/com/baidu/palo/analysis/CreateTableStmt.java
+++ b/fe/src/com/baidu/palo/analysis/CreateTableStmt.java
@@ -27,10 +27,11 @@
 import com.baidu.palo.catalog.KeysType;
 import com.baidu.palo.catalog.PartitionType;
 import com.baidu.palo.common.AnalysisException;
+import com.baidu.palo.common.Config;
 import com.baidu.palo.common.ErrorCode;
 import com.baidu.palo.common.ErrorReport;
-import com.baidu.palo.common.FeNameFormat;
 import com.baidu.palo.common.FeMetaVersion;
+import com.baidu.palo.common.FeNameFormat;
 import com.baidu.palo.common.InternalException;
 import com.baidu.palo.common.io.Text;
 import com.baidu.palo.common.io.Writable;
@@ -250,6 +251,7 @@ public void analyze(Analyzer analyzer) throws AnalysisException, InternalExcepti
             throw new AnalysisException("Kudu table does not support column num more than 300");
         }
 
+        int rowLengthBytes = 0;
         boolean hasHll = false;
         Set<String> columnSet = Sets.newTreeSet(String.CASE_INSENSITIVE_ORDER);
         for (Column col : columns) {
@@ -271,6 +273,13 @@ public void analyze(Analyzer analyzer) throws AnalysisException, InternalExcepti
             if (!columnSet.add(col.getName())) {
                 ErrorReport.reportAnalysisException(ErrorCode.ERR_DUP_FIELDNAME, col.getName());
             }
+
+            rowLengthBytes += col.getColumnType().getMemlayoutBytes();
+        }
+
+        if (rowLengthBytes > Config.max_layout_length_per_row) {
+            throw new AnalysisException("The size of a row (" + rowLengthBytes + ") exceed the maximal row size: "
+                    + Config.max_layout_length_per_row);
         }
 
         if (hasHll && keysDesc.getKeysType() != KeysType.AGG_KEYS) {
diff --git a/fe/src/com/baidu/palo/catalog/ColumnType.java b/fe/src/com/baidu/palo/catalog/ColumnType.java
index 4b330c68d603fb..17ceb751d797f2 100755
--- a/fe/src/com/baidu/palo/catalog/ColumnType.java
+++ b/fe/src/com/baidu/palo/catalog/ColumnType.java
@@ -20,16 +20,16 @@
 
 package com.baidu.palo.catalog;
 
-import java.io.DataInput;
-import java.io.DataOutput;
+import com.baidu.palo.common.AnalysisException;
+import com.baidu.palo.common.FeMetaVersion;
+import com.baidu.palo.common.io.Text;
+import com.baidu.palo.common.io.Writable;
+import com.baidu.palo.thrift.TColumnType;
+
+import java.io.DataInput;
+import java.io.DataOutput;
 import java.io.IOException;
 
-import com.baidu.palo.common.AnalysisException;
-import com.baidu.palo.common.FeMetaVersion;
-import com.baidu.palo.common.io.Text;
-import com.baidu.palo.common.io.Writable;
-import com.baidu.palo.thrift.TColumnType;
-
 /**
  * 这个是对Column类型的一个封装，对于大多数类型，primitive type足够了，这里有两个例外需要用到这个信息
  * 1. 对于decimal，character这种有一些附加信息的
@@ -197,6 +197,40 @@ public void setScale(int scale) {
 
     public boolean isString() {
         return type == PrimitiveType.CHAR || type == PrimitiveType.VARCHAR || type == PrimitiveType.HLL;
+    }
+
+    public int getMemlayoutBytes() {
+        switch (type) {
+            case BOOLEAN:
+                return 0;
+            case TINYINT:
+                return 1;
+            case SMALLINT:
+                return 2;
+            case INT:
+                return 4;
+            case BIGINT:
+                return 8;
+            case LARGEINT:
+                return 16;
+            case FLOAT:
+                return 4;
+            case DOUBLE:
+                return 12;
+            case DATE:
+                return 3;
+            case DATETIME:
+                return 8;
+            case DECIMAL:
+                return 40;
+            case CHAR:
+            case VARCHAR:
+                return len;
+            case HLL:
+                return 16385;
+            default:
+                return 0;
+        }
     }
 
     public void analyze() throws AnalysisException {
diff --git a/fe/src/com/baidu/palo/common/Config.java b/fe/src/com/baidu/palo/common/Config.java
index 6f6da33ee59b38..8c0e0407cd29eb 100644
--- a/fe/src/com/baidu/palo/common/Config.java
+++ b/fe/src/com/baidu/palo/common/Config.java
@@ -188,6 +188,23 @@ public class Config extends ConfigBase {
      *      the create table request will run at most (m * n * tablet_create_timeout_second) before timeout.
      */
     @ConfField public static int tablet_create_timeout_second = 1;
+    
+    /*
+     * Maximal memory layout length of a row. default is 100 KB.
+     * In BE, the maximal size of a RowBlock is 100MB(Configure as max_unpacked_row_block_size in be.conf).
+     * And each RowBlock contains 1024 rows. So the maximal size of a row is approximately 100 KB.
+     * 
+     * eg.
+     *      schema: k1(int), v1(decimal), v2(varchar(2000))
+     *      then the memory layout length of a row is: 8(int) + 40(decimal) + 2000(varchar) = 2048 (Bytes)
+     *      
+     * See memory layout length of all types, run 'help create table' in mysql-client.
+     * 
+     * If you want to increase this number to support more columns in a row, you also need to increase the 
+     * max_unpacked_row_block_size in be.conf. But the performance impact is unknown.
+     */
+    @ConfField
+    public static int max_layout_length_per_row = 100000; // 100k
 
     /*
      * Load checker's running interval.