diff --git a/dbcon/ddlpackage/ddl.y b/dbcon/ddlpackage/ddl.y
index 5c487a1ac..2d47e70c0 100644
--- a/dbcon/ddlpackage/ddl.y
+++ b/dbcon/ddlpackage/ddl.y
@@ -56,24 +56,42 @@ int ddllex(YYSTYPE* ddllval, void* yyscanner);
 void ddlerror(struct pass_to_bison* x, char const *s);
 char* copy_string(const char *str);
 
-void fix_column_length(SchemaObject* elem, const CHARSET_INFO* def_cs) {
+void fix_column_length_and_charset(SchemaObject* elem, const CHARSET_INFO* def_cs)
+{
     auto* column = dynamic_cast<ColumnDef*>(elem);
+
     if (column == NULL || column->fType == NULL)
     {
         return;
     }
 
-    if (column->fType->fType == DDL_VARCHAR ||
-         column->fType->fType == DDL_CHAR ||
-         (column->fType->fType == DDL_TEXT && column->fType->fExplicitLength))
+    if (column->fType->fType == DDL_BLOB ||
+        column->fType->fType == DDL_VARBINARY)
     {
-        unsigned mul = def_cs ? def_cs->mbmaxlen : 1;
-        if (column->fType->fCharset) {
-            const CHARSET_INFO* cs = get_charset_by_csname(column->fType->fCharset, MY_CS_PRIMARY, MYF(0));
-            if (cs)
-                mul = cs->mbmaxlen;
-        }
-        column->fType->fLength *= mul;
+        CHARSET_INFO* cs = &my_charset_bin;
+        column->fType->fCharset = cs->cs_name.str;
+        column->fType->fCollate = cs->coll_name.str;
+        column->fType->fCharsetNum = cs->number;
+        return;
+    }
+
+    if (column->fType->fType == DDL_VARCHAR ||
+        column->fType->fType == DDL_CHAR ||
+        column->fType->fType == DDL_TEXT)
+    {
+        CHARSET_INFO* cs = def_cs ? def_cs : &my_charset_latin1;
+
+        if (column->fType->fCollate)
+            cs = get_charset_by_name(column->fType->fCollate, MYF(0));
+        else if (column->fType->fCharset)
+            cs = get_charset_by_csname(column->fType->fCharset, MY_CS_PRIMARY, MYF(0));
+
+        column->fType->fCharset = cs->cs_name.str;
+        column->fType->fCollate = cs->coll_name.str;
+        column->fType->fCharsetNum = cs->number;
+
+        if ((column->fType->fType != DDL_TEXT) || column->fType->fExplicitLength)
+            column->fType->fLength *= cs->mbmaxlen;
     }
 
     if (column->fType->fType == DDL_TEXT && column->fType->fExplicitLength)
@@ -236,6 +254,7 @@ ZEROFILL
 %type <str>                  ident
 %type <str>                  opt_quoted_literal
 %type <str>                  opt_column_charset
+%type <str>                  opt_column_collate
 %%
 stmtblock:	stmtmulti { x->fParseTree = $1; }
 		;
@@ -319,7 +338,7 @@ create_table_statement:
 	{
         for (auto* elem : *$6)
         {
-            fix_column_length(elem, x->default_table_charset);
+            fix_column_length_and_charset(elem, x->default_table_charset);
         }
 		$$ = new CreateTableStatement(new TableDef($4, $6, $8));
 	}
@@ -500,9 +519,20 @@ table_options:
 	;
 
 opt_equal:
-	{} | '=' {}
+	  /* empty */ {}
+        | '=' {}
 	;
 
+opt_default:
+          /* empty */ {}
+        | DEFAULT {}
+        ;
+
+charset:
+          IDB_CHAR SET {}
+        | CHARSET {}
+        ;
+
 table_option:
  	ENGINE opt_equal ident {$$ = new pair<string,string>("engine", $3);}
 	|
@@ -515,19 +545,13 @@ table_option:
 	COMMENT string_literal {$$ = new pair<string,string>("comment", $2);}
  	|
 	AUTO_INCREMENT opt_equal ICONST
-    {
-       $$ = new pair<string,string>("auto_increment", $3);
-    }
+        {
+          $$ = new pair<string,string>("auto_increment", $3);
+        }
  	|
- 	DEFAULT CHARSET opt_equal ident {$$ = new pair<string,string>("default charset", $4);}
-    |
-    CHARSET opt_equal ident {$$ = new pair<string, string>("default charset", $3);}
+        opt_default charset opt_equal opt_quoted_literal {$$ = new pair<string,string>("default charset", $4);}
  	|
- 	DEFAULT IDB_CHAR SET opt_equal ident {$$ = new pair<string,string>("default charset", $5);}
-    |
-    DEFAULT COLLATE opt_equal opt_quoted_literal {$$ = new pair<string, string>("default collate", $4);}
-    |
-    COLLATE opt_equal opt_quoted_literal {$$ = new pair<string, string>("default collate", $3);}
+        opt_default COLLATE opt_equal opt_quoted_literal {$$ = new pair<string, string>("default collate", $4);}
 	;
 
 alter_table_statement:
@@ -678,17 +702,17 @@ ata_add_column:
     /* See the documentation for SchemaObject for an explanation of why we are using
      * dynamic_cast here.
      */
-	ADD column_def { fix_column_length($2, x->default_table_charset); $$ = new AtaAddColumn(dynamic_cast<ColumnDef*>($2));}
-	| ADD COLUMN column_def { fix_column_length($3, x->default_table_charset); $$ = new AtaAddColumn(dynamic_cast<ColumnDef*>($3));}
+	ADD column_def { fix_column_length_and_charset($2, x->default_table_charset); $$ = new AtaAddColumn(dynamic_cast<ColumnDef*>($2));}
+	| ADD COLUMN column_def { fix_column_length_and_charset($3, x->default_table_charset); $$ = new AtaAddColumn(dynamic_cast<ColumnDef*>($3));}
 	| ADD '(' table_element_list ')' {
         for (auto* elem : *$3) {
-            fix_column_length(elem, x->default_table_charset);
+            fix_column_length_and_charset(elem, x->default_table_charset);
         }
         $$ = new AtaAddColumns($3);
     }
 	| ADD COLUMN '(' table_element_list ')' {
         for (auto* elem : *$4) {
-            fix_column_length(elem, x->default_table_charset);
+            fix_column_length_and_charset(elem, x->default_table_charset);
         }
         $$ = new AtaAddColumns($4);
     }
@@ -780,18 +804,19 @@ optional_braces:
 opt_column_charset:
     /* empty */ { $$ = NULL; }
     |
-    IDB_CHAR SET opt_quoted_literal { $$ = $3; }
+    charset opt_quoted_literal { $$ = $2; }
     ;
 
 opt_column_collate:
-    /* empty */ {}
+    /* empty */ { $$ = NULL; }
     |
-    COLLATE opt_quoted_literal {}
+    COLLATE opt_quoted_literal { $$ = $2; }
     ;
 
 data_type:
 	character_string_type opt_column_charset opt_column_collate {
         $1->fCharset = $2;
+        $1->fCollate = $3;
         $$ = $1;
     }
 	| binary_string_type
@@ -800,6 +825,7 @@ data_type:
 	| blob_type
 	| text_type opt_column_charset opt_column_collate {
         $1->fCharset = $2;
+        $1->fCollate = $3;
         $$ = $1;
     }
 	| IDB_BLOB
diff --git a/dbcon/ddlpackage/ddlpkg.cpp b/dbcon/ddlpackage/ddlpkg.cpp
index 49b8abbc4..0cc258e29 100644
--- a/dbcon/ddlpackage/ddlpkg.cpp
+++ b/dbcon/ddlpackage/ddlpkg.cpp
@@ -64,13 +64,16 @@ ColumnType::ColumnType(int prec, int scale)
  , fScale(scale)
  , fWithTimezone(false)
  , fCharset(NULL)
+ , fCollate(NULL)
+ , fCharsetNum(0)
  , fExplicitLength(false)
 {
   fLength = utils::widthByPrecision(fPrecision);
 }
 
 ColumnType::ColumnType(int type)
- : fType(type), fLength(0), fScale(0), fWithTimezone(false), fCharset(NULL), fExplicitLength(false)
+ : fType(type), fLength(0), fScale(0), fWithTimezone(false),
+   fCharset(NULL), fCollate(NULL), fCharsetNum(0), fExplicitLength(false)
 {
   switch (type)
   {
diff --git a/dbcon/ddlpackage/ddlpkg.h b/dbcon/ddlpackage/ddlpkg.h
index bada82db3..59238a899 100644
--- a/dbcon/ddlpackage/ddlpkg.h
+++ b/dbcon/ddlpackage/ddlpkg.h
@@ -934,7 +934,7 @@ struct ColumnType
   EXPORT int serialize(messageqcpp::ByteStream& bs);
 
   /** @brief For deserialization. */
-  ColumnType() : fCharset(NULL), fExplicitLength(false)
+  ColumnType() : fCharset(NULL), fCollate(NULL), fCharsetNum(0), fExplicitLength(false)
   {
   }
 
@@ -978,6 +978,10 @@ struct ColumnType
 
   /** @brief Column charset (CHAR, VARCHAR and TEXT only) */
   const char* fCharset;
+  /** @brief Column collation (CHAR, VARCHAR and TEXT only) */
+  const char* fCollate;
+  /** @brief Column charset number (CHAR, VARCHAR and TEXT only) */
+  uint32_t fCharsetNum;
 
   /** @brief Is the TEXT column has explicit defined length, ie TEXT(1717) */
   bool fExplicitLength;
diff --git a/dbcon/ddlpackage/serialize.cpp b/dbcon/ddlpackage/serialize.cpp
index 796b57cd0..887eaae2a 100644
--- a/dbcon/ddlpackage/serialize.cpp
+++ b/dbcon/ddlpackage/serialize.cpp
@@ -1109,6 +1109,7 @@ int ColumnType::unserialize(ByteStream& bytestream)
   messageqcpp::ByteStream::quadbyte compressiontype;
   std::string autoincrement;
   messageqcpp::ByteStream::octbyte nextVal;
+  messageqcpp::ByteStream::quadbyte charsetNum;
 
   // read column types
   bytestream >> ftype;
@@ -1119,6 +1120,7 @@ int ColumnType::unserialize(ByteStream& bytestream)
   bytestream >> compressiontype;
   bytestream >> autoincrement;
   bytestream >> nextVal;
+  bytestream >> charsetNum;
 
   fType = ftype;
   fLength = length;
@@ -1128,6 +1130,7 @@ int ColumnType::unserialize(ByteStream& bytestream)
   fCompressiontype = compressiontype;
   fAutoincrement = autoincrement;
   fNextvalue = nextVal;
+  fCharsetNum = charsetNum;
 
   //	cout << "BS length = " << bytestream.length() << endl;
 
@@ -1147,6 +1150,7 @@ int ColumnType::serialize(ByteStream& bytestream)
   messageqcpp::ByteStream::quadbyte compressiontype = fCompressiontype;
   std::string autoincrement = fAutoincrement;
   messageqcpp::ByteStream::octbyte nextVal = fNextvalue;
+  messageqcpp::ByteStream::quadbyte charsetNum = fCharsetNum;
 
   // write column types
   bytestream << ftype;
@@ -1157,6 +1161,7 @@ int ColumnType::serialize(ByteStream& bytestream)
   bytestream << compressiontype;
   bytestream << autoincrement;
   bytestream << nextVal;
+  bytestream << charsetNum;
 
   //	cout << "BS length = " << bytestream.length() << endl;
 
diff --git a/dbcon/execplan/calpontsystemcatalog.cpp b/dbcon/execplan/calpontsystemcatalog.cpp
index 007024efd..1b9bad261 100644
--- a/dbcon/execplan/calpontsystemcatalog.cpp
+++ b/dbcon/execplan/calpontsystemcatalog.cpp
@@ -5912,317 +5912,132 @@ void CalpontSystemCatalog::updateColinfoCache(CalpontSystemCatalog::OIDNextvalMa
 }
 void CalpontSystemCatalog::buildSysColinfomap()
 {
-  ColType aCol;
-  // aCol.defaultValue = "";
-  aCol.scale = 0;
-  aCol.precision = 10;
-  aCol.compressionType = 0;
+  int32_t scale = 0, precision = 10, compressionType = 0, colPosition = 0;
 
   ResourceManager* rm = ResourceManager::instance();
 
   if (rm->useHdfs())
-    aCol.compressionType = 2;
+    compressionType = 2;
 
   DictOID notDict;
 
   // @bug 4433 - Increase object width from 64 to 128 for schema names, table names, and column names.
-  aCol.colWidth = 129;  // @bug 4433
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = VARCHAR;
-  aCol.ddn.dictOID = DICTOID_SYSTABLE_TABLENAME;
-  aCol.ddn.listOID = LISTOID_SYSTABLE_TABLENAME;
-  aCol.ddn.treeOID = TREEOID_SYSTABLE_TABLENAME;
-  aCol.ddn.compressionType = aCol.compressionType;
-  aCol.colPosition = 0;
-  aCol.columnOID = OID_SYSTABLE_TABLENAME;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_TABLENAME] = ColType(129, scale, precision, NOTNULL_CONSTRAINT,
+    DictOID(DICTOID_SYSTABLE_TABLENAME, LISTOID_SYSTABLE_TABLENAME, TREEOID_SYSTABLE_TABLENAME, compressionType),
+    colPosition++, compressionType, OID_SYSTABLE_TABLENAME, VARCHAR);
 
-  aCol.colWidth = 129;  // @bug 4433
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = VARCHAR;
-  aCol.ddn.dictOID = DICTOID_SYSTABLE_SCHEMA;
-  aCol.ddn.listOID = LISTOID_SYSTABLE_SCHEMA;
-  aCol.ddn.treeOID = TREEOID_SYSTABLE_SCHEMA;
-  aCol.ddn.compressionType = aCol.compressionType;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSTABLE_SCHEMA;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_SCHEMA] = ColType(129, scale, precision, NOTNULL_CONSTRAINT,
+    DictOID(DICTOID_SYSTABLE_SCHEMA, LISTOID_SYSTABLE_SCHEMA, TREEOID_SYSTABLE_SCHEMA, compressionType),
+    colPosition++, compressionType, OID_SYSTABLE_SCHEMA, VARCHAR);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSTABLE_OBJECTID;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_OBJECTID] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSTABLE_OBJECTID, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = DATE;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSTABLE_CREATEDATE;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_CREATEDATE] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSTABLE_CREATEDATE, DATE);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = DATE;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSTABLE_LASTUPDATE;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_LASTUPDATE] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSTABLE_LASTUPDATE, DATE);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSTABLE_INIT;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_INIT] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSTABLE_INIT, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSTABLE_NEXT;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_NEXT] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSTABLE_NEXT, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSTABLE_NUMOFROWS;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_NUMOFROWS] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSTABLE_NUMOFROWS, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSTABLE_AVGROWLEN;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_AVGROWLEN] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSTABLE_AVGROWLEN, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSTABLE_NUMOFBLOCKS;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_NUMOFBLOCKS] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSTABLE_NUMOFBLOCKS, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSTABLE_AUTOINCREMENT;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_AUTOINCREMENT] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSTABLE_AUTOINCREMENT, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSTABLE_AUXCOLUMNOID;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSTABLE_AUXCOLUMNOID] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSTABLE_AUXCOLUMNOID, INT);
 
   fTablemap[make_table(CALPONT_SCHEMA, SYSCOLUMN_TABLE)] = SYSCOLUMN_BASE;
 
-  aCol.colWidth = 129;  // @bug 4433
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = VARCHAR;
-  aCol.ddn.dictOID = DICTOID_SYSCOLUMN_SCHEMA;
-  aCol.ddn.listOID = LISTOID_SYSCOLUMN_SCHEMA;
-  aCol.ddn.treeOID = TREEOID_SYSCOLUMN_SCHEMA;
-  aCol.ddn.compressionType = aCol.compressionType;
-  aCol.colPosition = 0;
-  aCol.columnOID = OID_SYSCOLUMN_SCHEMA;
-  fColinfomap[aCol.columnOID] = aCol;
+  colPosition = 0;
 
-  aCol.colWidth = 129;  // @bug 4433
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = VARCHAR;
-  aCol.ddn.dictOID = DICTOID_SYSCOLUMN_TABLENAME;
-  aCol.ddn.listOID = LISTOID_SYSCOLUMN_TABLENAME;
-  aCol.ddn.treeOID = TREEOID_SYSCOLUMN_TABLENAME;
-  aCol.ddn.compressionType = aCol.compressionType;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_TABLENAME;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_SCHEMA] = ColType(129, scale, precision, NOTNULL_CONSTRAINT,
+    DictOID(DICTOID_SYSCOLUMN_SCHEMA, LISTOID_SYSCOLUMN_SCHEMA, TREEOID_SYSCOLUMN_SCHEMA, compressionType),
+    colPosition++, compressionType, OID_SYSCOLUMN_SCHEMA, VARCHAR);
 
-  aCol.colWidth = 129;  // @bug 4433
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = VARCHAR;
-  aCol.ddn.dictOID = DICTOID_SYSCOLUMN_COLNAME;
-  aCol.ddn.listOID = LISTOID_SYSCOLUMN_COLNAME;
-  aCol.ddn.treeOID = TREEOID_SYSCOLUMN_COLNAME;
-  aCol.ddn.compressionType = aCol.compressionType;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_COLNAME;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_TABLENAME] = ColType(129, scale, precision, NOTNULL_CONSTRAINT,
+    DictOID(DICTOID_SYSCOLUMN_TABLENAME, LISTOID_SYSCOLUMN_TABLENAME, TREEOID_SYSCOLUMN_TABLENAME, compressionType),
+    colPosition++, compressionType, OID_SYSCOLUMN_TABLENAME, VARCHAR);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_OBJECTID;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_COLNAME] = ColType(129, scale, precision, NOTNULL_CONSTRAINT,
+    DictOID(DICTOID_SYSCOLUMN_COLNAME, LISTOID_SYSCOLUMN_COLNAME, TREEOID_SYSCOLUMN_COLNAME, compressionType),
+    colPosition++, compressionType, OID_SYSCOLUMN_COLNAME, VARCHAR);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_DICTOID;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_OBJECTID] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_OBJECTID, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_LISTOBJID;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_DICTOID] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_DICTOID, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_TREEOBJID;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_LISTOBJID] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_LISTOBJID, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_DATATYPE;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_TREEOBJID] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_TREEOBJID, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_COLUMNLEN;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_DATATYPE] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_DATATYPE, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_COLUMNPOS;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_COLUMNLEN] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_COLUMNLEN, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = DATE;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_LASTUPDATE;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_COLUMNPOS] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_COLUMNPOS, INT);
 
-  aCol.colWidth = 64;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = VARCHAR;
-  aCol.ddn.dictOID = DICTOID_SYSCOLUMN_DEFAULTVAL;
-  aCol.ddn.listOID = LISTOID_SYSCOLUMN_DEFAULTVAL;
-  aCol.ddn.treeOID = TREEOID_SYSCOLUMN_DEFAULTVAL;
-  aCol.ddn.compressionType = aCol.compressionType;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_DEFAULTVAL;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_LASTUPDATE] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_LASTUPDATE, DATE);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_NULLABLE;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_DEFAULTVAL] = ColType(64, scale, precision, NO_CONSTRAINT,
+    DictOID(DICTOID_SYSCOLUMN_DEFAULTVAL, LISTOID_SYSCOLUMN_DEFAULTVAL, TREEOID_SYSCOLUMN_DEFAULTVAL, compressionType),
+    colPosition++, compressionType, OID_SYSCOLUMN_DEFAULTVAL, VARCHAR);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_SCALE;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_NULLABLE] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_NULLABLE, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_PRECISION;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_SCALE] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_SCALE, INT);
 
-  aCol.colWidth = 1;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = CHAR;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_AUTOINC;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_PRECISION] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_PRECISION, INT);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_DISTCOUNT;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_AUTOINC] = ColType(1, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_AUTOINC, CHAR);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_NULLCOUNT;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_DISTCOUNT] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_DISTCOUNT, INT);
 
-  aCol.colWidth = 65;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = VARCHAR;
-  aCol.ddn.dictOID = DICTOID_SYSCOLUMN_MINVALUE;
-  aCol.ddn.listOID = LISTOID_SYSCOLUMN_MINVALUE;
-  aCol.ddn.treeOID = TREEOID_SYSCOLUMN_MINVALUE;
-  aCol.ddn.compressionType = aCol.compressionType;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_MINVALUE;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_NULLCOUNT] = ColType(4, scale, precision, NO_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_NULLCOUNT, INT);
 
-  aCol.colWidth = 65;
-  aCol.constraintType = NO_CONSTRAINT;
-  aCol.colDataType = VARCHAR;
-  aCol.ddn.dictOID = DICTOID_SYSCOLUMN_MAXVALUE;
-  aCol.ddn.listOID = LISTOID_SYSCOLUMN_MAXVALUE;
-  aCol.ddn.treeOID = TREEOID_SYSCOLUMN_MAXVALUE;
-  aCol.ddn.compressionType = aCol.compressionType;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_MAXVALUE;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_MINVALUE] = ColType(65, scale, precision, NO_CONSTRAINT,
+    DictOID(DICTOID_SYSCOLUMN_MINVALUE, LISTOID_SYSCOLUMN_MINVALUE, TREEOID_SYSCOLUMN_MINVALUE, compressionType),
+    colPosition++, compressionType, OID_SYSCOLUMN_MINVALUE, VARCHAR);
 
-  aCol.colWidth = 4;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = INT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_COMPRESSIONTYPE;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_MAXVALUE] = ColType(65, scale, precision, NO_CONSTRAINT,
+    DictOID(DICTOID_SYSCOLUMN_MAXVALUE, LISTOID_SYSCOLUMN_MAXVALUE, TREEOID_SYSCOLUMN_MAXVALUE, compressionType),
+    colPosition++, compressionType, OID_SYSCOLUMN_MAXVALUE, VARCHAR);
 
-  aCol.colWidth = 8;
-  aCol.constraintType = NOTNULL_CONSTRAINT;
-  aCol.colDataType = UBIGINT;
-  aCol.ddn = notDict;
-  aCol.colPosition++;
-  aCol.columnOID = OID_SYSCOLUMN_NEXTVALUE;
-  fColinfomap[aCol.columnOID] = aCol;
+  fColinfomap[OID_SYSCOLUMN_COMPRESSIONTYPE] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_COMPRESSIONTYPE, INT);
+
+  fColinfomap[OID_SYSCOLUMN_NEXTVALUE] = ColType(8, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_NEXTVALUE, UBIGINT);
+
+  fColinfomap[OID_SYSCOLUMN_CHARSETNUM] = ColType(4, scale, precision, NOTNULL_CONSTRAINT,
+    notDict, colPosition++, compressionType, OID_SYSCOLUMN_CHARSETNUM, INT);
 }
 
 void CalpontSystemCatalog::buildSysOIDmap()
@@ -6261,6 +6076,7 @@ void CalpontSystemCatalog::buildSysOIDmap()
   fOIDmap[make_tcn(CALPONT_SCHEMA, SYSCOLUMN_TABLE, MAXVALUE_COL)] = OID_SYSCOLUMN_MAXVALUE;
   fOIDmap[make_tcn(CALPONT_SCHEMA, SYSCOLUMN_TABLE, COMPRESSIONTYPE_COL)] = OID_SYSCOLUMN_COMPRESSIONTYPE;
   fOIDmap[make_tcn(CALPONT_SCHEMA, SYSCOLUMN_TABLE, NEXTVALUE_COL)] = OID_SYSCOLUMN_NEXTVALUE;
+  fOIDmap[make_tcn(CALPONT_SCHEMA, SYSCOLUMN_TABLE, CHARSETNUM_COL)] = OID_SYSCOLUMN_CHARSETNUM;
 }
 
 void CalpontSystemCatalog::buildSysTablemap()
@@ -6326,6 +6142,21 @@ CalpontSystemCatalog::ColType::ColType(const ColType& rhs) : TypeHolderStd(rhs)
   cs = rhs.cs;
 }
 
+CalpontSystemCatalog::ColType::ColType(int32_t colWidth_, int32_t scale_, int32_t precision_,
+  const ConstraintType& constraintType_, const DictOID& ddn_, int32_t colPosition_,
+  int32_t compressionType_, OID columnOID_, const ColDataType& colDataType_)
+  : constraintType(constraintType_),
+    ddn(ddn_),
+    colPosition(colPosition_),
+    compressionType(compressionType_),
+    columnOID(columnOID_)
+{
+  colWidth = colWidth_;
+  scale = scale_;
+  precision = precision_;
+  colDataType = colDataType_;
+}
+
 CalpontSystemCatalog::ColType& CalpontSystemCatalog::ColType::operator=(const ColType& rhs)
 {
   TypeHolderStd::operator=(rhs);
diff --git a/dbcon/execplan/calpontsystemcatalog.h b/dbcon/execplan/calpontsystemcatalog.h
index f8d7d500a..baa10446f 100644
--- a/dbcon/execplan/calpontsystemcatalog.h
+++ b/dbcon/execplan/calpontsystemcatalog.h
@@ -134,6 +134,15 @@ class CalpontSystemCatalog : public datatypes::SystemCatalog
     DictOID() : dictOID(0), listOID(0), treeOID(0), compressionType(0)
     {
     }
+    DictOID(OID dictOID_, OID listOID_, OID treeOID_, int compressionType_) :
+      dictOID(dictOID_), listOID(listOID_), treeOID(treeOID_),
+      compressionType(compressionType_)
+    {
+    }
+    DictOID(const DictOID& rhs)
+    : dictOID(rhs.dictOID), listOID(rhs.listOID), treeOID(rhs.treeOID), compressionType(rhs.compressionType)
+    {
+    }
     OID dictOID;
     OID listOID;
     OID treeOID;
@@ -223,6 +232,9 @@ class CalpontSystemCatalog : public datatypes::SystemCatalog
    public:
     ColType();
     ColType(const ColType& rhs);
+    ColType(int32_t colWidth_, int32_t scale_, int32_t precision_,
+            const ConstraintType& constraintType_, const DictOID& ddn_, int32_t colPosition_,
+            int32_t compressionType_, OID columnOID_, const ColDataType& colDataType_);
     ColType& operator=(const ColType& rhs);
 
     CHARSET_INFO* getCharset();
@@ -1206,6 +1218,7 @@ const std::string MAXVALUE_COL = "maxvalue";
 const std::string COMPRESSIONTYPE_COL = "compressiontype";
 const std::string NEXTVALUE_COL = "nextvalue";
 const std::string AUXCOLUMNOID_COL = "auxcolumnoid";
+const std::string CHARSETNUM_COL = "charsetnum";
 
 /*****************************************************
  * System tables OID definition
@@ -1257,7 +1270,8 @@ const int OID_SYSCOLUMN_MINVALUE = SYSCOLUMN_BASE + 19;        /** @brief min va
 const int OID_SYSCOLUMN_MAXVALUE = SYSCOLUMN_BASE + 20;        /** @brief max value col */
 const int OID_SYSCOLUMN_COMPRESSIONTYPE = SYSCOLUMN_BASE + 21; /** @brief compression type */
 const int OID_SYSCOLUMN_NEXTVALUE = SYSCOLUMN_BASE + 22;       /** @brief next value */
-const int SYSCOLUMN_MAX = SYSCOLUMN_BASE + 23;                 // be sure this is one more than the highest #
+const int OID_SYSCOLUMN_CHARSETNUM = SYSCOLUMN_BASE + 23;      /** @brief character set number for the column */
+const int SYSCOLUMN_MAX = SYSCOLUMN_BASE + 24;                 // be sure this is one more than the highest #
 
 /*****************************************************
  * SYSTABLE columns dictionary OID definition
diff --git a/dbcon/mysql/ha_mcs_ddl.cpp b/dbcon/mysql/ha_mcs_ddl.cpp
index d3579e602..db2176708 100644
--- a/dbcon/mysql/ha_mcs_ddl.cpp
+++ b/dbcon/mysql/ha_mcs_ddl.cpp
@@ -2515,7 +2515,8 @@ int ha_mcs_impl_create_(const char* name, TABLE* table_arg, HA_CREATE_INFO* crea
         const CHARSET_INFO* field_cs = (*field)->charset();
         if (field_cs && (!share->table_charset || field_cs->number != share->table_charset->number))
         {
-          oss << " CHARACTER SET " << field_cs->cs_name.str;
+          oss << " CHARACTER SET " << field_cs->cs_name.str <<
+          " COLLATE " << field_cs->coll_name.str;
         }
       }
 
@@ -2555,7 +2556,8 @@ int ha_mcs_impl_create_(const char* name, TABLE* table_arg, HA_CREATE_INFO* crea
 
     if (share->table_charset)
     {
-      oss << " DEFAULT CHARSET=" << share->table_charset->cs_name.str;
+      oss << " DEFAULT CHARSET=" << share->table_charset->cs_name.str <<
+      " COLLATE=" << share->table_charset->coll_name.str;
     }
 
     // Process table level options such as MIN_ROWS, MAX_ROWS, COMMENT
diff --git a/dbcon/mysql/syscatalog_mysql.sql b/dbcon/mysql/syscatalog_mysql.sql
index 900a9d164..3e93db678 100644
--- a/dbcon/mysql/syscatalog_mysql.sql
+++ b/dbcon/mysql/syscatalog_mysql.sql
@@ -41,6 +41,7 @@ create table if not exists syscolumn (`schema` varchar(128),
                         minvalue varchar(64),
                         `maxvalue` varchar(64),
                         compressiontype integer,
-						nextvalue bigint) engine=columnstore comment='SCHEMA SYNC ONLY';
+                        nextvalue bigint,
+                        charsetnum int not null default 0) engine=columnstore comment='SCHEMA SYNC ONLY';
 
-DELIMITER ;
\ No newline at end of file
+DELIMITER ;
diff --git a/mysql-test/columnstore/basic/r/mcol-5005.result b/mysql-test/columnstore/basic/r/mcol-5005.result
new file mode 100644
index 000000000..957ff7153
--- /dev/null
+++ b/mysql-test/columnstore/basic/r/mcol-5005.result
@@ -0,0 +1,85 @@
+DROP DATABASE IF EXISTS mcol5005;
+CREATE DATABASE mcol5005;
+USE mcol5005;
+CREATE TABLE t1 (
+a VARCHAR(15) collate utf8mb4_romanian_ci,
+b VARCHAR(15) charset 'utf8mb3',
+c VARCHAR(15),
+d BLOB(15),
+e INT
+) ENGINE=columnstore collate=latin2_croatian_ci;
+ALTER TABLE t1 ADD COLUMN (f VARCHAR(15) collate 'utf8mb4_icelandic_ci');
+ALTER TABLE t1 ADD COLUMN (g VARCHAR(15));
+CREATE TABLE t2 ENGINE=columnstore AS SELECT * FROM t1;
+CREATE TABLE t3 LIKE t1;
+CREATE TABLE t4 (a varchar(15)) ENGINE=InnoDB charset=latin2;
+SHOW CREATE TABLE t4;
+Table	Create Table
+t4	CREATE TABLE `t4` (
+  `a` varchar(15) DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin2 COLLATE=latin2_general_ci
+ALTER TABLE t4 ENGINE=columnstore;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` varchar(15) CHARACTER SET utf8mb4 COLLATE utf8mb4_romanian_ci DEFAULT NULL,
+  `b` varchar(15) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci DEFAULT NULL,
+  `c` varchar(15) DEFAULT NULL,
+  `d` tinyblob DEFAULT NULL,
+  `e` int(11) DEFAULT NULL,
+  `f` varchar(15) CHARACTER SET utf8mb4 COLLATE utf8mb4_icelandic_ci DEFAULT NULL,
+  `g` varchar(15) DEFAULT NULL
+) ENGINE=Columnstore DEFAULT CHARSET=latin2 COLLATE=latin2_croatian_ci
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` varchar(15) CHARACTER SET utf8mb4 COLLATE utf8mb4_romanian_ci DEFAULT NULL,
+  `b` varchar(15) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci DEFAULT NULL,
+  `c` varchar(15) CHARACTER SET latin2 COLLATE latin2_croatian_ci DEFAULT NULL,
+  `d` tinyblob DEFAULT NULL,
+  `e` int(11) DEFAULT NULL,
+  `f` varchar(15) CHARACTER SET utf8mb4 COLLATE utf8mb4_icelandic_ci DEFAULT NULL,
+  `g` varchar(15) CHARACTER SET latin2 COLLATE latin2_croatian_ci DEFAULT NULL
+) ENGINE=Columnstore DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SHOW CREATE TABLE t3;
+Table	Create Table
+t3	CREATE TABLE `t3` (
+  `a` varchar(15) CHARACTER SET utf8mb4 COLLATE utf8mb4_romanian_ci DEFAULT NULL,
+  `b` varchar(15) CHARACTER SET utf8mb3 COLLATE utf8mb3_general_ci DEFAULT NULL,
+  `c` varchar(15) DEFAULT NULL,
+  `d` tinyblob DEFAULT NULL,
+  `e` int(11) DEFAULT NULL,
+  `f` varchar(15) CHARACTER SET utf8mb4 COLLATE utf8mb4_icelandic_ci DEFAULT NULL,
+  `g` varchar(15) DEFAULT NULL
+) ENGINE=Columnstore DEFAULT CHARSET=latin2 COLLATE=latin2_croatian_ci
+SHOW CREATE TABLE t4;
+Table	Create Table
+t4	CREATE TABLE `t4` (
+  `a` varchar(15) DEFAULT NULL
+) ENGINE=Columnstore DEFAULT CHARSET=latin2 COLLATE=latin2_general_ci
+SELECT `schema`, tablename, columnname, charsetnum FROM calpontsys.syscolumn
+WHERE `schema`='mcol5005' AND tablename in ('t1', 't2', 't3', 't4');
+schema	tablename	columnname	charsetnum
+mcol5005	t1	a	227
+mcol5005	t1	b	33
+mcol5005	t1	c	27
+mcol5005	t1	d	63
+mcol5005	t1	e	0
+mcol5005	t1	f	225
+mcol5005	t1	g	27
+mcol5005	t2	a	227
+mcol5005	t2	b	33
+mcol5005	t2	c	27
+mcol5005	t2	d	63
+mcol5005	t2	e	0
+mcol5005	t2	f	225
+mcol5005	t2	g	27
+mcol5005	t3	a	227
+mcol5005	t3	b	33
+mcol5005	t3	c	27
+mcol5005	t3	d	63
+mcol5005	t3	e	0
+mcol5005	t3	f	225
+mcol5005	t3	g	27
+mcol5005	t4	a	9
+DROP DATABASE mcol5005;
diff --git a/mysql-test/columnstore/basic/t/mcol-5005.test b/mysql-test/columnstore/basic/t/mcol-5005.test
new file mode 100644
index 000000000..1cbca1eb1
--- /dev/null
+++ b/mysql-test/columnstore/basic/t/mcol-5005.test
@@ -0,0 +1,67 @@
+#
+# MCOL-5005 Add charset number to the calpontsys.syscolumn system
+# catalog table
+#
+--source ../include/have_columnstore.inc
+--source include/have_innodb.inc
+
+#
+# If the calpontsys database does not exist, let's create it.
+# It's possible if we're running mtr without --extern.
+#
+let $calpontsys_exists=`SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='calpontsys' AND TABLE_NAME='syscolumn';`;
+--disable_query_log
+if (!$calpontsys_exists)
+{
+  --exec $MYSQL < $MCS_SYSCATALOG_MYSQL_SQL
+  use test;
+}
+--enable_query_log
+
+
+--disable_warnings
+DROP DATABASE IF EXISTS mcol5005;
+--enable_warnings
+
+CREATE DATABASE mcol5005;
+
+USE mcol5005;
+
+CREATE TABLE t1 (
+  a VARCHAR(15) collate utf8mb4_romanian_ci,
+  b VARCHAR(15) charset 'utf8mb3',
+  c VARCHAR(15),
+  d BLOB(15),
+  e INT
+) ENGINE=columnstore collate=latin2_croatian_ci;
+
+ALTER TABLE t1 ADD COLUMN (f VARCHAR(15) collate 'utf8mb4_icelandic_ci');
+ALTER TABLE t1 ADD COLUMN (g VARCHAR(15));
+
+CREATE TABLE t2 ENGINE=columnstore AS SELECT * FROM t1;
+
+CREATE TABLE t3 LIKE t1;
+
+CREATE TABLE t4 (a varchar(15)) ENGINE=InnoDB charset=latin2;
+SHOW CREATE TABLE t4;
+ALTER TABLE t4 ENGINE=columnstore;
+
+SHOW CREATE TABLE t1;
+SHOW CREATE TABLE t2;
+SHOW CREATE TABLE t3;
+SHOW CREATE TABLE t4;
+
+SELECT `schema`, tablename, columnname, charsetnum FROM calpontsys.syscolumn
+WHERE `schema`='mcol5005' AND tablename in ('t1', 't2', 't3', 't4');
+
+DROP DATABASE mcol5005;
+
+
+--disable_query_log
+if (!$calpontsys_exists)
+{
+  drop table calpontsys.systable restrict;
+  drop table calpontsys.syscolumn restrict;
+  drop database calpontsys;
+}
+--enable_query_log
diff --git a/tools/dbbuilder/dbbuilder.cpp b/tools/dbbuilder/dbbuilder.cpp
index 17d3da59e..973443fac 100644
--- a/tools/dbbuilder/dbbuilder.cpp
+++ b/tools/dbbuilder/dbbuilder.cpp
@@ -203,13 +203,18 @@ int main(int argc, char* argv[])
                                                     // note that the candidate and reference OID
                                                     // datatypes and colwidths are assumed to be the
                                                     // same in SystemCatalog::upgrade().
+    upgradeOidMap[OID_SYSCOLUMN_CHARSETNUM] =
+      std::make_pair(OID_SYSCOLUMN_OBJECTID, false);
 
     std::unordered_map<int, OidTypeT> upgradeOidTypeMap;
     upgradeOidTypeMap[OID_SYSTABLE_AUXCOLUMNOID] =
       std::make_pair(CalpontSystemCatalog::INT, 4);
+    upgradeOidTypeMap[OID_SYSCOLUMN_CHARSETNUM] =
+      std::make_pair(CalpontSystemCatalog::INT, 4);
 
     std::unordered_map<int, std::string> upgradeOidDefaultValStrMap;
     upgradeOidDefaultValStrMap[OID_SYSTABLE_AUXCOLUMNOID] = "0";
+    upgradeOidDefaultValStrMap[OID_SYSCOLUMN_CHARSETNUM] = "0";
 
     try
     {
@@ -300,7 +305,7 @@ int main(int argc, char* argv[])
       }
       else
       {
-        sysCatalog.upgrade(upgradeOidMap, upgradeOidTypeMap,upgradeOidDefaultValStrMap);
+        sysCatalog.upgrade(upgradeOidMap, upgradeOidTypeMap, upgradeOidDefaultValStrMap);
       }
 
       std::string cmd = "echo 'OK: buildOption=" + oam.itoa(buildOption) + "' > " + logFile;
diff --git a/tools/dbbuilder/systemcatalog.cpp b/tools/dbbuilder/systemcatalog.cpp
index c55edc69e..1f54088f2 100644
--- a/tools/dbbuilder/systemcatalog.cpp
+++ b/tools/dbbuilder/systemcatalog.cpp
@@ -272,6 +272,7 @@ void SystemCatalog::build()
     oids[DICTOID_SYSCOLUMN_MAXVALUE] = DICTOID_SYSCOLUMN_MAXVALUE;
     oids[OID_SYSCOLUMN_COMPRESSIONTYPE] = OID_SYSCOLUMN_COMPRESSIONTYPE;
     oids[OID_SYSCOLUMN_NEXTVALUE] = OID_SYSCOLUMN_NEXTVALUE;
+    oids[OID_SYSCOLUMN_CHARSETNUM] = OID_SYSCOLUMN_CHARSETNUM;
   }
 
   cout << endl;
@@ -573,6 +574,17 @@ void SystemCatalog::build()
 
   msg.str("");
 
+  // charsetnum
+  msg << "  Creating CHARSETNUM column OID: " << OID_SYSCOLUMN_CHARSETNUM;
+  cout << msg.str() << endl;
+  rc = fWriteEngine.createColumn(txnID, OID_SYSCOLUMN_CHARSETNUM, CalpontSystemCatalog::INT, 4, dbRoot,
+                                 partition, compressionType);
+
+  if (rc)
+    throw runtime_error(msg.str() + ec.errorString(rc));
+
+  msg.str("");
+
   // flush data files
   fWriteEngine.flushDataFiles(rc, 1, oids);
   // save brm
diff --git a/writeengine/server/we_ddlcommandproc.cpp b/writeengine/server/we_ddlcommandproc.cpp
index 2e385f20a..e5aed232e 100644
--- a/writeengine/server/we_ddlcommandproc.cpp
+++ b/writeengine/server/we_ddlcommandproc.cpp
@@ -645,6 +645,10 @@ uint8_t WE_DDLCommandProc::writeCreateSyscolumn(ByteStream& bs, std::string& err
         {
           colTuple.data = colDefPtr->fType->fNextvalue;
         }
+        else if (CHARSETNUM_COL == column.tableColName.column)
+        {
+          colTuple.data = colDefPtr->fType->fCharsetNum;
+        }
         else
         {
           colTuple.data = column.colType.getNullValueForType();
@@ -1031,6 +1035,10 @@ uint8_t WE_DDLCommandProc::writeSyscolumn(ByteStream& bs, std::string& err)
       {
         colTuple.data = colDefPtr->fType->fNextvalue;
       }
+      else if (CHARSETNUM_COL == column.tableColName.column)
+      {
+        colTuple.data = colDefPtr->fType->fCharsetNum;
+      }
       else
       {
         colTuple.data = column.colType.getNullValueForType();