backport to 5.5 dyncol changes and names support

19 files changed, 1424 insertions, 778 deletions

diff --git a/include/ma_dyncol.h b/include/ma_dyncol.h
index 2264ec6f53f..78d3f15978c 100644
--- a/include/ma_dyncol.h
+++ b/include/ma_dyncol.h
@@ -38,12 +38,13 @@
   how the offset are stored.
 */
 #define MAX_DYNAMIC_COLUMN_LENGTH 0X1FFFFFFFL
+#define MAX_DYNAMIC_COLUMN_LENGTH_NM 0XFFFFFFFFFL
 
 /*
   Limits of implementation
 */
-#define MAX_NAME_LENGTH 255
 #define MAX_TOTAL_NAME_LENGTH 65535
+#define MAX_NAME_LENGTH (MAX_TOTAL_NAME_LENGTH/4)
 
 /* NO and OK is the same used just to show semantics */
 #define ER_DYNCOL_NO ER_DYNCOL_OK
@@ -72,7 +73,8 @@ enum enum_dynamic_column_type
   DYN_COL_DECIMAL,
   DYN_COL_DATETIME,
   DYN_COL_DATE,
-  DYN_COL_TIME
+  DYN_COL_TIME,
+  DYN_COL_DYNCOL
 };
 
 typedef enum enum_dynamic_column_type DYNAMIC_COLUMN_TYPE;
@@ -88,7 +90,6 @@ struct st_dynamic_column_value
     struct {
       LEX_STRING value;
       CHARSET_INFO *charset;
-      my_bool nonfreeable;
     } string;
     struct {
       decimal_digit_t buffer[DECIMAL_BUFF_LENGTH];
@@ -100,6 +101,8 @@ struct st_dynamic_column_value
 
 typedef struct st_dynamic_column_value DYNAMIC_COLUMN_VALUE;
 
+/* old functions (deprecated) */
+#ifdef MADYNCOL_DEPRECATED
 enum enum_dyncol_func_result
 dynamic_column_create(DYNAMIC_COLUMN *str,
                       uint column_nr, DYNAMIC_COLUMN_VALUE *value);
@@ -109,21 +112,6 @@ dynamic_column_create_many(DYNAMIC_COLUMN *str,
                            uint column_count,
                            uint *column_numbers,
                            DYNAMIC_COLUMN_VALUE *values);
-
-enum enum_dyncol_func_result
-dynamic_column_create_many_fmt(DYNAMIC_COLUMN *str,
-                               uint column_count,
-                               uchar *column_keys,
-                               DYNAMIC_COLUMN_VALUE *values,
-                               my_bool names);
-enum enum_dyncol_func_result
-dynamic_column_create_many_internal_fmt(DYNAMIC_COLUMN *str,
-                                        uint column_count,
-                                        void *column_keys,
-                                        DYNAMIC_COLUMN_VALUE *values,
-                                        my_bool new_str,
-                                        my_bool string_keys);
-
 enum enum_dyncol_func_result
 dynamic_column_update(DYNAMIC_COLUMN *org, uint column_nr,
                       DYNAMIC_COLUMN_VALUE *value);
@@ -133,73 +121,105 @@ dynamic_column_update_many(DYNAMIC_COLUMN *str,
                            uint *column_numbers,
                            DYNAMIC_COLUMN_VALUE *values);
 enum enum_dyncol_func_result
-dynamic_column_update_many_fmt(DYNAMIC_COLUMN *str,
-                               uint add_column_count,
-                               void *column_keys,
-                               DYNAMIC_COLUMN_VALUE *values,
-                               my_bool string_keys);
-
-enum enum_dyncol_func_result
 dynamic_column_delete(DYNAMIC_COLUMN *org, uint column_nr);
 
 enum enum_dyncol_func_result
 dynamic_column_exists(DYNAMIC_COLUMN *org, uint column_nr);
+
 enum enum_dyncol_func_result
-dynamic_column_exists_str(DYNAMIC_COLUMN *str, LEX_STRING *name);
+dynamic_column_list(DYNAMIC_COLUMN *org, DYNAMIC_ARRAY *array_of_uint);
+
 enum enum_dyncol_func_result
-dynamic_column_exists_fmt(DYNAMIC_COLUMN *str, void *key, my_bool string_keys);
+dynamic_column_get(DYNAMIC_COLUMN *org, uint column_nr,
+                   DYNAMIC_COLUMN_VALUE *store_it_here);
+#endif
 
-/* List of not NULL columns */
+/* new functions */
 enum enum_dyncol_func_result
-dynamic_column_list(DYNAMIC_COLUMN *org, DYNAMIC_ARRAY *array_of_uint);
+mariadb_dyncol_create_many(DYNAMIC_COLUMN *str,
+                           uint column_count,
+                           uint *column_numbers,
+                           DYNAMIC_COLUMN_VALUE *values,
+                           my_bool new_string);
 enum enum_dyncol_func_result
-dynamic_column_list_str(DYNAMIC_COLUMN *str, DYNAMIC_ARRAY *array_of_lexstr);
+mariadb_dyncol_create_many_named(DYNAMIC_COLUMN *str,
+                                 uint column_count,
+                                 LEX_STRING *column_keys,
+                                 DYNAMIC_COLUMN_VALUE *values,
+                                 my_bool new_string);
+
+
+enum enum_dyncol_func_result
+mariadb_dyncol_update_many(DYNAMIC_COLUMN *str,
+                           uint add_column_count,
+                           uint *column_keys,
+                           DYNAMIC_COLUMN_VALUE *values);
+enum enum_dyncol_func_result
+mariadb_dyncol_update_many_named(DYNAMIC_COLUMN *str,
+                                 uint add_column_count,
+                                 LEX_STRING *column_keys,
+                                 DYNAMIC_COLUMN_VALUE *values);
+
+
+enum enum_dyncol_func_result
+mariadb_dyncol_exists(DYNAMIC_COLUMN *org, uint column_nr);
 enum enum_dyncol_func_result
-dynamic_column_list_fmt(DYNAMIC_COLUMN *str, DYNAMIC_ARRAY *array, my_bool string_keys);
+mariadb_dyncol_exists_named(DYNAMIC_COLUMN *str, LEX_STRING *name);
+
+/* List of not NULL columns */
+enum enum_dyncol_func_result
+mariadb_dyncol_list(DYNAMIC_COLUMN *org, DYNAMIC_ARRAY *array_of_uint);
+enum enum_dyncol_func_result
+mariadb_dyncol_list_named(DYNAMIC_COLUMN *str, uint *count, LEX_STRING **names);
 
 /*
    if the column do not exists it is NULL
 */
 enum enum_dyncol_func_result
-dynamic_column_get(DYNAMIC_COLUMN *org, uint column_nr,
+mariadb_dyncol_get(DYNAMIC_COLUMN *org, uint column_nr,
                    DYNAMIC_COLUMN_VALUE *store_it_here);
 enum enum_dyncol_func_result
-dynamic_column_get_str(DYNAMIC_COLUMN *str, LEX_STRING *name,
-                           DYNAMIC_COLUMN_VALUE *store_it_here);
+mariadb_dyncol_get_named(DYNAMIC_COLUMN *str, LEX_STRING *name,
+                         DYNAMIC_COLUMN_VALUE *store_it_here);
 
-my_bool dynamic_column_has_names(DYNAMIC_COLUMN *str);
+my_bool mariadb_dyncol_has_names(DYNAMIC_COLUMN *str);
 
 enum enum_dyncol_func_result
-dynamic_column_check(DYNAMIC_COLUMN *str);
+mariadb_dyncol_check(DYNAMIC_COLUMN *str);
 
 enum enum_dyncol_func_result
-dynamic_column_json(DYNAMIC_COLUMN *str, DYNAMIC_STRING *json);
+mariadb_dyncol_json(DYNAMIC_COLUMN *str, DYNAMIC_STRING *json);
 
 #define dynamic_column_initialize(A) memset((A), 0, sizeof(*(A)))
 #define dynamic_column_column_free(V) dynstr_free(V)
 
 /* conversion of values to 3 base types */
 enum enum_dyncol_func_result
-dynamic_column_val_str(DYNAMIC_STRING *str, DYNAMIC_COLUMN_VALUE *val,
+mariadb_dyncol_val_str(DYNAMIC_STRING *str, DYNAMIC_COLUMN_VALUE *val,
                        CHARSET_INFO *cs, my_bool quote);
 enum enum_dyncol_func_result
-dynamic_column_val_long(longlong *ll, DYNAMIC_COLUMN_VALUE *val);
+mariadb_dyncol_val_long(longlong *ll, DYNAMIC_COLUMN_VALUE *val);
+enum enum_dyncol_func_result
+mariadb_dyncol_val_double(double *dbl, DYNAMIC_COLUMN_VALUE *val);
+
+
 enum enum_dyncol_func_result
-dynamic_column_val_double(double *dbl, DYNAMIC_COLUMN_VALUE *val);
+mariadb_dyncol_unpack(DYNAMIC_COLUMN *str,
+                      uint *count,
+                      LEX_STRING **names, DYNAMIC_COLUMN_VALUE **vals);
 
+int mariadb_dyncol_column_cmp_named(const LEX_STRING *s1, const LEX_STRING *s2);
 
 enum enum_dyncol_func_result
-dynamic_column_vals(DYNAMIC_COLUMN *str,
-                    DYNAMIC_ARRAY *names, DYNAMIC_ARRAY *vals,
-                    char **free_names);
+mariadb_dyncol_column_count(DYNAMIC_COLUMN *str, uint *column_count);
 
 /***************************************************************************
  Internal functions, don't use if you don't know what you are doing...
 ***************************************************************************/
 
-#define dynamic_column_reassociate(V,P,L, A) dynstr_reassociate((V),(P),(L),(A))
+#define mariadb_dyncol_reassociate(V,P,L, A) dynstr_reassociate((V),(P),(L),(A))
 
-#define dynamic_column_value_init(V) (V)->type= DYN_COL_NULL
+#define dyncol_value_init(V) (V)->type= DYN_COL_NULL
 
 /*
   Prepare value for using as decimal
diff --git a/include/my_sys.h b/include/my_sys.h
index 58a343bb789..f56c29afeae 100644
--- a/include/my_sys.h
+++ b/include/my_sys.h
@@ -794,11 +794,16 @@ my_bool dynstr_append_mem(DYNAMIC_STRING *str, const char *append,
 extern my_bool dynstr_append_os_quoted(DYNAMIC_STRING *str, const char *append,
                                        ...);
 extern my_bool dynstr_append_quoted(DYNAMIC_STRING *str,
-                                    const char *append, size_t len);
+                                    const char *append, size_t len,
+                                    char quote);
 extern my_bool dynstr_set(DYNAMIC_STRING *str, const char *init_str);
 extern my_bool dynstr_realloc(DYNAMIC_STRING *str, size_t additional_size);
 extern my_bool dynstr_trunc(DYNAMIC_STRING *str, size_t n);
 extern void dynstr_free(DYNAMIC_STRING *str);
+extern uint32 copy_and_convert_extended(char *to, uint32 to_length,
+                                        CHARSET_INFO *to_cs,
+                                        const char *from, uint32 from_length,
+                                        CHARSET_INFO *from_cs, uint *errors);
 extern void dynstr_reassociate(DYNAMIC_STRING *str, char **res, size_t *length,
                                size_t *alloc_length);
 extern uint32 copy_and_convert_extended(char *to, uint32 to_length,
diff --git a/mysql-test/r/cassandra.result b/mysql-test/r/cassandra.result
index 55607726458..b5ff3194480 100644
--- a/mysql-test/r/cassandra.result
+++ b/mysql-test/r/cassandra.result
@@ -532,7 +532,7 @@ delete from t1;
 drop table t1;
 CREATE TABLE t1 (rowkey varchar(10) PRIMARY KEY, dyn blob DYNAMIC_COLUMN_STORAGE=yes) ENGINE=CASSANDRA thrift_host='localhost' keyspace='mariadbtest2' column_family = 'cfd1';
 select * from t1;
-ERROR HY000: Internal error: 'Unable to convert value for field `dyn` from Cassandra's data format. Name length exceed limit of 255: 'very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_ver'
+ERROR HY000: Internal error: 'Unable to convert value for field `dyn` from Cassandra's data format. Name length exceed limit of 16383: 'very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_v'
 drop table t1;
 CREATE TABLE t1 (rowkey int PRIMARY KEY, dyn blob DYNAMIC_COLUMN_STORAGE=yes) 
 ENGINE=CASSANDRA thrift_host='localhost' keyspace='mariadbtest2' column_family = 'cfd2';
diff --git a/mysql-test/r/dyncol.result b/mysql-test/r/dyncol.result
index 2e9a8462eee..c8d23d75d6c 100644
--- a/mysql-test/r/dyncol.result
+++ b/mysql-test/r/dyncol.result
@@ -1342,22 +1342,22 @@ hex(COLUMN_CREATE(0, 0.0 as decimal))
 set names utf8;
 select hex(column_create("адын", 1212));
 hex(column_create("адын", 1212))
-040100080008000000D0B0D0B4D18BD0BD7809
+040100080000000000D0B0D0B4D18BD0BD7809
 select hex(column_create("1212", 1212));
 hex(column_create("1212", 1212))
-040100040004000000313231327809
+040100040000000000313231327809
 select hex(column_create(1212, 2, "www", 3));
 hex(column_create(1212, 2, "www", 3))
-04020007000300000004030008777777313231320604
+04020007000000000003001000777777313231320604
 select hex(column_create("1212", 2, "www", 3));
 hex(column_create("1212", 2, "www", 3))
-04020007000300000004030008777777313231320604
+04020007000000000003001000777777313231320604
 select hex(column_create("1212", 2, 3, 3));
 hex(column_create("1212", 2, 3, 3))
-0402000500010000000401000833313231320604
+0402000500000000000100100033313231320604
 select hex(column_create("1212", 2, "адын", 1, 3, 3));
 hex(column_create("1212", 2, "адын", 1, 3, 3))
-0403000D000100000004010008080500103331323132D0B0D0B4D18BD0BD060402
+0403000D000000000001001000050020003331323132D0B0D0B4D18BD0BD060402
 set names default;
 # fetching column test (names)
 set names utf8;
@@ -1413,10 +1413,10 @@ set names default;
 # column changing test (names)
 select hex(column_add(column_create(1, "AAA"), "b", "BBB"));
 hex(column_add(column_create(1, "AAA"), "b", "BBB"))
-0402000200010000030101002331620841414108424242
+0402000200000003000100430031620841414108424242
 select hex(column_add(column_create("1", "AAA"), "b", "BBB"));
 hex(column_add(column_create("1", "AAA"), "b", "BBB"))
-0402000200010000030101002331620841414108424242
+0402000200000003000100430031620841414108424242
 select column_get(column_add(column_create(1, "AAA"), "b", "BBB"), 1 as char);
 column_get(column_add(column_create(1, "AAA"), "b", "BBB"), 1 as char)
 AAA
@@ -1425,25 +1425,25 @@ column_get(column_add(column_create(1, "AAA"), "b", "BBB"), "b" as char)
 BBB
 select hex(column_add(column_create("a", "AAA"), 1, "BBB"));
 hex(column_add(column_create("a", "AAA"), 1, "BBB"))
-0402000200010000030101002331610842424208414141
+0402000200000003000100430031610842424208414141
 select hex(column_add(column_create("a", "AAA"), "1", "BBB"));
 hex(column_add(column_create("a", "AAA"), "1", "BBB"))
-0402000200010000030101002331610842424208414141
+0402000200000003000100430031610842424208414141
 select hex(column_add(column_create("a", 1212 as integer), "b", "1212" as integer));
 hex(column_add(column_create("a", 1212 as integer), "b", "1212" as integer))
-04020002000100000001010010616278097809
+04020002000000000001002000616278097809
 select hex(column_add(column_create("a", 1212 as integer), "a", "1212" as integer));
 hex(column_add(column_create("a", 1212 as integer), "a", "1212" as integer))
-040100010001000000617809
+040100010000000000617809
 select hex(column_add(column_create("a", 1212 as integer), "a", NULL as integer));
 hex(column_add(column_create("a", 1212 as integer), "a", NULL as integer))
 0400000000
 select hex(column_add(column_create("a", 1212 as integer), "b", NULL as integer));
 hex(column_add(column_create("a", 1212 as integer), "b", NULL as integer))
-040100010001000000617809
+040100010000000000617809
 select hex(column_add(column_create("a", 1212 as integer), "b", 1212 as integer, "a", 11 as integer));
 hex(column_add(column_create("a", 1212 as integer), "b", 1212 as integer, "a", 11 as integer))
-040200020001000000010100086162167809
+040200020000000000010010006162167809
 select column_get(column_add(column_create("a", 1212 as integer), "b", 1212 as integer, "a", 11 as integer), "a" as integer);
 column_get(column_add(column_create("a", 1212 as integer), "b", 1212 as integer, "a", 11 as integer), "a" as integer)
 11
@@ -1452,13 +1452,13 @@ column_get(column_add(column_create("a", 1212 as integer), "b", 1212 as integer,
 1212
 select hex(column_add(column_create("a", 1212 as integer), "a", 1212 as integer, "b", 11 as integer));
 hex(column_add(column_create("a", 1212 as integer), "a", 1212 as integer, "b", 11 as integer))
-040200020001000000010100106162780916
+040200020000000000010020006162780916
 select hex(column_add(column_create("a", NULL as integer), "a", 1212 as integer, "b", 11 as integer));
 hex(column_add(column_create("a", NULL as integer), "a", 1212 as integer, "b", 11 as integer))
-040200020001000000010100106162780916
+040200020000000000010020006162780916
 select hex(column_add(column_create("a", 1212 as integer, "b", 1212 as integer), "a", 11 as integer));
 hex(column_add(column_create("a", 1212 as integer, "b", 1212 as integer), "a", 11 as integer))
-040200020001000000010100086162167809
+040200020000000000010010006162167809
 select hex(column_add(column_create("a", 1), "a", null));
 hex(column_add(column_create("a", 1), "a", null))
 0400000000
@@ -1470,29 +1470,29 @@ column_list(column_add(column_create("a", 1), "a", ""))
 `a`
 select hex(column_add("", "a", 1));
 hex(column_add("", "a", 1))
-0401000100010000006102
+0401000100000000006102
 # column delete (names)
 select hex(column_delete(column_create("a", 1212 as integer, "b", 1212 as integer), "a"));
 hex(column_delete(column_create("a", 1212 as integer, "b", 1212 as integer), "a"))
-040100010001000000627809
+040100010000000000627809
 select hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "b"));
 hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "b"))
-0402000200010000000101000861630206
+0402000200000000000100100061630206
 select hex(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer));
 hex(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer))
-0403000300010000000101000801020010616263020406
+0403000300000000000100100002002000616263020406
 select hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "c"));
 hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "c"))
-0402000200010000000101000861620204
+0402000200000000000100100061620204
 select hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "d"));
 hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "d"))
-0403000300010000000101000801020010616263020406
+0403000300000000000100100002002000616263020406
 select hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "b", "a"));
 hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "b", "a"))
-0401000100010000006306
+0401000100000000006306
 select hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "b", "c"));
 hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "b", "c"))
-0401000100010000006102
+0401000100000000006102
 select hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "a", "b", "c"));
 hex(column_delete(column_create("a", 1 as integer, "b", 2 as integer, "c", 3 as integer), "a", "b", "c"))
 0400000000
@@ -1517,11 +1517,21 @@ ERROR 42S22: Unknown column 'color' in 'field list'
 #
 CREATE TABLE t1 (f1 tinyblob);
 INSERT INTO t1 VALUES (COLUMN_CREATE('col1', REPEAT('a',30)));
+select column_check(f1) from t1;
+column_check(f1)
+1
 UPDATE t1 SET f1 = COLUMN_ADD( f1, REPEAT('b',211), 'val2' );
 Warnings:
 Warning	1265	Data truncated for column 'f1' at row 1
+select column_check(f1) from t1;
+column_check(f1)
+0
 UPDATE t1 SET f1 = COLUMN_ADD( f1, REPEAT('c',211), 'val3' );
-ERROR HY000: Encountered illegal format of dynamic column string
+Warnings:
+Warning	1265	Data truncated for column 'f1' at row 1
+select column_check(f1) from t1;
+column_check(f1)
+0
 drop table t1;
 #
 # MDEV-490/MDEV-491 null as arguments
@@ -1550,8 +1560,8 @@ NULL
 #
 SELECT hex(COLUMN_CREATE(REPEAT('a',255),1));
 hex(COLUMN_CREATE(REPEAT('a',255),1))
-040100FF00FF00000061616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616102
-SELECT hex(COLUMN_CREATE(REPEAT('a',256),1));
+040100FF000000000061616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616161616102
+SELECT hex(COLUMN_CREATE(REPEAT('a',65536),1));
 ERROR 22007: Illegal value used as argument of dynamic column function
 #
 # JSON conversion
@@ -1577,3 +1587,53 @@ COLUMN_CHECK('')
 SELECT COLUMN_CHECK(NULL);
 COLUMN_CHECK(NULL)
 NULL
+#
+# escaping check
+#
+select column_json(column_create("string", "'\"/\\`.,whatever")),hex(column_create("string", "'\"/\\`.,whatever"));
+column_json(column_create("string", "'\"/\\`.,whatever"))	hex(column_create("string", "'\"/\\`.,whatever"))
+[{"string":"'\"/\\`.,whatever"}]	040100060000000300737472696E670827222F5C602E2C7768617465766572
+#
+# embedding test
+#
+select column_json(column_create("val", "val", "emb", column_create("val2", "val2")));
+column_json(column_create("val", "val", "emb", column_create("val2", "val2")))
+[{"emb":[{"val2":"val2"}],{"val":"val"}]
+select column_json(column_create(1, "val", 2, column_create(3, "val2")));
+column_json(column_create(1, "val", 2, column_create(3, "val2")))
+[{"1":"val"},{"2":[{"3":"val2"}]]
+#
+# Time encoding
+#
+select hex(column_create("t", "800:46:06.23434" AS time)) as hex,
+column_json(column_create("t", "800:46:06.23434" AS time)) as json;
+hex	json
+04010001000000070074649363B82003	[{"t":"800:46:06.234340"}]
+select hex(column_create(1, "800:46:06.23434" AS time)) as hex,
+column_json(column_create(1, "800:46:06.23434" AS time)) as json;
+hex	json
+000100010007649363B82003	[{"1":"800:46:06.234340"}]
+select hex(column_create("t", "800:46:06" AS time)) as hex,
+column_json(column_create("t", "800:46:06" AS time)) as json;
+hex	json
+04010001000000070074860B32	[{"t":"800:46:06"}]
+select hex(column_create(1, "800:46:06" AS time)) as hex,
+column_json(column_create(1, "800:46:06" AS time)) as json;
+hex	json
+000100010007000060B82003	[{"1":"800:46:06"}]
+select hex(column_create("t", "2012-12-21 10:46:06.23434" AS datetime)) as hex,
+column_json(column_create("t", "2012-12-21 10:46:06.23434" AS datetime)) as json;
+hex	json
+0401000100000005007495B90F649363B80A00	[{"t":"2012-12-21 10:46:06.234340"}]
+select hex(column_create(1, "2012-12-21 10:46:06.23434" AS datetime)) as hex,
+column_json(column_create(1, "2012-12-21 10:46:06.23434" AS datetime)) as json;
+hex	json
+00010001000595B90F649363B80A00	[{"1":"2012-12-21 10:46:06.234340"}]
+select hex(column_create("t", "2012-12-21 10:46:06" AS datetime)) as hex,
+column_json(column_create("t", "2012-12-21 10:46:06" AS datetime)) as json;
+hex	json
+0401000100000005007495B90F86AB00	[{"t":"2012-12-21 10:46:06"}]
+select hex(column_create(1, "2012-12-21 10:46:06" AS datetime)) as hex,
+column_json(column_create(1, "2012-12-21 10:46:06" AS datetime)) as json;
+hex	json
+00010001000595B90F000060B80A00	[{"1":"2012-12-21 10:46:06"}]
diff --git a/mysql-test/t/cassandra.test b/mysql-test/t/cassandra.test
index 542987a869a..93c81086de8 100644
--- a/mysql-test/t/cassandra.test
+++ b/mysql-test/t/cassandra.test
@@ -99,7 +99,7 @@ CREATE COLUMN FAMILY cfd1
   WITH comparator = UTF8Type
   AND key_validation_class=UTF8Type 
   AND default_validation_class = UTF8Type;
-SET cfd1['1']['very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_long_name']='1';
+SET cfd1['1']['very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_very_long_name']='1';
 
 CREATE COLUMN FAMILY cfd2 
   WITH comparator = UTF8Type
diff --git a/mysql-test/t/dyncol.test b/mysql-test/t/dyncol.test
index 143a833fe8d..de30cac610a 100644
--- a/mysql-test/t/dyncol.test
+++ b/mysql-test/t/dyncol.test
@@ -636,9 +636,14 @@ select COLUMN_CREATE(color, "black");
 CREATE TABLE t1 (f1 tinyblob);
 
 INSERT INTO t1 VALUES (COLUMN_CREATE('col1', REPEAT('a',30)));
+select column_check(f1) from t1;
 UPDATE t1 SET f1 = COLUMN_ADD( f1, REPEAT('b',211), 'val2' );
---error ER_DYN_COL_WRONG_FORMAT
+# we can't detect last string cut with 100% probability,
+# because we detect it by string end
+select column_check(f1) from t1;
 UPDATE t1 SET f1 = COLUMN_ADD( f1, REPEAT('c',211), 'val3' );
+select column_check(f1) from t1;
+
 drop table t1;
 
 --echo #
@@ -657,7 +662,7 @@ SELECT COLUMN_ADD( NULL, 'val', 'col');
 --echo #
 SELECT hex(COLUMN_CREATE(REPEAT('a',255),1));
 --error ER_DYN_COL_DATA
-SELECT hex(COLUMN_CREATE(REPEAT('a',256),1));
+SELECT hex(COLUMN_CREATE(REPEAT('a',65536),1));
 
 --echo #
 --echo # JSON conversion
@@ -672,3 +677,37 @@ SELECT COLUMN_CHECK(COLUMN_CREATE(1,'a'));
 SELECT COLUMN_CHECK('abracadabra');
 SELECT COLUMN_CHECK('');
 SELECT COLUMN_CHECK(NULL);
+
+--echo #
+--echo # escaping check
+--echo #
+select column_json(column_create("string", "'\"/\\`.,whatever")),hex(column_create("string", "'\"/\\`.,whatever"));
+
+--echo #
+--echo # embedding test
+--echo #
+select column_json(column_create("val", "val", "emb", column_create("val2", "val2")));
+select column_json(column_create(1, "val", 2, column_create(3, "val2")));
+
+--echo #
+--echo # Time encoding
+--echo #
+select hex(column_create("t", "800:46:06.23434" AS time)) as hex,
+       column_json(column_create("t", "800:46:06.23434" AS time)) as json;
+select hex(column_create(1, "800:46:06.23434" AS time)) as hex,
+       column_json(column_create(1, "800:46:06.23434" AS time)) as json;
+
+select hex(column_create("t", "800:46:06" AS time)) as hex,
+       column_json(column_create("t", "800:46:06" AS time)) as json;
+select hex(column_create(1, "800:46:06" AS time)) as hex,
+       column_json(column_create(1, "800:46:06" AS time)) as json;
+
+select hex(column_create("t", "2012-12-21 10:46:06.23434" AS datetime)) as hex,
+       column_json(column_create("t", "2012-12-21 10:46:06.23434" AS datetime)) as json;
+select hex(column_create(1, "2012-12-21 10:46:06.23434" AS datetime)) as hex,
+       column_json(column_create(1, "2012-12-21 10:46:06.23434" AS datetime)) as json;
+
+select hex(column_create("t", "2012-12-21 10:46:06" AS datetime)) as hex,
+       column_json(column_create("t", "2012-12-21 10:46:06" AS datetime)) as json;
+select hex(column_create(1, "2012-12-21 10:46:06" AS datetime)) as hex,
+       column_json(column_create(1, "2012-12-21 10:46:06" AS datetime)) as json;
diff --git a/mysys/ma_dyncol.c b/mysys/ma_dyncol.c
index e7c9b835454..34bb8b9de3d 100644
--- a/mysys/ma_dyncol.c
+++ b/mysys/ma_dyncol.c
@@ -1,5 +1,5 @@
-/* Copyright (c) 2011, Monty Program Ab
-   Copyright (c) 2011, Oleksandr Byelkin
+/* Copyright (c) 2011,2012  Monty Program Ab;
+   Copyright (c) 2011,2012 Oleksandr Byelkin
 
    Redistribution and use in source and binary forms, with or without
    modification, are permitted provided that the following conditions are
@@ -26,6 +26,40 @@
    SUCH DAMAGE.
 */
 
+/*
+ Numeric format:
+ ===============
+  * Fixed header part
+    1 byte flags:
+      0,1 bits - <offset size> - 1
+      2-7 bits - 0
+    2 bytes column counter
+  * Columns directory sorted by column number, each entry contains of:
+    2 bytes column number
+    <offset size> bytes (1-4) combined offset from beginning of
+      the data segment + 3 bit type
+  * Data of above columns size of data and length depend on type
+
+ Columns with names:
+ ===================
+  * Fixed header part
+    1 byte flags:
+      0,1 bits - <offset size> - 2
+      2 bit    - 1 (means format with names)
+      3,4 bits - 00 (means <names offset size> - 2,
+                     now 2 is the only supported size)
+      5-7 bits - 0
+    2 bytes column counter
+  * Variable header part (now it is actually fixed part)
+    <names offset size> (2) bytes size of stored names pool
+  * Column directory sorted by names, each consists of
+    <names offset size> (2) bytes offset of name
+    <offset size> bytes (2-5)bytes combined offset from beginning of
+      the data segment + 4 bit type
+  * Names stored one after another
+  * Data of above columns size of data and length depend on type
+*/
+
 #include "mysys_priv.h"
 #include <m_string.h>
 #include <ma_dyncol.h>
@@ -40,14 +74,22 @@ uint32 copy_and_convert(char *to, uint32 to_length, CHARSET_INFO *to_cs,
   2 bits which determinate size of offset in the header -1
 */
 /* mask to get above bits */
-#define DYNCOL_FLG_OFFSET  3
-#define DYNCOL_FLG_NAMES  4
-/* All known flags mask */
-#define DYNCOL_FLG_KNOWN  7
+#define DYNCOL_FLG_OFFSET   (1|2)
+#define DYNCOL_FLG_NAMES    4
+#define DYNCOL_FLG_NMOFFSET (8|16)
+/**
+  All known flags mask that could be set.
+
+  @note DYNCOL_FLG_NMOFFSET should be 0 for now.
+*/
+#define DYNCOL_FLG_KNOWN  (1|2|4)
 
 /* formats */
-#define DYNCOL_FMT_NUM 0
-#define DYNCOL_FMT_STR 1
+enum enum_dyncol_format
+{
+  dyncol_fmt_num= 0,
+  dyncol_fmt_str= 1
+};
 
 /* dynamic column size reserve */
 #define DYNCOL_SYZERESERVE 80
@@ -63,14 +105,15 @@ uint32 copy_and_convert(char *to, uint32 to_length, CHARSET_INFO *to_cs,
 #define FIXED_HEADER_SIZE_NM 5
 
 #define COLUMN_NUMBER_SIZE 2
-/* 1 byte name length + 2 bytes offset from the name pool */
-#define COLUMN_NAMEPTR_SIZE 3
+/* 2 bytes offset from the name pool */
+#define COLUMN_NAMEPTR_SIZE 2
 
-#define MAX_OFFSET_LENGTH  5
+#define MAX_OFFSET_LENGTH    4
+#define MAX_OFFSET_LENGTH_NM 5
 
 #define DYNCOL_NUM_CHAR 6
 
-my_bool dynamic_column_has_names(DYNAMIC_COLUMN *str)
+my_bool mariadb_dyncol_has_names(DYNAMIC_COLUMN *str)
 {
   if (str->length < 1)
     return FALSE;
@@ -79,7 +122,7 @@ my_bool dynamic_column_has_names(DYNAMIC_COLUMN *str)
 
 static enum enum_dyncol_func_result
 dynamic_column_time_store(DYNAMIC_COLUMN *str,
-                          MYSQL_TIME *value);
+                          MYSQL_TIME *value, enum enum_dyncol_format format);
 static enum enum_dyncol_func_result
 dynamic_column_date_store(DYNAMIC_COLUMN *str,
                           MYSQL_TIME *value);
@@ -96,14 +139,14 @@ dynamic_column_get_internal(DYNAMIC_COLUMN *str,
 static enum enum_dyncol_func_result
 dynamic_column_exists_internal(DYNAMIC_COLUMN *str, uint num_key,
                                LEX_STRING *str_key);
-enum enum_dyncol_func_result
+static enum enum_dyncol_func_result
 dynamic_column_update_many_fmt(DYNAMIC_COLUMN *str,
                                uint add_column_count,
                                void *column_keys,
                                DYNAMIC_COLUMN_VALUE *values,
                                my_bool string_keys);
 static int plan_sort_num(const void *a, const void *b);
-static int plan_sort_str(const void *a, const void *b);
+static int plan_sort_named(const void *a, const void *b);
 
 /*
   Structure to hold information about dynamic columns record and
@@ -118,8 +161,8 @@ struct st_dyn_header
   size_t header_size;
   size_t nmpool_size;
   size_t data_size;
-  /* DYNCOL_FMT_NUM - numeric columns, DYNCOL_FMT_STR - column names */
-  uint format;
+  /* dyncol_fmt_num - numeric columns, dyncol_fmt_str - column names */
+  enum enum_dyncol_format format;
   uint column_count;
 
   uchar *entry, *data, *name;
@@ -135,9 +178,6 @@ static inline my_bool read_fixed_header(DYN_HEADER *hdr,
 static void set_fixed_header(DYNAMIC_COLUMN *str,
                              uint offset_size,
                              uint column_count);
-static my_bool type_and_offset_store(uchar *place, size_t offset_size,
-                                     DYNAMIC_COLUMN_TYPE type,
-                                     size_t offset);
 
 /*
   Calculate entry size (E) and header size (H) by offset size (O) and column
@@ -154,8 +194,8 @@ static my_bool type_and_offset_store(uchar *place, size_t offset_size,
   Name pool size functions, for numeric format it is 0
 */
 
-size_t name_size_num(void *keys __attribute__((unused)),
-                     uint i __attribute__((unused)))
+static size_t name_size_num(void *keys __attribute__((unused)),
+                            uint i __attribute__((unused)))
 {
   return 0;
 }
@@ -164,7 +204,7 @@ size_t name_size_num(void *keys __attribute__((unused)),
 /**
   Name pool size functions.
 */
-size_t name_size_str(void *keys, uint i)
+static size_t name_size_named(void *keys, uint i)
 {
   return ((LEX_STRING *) keys)[i].length;
 }
@@ -180,24 +220,34 @@ static int column_sort_num(const void *a, const void *b)
   return **((uint **)a) - **((uint **)b);
 }
 
-
 /**
   Comparator function for references on column numbers for qsort
   (names format)
 */
 
-static int column_sort_str(const void *a, const void *b)
+int mariadb_dyncol_column_cmp_named(const LEX_STRING *s1, const LEX_STRING *s2)
 {
-  LEX_STRING *s1= *((LEX_STRING **)a);
-  LEX_STRING *s2= *((LEX_STRING **)b);
   int rc= s1->length - s2->length;
   if (rc == 0)
-    rc= memcmp((void *)s1->str, (void *)s2->str, (size_t) s1->length);
+    rc= memcmp((void *)s1->str, (void *)s2->str,
+               (size_t) s1->length);
   return rc;
 }
 
 
 /**
+  Comparator function for references on column numbers for qsort
+  (names format)
+*/
+
+static int column_sort_named(const void *a, const void *b)
+{
+  return mariadb_dyncol_column_cmp_named(*((LEX_STRING **)a),
+                                         *((LEX_STRING **)b));
+}
+
+
+/**
   Check limit function (numeric format)
 */
 
@@ -211,7 +261,7 @@ static my_bool check_limit_num(const void *val)
   Check limit function (names format)
 */
 
-static my_bool check_limit_str(const void *val)
+static my_bool check_limit_named(const void *val)
 {
   return (*((LEX_STRING **)val))->length > MAX_NAME_LENGTH;
 }
@@ -221,7 +271,7 @@ static my_bool check_limit_str(const void *val)
   Write numeric format static header part.
 */
 
-void set_fixed_header_num(DYNAMIC_COLUMN *str, DYN_HEADER *hdr)
+static void set_fixed_header_num(DYNAMIC_COLUMN *str, DYN_HEADER *hdr)
 {
   set_fixed_header(str, hdr->offset_size, hdr->column_count);
   hdr->header= (uchar *)str->str + FIXED_HEADER_SIZE;
@@ -233,10 +283,14 @@ void set_fixed_header_num(DYNAMIC_COLUMN *str, DYN_HEADER *hdr)
   Write names format static header part.
 */
 
-void set_fixed_header_str(DYNAMIC_COLUMN *str, DYN_HEADER *hdr)
+static void set_fixed_header_named(DYNAMIC_COLUMN *str, DYN_HEADER *hdr)
 {
-  set_fixed_header(str, hdr->offset_size, hdr->column_count);
-  str->str[0]|= DYNCOL_FLG_NAMES;
+  DBUG_ASSERT(hdr->column_count <= 0xffff);
+  DBUG_ASSERT(hdr->offset_size <= MAX_OFFSET_LENGTH_NM);
+  /* size of data offset, named format flag, size of names offset (0 means 2) */
+  str->str[0]= ((str->str[0] & ~(DYNCOL_FLG_OFFSET | DYNCOL_FLG_NMOFFSET)) |
+                (hdr->offset_size - 2) | DYNCOL_FLG_NAMES);
+  int2store(str->str + 1, hdr->column_count);        /* columns number */
   int2store(str->str + 3, hdr->nmpool_size);
   hdr->header= (uchar *)str->str + FIXED_HEADER_SIZE_NM;
   hdr->nmpool= hdr->header + hdr->header_size;
@@ -245,6 +299,94 @@ void set_fixed_header_str(DYNAMIC_COLUMN *str, DYN_HEADER *hdr)
 
 
 /**
+  Store offset and type information in the given place
+
+  @param place           Beginning of the index entry
+  @param offset_size     Size of offset field in bytes
+  @param type            Type to be written
+  @param offset          Offset to be written
+*/
+
+static my_bool type_and_offset_store_num(uchar *place, size_t offset_size,
+                                         DYNAMIC_COLUMN_TYPE type,
+                                         size_t offset)
+{
+  ulong val = (((ulong) offset) << 3) | (type - 1);
+  DBUG_ASSERT(type != DYN_COL_NULL);
+  DBUG_ASSERT(((type - 1) & (~7)) == 0); /* fit in 3 bits */
+  DBUG_ASSERT(offset_size >= 1 && offset_size <= 4);
+
+  /* Index entry starts with column number; jump over it */
+  place+= COLUMN_NUMBER_SIZE;
+
+  switch (offset_size) {
+  case 1:
+    if (offset >= 0x1f)          /* all 1 value is reserved */
+      return TRUE;
+    place[0]= (uchar)val;
+    break;
+  case 2:
+    if (offset >= 0x1fff)        /* all 1 value is reserved */
+      return TRUE;
+    int2store(place, val);
+    break;
+  case 3:
+    if (offset >= 0x1fffff)      /* all 1 value is reserved */
+      return TRUE;
+    int3store(place, val);
+    break;
+  case 4:
+    if (offset >= 0x1fffffff)    /* all 1 value is reserved */
+      return TRUE;
+    int4store(place, val);
+    break;
+  default:
+      return TRUE;
+  }
+  return FALSE;
+}
+
+
+static my_bool type_and_offset_store_named(uchar *place, size_t offset_size,
+                                           DYNAMIC_COLUMN_TYPE type,
+                                           size_t offset)
+{
+  ulong val = (((ulong) offset) << 4) | (type - 1);
+  DBUG_ASSERT(type != DYN_COL_NULL);
+  DBUG_ASSERT(((type - 1) & (~0xf)) == 0); /* fit in 4 bits */
+  DBUG_ASSERT(offset_size >= 2 && offset_size <= 5);
+
+  /* Index entry starts with name offset; jump over it */
+  place+= COLUMN_NAMEPTR_SIZE;
+  switch (offset_size) {
+  case 2:
+    if (offset >= 0xfff)          /* all 1 value is reserved */
+      return TRUE;
+    int2store(place, val);
+    break;
+  case 3:
+    if (offset >= 0xfffff)        /* all 1 value is reserved */
+      return TRUE;
+    int3store(place, val);
+    break;
+  case 4:
+    if (offset >= 0xfffffff)      /* all 1 value is reserved */
+      return TRUE;
+    int4store(place, val);
+    break;
+  case 5:
+    if (offset >= 0xfffffffff)    /* all 1 value is reserved */
+      return TRUE;
+    int5store(place, val);
+    break;
+  case 1:
+  default:
+      return TRUE;
+  }
+  return FALSE;
+}
+
+/**
   Write numeric format header entry
    2 bytes - column number
    1-4 bytes - data offset combined with type
@@ -255,17 +397,17 @@ void set_fixed_header_str(DYNAMIC_COLUMN *str, DYN_HEADER *hdr)
   @param offset          offset of the data
 */
 
-my_bool put_header_entry_num(DYN_HEADER *hdr,
-                             void *column_key,
-                             DYNAMIC_COLUMN_VALUE *value,
-                             size_t offset)
+static my_bool put_header_entry_num(DYN_HEADER *hdr,
+                                    void *column_key,
+                                    DYNAMIC_COLUMN_VALUE *value,
+                                    size_t offset)
 {
   uint *column_number= (uint *)column_key;
   int2store(hdr->entry, *column_number);
   DBUG_ASSERT(hdr->nmpool_size == 0);
-  if (type_and_offset_store(hdr->entry, hdr->offset_size,
-                            value->type,
-                            offset))
+  if (type_and_offset_store_num(hdr->entry, hdr->offset_size,
+                                value->type,
+                                offset))
       return TRUE;
   hdr->entry= hdr->entry + hdr->entry_size;
   return FALSE;
@@ -284,21 +426,20 @@ my_bool put_header_entry_num(DYN_HEADER *hdr,
   @param offset          offset of the data
 */
 
-my_bool put_header_entry_str(DYN_HEADER *hdr,
-                             void *column_key,
-                             DYNAMIC_COLUMN_VALUE *value,
-                             size_t offset)
+static my_bool put_header_entry_named(DYN_HEADER *hdr,
+                                      void *column_key,
+                                      DYNAMIC_COLUMN_VALUE *value,
+                                      size_t offset)
 {
   LEX_STRING *column_name= (LEX_STRING *)column_key;
   DBUG_ASSERT(column_name->length <= MAX_NAME_LENGTH);
-  hdr->entry[0]= column_name->length;
   DBUG_ASSERT(hdr->name - hdr->nmpool < (long) 0x10000L);
-  int2store(hdr->entry + 1, hdr->name - hdr->nmpool);
+  int2store(hdr->entry, hdr->name - hdr->nmpool);
   memcpy(hdr->name, column_name->str, column_name->length);
   DBUG_ASSERT(hdr->nmpool_size != 0 || column_name->length == 0);
-  if (type_and_offset_store(hdr->entry + 1, hdr->offset_size,
-                            value->type,
-                            offset))
+  if (type_and_offset_store_named(hdr->entry, hdr->offset_size,
+                                  value->type,
+                                  offset))
     return TRUE;
   hdr->entry+= hdr->entry_size;
   hdr->name+= column_name->length;
@@ -307,6 +448,119 @@ my_bool put_header_entry_str(DYN_HEADER *hdr,
 
 
 /**
+  Calculate length of offset field for given data length
+
+  @param data_length     Length of the data segment
+
+  @return number of bytes
+*/
+
+static size_t dynamic_column_offset_bytes_num(size_t data_length)
+{
+  if (data_length < 0x1f)                /* all 1 value is reserved */
+    return 1;
+  if (data_length < 0x1fff)              /* all 1 value is reserved */
+    return 2;
+  if (data_length < 0x1fffff)            /* all 1 value is reserved */
+    return 3;
+  if (data_length < 0x1fffffff)          /* all 1 value is reserved */
+    return 4;
+  return MAX_OFFSET_LENGTH + 1;          /* For an error generation*/
+}
+
+static size_t dynamic_column_offset_bytes_named(size_t data_length)
+{
+  if (data_length < 0xfff)                /* all 1 value is reserved */
+    return 2;
+  if (data_length < 0xfffff)              /* all 1 value is reserved */
+    return 3;
+  if (data_length < 0xfffffff)            /* all 1 value is reserved */
+    return 4;
+  if (data_length < 0xfffffffff)          /* all 1 value is reserved */
+    return 5;
+  return MAX_OFFSET_LENGTH_NM + 1;        /* For an error generation */
+}
+
+/**
+  Read offset and type information from index entry
+
+  @param type            Where to put type info
+  @param offset          Where to put offset info
+  @param place           beginning of the type and offset
+  @param offset_size     Size of offset field in bytes
+*/
+
+static my_bool type_and_offset_read_num(DYNAMIC_COLUMN_TYPE *type,
+                                        size_t *offset,
+                                        uchar *place, size_t offset_size)
+{
+  ulong UNINIT_VAR(val);
+  ulong UNINIT_VAR(lim);
+
+  DBUG_ASSERT(offset_size >= 1 && offset_size <= 4);
+
+  switch (offset_size) {
+  case 1:
+    val= (ulong)place[0];
+    lim= 0x1f;
+    break;
+  case 2:
+    val= uint2korr(place);
+    lim= 0x1fff;
+    break;
+  case 3:
+    val= uint3korr(place);
+    lim= 0x1fffff;
+    break;
+  case 4:
+    val= uint4korr(place);
+    lim= 0x1fffffff;
+    break;
+  default:
+    DBUG_ASSERT(0);                             /* impossible */
+    return 1;
+  }
+  *type= (val & 0x7) + 1;
+  *offset= val >> 3;
+  return (*offset >= lim);
+}
+
+static my_bool type_and_offset_read_named(DYNAMIC_COLUMN_TYPE *type,
+                                          size_t *offset,
+                                          uchar *place, size_t offset_size)
+{
+  ulong UNINIT_VAR(val);
+  ulong UNINIT_VAR(lim);
+  DBUG_ASSERT(offset_size >= 2 && offset_size <= 5);
+
+  switch (offset_size) {
+  case 2:
+    val= uint2korr(place);
+    lim= 0xfff;
+    break;
+  case 3:
+    val= uint3korr(place);
+    lim= 0xfffff;
+    break;
+  case 4:
+    val= uint4korr(place);
+    lim= 0xfffffff;
+    break;
+  case 5:
+    val= uint5korr(place);
+    lim= 0xfffffffff;
+    break;
+  case 1:
+  default:
+    DBUG_ASSERT(0);                             /* impossible */
+    return 1;
+  }
+  *type= (val & 0xf) + 1;
+  *offset= val >> 4;
+  return (*offset >= lim);
+}
+
+/**
   Format descriptor, contain constants and function references for
   format processing
 */
@@ -321,6 +575,9 @@ struct st_service_funcs
   /*size of array element which stores keys */
   uint key_size_in_array;
 
+  /* Maximum data offset size in bytes */
+  size_t max_offset_size;
+
   size_t (*name_size)
     (void *, uint);
   int (*column_sort)
@@ -334,6 +591,11 @@ struct st_service_funcs
                               DYNAMIC_COLUMN_VALUE *value,
                               size_t offset);
   int (*plan_sort)(const void *a, const void *b);
+  size_t (*dynamic_column_offset_bytes)(size_t data_length);
+  my_bool (*type_and_offset_read)(DYNAMIC_COLUMN_TYPE *type,
+                                  size_t *offset,
+                                  uchar *place, size_t offset_size);
+
 };
 
 
@@ -347,23 +609,29 @@ static struct st_service_funcs fmt_data[2]=
     FIXED_HEADER_SIZE,
     COLUMN_NUMBER_SIZE,
     sizeof(uint),
+    MAX_OFFSET_LENGTH,
     &name_size_num,
     &column_sort_num,
     &check_limit_num,
     &set_fixed_header_num,
     &put_header_entry_num,
-    &plan_sort_num
+    &plan_sort_num,
+    &dynamic_column_offset_bytes_num,
+    &type_and_offset_read_num
   },
   {
     FIXED_HEADER_SIZE_NM,
     COLUMN_NAMEPTR_SIZE,
     sizeof(LEX_STRING),
-    &name_size_str,
-    &column_sort_str,
-    &check_limit_str,
-    &set_fixed_header_str,
-    &put_header_entry_str,
-    &plan_sort_str
+    MAX_OFFSET_LENGTH_NM,
+    &name_size_named,
+    &column_sort_named,
+    &check_limit_named,
+    &set_fixed_header_named,
+    &put_header_entry_named,
+    &plan_sort_named,
+    &dynamic_column_offset_bytes_named,
+    &type_and_offset_read_named
   }
 };
 
@@ -377,7 +645,7 @@ static struct st_service_funcs fmt_data[2]=
   @return ER_DYNCOL_* return code
 */
 
-enum enum_dyncol_func_result
+static enum enum_dyncol_func_result
 init_read_hdr(DYN_HEADER *hdr, DYNAMIC_COLUMN *str)
 {
   if (read_fixed_header(hdr, str))
@@ -405,7 +673,7 @@ init_read_hdr(DYN_HEADER *hdr, DYNAMIC_COLUMN *str)
   @retval TRUE  error
 */
 
-static my_bool dynamic_column_init_str(DYNAMIC_COLUMN *str, size_t size)
+static my_bool dynamic_column_init_named(DYNAMIC_COLUMN *str, size_t size)
 {
   DBUG_ASSERT(size != 0);
 
@@ -647,7 +915,8 @@ dynamic_column_sint_read(DYNAMIC_COLUMN_VALUE *store_it_here,
 */
 
 static size_t
-dynamic_column_value_len(DYNAMIC_COLUMN_VALUE *value)
+dynamic_column_value_len(DYNAMIC_COLUMN_VALUE *value,
+                         enum enum_dyncol_format format)
 {
   switch (value->type) {
   case DYN_COL_NULL:
@@ -689,14 +958,22 @@ dynamic_column_value_len(DYNAMIC_COLUMN_VALUE *value)
             decimal_bin_size(precision, scale));
   }
   case DYN_COL_DATETIME:
-    /* date+time in bits: 14 + 4 + 5 + 10 + 6 + 6 + 20 + 1 66bits ~= 9 bytes */
-    return 9;
+    if (format == dyncol_fmt_num || value->x.time_value.second_part)
+      /* date+time in bits: 14 + 4 + 5 + 10 + 6 + 6 + 20 + 1 66bits ~= 9 bytes*/
+      return 9;
+    else
+      return 6;
   case DYN_COL_DATE:
     /* date in dits: 14 + 4 + 5 = 23bits ~= 3bytes*/
     return 3;
   case DYN_COL_TIME:
-    /* time in bits: 10 + 6 + 6 + 20 + 1 = 43bits ~= 6bytes*/
-    return 6;
+    if (format == dyncol_fmt_num || value->x.time_value.second_part)
+      /* time in bits: 10 + 6 + 6 + 20 + 1 = 43bits ~= 6bytes*/
+      return 6;
+    else
+      return 3;
+  case DYN_COL_DYNCOL:
+    return value->x.string.value.length;
   }
   DBUG_ASSERT(0);
   return 0;
@@ -765,6 +1042,22 @@ dynamic_column_string_store(DYNAMIC_COLUMN *str, LEX_STRING *string,
   return ER_DYNCOL_OK;
 }
 
+/**
+  Append the string with given string value.
+
+  @param str             the string where to put the value
+  @param val             the value to put in the string
+
+  @return ER_DYNCOL_* return code
+*/
+
+static enum enum_dyncol_func_result
+dynamic_column_dyncol_store(DYNAMIC_COLUMN *str, LEX_STRING *string)
+{
+  if (dynstr_append_mem(str, string->str, string->length))
+    return ER_DYNCOL_RESOURCE;
+  return ER_DYNCOL_OK;
+}
 
 /**
   Read string value of given length from the packed string
@@ -793,6 +1086,26 @@ dynamic_column_string_read(DYNAMIC_COLUMN_VALUE *store_it_here,
   return ER_DYNCOL_OK;
 }
 
+/**
+  Read Dynamic columns packet string value of given length
+  from the packed string
+
+  @param store_it_here   The structure to store the value
+  @param data            The packed string which should be read
+  @param length          The length (in bytes) of the value in nthe string
+
+  @return ER_DYNCOL_* return code
+*/
+
+static enum enum_dyncol_func_result
+dynamic_column_dyncol_read(DYNAMIC_COLUMN_VALUE *store_it_here,
+                           uchar *data, size_t length)
+{
+  store_it_here->x.string.charset= &my_charset_bin;
+  store_it_here->x.string.value.length= length;
+  store_it_here->x.string.value.str= (char*) data;
+  return ER_DYNCOL_OK;
+}
 
 /**
   Append the string with given decimal value.
@@ -835,7 +1148,7 @@ dynamic_column_decimal_store(DYNAMIC_COLUMN *str,
   @param value           The value structure which sould be setup.
 */
 
-void dynamic_column_prepare_decimal(DYNAMIC_COLUMN_VALUE *value)
+void mariadb_dyncol_prepare_decimal(DYNAMIC_COLUMN_VALUE *value)
 {
   value->x.decimal.value.buf= value->x.decimal.buffer;
   value->x.decimal.value.len= DECIMAL_BUFF_LENGTH;
@@ -844,6 +1157,12 @@ void dynamic_column_prepare_decimal(DYNAMIC_COLUMN_VALUE *value)
   decimal_make_zero(&value->x.decimal.value);
 }
 
+void dynamic_column_prepare_decimal(DYNAMIC_COLUMN_VALUE *value)
+{
+  mariadb_dyncol_prepare_decimal(value);
+}
+
+
 
 /**
   Read decimal value of given length from the string
@@ -899,7 +1218,8 @@ dynamic_column_decimal_read(DYNAMIC_COLUMN_VALUE *store_it_here,
 */
 
 static enum enum_dyncol_func_result
-dynamic_column_date_time_store(DYNAMIC_COLUMN *str, MYSQL_TIME *value)
+dynamic_column_date_time_store(DYNAMIC_COLUMN *str, MYSQL_TIME *value,
+                               enum enum_dyncol_format format)
 {
   enum enum_dyncol_func_result rc;
   /*
@@ -908,7 +1228,7 @@ dynamic_column_date_time_store(DYNAMIC_COLUMN *str, MYSQL_TIME *value)
     <123456><123456><123456><123456><123456><123456><123456><123456><123456>
   */
   if ((rc= dynamic_column_date_store(str, value)) ||
-      (rc= dynamic_column_time_store(str, value)))
+      (rc= dynamic_column_time_store(str, value, format)))
     return rc;
   return ER_DYNCOL_OK;
 }
@@ -934,11 +1254,12 @@ dynamic_column_date_time_read(DYNAMIC_COLUMN_VALUE *store_it_here,
      12345678901234123412345     1123456789012345612345612345678901234567890
     <123456><123456><123456><123456><123456><123456><123456><123456><123456>
   */
-  if (length != 9)
+  if (length != 9 && length != 6)
     goto err;
   store_it_here->x.time_value.time_type= MYSQL_TIMESTAMP_DATETIME;
   if ((rc= dynamic_column_date_read_internal(store_it_here, data, 3)) ||
-      (rc= dynamic_column_time_read_internal(store_it_here, data + 3, 6)))
+      (rc= dynamic_column_time_read_internal(store_it_here, data + 3,
+                                             length - 3)))
     goto err;
   return ER_DYNCOL_OK;
 
@@ -958,7 +1279,8 @@ err:
 */
 
 static enum enum_dyncol_func_result
-dynamic_column_time_store(DYNAMIC_COLUMN *str, MYSQL_TIME *value)
+dynamic_column_time_store(DYNAMIC_COLUMN *str, MYSQL_TIME *value,
+                          enum enum_dyncol_format format)
 {
   uchar *buf;
   if (dynstr_realloc(str, 6))
@@ -980,19 +1302,35 @@ dynamic_column_time_store(DYNAMIC_COLUMN *str, MYSQL_TIME *value)
   DBUG_ASSERT(value->minute <= 59);
   DBUG_ASSERT(value->second <= 59);
   DBUG_ASSERT(value->second_part <= 999999);
+  if (format == dyncol_fmt_num || value->second_part)
+  {
   /*
     00000!<-hours--><min-><sec-><---microseconds--->
          1123456789012345612345612345678901234567890
     <123456><123456><123456><123456><123456><123456>
   */
-  buf[0]= (value->second_part & 0xff);
-  buf[1]= ((value->second_part & 0xff00) >> 8);
-  buf[2]= (uchar)(((value->second & 0xf) << 4) |
-           ((value->second_part & 0xf0000) >> 16));
-  buf[3]= ((value->minute << 2) | ((value->second & 0x30) >> 4));
-  buf[4]= (value->hour & 0xff);
-  buf[5]= ((value->neg ? 0x4 : 0) | (value->hour >> 8));
-  str->length+= 6;
+    buf[0]= (value->second_part & 0xff);
+    buf[1]= ((value->second_part & 0xff00) >> 8);
+    buf[2]= (uchar)(((value->second & 0xf) << 4) |
+                    ((value->second_part & 0xf0000) >> 16));
+    buf[3]= ((value->minute << 2) | ((value->second & 0x30) >> 4));
+    buf[4]= (value->hour & 0xff);
+    buf[5]= ((value->neg ? 0x4 : 0) | (value->hour >> 8));
+    str->length+= 6;
+  }
+  else
+  {
+  /*
+     !<-hours--><min-><sec->
+     11234567890123456123456
+    <123456><123456><123456>
+  */
+    buf[0]= (value->second) | ((value->minute & 0x3) << 6);
+    buf[1]= (value->minute >> 2) | ((value->hour & 0xf) << 4);
+    buf[2]= (value->hour >> 4) | (value->neg ? 0x80 : 0);
+    str->length+= 3;
+  }
+
   return ER_DYNCOL_OK;
 }
 
@@ -1031,21 +1369,37 @@ static enum enum_dyncol_func_result
 dynamic_column_time_read_internal(DYNAMIC_COLUMN_VALUE *store_it_here,
                                   uchar *data, size_t length)
 {
-  if (length != 6)
+  if (length != 6 && length != 3)
     goto err;
-  /*
-    00000!<-hours--><min-><sec-><---microseconds--->
-         1123456789012345612345612345678901234567890
-    <123456><123456><123456><123456><123456><123456>
-  */
-  store_it_here->x.time_value.second_part= (data[0] |
-                                          (data[1] << 8) |
-                                          ((data[2] & 0xf) << 16));
-  store_it_here->x.time_value.second= ((data[2] >> 4) |
-                                     ((data[3] & 0x3) << 4));
-  store_it_here->x.time_value.minute= (data[3] >> 2);
-  store_it_here->x.time_value.hour= (((((uint)data[5]) & 0x3 ) << 8) | data[4]);
-  store_it_here->x.time_value.neg= ((data[5] & 0x4) ? 1 : 0);
+  if (length == 6)
+  {
+    /*
+      00000!<-hours--><min-><sec-><---microseconds--->
+      1123456789012345612345612345678901234567890
+      <123456><123456><123456><123456><123456><123456>
+    */
+    store_it_here->x.time_value.second_part= (data[0] |
+                                              (data[1] << 8) |
+                                              ((data[2] & 0xf) << 16));
+    store_it_here->x.time_value.second= ((data[2] >> 4) |
+                                         ((data[3] & 0x3) << 4));
+    store_it_here->x.time_value.minute= (data[3] >> 2);
+    store_it_here->x.time_value.hour= (((((uint)data[5]) & 0x3 ) << 8) | data[4]);
+    store_it_here->x.time_value.neg= ((data[5] & 0x4) ? 1 : 0);
+  }
+  else
+  {
+    /*
+     !<-hours--><min-><sec->
+     11234567890123456123456
+    <123456><123456><123456>
+   */
+    store_it_here->x.time_value.second_part= 0;
+    store_it_here->x.time_value.second= (data[0] & 0x3f);
+    store_it_here->x.time_value.minute= (data[0] >> 6) | ((data[1] & 0xf) << 2);
+    store_it_here->x.time_value.hour= (data[1] >> 4) | ((data[2] & 0x3f) << 4);
+    store_it_here->x.time_value.neg= ((data[2] & 0x80) ? 1 : 0);
+  }
   if (store_it_here->x.time_value.second > 59 ||
       store_it_here->x.time_value.minute > 59 ||
       store_it_here->x.time_value.hour > 838 ||
@@ -1170,7 +1524,8 @@ err:
 */
 
 static enum enum_dyncol_func_result
-data_store(DYNAMIC_COLUMN *str, DYNAMIC_COLUMN_VALUE *value)
+data_store(DYNAMIC_COLUMN *str, DYNAMIC_COLUMN_VALUE *value,
+           enum enum_dyncol_format format)
 {
   switch (value->type) {
   case DYN_COL_INT:
@@ -1186,13 +1541,15 @@ data_store(DYNAMIC_COLUMN *str, DYNAMIC_COLUMN_VALUE *value)
     return dynamic_column_decimal_store(str, &value->x.decimal.value);
   case DYN_COL_DATETIME:
     /* date+time in bits: 14 + 4 + 5 + 5 + 6 + 6 40bits = 5 bytes */
-    return dynamic_column_date_time_store(str, &value->x.time_value);
+    return dynamic_column_date_time_store(str, &value->x.time_value, format);
   case DYN_COL_DATE:
     /* date in dits: 14 + 4 + 5 = 23bits ~= 3bytes*/
     return dynamic_column_date_store(str, &value->x.time_value);
   case DYN_COL_TIME:
     /* time in bits: 5 + 6 + 6 = 17bits ~= 3bytes*/
-    return dynamic_column_time_store(str, &value->x.time_value);
+    return dynamic_column_time_store(str, &value->x.time_value, format);
+  case DYN_COL_DYNCOL:
+    return dynamic_column_dyncol_store(str, &value->x.string.value);
   case DYN_COL_NULL:
     break;                                      /* Impossible */
   }
@@ -1202,117 +1559,6 @@ data_store(DYNAMIC_COLUMN *str, DYNAMIC_COLUMN_VALUE *value)
 
 
 /**
-  Calculate length of offset field for given data length
-
-  @param data_length     Length of the data segment
-
-  @return number of bytes
-*/
-
-static size_t dynamic_column_offset_bytes(size_t data_length)
-{
-  if (data_length < 0x1f)                /* all 1 value is reserved */
-    return 1;
-  if (data_length < 0x1fff)              /* all 1 value is reserved */
-    return 2;
-  if (data_length < 0x1fffff)            /* all 1 value is reserved */
-    return 3;
-  if (data_length < 0x1fffffff)          /* all 1 value is reserved */
-    return 4;
-  return MAX_OFFSET_LENGTH;              /* For future */
-}
-
-/**
-  Store offset and type information in the given place
-
-  @param place           Beginning of the index entry
-  @param offset_size     Size of offset field in bytes
-  @param type            Type to be written
-  @param offset          Offset to be written
-*/
-
-static my_bool type_and_offset_store(uchar *place, size_t offset_size,
-                                     DYNAMIC_COLUMN_TYPE type,
-                                     size_t offset)
-{
-  ulong val = (((ulong) offset) << 3) | (type - 1);
-  DBUG_ASSERT(type != DYN_COL_NULL);
-  DBUG_ASSERT(((type - 1) & (~7)) == 0); /* fit in 3 bits */
-
-  /* Index entry starts with column number; Jump over it */
-  place+= COLUMN_NUMBER_SIZE;
-
-  switch (offset_size) {
-  case 1:
-    if (offset >= 0x1f)          /* all 1 value is reserved */
-      return TRUE;
-    place[0]= (uchar)val;
-    break;
-  case 2:
-    if (offset >= 0x1fff)        /* all 1 value is reserved */
-      return TRUE;
-    int2store(place, val);
-    break;
-  case 3:
-    if (offset >= 0x1fffff)      /* all 1 value is reserved */
-      return TRUE;
-    int3store(place, val);
-    break;
-  case 4:
-    if (offset >= 0x1fffffff)    /* all 1 value is reserved */
-      return TRUE;
-    int4store(place, val);
-    break;
-  default:
-      return TRUE;
-  }
-  return FALSE;
-}
-
-
-/**
-  Read offset and type information from index entry
-
-  @param type            Where to put type info
-  @param offset          Where to put offset info
-  @param place           beginning of the type and offset
-  @param offset_size     Size of offset field in bytes
-*/
-
-static my_bool type_and_offset_read(DYNAMIC_COLUMN_TYPE *type,
-                                    size_t *offset,
-                                    uchar *place, size_t offset_size)
-{
-  ulong UNINIT_VAR(val);
-  ulong UNINIT_VAR(lim);
-
-  switch (offset_size) {
-  case 1:
-    val= (ulong)place[0];
-    lim= 0x1f;
-    break;
-  case 2:
-    val= uint2korr(place);
-    lim= 0x1fff;
-    break;
-  case 3:
-    val= uint3korr(place);
-    lim= 0x1fffff;
-    break;
-  case 4:
-    val= uint4korr(place);
-    lim= 0x1fffffff;
-    break;
-  default:
-    DBUG_ASSERT(0);                             /* impossible */
-  }
-  *type= (val & 0x7) + 1;
-  *offset= val >> 3;
-  return (*offset >= lim);
-}
-
-
-/**
   Write information to the fixed header
 
   @param str             String where to write the header
@@ -1325,7 +1571,7 @@ static void set_fixed_header(DYNAMIC_COLUMN *str,
                              uint column_count)
 {
   DBUG_ASSERT(column_count <= 0xffff);
-  DBUG_ASSERT(offset_size <= 4);
+  DBUG_ASSERT(offset_size <= MAX_OFFSET_LENGTH);
   str->str[0]= ((str->str[0] & ~DYNCOL_FLG_OFFSET) |
                 (offset_size - 1));             /* size of offset */
   int2store(str->str + 1, column_count);        /* columns number */
@@ -1364,12 +1610,12 @@ dynamic_new_column_store(DYNAMIC_COLUMN *str,
     return ER_DYNCOL_RESOURCE;
   if (new_str)
   {
-    if (dynamic_column_init_str(str,
-                                fmt->fixed_hdr +
-                                hdr->header_size +
-                                hdr->nmpool_size +
-                                hdr->data_size +
-                                DYNCOL_SYZERESERVE))
+    if (dynamic_column_init_named(str,
+                                  fmt->fixed_hdr +
+                                  hdr->header_size +
+                                  hdr->nmpool_size +
+                                  hdr->data_size +
+                                  DYNCOL_SYZERESERVE))
       goto err;
   }
   else
@@ -1438,7 +1684,7 @@ dynamic_new_column_store(DYNAMIC_COLUMN *str,
       }
 
       /* Store value in 'str + str->length' and increase str->length */
-      if ((rc= data_store(str, values + ord)))
+      if ((rc= data_store(str, values + ord, hdr->format)))
         goto err;
     }
   }
@@ -1476,15 +1722,19 @@ calc_var_sizes(DYN_HEADER *hdr,
     {
       size_t tmp;
       hdr->column_count++;
-      hdr->data_size+= (tmp= dynamic_column_value_len(values + i));
+      hdr->data_size+= (tmp= dynamic_column_value_len(values + i,
+                        hdr->format));
       if (tmp == (size_t) ~0)
         return ER_DYNCOL_DATA;
       hdr->nmpool_size+= (*fmt->name_size)(column_keys, i);
     }
   }
-  /* We can handle data up to 1fffffff = 536870911 bytes now */
-  if ((hdr->offset_size= dynamic_column_offset_bytes(hdr->data_size)) >=
-      MAX_OFFSET_LENGTH)
+  /*
+    We can handle data up to 0x1fffffff (old format) and
+    0xfffffffff (new format) bytes now.
+  */
+  if ((hdr->offset_size= fmt->dynamic_column_offset_bytes(hdr->data_size)) >=
+      fmt->max_offset_size)
     return ER_DYNCOL_LIMIT;
 
   /* header entry is column number or string pointer + offset & type */
@@ -1506,7 +1756,7 @@ calc_var_sizes(DYN_HEADER *hdr,
   @return ER_DYNCOL_* return code
 */
 
-enum enum_dyncol_func_result
+static enum enum_dyncol_func_result
 dynamic_column_create_many_internal_fmt(DYNAMIC_COLUMN *str,
                                         uint column_count,
                                         void *column_keys,
@@ -1563,24 +1813,49 @@ dynamic_column_create_many(DYNAMIC_COLUMN *str,
 
   @param str             String where to write the data
   @param column_count    Number of columns in the arrays
+  @param column_numbers  Array of columns numbers
+  @param values          Array of columns values
+  @param new_string      True if we need allocate new string
+
+  @return ER_DYNCOL_* return code
+*/
+
+enum enum_dyncol_func_result
+mariadb_dyncol_create_many(DYNAMIC_COLUMN *str,
+                           uint column_count,
+                           uint *column_numbers,
+                           DYNAMIC_COLUMN_VALUE *values,
+                           my_bool new_string)
+{
+  DBUG_ENTER("mariadb_dyncol_create_many");
+  DBUG_RETURN(dynamic_column_create_many_internal_fmt(str, column_count,
+                                                      column_numbers, values,
+                                                      new_string, FALSE));
+}
+
+/**
+  Create packed string which contains given columns
+
+  @param str             String where to write the data
+  @param column_count    Number of columns in the arrays
   @param column_keys     Array of columns keys
   @param values          Array of columns value
-  @param names           use string names as keys
+  @param new_string      True if we need allocate new string
 
   @return ER_DYNCOL_* return code
 */
 
 enum enum_dyncol_func_result
-dynamic_column_create_many_fmt(DYNAMIC_COLUMN *str,
-                               uint column_count,
-                               uchar *column_keys,
-                               DYNAMIC_COLUMN_VALUE *values,
-                               my_bool names)
+mariadb_dyncol_create_many_named(DYNAMIC_COLUMN *str,
+                                 uint column_count,
+                                 LEX_STRING *column_keys,
+                                 DYNAMIC_COLUMN_VALUE *values,
+                                 my_bool new_string)
 {
-  DBUG_ENTER("dynamic_column_create_many");
+  DBUG_ENTER("mariadb_dyncol_create_many_named");
   DBUG_RETURN(dynamic_column_create_many_internal_fmt(str, column_count,
                                                       column_keys, values,
-                                                      TRUE, names));
+                                                      new_string, TRUE));
 }
 
 /**
@@ -1622,13 +1897,13 @@ static size_t get_length_interval(uchar *entry, uchar *entry_next,
   DYNAMIC_COLUMN_TYPE type, type_next;
   DBUG_ASSERT(entry < entry_next);
 
-  if (type_and_offset_read(&type, &offset, entry + COLUMN_NUMBER_SIZE,
-                           offset_size))
+  if (type_and_offset_read_num(&type, &offset, entry + COLUMN_NUMBER_SIZE,
+                               offset_size))
       return DYNCOL_OFFSET_ERROR;
   if (entry_next >= header_end)
     return (last_offset - offset);
-  if (type_and_offset_read(&type_next, &offset_next,
-                           entry_next + COLUMN_NUMBER_SIZE, offset_size))
+  if (type_and_offset_read_num(&type_next, &offset_next,
+                               entry_next + COLUMN_NUMBER_SIZE, offset_size))
     return DYNCOL_OFFSET_ERROR;
   return (offset_next - offset);
 }
@@ -1653,13 +1928,15 @@ static size_t hdr_interval_length(DYN_HEADER *hdr, uchar *next_entry)
   DBUG_ASSERT(hdr->entry >= hdr->header);
   DBUG_ASSERT(next_entry <= hdr->header + hdr->header_size);
 
-  if (type_and_offset_read(&hdr->type, &hdr->offset,
-                           hdr->entry + fmt->fixed_hdr_entry, hdr->offset_size))
+  if ((*fmt->type_and_offset_read)(&hdr->type, &hdr->offset,
+                                   hdr->entry + fmt->fixed_hdr_entry,
+                                   hdr->offset_size))
     return DYNCOL_OFFSET_ERROR;
   if (next_entry == hdr->header + hdr->header_size)
     return hdr->data_size - hdr->offset;
-  if (type_and_offset_read(&next_entry_type, &next_entry_offset,
-                           next_entry + fmt->fixed_hdr_entry, hdr->offset_size))
+  if ((*fmt->type_and_offset_read)(&next_entry_type, &next_entry_offset,
+                                   next_entry + fmt->fixed_hdr_entry,
+                                   hdr->offset_size))
     return DYNCOL_OFFSET_ERROR;
   return (next_entry_offset - hdr->offset);
 }
@@ -1688,7 +1965,7 @@ static int header_compar_num(const void *a, const void *b)
 static uchar *find_entry_num(DYN_HEADER *hdr, uint key)
 {
   uchar header_entry[2+4];
-  DBUG_ASSERT(hdr->format == DYNCOL_FMT_NUM);
+  DBUG_ASSERT(hdr->format == dyncol_fmt_num);
   int2store(header_entry, key);
   return hdr->entry= bsearch(header_entry, hdr->header,
                              (size_t)hdr->column_count,
@@ -1697,6 +1974,39 @@ static uchar *find_entry_num(DYN_HEADER *hdr, uint key)
 
 
 /**
+  Read name from header entry
+
+  @param hdr             descriptor of dynamic column record
+  @param entry           pointer to the header entry
+  @param name            where to put name
+
+  @return 0 ok
+  @return 1 error in data
+*/
+
+static my_bool read_name(DYN_HEADER *hdr, uchar *entry, LEX_STRING *name)
+{
+  size_t nmoffset= uint2korr(entry);
+  uchar *next_entry= entry + hdr->entry_size;
+
+  if (nmoffset > hdr->nmpool_size)
+    return 1;
+
+  name->str= (char *)hdr->nmpool + nmoffset;
+  if (next_entry == hdr->header + hdr->header_size)
+    name->length= hdr->nmpool_size - nmoffset;
+  else
+  {
+    size_t next_nmoffset= uint2korr(next_entry);
+    if (next_nmoffset > hdr->nmpool_size)
+      return 1;
+    name->length= next_nmoffset - nmoffset;
+  }
+  return 0;
+}
+
+
+/**
   Find entry in the names format header by the column number
 
   @param hdr             descriptor of dynamic column record
@@ -1704,22 +2014,24 @@ static uchar *find_entry_num(DYN_HEADER *hdr, uint key)
 
   @return pointer to the entry or NULL
 */
-static uchar *find_entry_str(DYN_HEADER *hdr, LEX_STRING *key)
+static uchar *find_entry_named(DYN_HEADER *hdr, LEX_STRING *key)
 {
   uchar *min= hdr->header;
   uchar *max= hdr->header + (hdr->column_count - 1) * hdr->entry_size;
   uchar *mid;
-  DBUG_ASSERT(hdr->format == DYNCOL_FMT_STR);
+  DBUG_ASSERT(hdr->format == dyncol_fmt_str);
   DBUG_ASSERT(hdr->nmpool != NULL);
   while (max >= min)
   {
-    uint len;
+    LEX_STRING name;
     int cmp;
-    mid= hdr->header + ((min - hdr->header) + (max - hdr->header)) / 2 / hdr->entry_size * hdr->entry_size;
-    len= mid[0];
-    cmp= len - key->length;
-    if (cmp == 0)
-      cmp= memcmp(hdr->nmpool +  uint2korr(mid + 1), key->str, len);
+    mid= hdr->header + ((min - hdr->header) +
+                        (max - hdr->header)) /
+      2 /
+      hdr->entry_size * hdr->entry_size;
+    if (read_name(hdr, mid, &name))
+      return NULL;
+    cmp= mariadb_dyncol_column_cmp_named(&name, key);
     if (cmp < 0)
       min= mid + hdr->entry_size;
     else if (cmp > 0)
@@ -1773,7 +2085,7 @@ find_column(DYN_HEADER *hdr, uint numkey, LEX_STRING *strkey)
     return TRUE;
 
   /* fix key */
-  if (hdr->format == DYNCOL_FMT_NUM && strkey != NULL)
+  if (hdr->format == dyncol_fmt_num && strkey != NULL)
   {
     char *end;
     numkey= (uint) strtoul(strkey->str, &end, 10);
@@ -1784,16 +2096,16 @@ find_column(DYN_HEADER *hdr, uint numkey, LEX_STRING *strkey)
       return 0;
     }
   }
-  else if (hdr->format == DYNCOL_FMT_STR && strkey == NULL)
+  else if (hdr->format == dyncol_fmt_str && strkey == NULL)
   {
     nmkey.str= backwritenum(nmkeybuff + sizeof(nmkeybuff), numkey);
     nmkey.length= (nmkeybuff + sizeof(nmkeybuff)) - nmkey.str;
     strkey= &nmkey;
   }
-  if (hdr->format == DYNCOL_FMT_NUM)
+  if (hdr->format == dyncol_fmt_num)
     hdr->entry= find_entry_num(hdr, numkey);
   else
-    hdr->entry= find_entry_str(hdr, strkey);
+    hdr->entry= find_entry_named(hdr, strkey);
 
   if (!hdr->entry)
   {
@@ -1837,14 +2149,15 @@ static inline my_bool read_fixed_header(DYN_HEADER *hdr,
       (str->str[0] & (~DYNCOL_FLG_KNOWN)))
     return 1;
   hdr->format= ((str->str[0] & DYNCOL_FLG_NAMES) ?
-                DYNCOL_FMT_STR:
-                DYNCOL_FMT_NUM);
+                dyncol_fmt_str:
+                dyncol_fmt_num);
   if ((str->length < fmt_data[hdr->format].fixed_hdr))
     return 1;                                   /* Wrong header */
-  hdr->offset_size= (str->str[0] & DYNCOL_FLG_OFFSET) + 1;
+  hdr->offset_size= (str->str[0] & DYNCOL_FLG_OFFSET) + 1 +
+    (hdr->format == dyncol_fmt_str ? 1 : 0);
   hdr->column_count= uint2korr(str->str + 1);
-  if (hdr->format == DYNCOL_FMT_STR)
-    hdr->nmpool_size= uint2korr(str->str + 3);
+  if (hdr->format == dyncol_fmt_str)
+    hdr->nmpool_size= uint2korr(str->str + 3); // only 2 bytes supported for now
   else
     hdr->nmpool_size= 0;
   return 0;
@@ -1868,6 +2181,13 @@ dynamic_column_get(DYNAMIC_COLUMN *str, uint column_nr,
   return dynamic_column_get_internal(str, store_it_here, column_nr, NULL);
 }
 
+enum enum_dyncol_func_result
+mariadb_dyncol_get(DYNAMIC_COLUMN *str, uint column_nr,
+                   DYNAMIC_COLUMN_VALUE *store_it_here)
+{
+  return dynamic_column_get_internal(str, store_it_here, column_nr, NULL);
+}
+
 
 /**
   Get dynamic column value by name
@@ -1880,39 +2200,14 @@ dynamic_column_get(DYNAMIC_COLUMN *str, uint column_nr,
 */
 
 enum enum_dyncol_func_result
-dynamic_column_get_str(DYNAMIC_COLUMN *str, LEX_STRING *name,
-                           DYNAMIC_COLUMN_VALUE *store_it_here)
+mariadb_dyncol_get_named(DYNAMIC_COLUMN *str, LEX_STRING *name,
+                         DYNAMIC_COLUMN_VALUE *store_it_here)
 {
   DBUG_ASSERT(name != NULL);
   return dynamic_column_get_internal(str, store_it_here, 0, name);
 }
 
 
-/**
-  Get dynamic column value by number or name
-
-  @param str             The packed string to extract the column
-  @param key             Name or number of column to fetch
-                         (depends on string_key)
-  @param store_it_here   Where to store the extracted value
-  @param string_key      True if we gave pointer to LEX_STRING.
-
-  @return ER_DYNCOL_* return code
-*/
-
-enum enum_dyncol_func_result
-dynamic_column_get_fmt(DYNAMIC_COLUMN *str, void *key,
-                           DYNAMIC_COLUMN_VALUE *store_it_here,
-                           my_bool string_key)
-{
-  DBUG_ASSERT(key != NULL);
-  if (string_key)
-    return dynamic_column_get_internal(str, store_it_here,
-                                       0, (LEX_STRING *)key);
-  return dynamic_column_get_internal(str, store_it_here,
-                                     *((uint *)key), NULL);
-}
-
 static enum enum_dyncol_func_result
 dynamic_column_get_value(DYN_HEADER *hdr, DYNAMIC_COLUMN_VALUE *store_it_here)
 {
@@ -1946,6 +2241,9 @@ dynamic_column_get_value(DYN_HEADER *hdr, DYNAMIC_COLUMN_VALUE *store_it_here)
   case DYN_COL_NULL:
     rc= ER_DYNCOL_OK;
     break;
+  case DYN_COL_DYNCOL:
+    rc= dynamic_column_dyncol_read(store_it_here, hdr->data, hdr->length);
+    break;
   default:
     rc= ER_DYNCOL_FORMAT;
     store_it_here->type= DYN_COL_NULL;
@@ -2012,6 +2310,11 @@ dynamic_column_exists(DYNAMIC_COLUMN *str, uint column_nr)
   return dynamic_column_exists_internal(str, column_nr, NULL);
 }
 
+enum enum_dyncol_func_result
+mariadb_dyncol_exists(DYNAMIC_COLUMN *str, uint column_nr)
+{
+  return dynamic_column_exists_internal(str, column_nr, NULL);
+}
 
 /**
   Check existence of the column in the packed string (by name)
@@ -2023,7 +2326,7 @@ dynamic_column_exists(DYNAMIC_COLUMN *str, uint column_nr)
 */
 
 enum enum_dyncol_func_result
-dynamic_column_exists_str(DYNAMIC_COLUMN *str, LEX_STRING *name)
+mariadb_dyncol_exists_named(DYNAMIC_COLUMN *str, LEX_STRING *name)
 {
   DBUG_ASSERT(name != NULL);
   return dynamic_column_exists_internal(str, 0, name);
@@ -2034,27 +2337,6 @@ dynamic_column_exists_str(DYNAMIC_COLUMN *str, LEX_STRING *name)
   Check existence of the column in the packed string (by name of number)
 
   @param str             The packed string to check the column
-  @param key             Name or number of column to fetch
-                         (depends on string_key)
-  @param string_key      True if we gave pointer to LEX_STRING.
-
-  @return ER_DYNCOL_* return code
-*/
-
-enum enum_dyncol_func_result
-dynamic_column_exists_fmt(DYNAMIC_COLUMN *str, void *key, my_bool string_key)
-{
-  DBUG_ASSERT(key != NULL);
-  if (string_key)
-    return dynamic_column_exists_internal(str, 0, (LEX_STRING *) key);
-  return dynamic_column_exists_internal(str, *((uint *)key), NULL);
-}
-
-
-/**
-  Check existence of the column in the packed string (by name of number)
-
-  @param str             The packed string to check the column
   @param num_key         Number of the column to fetch (if strkey is NULL)
   @param str_key         Name of the column to fetch (or NULL)
 
@@ -2093,10 +2375,15 @@ dynamic_column_exists_internal(DYNAMIC_COLUMN *str, uint num_key,
 
   @return ER_DYNCOL_* return code
 */
-
 enum enum_dyncol_func_result
 dynamic_column_list(DYNAMIC_COLUMN *str, DYNAMIC_ARRAY *array_of_uint)
 {
+  return mariadb_dyncol_list(str, array_of_uint);
+}
+
+enum enum_dyncol_func_result
+mariadb_dyncol_list(DYNAMIC_COLUMN *str, DYNAMIC_ARRAY *array_of_uint)
+{
   DYN_HEADER header;
   uchar *read;
   uint i;
@@ -2109,7 +2396,7 @@ dynamic_column_list(DYNAMIC_COLUMN *str, DYNAMIC_ARRAY *array_of_uint)
   if ((rc= init_read_hdr(&header, str)) < 0)
     return rc;
 
-  if (header.format != DYNCOL_FMT_NUM)
+  if (header.format != dyncol_fmt_num)
     return ER_DYNCOL_FORMAT;
 
   if (header.entry_size * header.column_count + FIXED_HEADER_SIZE >
@@ -2135,21 +2422,24 @@ dynamic_column_list(DYNAMIC_COLUMN *str, DYNAMIC_ARRAY *array_of_uint)
   List not-null columns in the packed string (any format)
 
   @param str             The packed string
-  @param array_of_lexstr Where to put reference on created array
+  @param count           Number of names in the list
+  @param names           Where to put names list (should be freed)
 
   @return ER_DYNCOL_* return code
 */
 
 enum enum_dyncol_func_result
-dynamic_column_list_str(DYNAMIC_COLUMN *str, DYNAMIC_ARRAY *array_of_lexstr)
+mariadb_dyncol_list_named(DYNAMIC_COLUMN *str, uint *count, LEX_STRING **names)
 {
   DYN_HEADER header;
   uchar *read;
+  char *pool;
   struct st_service_funcs *fmt;
   uint i;
   enum enum_dyncol_func_result rc;
 
-  bzero(array_of_lexstr, sizeof(*array_of_lexstr)); /* In case of errors */
+  (*names)= 0; (*count)= 0;
+
   if (str->length == 0)
     return ER_DYNCOL_OK;                        /* no columns */
 
@@ -2162,36 +2452,41 @@ dynamic_column_list_str(DYNAMIC_COLUMN *str, DYNAMIC_ARRAY *array_of_lexstr)
       str->length)
     return ER_DYNCOL_FORMAT;
 
-  if (init_dynamic_array(array_of_lexstr, sizeof(LEX_STRING),
-                         header.column_count, 0))
+  if (header.format == dyncol_fmt_num)
+    *names= my_malloc(sizeof(LEX_STRING) * header.column_count +
+                      DYNCOL_NUM_CHAR * header.column_count, MYF(0));
+  else
+    *names= my_malloc(sizeof(LEX_STRING) * header.column_count +
+                      header.nmpool_size + header.column_count, MYF(0));
+  if (!(*names))
     return ER_DYNCOL_RESOURCE;
+  pool= ((char *)(*names)) + sizeof(LEX_STRING) * header.column_count;
 
   for (i= 0, read= header.header;
        i < header.column_count;
        i++, read+= header.entry_size)
   {
-    LEX_STRING tmp;
-    if (header.format == DYNCOL_FMT_NUM)
+    if (header.format == dyncol_fmt_num)
     {
       uint nm= uint2korr(read);
-      tmp.str= my_malloc(DYNCOL_NUM_CHAR, MYF(0));
-      if (!tmp.str)
-        return ER_DYNCOL_RESOURCE;
-      tmp.length= snprintf(tmp.str, DYNCOL_NUM_CHAR, "%u", nm);
+      (*names)[i].str= pool;
+      pool+= DYNCOL_NUM_CHAR;
+      (*names)[i].length=
+        longlong2str(nm, (*names)[i].str, 10) - (*names)[i].str;
     }
     else
     {
-      tmp.length= read[0];
-      tmp.str= my_malloc(tmp.length + 1, MYF(0));
-      if(!tmp.str)
-        return ER_DYNCOL_RESOURCE;
-      memcpy(tmp.str, (const void *)header.nmpool + uint2korr(read + 1),
-             tmp.length);
-      tmp.str[tmp.length]= '\0'; // just for safety
+      LEX_STRING tmp;
+      if (read_name(&header, read, &tmp))
+        return ER_DYNCOL_FORMAT;
+      (*names)[i].length= tmp.length;
+      (*names)[i].str= pool;
+      pool+= tmp.length + 1;
+      memcpy((*names)[i].str, (const void *)tmp.str, tmp.length);
+      (*names)[i].str[tmp.length]= '\0'; // just for safety
     }
-    /* Insert can't never fail as it's pre-allocated above */
-    (void) insert_dynamic(array_of_lexstr, (uchar *)&tmp);
   }
+  (*count)= header.column_count;
   return ER_DYNCOL_OK;
 }
 
@@ -2211,15 +2506,14 @@ static my_bool
 find_place(DYN_HEADER *hdr, void *key, my_bool string_keys)
 {
   uint mid, start, end, val;
-  int flag;
+  int UNINIT_VAR(flag);
   LEX_STRING str;
   char buff[DYNCOL_NUM_CHAR];
-  my_bool need_conversion= ((string_keys ? DYNCOL_FMT_STR : DYNCOL_FMT_NUM) !=
+  my_bool need_conversion= ((string_keys ? dyncol_fmt_str : dyncol_fmt_num) !=
                             hdr->format);
-  LINT_INIT(flag);                              /* 100 % safe */
   /* new format can't be numeric if the old one is names */
   DBUG_ASSERT(string_keys ||
-              hdr->format == DYNCOL_FMT_NUM);
+              hdr->format == dyncol_fmt_num);
 
   start= 0;
   end= hdr->column_count -1;
@@ -2243,13 +2537,11 @@ find_place(DYN_HEADER *hdr, void *key, my_bool string_keys)
       }
       else
       {
-        DBUG_ASSERT(hdr->format == DYNCOL_FMT_STR);
-        str.length= hdr->entry[0];
-        str.str= (char *)hdr->nmpool + uint2korr(hdr->entry + 1);
+        DBUG_ASSERT(hdr->format == dyncol_fmt_str);
+        if (read_name(hdr, hdr->entry, &str))
+          return 0;
       }
-      flag= ((LEX_STRING *) key)->length - str.length;
-      if (flag == 0)
-        flag= memcmp(((LEX_STRING *) key)->str, str.str, str.length);
+      flag= mariadb_dyncol_column_cmp_named((LEX_STRING *)key, &str);
     }
     if (flag <= 0)
       end= mid;
@@ -2273,13 +2565,11 @@ find_place(DYN_HEADER *hdr, void *key, my_bool string_keys)
       }
       else
       {
-        DBUG_ASSERT(hdr->format == DYNCOL_FMT_STR);
-        str.length= hdr->entry[0];
-        str.str= (char*) hdr->nmpool + uint2korr(hdr->entry + 1);
+        DBUG_ASSERT(hdr->format == dyncol_fmt_str);
+        if (read_name(hdr, hdr->entry, &str))
+          return 0;
       }
-      flag= ((LEX_STRING *) key)->length - str.length;
-      if (flag == 0)
-        flag= memcmp(((LEX_STRING *) key)->str, str.str, str.length);
+      flag= mariadb_dyncol_column_cmp_named((LEX_STRING *)key, &str);
     }
   }
   if (flag > 0)
@@ -2289,7 +2579,7 @@ find_place(DYN_HEADER *hdr, void *key, my_bool string_keys)
 
 
 /*
-  It is internal structure which describes plan of chenging the record
+  It is internal structure which describes a plan of changing the record
   of dynamic columns
 */
 
@@ -2321,15 +2611,10 @@ static int plan_sort_num(const void *a, const void *b)
   Sort function for plan by column name
 */
 
-static int plan_sort_str(const void *a, const void *b)
+static int plan_sort_named(const void *a, const void *b)
 {
-  int res= (((LEX_STRING *)((PLAN *)a)->key)->length -
-            ((LEX_STRING *)((PLAN *)b)->key)->length);
-  if (res == 0)
-    res= memcmp(((LEX_STRING *)((PLAN *)a)->key)->str,
-                ((LEX_STRING *)((PLAN *)b)->key)->str,
-                ((LEX_STRING *)((PLAN *)a)->key)->length);
-  return res;
+  return mariadb_dyncol_column_cmp_named((LEX_STRING *)((PLAN *)a)->key,
+                                         (LEX_STRING *)((PLAN *)b)->key);
 }
 
 #define DELTA_CHECK(S, D, C)        \
@@ -2354,7 +2639,7 @@ static int plan_sort_str(const void *a, const void *b)
   @return ER_DYNCOL_* return code
 */
 
-enum enum_dyncol_func_result
+static enum enum_dyncol_func_result
 dynamic_column_update_copy(DYNAMIC_COLUMN *str, PLAN *plan,
                            uint add_column_count,
                            DYN_HEADER *hdr, DYN_HEADER *new_hdr,
@@ -2366,7 +2651,7 @@ dynamic_column_update_copy(DYNAMIC_COLUMN *str, PLAN *plan,
   uint i, j, k;
   size_t all_headers_size;
 
-  if (dynamic_column_init_str(&tmp,
+  if (dynamic_column_init_named(&tmp,
                               (new_fmt->fixed_hdr + new_hdr->header_size +
                                new_hdr->nmpool_size +
                                new_hdr->data_size + DYNCOL_SYZERESERVE)))
@@ -2432,7 +2717,7 @@ dynamic_column_update_copy(DYNAMIC_COLUMN *str, PLAN *plan,
       DYNAMIC_COLUMN_TYPE tp;
       char buff[DYNCOL_NUM_CHAR];
 
-      if (hdr->format == DYNCOL_FMT_NUM)
+      if (hdr->format == dyncol_fmt_num)
       {
         if (convert)
         {
@@ -2448,12 +2733,13 @@ dynamic_column_update_copy(DYNAMIC_COLUMN *str, PLAN *plan,
       }
       else
       {
-        name.length= read[0];
-        name.str= (char *) hdr->nmpool +  uint2korr(read + 1);
+        if (read_name(hdr, read, &name))
+          goto err;
         key= &name;
       }
-      if (type_and_offset_read(&tp, &offs,
-                               read + fmt->fixed_hdr_entry, hdr->offset_size))
+      if (fmt->type_and_offset_read(&tp, &offs,
+                                    read + fmt->fixed_hdr_entry,
+                                    hdr->offset_size))
           goto err;
       if (k == start)
         first_offset= offs;
@@ -2496,7 +2782,7 @@ dynamic_column_update_copy(DYNAMIC_COLUMN *str, PLAN *plan,
                                          plan[i].val,
                                          tmp.length - all_headers_size))
           goto err;
-        data_store(&tmp, plan[i].val);        /* Append new data */
+        data_store(&tmp, plan[i].val, new_hdr->format); /* Append new data */
       }
     }
   }
@@ -2508,7 +2794,7 @@ err:
   return ER_DYNCOL_FORMAT;
 }
 
-enum enum_dyncol_func_result
+static enum enum_dyncol_func_result
 dynamic_column_update_move_left(DYNAMIC_COLUMN *str, PLAN *plan,
                                 size_t offset_size,
                                 size_t entry_size,
@@ -2570,9 +2856,9 @@ dynamic_column_update_move_left(DYNAMIC_COLUMN *str, PLAN *plan,
 
     {
       DYNAMIC_COLUMN_TYPE tp;
-      if (type_and_offset_read(&tp, &first_offset,
-                               header_base + start * entry_size +
-                               COLUMN_NUMBER_SIZE, offset_size))
+      if (type_and_offset_read_num(&tp, &first_offset,
+                                   header_base + start * entry_size +
+                                   COLUMN_NUMBER_SIZE, offset_size))
         return ER_DYNCOL_FORMAT;
     }
     /* find data to be moved */
@@ -2617,8 +2903,8 @@ dynamic_column_update_move_left(DYNAMIC_COLUMN *str, PLAN *plan,
         DYNAMIC_COLUMN_TYPE tp;
 
         nm= uint2korr(read);                    /* Column nummber */
-        if (type_and_offset_read(&tp, &offs, read + COLUMN_NUMBER_SIZE,
-                                 offset_size))
+        if (type_and_offset_read_num(&tp, &offs, read + COLUMN_NUMBER_SIZE,
+                                     offset_size))
           return ER_DYNCOL_FORMAT;
 
         if (k > start && offs < first_offset)
@@ -2630,7 +2916,7 @@ dynamic_column_update_move_left(DYNAMIC_COLUMN *str, PLAN *plan,
         offs+= plan[i].ddelta;
         int2store(write, nm);
         /* write rest of data at write + COLUMN_NUMBER_SIZE */
-        type_and_offset_store(write, new_offset_size, tp, offs);
+        type_and_offset_store_num(write, new_offset_size, tp, offs);
         write+= new_entry_size;
       }
     }
@@ -2641,9 +2927,9 @@ dynamic_column_update_move_left(DYNAMIC_COLUMN *str, PLAN *plan,
       if( plan[i].act == PLAN_ADD || plan[i].act == PLAN_REPLACE)
       {
         int2store(write, *((uint *)plan[i].key));
-        type_and_offset_store(write, new_offset_size,
-                              plan[i].val[0].type,
-                              curr_offset);
+        type_and_offset_store_num(write, new_offset_size,
+                                  plan[i].val[0].type,
+                                  curr_offset);
         write+= new_entry_size;
         curr_offset+= plan[i].length;
       }
@@ -2690,14 +2976,15 @@ dynamic_column_update_move_left(DYNAMIC_COLUMN *str, PLAN *plan,
     {
       if( plan[i].act == PLAN_ADD || plan[i].act == PLAN_REPLACE)
       {
-        data_store(str, plan[i].val);        /* Append new data */
+        data_store(str, plan[i].val, dyncol_fmt_num);/* Append new data */
       }
     }
   }
   return ER_DYNCOL_OK;
 }
 
-enum enum_dyncol_func_result
+#ifdef UNUSED
+static enum enum_dyncol_func_result
 dynamic_column_update_move_right(DYNAMIC_COLUMN *str, PLAN *plan,
                                  size_t offset_size,
                                  size_t entry_size,
@@ -2759,8 +3046,10 @@ dynamic_column_update_move_right(DYNAMIC_COLUMN *str, PLAN *plan,
 
     {
       DYNAMIC_COLUMN_TYPE tp;
-      type_and_offset_read(&tp, &first_offset,
-                           header_base + start * entry_size + COLUMN_NUMBER_SIZE, offset_size);
+      type_and_offset_read_num(&tp, &first_offset,
+                               header_base +
+                               start * entry_size + COLUMN_NUMBER_SIZE,
+                               offset_size);
     }
     /* find data to be moved */
     if (start < end)
@@ -2804,7 +3093,8 @@ dynamic_column_update_move_right(DYNAMIC_COLUMN *str, PLAN *plan,
         DYNAMIC_COLUMN_TYPE tp;
 
         nm= uint2korr(read);                    /* Column nummber */
-        type_and_offset_read(&tp, &offs, read + COLUMN_NUMBER_SIZE, offset_size);
+        type_and_offset_read_num(&tp, &offs, read + COLUMN_NUMBER_SIZE,
+                                 offset_size);
         if (k > start && offs < first_offset)
         {
           str->length= 0; // just something valid
@@ -2814,7 +3104,7 @@ dynamic_column_update_move_right(DYNAMIC_COLUMN *str, PLAN *plan,
         offs+= plan[i].ddelta;
         int2store(write, nm);
         /* write rest of data at write + COLUMN_NUMBER_SIZE */
-        if (type_and_offset_store(write, new_offset_size, tp, offs))
+        if (type_and_offset_store_num(write, new_offset_size, tp, offs))
         {
           str->length= 0; // just something valid
           return ER_DYNCOL_FORMAT;
@@ -2829,9 +3119,9 @@ dynamic_column_update_move_right(DYNAMIC_COLUMN *str, PLAN *plan,
       if( plan[i].act == PLAN_ADD || plan[i].act == PLAN_REPLACE)
       {
         int2store(write, *((uint *)plan[i].key));
-        if (type_and_offset_store(write, new_offset_size,
-                                  plan[i].val[0].type,
-                                  curr_offset))
+        if (type_and_offset_store_num(write, new_offset_size,
+                                      plan[i].val[0].type,
+                                      curr_offset))
         {
           str->length= 0; // just something valid
           return ER_DYNCOL_FORMAT;
@@ -2882,13 +3172,13 @@ dynamic_column_update_move_right(DYNAMIC_COLUMN *str, PLAN *plan,
     {
       if( plan[i].act == PLAN_ADD || plan[i].act == PLAN_REPLACE)
       {
-        data_store(str, plan[i].val);        /* Append new data */
+        data_store(str, plan[i].val, dyncol_fmt_num); /* Append new data */
       }
     }
   }
   return ER_DYNCOL_OK;
 }
-
+#endif
 
 /**
   Update the packed string with the given columns
@@ -2913,7 +3203,27 @@ dynamic_column_update_many(DYNAMIC_COLUMN *str,
                                         values, FALSE);
 }
 
-uint numlen(uint val)
+enum enum_dyncol_func_result
+mariadb_dyncol_update_many(DYNAMIC_COLUMN *str,
+                           uint add_column_count,
+                           uint *column_numbers,
+                           DYNAMIC_COLUMN_VALUE *values)
+{
+  return dynamic_column_update_many_fmt(str, add_column_count, column_numbers,
+                                        values, FALSE);
+}
+
+enum enum_dyncol_func_result
+mariadb_dyncol_update_many_named(DYNAMIC_COLUMN *str,
+                                 uint add_column_count,
+                                 LEX_STRING *column_names,
+                                 DYNAMIC_COLUMN_VALUE *values)
+{
+  return dynamic_column_update_many_fmt(str, add_column_count, column_names,
+                                        values, TRUE);
+}
+
+static uint numlen(uint val)
 {
   uint res;
   if (val == 0)
@@ -2927,7 +3237,7 @@ uint numlen(uint val)
   return res;
 }
 
-enum enum_dyncol_func_result
+static enum enum_dyncol_func_result
 dynamic_column_update_many_fmt(DYNAMIC_COLUMN *str,
                                uint add_column_count,
                                void *column_keys,
@@ -2952,7 +3262,7 @@ dynamic_column_update_many_fmt(DYNAMIC_COLUMN *str,
 
   bzero(&header, sizeof(header));
   bzero(&new_header, sizeof(new_header));
-  new_header.format= (string_keys ? DYNCOL_FMT_STR : DYNCOL_FMT_NUM);
+  new_header.format= (string_keys ? dyncol_fmt_str : dyncol_fmt_num);
   new_fmt= fmt_data + new_header.format;
 
   /*
@@ -2997,8 +3307,8 @@ dynamic_column_update_many_fmt(DYNAMIC_COLUMN *str,
     goto end;
   fmt= fmt_data + header.format;
   /* new format can't be numeric if the old one is names */
-  DBUG_ASSERT(new_header.format == DYNCOL_FMT_STR ||
-              header.format == DYNCOL_FMT_NUM);
+  DBUG_ASSERT(new_header.format == dyncol_fmt_str ||
+              header.format == dyncol_fmt_num);
   if (header.column_count == 0)
     goto create_new_string;
 
@@ -3006,8 +3316,8 @@ dynamic_column_update_many_fmt(DYNAMIC_COLUMN *str,
 
   new_header.column_count= header.column_count;
   new_header.nmpool_size= header.nmpool_size;
-  if ((convert= (new_header.format == DYNCOL_FMT_STR &&
-                 header.format == DYNCOL_FMT_NUM)))
+  if ((convert= (new_header.format == dyncol_fmt_str &&
+                 header.format == dyncol_fmt_num)))
   {
     DBUG_ASSERT(new_header.nmpool_size == 0);
     for(i= 0, header.entry= header.header;
@@ -3064,12 +3374,18 @@ dynamic_column_update_many_fmt(DYNAMIC_COLUMN *str,
         goto end;
       }
 
-        //get_length(header.entry, header.dtpool, header.offset_size,
-        //header.data_size);
-      if (new_header.format == DYNCOL_FMT_STR)
+      if (new_header.format == dyncol_fmt_str)
       {
-        if (header.format == DYNCOL_FMT_STR)
-          entry_name_size= header.entry[0];
+        if (header.format == dyncol_fmt_str)
+        {
+          LEX_STRING name;
+          if (read_name(&header, header.entry, &name))
+          {
+            rc= ER_DYNCOL_FORMAT;
+            goto end;
+          }
+          entry_name_size= name.length;
+        }
         else
           entry_name_size= numlen(uint2korr(header.entry));
       }
@@ -3089,14 +3405,15 @@ dynamic_column_update_many_fmt(DYNAMIC_COLUMN *str,
 
         plan[i].act= PLAN_REPLACE;
         /* get data delta in bytes */
-        if ((plan[i].length= dynamic_column_value_len(plan[i].val)) ==
+        if ((plan[i].length= dynamic_column_value_len(plan[i].val,
+                                                      new_header.format)) ==
             (size_t) ~0)
         {
           rc= ER_DYNCOL_DATA;
           goto end;
         }
         data_delta+= plan[i].length - entry_data_size;
-        if (new_header.format == DYNCOL_FMT_STR)
+        if (new_header.format == dyncol_fmt_str)
         {
           name_delta+= ((LEX_STRING *)(plan[i].key))->length - entry_name_size;
         }
@@ -3117,14 +3434,15 @@ dynamic_column_update_many_fmt(DYNAMIC_COLUMN *str,
         plan[i].act= PLAN_ADD;
         header_delta++;                         /* One more row in header */
         /* get data delta in bytes */
-        if ((plan[i].length= dynamic_column_value_len(plan[i].val)) ==
+        if ((plan[i].length= dynamic_column_value_len(plan[i].val,
+                                                      new_header.format)) ==
             (size_t) ~0)
         {
           rc= ER_DYNCOL_DATA;
           goto end;
         }
         data_delta+= plan[i].length;
-        if (new_header.format == DYNCOL_FMT_STR)
+        if (new_header.format == dyncol_fmt_str)
           name_delta+= ((LEX_STRING *)plan[i].key)->length;
       }
     }
@@ -3143,18 +3461,18 @@ dynamic_column_update_many_fmt(DYNAMIC_COLUMN *str,
   */
   new_header.data_size= header.data_size + data_delta;
   new_header.nmpool_size= new_header.nmpool_size + name_delta;
-  DBUG_ASSERT(new_header.format != DYNCOL_FMT_NUM ||
+  DBUG_ASSERT(new_header.format != dyncol_fmt_num ||
               new_header.nmpool_size == 0);
   if ((new_header.offset_size=
-       dynamic_column_offset_bytes(new_header.data_size)) >=
-      MAX_OFFSET_LENGTH)
+       new_fmt->dynamic_column_offset_bytes(new_header.data_size)) >=
+      new_fmt->max_offset_size)
   {
     rc= ER_DYNCOL_LIMIT;
     goto end;
   }
 
   copy= ((header.format != new_header.format) ||
-         (new_header.format == DYNCOL_FMT_STR));
+         (new_header.format == dyncol_fmt_str));
   /* if (new_header.offset_size!=offset_size) then we have to rewrite header */
   header_delta_sign=
     ((int)new_header.offset_size + new_fmt->fixed_hdr_entry) -
@@ -3173,14 +3491,14 @@ dynamic_column_update_many_fmt(DYNAMIC_COLUMN *str,
 
   /*
     Need copy because:
-    1. Header/data parts moved in different directions.
+    1, Header/data parts moved in different directions.
     2. There is no enough allocated space in the string.
     3. Header and data moved in different directions.
   */
-  if (copy || /*1*/
-      str->max_length < str->length + header_delta + data_delta || /*2*/
+  if (copy || /*1.*/
+      str->max_length < str->length + header_delta + data_delta || /*2.*/
       ((header_delta_sign < 0 && data_delta_sign > 0) ||
-       (header_delta_sign > 0 && data_delta_sign < 0))) /*3*/
+       (header_delta_sign > 0 && data_delta_sign < 0))) /*3.*/
     rc= dynamic_column_update_copy(str, plan, add_column_count,
                                    &header, &new_header,
                                    convert);
@@ -3246,7 +3564,7 @@ int dynamic_column_update(DYNAMIC_COLUMN *str, uint column_nr,
 
 
 enum enum_dyncol_func_result
-dynamic_column_check(DYNAMIC_COLUMN *str)
+mariadb_dyncol_check(DYNAMIC_COLUMN *str)
 {
   struct st_service_funcs *fmt;
   enum enum_dyncol_func_result rc= ER_DYNCOL_FORMAT;
@@ -3297,7 +3615,7 @@ dynamic_column_check(DYNAMIC_COLUMN *str)
     header.header_size - header.nmpool_size;
 
   /* read and check headers */
-  if (header.format == DYNCOL_FMT_NUM)
+  if (header.format == dyncol_fmt_num)
   {
     key= &num;
     prev_key= &prev_num;
@@ -3312,20 +3630,28 @@ dynamic_column_check(DYNAMIC_COLUMN *str)
        i++, header.entry+= header.entry_size)
   {
 
-    if (header.format == DYNCOL_FMT_NUM)
+    if (header.format == dyncol_fmt_num)
     {
        num= uint2korr(header.entry);
     }
     else
     {
-      DBUG_ASSERT(header.format == DYNCOL_FMT_STR);
-      name.length= header.entry[0];
-      name_offset= uint2korr(header.entry + 1);
-      name.str= (char *)header.nmpool + name_offset;
+      DBUG_ASSERT(header.format == dyncol_fmt_str);
+      if (read_name(&header, header.entry, &name))
+      {
+        DBUG_PRINT("info", ("Reading name failed: Field order: %u"
+                            "  Name offset: %u"
+                            "  Name pool size: %u",
+                            (uint) i,
+                            uint2korr(header.entry),
+                            (uint)header.nmpool_size));
+        goto end;
+      }
+      name_offset= name.str - (char *)header.nmpool;
     }
-    if (type_and_offset_read(&type, &data_offset,
-                             header.entry + fmt->fixed_hdr_entry,
-                             header.offset_size))
+    if ((*fmt->type_and_offset_read)(&type, &data_offset,
+                                     header.entry + fmt->fixed_hdr_entry,
+                                     header.offset_size))
       goto end;
 
     DBUG_ASSERT(type != DYN_COL_NULL);
@@ -3334,17 +3660,8 @@ dynamic_column_check(DYNAMIC_COLUMN *str)
       DBUG_PRINT("info", ("Field order: %u  Data offset: %u"
                           " > Data pool size: %u",
                           (uint)i,
-                          (uint)name_offset,
-                          (uint)header.nmpool_size));
-      goto end;
-    }
-    if (name_offset > header.nmpool_size)
-    {
-      DBUG_PRINT("info", ("Field order: %u  Name offset: %u"
-                          " > Name pool size: %u",
-                          (uint)i,
-                          (uint)name_offset,
-                          (uint)header.nmpool_size));
+                          (uint)data_offset,
+                          (uint)header.data_size));
       goto end;
     }
     if (prev_type != DYN_COL_NULL)
@@ -3389,9 +3706,9 @@ dynamic_column_check(DYNAMIC_COLUMN *str)
   {
     DYNAMIC_COLUMN_VALUE store;
     // already checked by previouse pass
-    type_and_offset_read(&header.type, &header.offset,
-                         header.entry + fmt->fixed_hdr_entry,
-                         header.offset_size);
+    (*fmt->type_and_offset_read)(&header.type, &header.offset,
+                                 header.entry + fmt->fixed_hdr_entry,
+                                 header.offset_size);
     header.length=
       hdr_interval_length(&header, header.entry + header.entry_size);
     header.data= header.dtpool + header.offset;
@@ -3421,6 +3738,9 @@ dynamic_column_check(DYNAMIC_COLUMN *str)
     case DYN_COL_TIME:
       rc= dynamic_column_time_read(&store, header.data, header.length);
       break;
+    case DYN_COL_DYNCOL:
+      rc= dynamic_column_dyncol_read(&store, header.data, header.length);
+      break;
     case DYN_COL_NULL:
     default:
       rc= ER_DYNCOL_FORMAT;
@@ -3442,13 +3762,13 @@ end:
 
 
 enum enum_dyncol_func_result
-dynamic_column_val_str(DYNAMIC_STRING *str, DYNAMIC_COLUMN_VALUE *val,
-                       CHARSET_INFO *cs, my_bool quote)
+mariadb_dyncol_val_str(DYNAMIC_STRING *str, DYNAMIC_COLUMN_VALUE *val,
+                       CHARSET_INFO *cs, char quote)
 {
   char buff[40];
   int len;
   switch (val->type) {
-  case DYN_COL_INT:
+    case DYN_COL_INT:
       len= snprintf(buff, sizeof(buff), "%lld", val->x.long_value);
       if (dynstr_append_mem(str, buff, len))
         return ER_DYNCOL_RESOURCE;
@@ -3463,11 +3783,12 @@ dynamic_column_val_str(DYNAMIC_STRING *str, DYNAMIC_COLUMN_VALUE *val,
       if (dynstr_realloc(str, len + (quote ? 2 : 0)))
         return ER_DYNCOL_RESOURCE;
       if (quote)
-        str->str[str->length++]= '"';
+        str->str[str->length++]= quote;
       dynstr_append_mem(str, buff, len);
       if (quote)
-        str->str[str->length++]= '"';
+        str->str[str->length++]= quote;
       break;
+    case DYN_COL_DYNCOL:
     case DYN_COL_STRING:
       {
         char *alloc= NULL;
@@ -3506,7 +3827,7 @@ dynamic_column_val_str(DYNAMIC_STRING *str, DYNAMIC_COLUMN_VALUE *val,
             return ER_DYNCOL_RESOURCE;
         }
         if (quote)
-          rc= dynstr_append_quoted(str, from, len);
+          rc= dynstr_append_quoted(str, from, len, quote);
         else
           rc= dynstr_append_mem(str, from, len);
         if (alloc)
@@ -3547,7 +3868,7 @@ dynamic_column_val_str(DYNAMIC_STRING *str, DYNAMIC_COLUMN_VALUE *val,
 
 
 enum enum_dyncol_func_result
-dynamic_column_val_long(longlong *ll, DYNAMIC_COLUMN_VALUE *val)
+mariadb_dyncol_val_long(longlong *ll, DYNAMIC_COLUMN_VALUE *val)
 {
   enum enum_dyncol_func_result rc= ER_DYNCOL_OK;
   *ll= 0;
@@ -3619,6 +3940,7 @@ dynamic_column_val_long(longlong *ll, DYNAMIC_COLUMN_VALUE *val)
             val->x.time_value.second) *
         (val->x.time_value.neg ? -1 : 1);
       break;
+    case DYN_COL_DYNCOL:
     case DYN_COL_NULL:
       rc= ER_DYNCOL_TRUNCATED;
       break;
@@ -3630,7 +3952,7 @@ dynamic_column_val_long(longlong *ll, DYNAMIC_COLUMN_VALUE *val)
 
 
 enum enum_dyncol_func_result
-dynamic_column_val_double(double *dbl, DYNAMIC_COLUMN_VALUE *val)
+mariadb_dyncol_val_double(double *dbl, DYNAMIC_COLUMN_VALUE *val)
 {
   enum enum_dyncol_func_result rc= ER_DYNCOL_OK;
   *dbl= 0;
@@ -3684,6 +4006,7 @@ dynamic_column_val_double(double *dbl, DYNAMIC_COLUMN_VALUE *val)
                      val->x.time_value.second) *
         (val->x.time_value.neg ? -1 : 1);
       break;
+    case DYN_COL_DYNCOL:
     case DYN_COL_NULL:
       rc= ER_DYNCOL_TRUNCATED;
       break;
@@ -3703,30 +4026,40 @@ dynamic_column_val_double(double *dbl, DYNAMIC_COLUMN_VALUE *val)
   @return ER_DYNCOL_* return code
 */
 
-enum enum_dyncol_func_result
-dynamic_column_json(DYNAMIC_COLUMN *str, DYNAMIC_STRING *json)
+#define JSON_STACK_PROTECTION 10
+
+static enum enum_dyncol_func_result
+mariadb_dyncol_json_internal(DYNAMIC_COLUMN *str, DYNAMIC_STRING *json,
+                             uint lvl)
 {
   DYN_HEADER header;
   uint i;
   enum enum_dyncol_func_result rc;
 
-  bzero(json, sizeof(DYNAMIC_STRING));          /* In case of errors */
+  if (lvl >= JSON_STACK_PROTECTION)
+  {
+    rc= ER_DYNCOL_RESOURCE;
+    goto err;
+  }
+
+
   if (str->length == 0)
     return ER_DYNCOL_OK;                        /* no columns */
 
   if ((rc= init_read_hdr(&header, str)) < 0)
-    return rc;
+    goto err;
 
   if (header.entry_size * header.column_count + FIXED_HEADER_SIZE >
       str->length)
-    return ER_DYNCOL_FORMAT;
+  {
+    rc= ER_DYNCOL_FORMAT;
+    goto err;
+  }
 
-  if (init_dynamic_string(json, NULL, str->length * 2, 100))
-    return ER_DYNCOL_RESOURCE;
+  rc= ER_DYNCOL_RESOURCE;
 
   if (dynstr_append_mem(json, "[", 1))
-    return ER_DYNCOL_RESOURCE;
-  rc= ER_DYNCOL_RESOURCE;
+    goto err;
   for (i= 0, header.entry= header.header;
        i < header.column_count;
        i++, header.entry+= header.entry_size)
@@ -3750,7 +4083,7 @@ dynamic_column_json(DYNAMIC_COLUMN *str, DYNAMIC_STRING *json)
     if ((rc= dynamic_column_get_value(&header, &val)) < 0 ||
         dynstr_append_mem(json, "{", 1))
       goto err;
-    if (header.format == DYNCOL_FMT_NUM)
+    if (header.format == dyncol_fmt_num)
     {
       uint nm= uint2korr(header.entry);
       if (dynstr_realloc(json, DYNCOL_NUM_CHAR + 3))
@@ -3761,23 +4094,47 @@ dynamic_column_json(DYNAMIC_COLUMN *str, DYNAMIC_STRING *json)
     }
     else
     {
-      uint len= header.entry[0];
-      if (dynstr_realloc(json, len + 3))
+      LEX_STRING name;
+      if (read_name(&header, header.entry, &name))
+      {
+        rc= ER_DYNCOL_FORMAT;
+        goto err;
+      }
+      if (dynstr_realloc(json, name.length + 3))
         goto err;
       json->str[json->length++]= '"';
-      memcpy(json->str + json->length, (const void *)header.nmpool +
-             uint2korr(header.entry + 1), len);
-      json->length+= len;
+      memcpy(json->str + json->length, name.str, name.length);
+      json->length+= name.length;
     }
     json->str[json->length++]= '"';
     json->str[json->length++]= ':';
-    if ((rc= dynamic_column_val_str(json, &val,
-                                    &my_charset_utf8_general_ci, TRUE)) < 0 ||
-        dynstr_append_mem(json, "}", 1))
-      goto err;
+    if (val.type == DYN_COL_DYNCOL)
+    {
+      /* here we use it only for read so can cheat a bit */
+      DYNAMIC_COLUMN dc;
+      bzero(&dc, sizeof(dc));
+      dc.str= val.x.string.value.str;
+      dc.length= val.x.string.value.length;
+      if (mariadb_dyncol_json_internal(&dc, json, lvl + 1) < 0)
+      {
+        dc.str= NULL; dc.length= 0;
+        goto err;
+      }
+      dc.str= NULL; dc.length= 0;
+    }
+    else
+    {
+      if ((rc= mariadb_dyncol_val_str(json, &val,
+                                &my_charset_utf8_general_ci, '"')) < 0 ||
+          dynstr_append_mem(json, "}", 1))
+        goto err;
+    }
   }
   if (dynstr_append_mem(json, "]", 1))
-    return ER_DYNCOL_RESOURCE;
+  {
+    rc= ER_DYNCOL_RESOURCE;
+    goto err;
+  }
   return ER_DYNCOL_OK;
 
 err:
@@ -3785,59 +4142,73 @@ err:
   return rc;
 }
 
+enum enum_dyncol_func_result
+mariadb_dyncol_json(DYNAMIC_COLUMN *str, DYNAMIC_STRING *json)
+{
+
+  if (init_dynamic_string(json, NULL, str->length * 2, 100))
+    return ER_DYNCOL_RESOURCE;
+
+  return mariadb_dyncol_json_internal(str, json, 1);
+}
+
 
 /**
   Convert to DYNAMIC_COLUMN_VALUE values and names (LEX_STING) dynamic array
 
   @param str             The packed string
-  @param names           Where to put names
-  @param vals            Where to put values
-  @param free_names      pointer to free names buffer if there is it.
+  @param count           number of elements in the arrays
+  @param names           Where to put names (should be free by user)
+  @param vals            Where to put values (should be free by user)
 
   @return ER_DYNCOL_* return code
 */
 
 enum enum_dyncol_func_result
-dynamic_column_vals(DYNAMIC_COLUMN *str,
-                    DYNAMIC_ARRAY *names, DYNAMIC_ARRAY *vals,
-                    char **free_names)
+mariadb_dyncol_unpack(DYNAMIC_COLUMN *str,
+                      uint *count,
+                      LEX_STRING **names, DYNAMIC_COLUMN_VALUE **vals)
 {
   DYN_HEADER header;
   char *nm;
   uint i;
   enum enum_dyncol_func_result rc;
 
-  *free_names= 0;
-  bzero(names, sizeof(DYNAMIC_ARRAY));        /* In case of errors */
-  bzero(vals, sizeof(DYNAMIC_ARRAY));         /* In case of errors */
+  *count= 0; *names= 0; *vals= 0;
+
   if (str->length == 0)
     return ER_DYNCOL_OK;                      /* no columns */
 
   if ((rc= init_read_hdr(&header, str)) < 0)
     return rc;
 
+
   if (header.entry_size * header.column_count + FIXED_HEADER_SIZE >
       str->length)
     return ER_DYNCOL_FORMAT;
 
-  if (init_dynamic_array(names, sizeof(LEX_STRING),
-                         header.column_count, 0) ||
-      init_dynamic_array(vals, sizeof(DYNAMIC_COLUMN_VALUE),
-                         header.column_count, 0) ||
-      (header.format == DYNCOL_FMT_NUM &&
-       !(*free_names= (char *)malloc(DYNCOL_NUM_CHAR * header.column_count))))
+  *vals= my_malloc(sizeof(DYNAMIC_COLUMN_VALUE)* header.column_count, MYF(0));
+  if (header.format == dyncol_fmt_num)
+  {
+    *names= my_malloc(sizeof(LEX_STRING) * header.column_count +
+                      DYNCOL_NUM_CHAR * header.column_count, MYF(0));
+    nm= (char *)(names + sizeof(LEX_STRING) * header.column_count);
+  }
+  else
+  {
+    *names= my_malloc(sizeof(LEX_STRING) * header.column_count, MYF(0));
+    nm= 0;
+  }
+  if (!(*vals) || !(*names))
   {
     rc= ER_DYNCOL_RESOURCE;
     goto err;
   }
-  nm= *free_names;
 
   for (i= 0, header.entry= header.header;
        i < header.column_count;
        i++, header.entry+= header.entry_size)
   {
-    DYNAMIC_COLUMN_VALUE val;
-    LEX_STRING name;
     header.length=
       hdr_interval_length(&header, header.entry + header.entry_size);
     header.data= header.dtpool + header.offset;
@@ -3851,32 +4222,60 @@ dynamic_column_vals(DYNAMIC_COLUMN *str,
       rc= ER_DYNCOL_FORMAT;
       goto err;
     }
-    if ((rc= dynamic_column_get_value(&header, &val)) < 0)
+    if ((rc= dynamic_column_get_value(&header, (*vals) + i)) < 0)
       goto err;
 
-    if (header.format == DYNCOL_FMT_NUM)
+    if (header.format == dyncol_fmt_num)
     {
       uint num= uint2korr(header.entry);
-      name.str= nm;
-      name.length= snprintf(nm, DYNCOL_NUM_CHAR, "%u", num);
-      nm+= name.length + 1;
+      (*names)[i].str= nm;
+      (*names)[i].length= snprintf(nm, DYNCOL_NUM_CHAR, "%u", num);
+      nm+= (*names)[i].length + 1;
     }
     else
     {
-      name.length= header.entry[0];
-      name.str= (char *)header.nmpool + uint2korr(header.entry + 1);
+      if (read_name(&header, header.entry, (*names) + i))
+      {
+        rc= ER_DYNCOL_FORMAT;
+        goto err;
+      }
     }
-    /* following is preallocated and so do not fail */
-    (void) insert_dynamic(names, (uchar *)&name);
-    (void) insert_dynamic(vals, (uchar *)&val);
   }
+
+  *count= header.column_count;
   return ER_DYNCOL_OK;
 
 err:
-  delete_dynamic(names);
-  delete_dynamic(vals);
-  if (*free_names)
-    my_free(*free_names);
-  *free_names= 0;
+  if (*vals)
+  {
+    my_free(*vals);
+    *vals= 0;
+  }
+  if (*names)
+  {
+    my_free(*names);
+    *names= 0;
+  }
+  return rc;
+}
+
+/**
+  Get not NULL column count
+
+  @param str             The packed string
+  @param column_count    Where to put column count
+
+  @return ER_DYNCOL_* return code
+*/
+
+enum enum_dyncol_func_result
+mariadb_dyncol_column_count(DYNAMIC_COLUMN *str, uint *column_count)
+{
+  DYN_HEADER header;
+  enum enum_dyncol_func_result rc;
+
+  if ((rc= init_read_hdr(&header, str)) < 0)
+    return rc;
+  *column_count= header.column_count;
   return rc;
 }
diff --git a/mysys/string.c b/mysys/string.c
index 0cf8f939260..1263e7824f9 100644
--- a/mysys/string.c
+++ b/mysys/string.c
@@ -176,18 +176,19 @@ my_bool dynstr_append_os_quoted(DYNAMIC_STRING *str, const char *append, ...)
 }
 
 my_bool dynstr_append_quoted(DYNAMIC_STRING *str,
-                             const char *append, size_t len)
+                             const char *append, size_t len,
+                             char quote)
 {
   uint additional= (str->alloc_increment ? str->alloc_increment : 10);
   uint lim= additional;
   uint i;
   if (dynstr_realloc(str, len + additional + 2))
     return TRUE;
-  str->str[str->length++]= '"';
+  str->str[str->length++]= quote;
   for (i= 0; i < len; i++)
   {
     register char c= append[i];
-    if (c == '"' || c == '\\')
+    if (c == quote || c == '\\')
     {
       if (!lim)
       {
@@ -200,10 +201,11 @@ my_bool dynstr_append_quoted(DYNAMIC_STRING *str,
     }
     str->str[str->length++]= c;
   }
-  str->str[str->length++]= '"';
+  str->str[str->length++]= quote;
   return FALSE;
 }
 
+
 void dynstr_free(DYNAMIC_STRING *str)
 {
   my_free(str->str);
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 72d28bb2e7e..ad01f9f83fc 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -6048,6 +6048,7 @@ Item* Item_equal::get_first(JOIN_TAB *context, Item *field_item)
   return NULL;
 }
 
+
 longlong Item_func_dyncol_check::val_int()
 {
   char buff[STRING_BUFFER_USUAL_SIZE];
@@ -6062,7 +6063,7 @@ longlong Item_func_dyncol_check::val_int()
   col.length= str->length();
   /* We do not change the string, so could do this trick */
   col.str= (char *)str->ptr();
-  rc= dynamic_column_check(&col);
+  rc= mariadb_dyncol_check(&col);
   if (rc < 0 && rc != ER_DYNCOL_FORMAT)
   {
     dynamic_column_error_message(rc);
@@ -6127,8 +6128,8 @@ longlong Item_func_dyncol_exists::val_int()
   /* We do not change the string, so could do this trick */
   col.str= (char *)str->ptr();
   rc= ((name == NULL) ?
-       dynamic_column_exists(&col, (uint) num) :
-       dynamic_column_exists_str(&col, name));
+       mariadb_dyncol_exists(&col, (uint) num) :
+       mariadb_dyncol_exists_named(&col, name));
   if (rc < 0)
   {
     dynamic_column_error_message(rc);
diff --git a/sql/item_create.cc b/sql/item_create.cc
index f886b3d598b..bc449b0aaf2 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -526,6 +526,54 @@ protected:
   virtual ~Create_func_coercibility() {}
 };
 
+class Create_func_dyncol_check : public Create_func_arg1
+{
+public:
+  virtual Item *create_1_arg(THD *thd, Item *arg1);
+
+  static Create_func_dyncol_check s_singleton;
+
+protected:
+  Create_func_dyncol_check() {}
+  virtual ~Create_func_dyncol_check() {}
+};
+
+class Create_func_dyncol_exists : public Create_func_arg2
+{
+public:
+  virtual Item *create_2_arg(THD *thd, Item *arg1, Item *arg2);
+
+  static Create_func_dyncol_exists s_singleton;
+
+protected:
+  Create_func_dyncol_exists() {}
+  virtual ~Create_func_dyncol_exists() {}
+};
+
+class Create_func_dyncol_list : public Create_func_arg1
+{
+public:
+  virtual Item *create_1_arg(THD *thd, Item *arg1);
+
+  static Create_func_dyncol_list s_singleton;
+
+protected:
+  Create_func_dyncol_list() {}
+  virtual ~Create_func_dyncol_list() {}
+};
+
+class Create_func_dyncol_json : public Create_func_arg1
+{
+public:
+  virtual Item *create_1_arg(THD *thd, Item *arg1);
+
+  static Create_func_dyncol_json s_singleton;
+
+protected:
+  Create_func_dyncol_json() {}
+  virtual ~Create_func_dyncol_json() {}
+};
+
 
 class Create_func_compress : public Create_func_arg1
 {
@@ -3108,6 +3156,38 @@ Create_func_coercibility::create_1_arg(THD *thd, Item *arg1)
 }
 
 
+Create_func_dyncol_check Create_func_dyncol_check::s_singleton;
+
+Item*
+Create_func_dyncol_check::create_1_arg(THD *thd, Item *arg1)
+{
+  return new (thd->mem_root) Item_func_dyncol_check(arg1);
+}
+
+Create_func_dyncol_exists Create_func_dyncol_exists::s_singleton;
+
+Item*
+Create_func_dyncol_exists::create_2_arg(THD *thd, Item *arg1, Item *arg2)
+{
+  return new (thd->mem_root) Item_func_dyncol_exists(arg1, arg2);
+}
+
+Create_func_dyncol_list Create_func_dyncol_list::s_singleton;
+
+Item*
+Create_func_dyncol_list::create_1_arg(THD *thd, Item *arg1)
+{
+  return new (thd->mem_root) Item_func_dyncol_list(arg1);
+}
+
+Create_func_dyncol_json Create_func_dyncol_json::s_singleton;
+
+Item*
+Create_func_dyncol_json::create_1_arg(THD *thd, Item *arg1)
+{
+  return new (thd->mem_root) Item_func_dyncol_json(arg1);
+}
+
 Create_func_concat Create_func_concat::s_singleton;
 
 Item*
@@ -5245,6 +5325,10 @@ static Native_func_registry func_array[] =
   { { C_STRING_WITH_LEN("CHARACTER_LENGTH") }, BUILDER(Create_func_char_length)},
   { { C_STRING_WITH_LEN("CHAR_LENGTH") }, BUILDER(Create_func_char_length)},
   { { C_STRING_WITH_LEN("COERCIBILITY") }, BUILDER(Create_func_coercibility)},
+  { { C_STRING_WITH_LEN("COLUMN_CHECK") }, BUILDER(Create_func_dyncol_check)},
+  { { C_STRING_WITH_LEN("COLUMN_EXISTS") }, BUILDER(Create_func_dyncol_exists)},
+  { { C_STRING_WITH_LEN("COLUMN_LIST") }, BUILDER(Create_func_dyncol_list)},
+  { { C_STRING_WITH_LEN("COLUMN_JSON") }, BUILDER(Create_func_dyncol_json)},
   { { C_STRING_WITH_LEN("COMPRESS") }, BUILDER(Create_func_compress)},
   { { C_STRING_WITH_LEN("CONCAT") }, BUILDER(Create_func_concat)},
   { { C_STRING_WITH_LEN("CONCAT_WS") }, BUILDER(Create_func_concat_ws)},
@@ -5720,11 +5804,6 @@ Item *create_func_dyncol_create(THD *thd, List<DYNCALL_CREATE_DEF> &list)
   return new (thd->mem_root) Item_func_dyncol_create(*args, dfs);
 }
 
-Item *create_func_dyncol_json(THD *thd, Item *str)
-{
-  return new (thd->mem_root) Item_func_dyncol_json(str);
-}
-
 Item *create_func_dyncol_add(THD *thd, Item *str,
                              List<DYNCALL_CREATE_DEF> &list)
 {
diff --git a/sql/item_func.h b/sql/item_func.h
index cb9c1929d7d..ccb86fd03e5 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -58,7 +58,7 @@ public:
                   NOW_FUNC, TRIG_COND_FUNC,
                   SUSERVAR_FUNC, GUSERVAR_FUNC, COLLATE_FUNC,
                   EXTRACT_FUNC, CHAR_TYPECAST_FUNC, FUNC_SP, UDF_FUNC,
-                  NEG_FUNC, GSYSVAR_FUNC };
+                  NEG_FUNC, GSYSVAR_FUNC, DYNCOL };
   enum optimize_type { OPTIMIZE_NONE,OPTIMIZE_KEY,OPTIMIZE_OP, OPTIMIZE_NULL,
                        OPTIMIZE_EQUAL };
   enum Type type() const { return FUNC_ITEM; }
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index ca6a6b2cea3..706fbbff94d 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -3760,8 +3760,8 @@ String *Item_func_uuid::val_str(String *str)
 
 Item_func_dyncol_create::Item_func_dyncol_create(List<Item> &args,
                                                  DYNCALL_CREATE_DEF *dfs)
-  : Item_str_func(args), defs(dfs), vals(0), keys(NULL), names(FALSE),
-  force_names(FALSE)
+  : Item_str_func(args), defs(dfs), vals(0), keys_num(NULL), keys_str(NULL),
+  names(FALSE), force_names(FALSE)
 {
   DBUG_ASSERT((args.elements & 0x1) == 0); // even number of arguments
 }
@@ -3782,13 +3782,15 @@ bool Item_func_dyncol_create::fix_fields(THD *thd, Item **ref)
       names= TRUE;
     }
 
-    keys= (uchar *) alloc_root(thd->mem_root,
+    keys_num= (uint *) alloc_root(thd->mem_root,
                                (sizeof(LEX_STRING) > sizeof(uint) ?
                                 sizeof(LEX_STRING) :
                                 sizeof(uint)) *
                                (arg_count / 2));
+    keys_str= (LEX_STRING *) keys_num;
+    //status_var_increment(thd->status_var.feature_dynamic_columns);
   }
-  return res || vals == 0 || keys == 0;
+  return res || vals == 0 || keys_num == 0;
 }
 
 
@@ -3808,6 +3810,41 @@ bool Item_func_dyncol_create::prepare_arguments(bool force_names_arg)
   my_decimal dtmp, *dres;
   force_names= force_names_arg;
 
+  if (!(names || force_names))
+  {
+    for (i= 0; i < column_count; i++)
+    {
+      uint valpos= i * 2 + 1;
+      DYNAMIC_COLUMN_TYPE type= defs[i].type;
+      if (type == DYN_COL_NULL)
+        switch (args[valpos]->field_type())
+        {
+        case MYSQL_TYPE_VARCHAR:
+        case MYSQL_TYPE_ENUM:
+        case MYSQL_TYPE_SET:
+        case MYSQL_TYPE_TINY_BLOB:
+        case MYSQL_TYPE_MEDIUM_BLOB:
+        case MYSQL_TYPE_LONG_BLOB:
+        case MYSQL_TYPE_BLOB:
+        case MYSQL_TYPE_VAR_STRING:
+        case MYSQL_TYPE_STRING:
+        case MYSQL_TYPE_GEOMETRY:
+          type= DYN_COL_STRING;
+          break;
+        default:
+          break;
+        }
+
+      if (type == DYN_COL_STRING &&
+          args[valpos]->type() == Item::FUNC_ITEM &&
+          ((Item_func *)args[valpos])->functype() == DYNCOL)
+      {
+        force_names= 1;
+        break;
+      }
+    }
+  }
+
   /* get values */
   for (i= 0; i < column_count; i++)
   {
@@ -3865,6 +3902,13 @@ bool Item_func_dyncol_create::prepare_arguments(bool force_names_arg)
         break;
       }
     }
+    if (type == DYN_COL_STRING &&
+        args[valpos]->type() == Item::FUNC_ITEM &&
+        ((Item_func *)args[valpos])->functype() == DYNCOL)
+    {
+      DBUG_ASSERT(names || force_names);
+      type= DYN_COL_DYNCOL;
+    }
     if (names || force_names)
     {
       res= args[i * 2]->val_str(&tmp);
@@ -3873,10 +3917,8 @@ bool Item_func_dyncol_create::prepare_arguments(bool force_names_arg)
         // guaranty UTF-8 string for names
         if (my_charset_same(res->charset(), &my_charset_utf8_general_ci))
         {
-          ((LEX_STRING *)keys)[i].length= res->length();
-          if (!(((LEX_STRING *)keys)[i].str=
-                (char *)sql_memdup(res->ptr(), res->length())))
-            ((LEX_STRING *)keys)[i].length= 0;
+          keys_str[i].length= res->length();
+          keys_str[i].str= sql_strmake(res->ptr(), res->length());
         }
         else
         {
@@ -3887,25 +3929,25 @@ bool Item_func_dyncol_create::prepare_arguments(bool force_names_arg)
                                my_charset_utf8_general_ci.mbmaxlen + 1));
           if (str)
           {
-            ((LEX_STRING *)keys)[i].length=
+            keys_str[i].length=
               copy_and_convert(str, strlen, &my_charset_utf8_general_ci,
                                res->ptr(), res->length(), res->charset(),
                                &dummy_errors);
-              ((LEX_STRING *)keys)[i].str= str;
+              keys_str[i].str= str;
           }
           else
-            ((LEX_STRING *)keys)[i].length= 0;
+            keys_str[i].length= 0;
 
         }
       }
       else
       {
-        ((LEX_STRING *)keys)[i].length= 0;
-        ((LEX_STRING *)keys)[i].str= NULL;
+        keys_str[i].length= 0;
+        keys_str[i].str= NULL;
       }
     }
     else
-      ((uint *)keys)[i]= (uint) args[i * 2]->val_int();
+      keys_num[i]= (uint) args[i * 2]->val_int();
     if (args[i * 2]->null_value)
     {
       /* to make cleanup possible */
@@ -3927,11 +3969,11 @@ bool Item_func_dyncol_create::prepare_arguments(bool force_names_arg)
     case DYN_COL_DOUBLE:
       vals[i].x.double_value= args[valpos]->val_real();
       break;
+    case DYN_COL_DYNCOL:
     case DYN_COL_STRING:
       res= args[valpos]->val_str(&tmp);
       if (res &&
-          (vals[i].x.string.value.str= my_strndup(res->ptr(), res->length(),
-                                                MYF(MY_WME))))
+          (vals[i].x.string.value.str= sql_strmake(res->ptr(), res->length())))
       {
 	vals[i].x.string.value.length= res->length();
 	vals[i].x.string.charset= res->charset();
@@ -3975,26 +4017,12 @@ bool Item_func_dyncol_create::prepare_arguments(bool force_names_arg)
     }
     if (vals[i].type != DYN_COL_NULL && args[valpos]->null_value)
     {
-      if (vals[i].type == DYN_COL_STRING)
-        my_free(vals[i].x.string.value.str);
       vals[i].type= DYN_COL_NULL;
     }
   }
   return FALSE;
 }
 
-void Item_func_dyncol_create::cleanup_arguments()
-{
-  uint column_count= (arg_count / 2);
-  uint i;
-
-  for (i= 0; i < column_count; i++)
-  {
-    if (vals[i].type == DYN_COL_STRING)
-      my_free(vals[i].x.string.value.str);
-  }
-  force_names= FALSE;
-}
 
 String *Item_func_dyncol_create::val_str(String *str)
 {
@@ -4011,8 +4039,11 @@ String *Item_func_dyncol_create::val_str(String *str)
   }
   else
   {
-    if ((rc= dynamic_column_create_many_fmt(&col, column_count, keys,
-                                            vals, names || force_names)))
+    if ((rc= ((names || force_names) ?
+              mariadb_dyncol_create_many_named(&col, column_count, keys_str,
+                                               vals, TRUE) :
+              mariadb_dyncol_create_many(&col, column_count, keys_num,
+                                         vals, TRUE))))
     {
       dynamic_column_error_message(rc);
       dynamic_column_column_free(&col);
@@ -4024,7 +4055,7 @@ String *Item_func_dyncol_create::val_str(String *str)
       /* Move result from DYNAMIC_COLUMN to str_value */
       char *ptr;
       size_t length, alloc_length;
-      dynamic_column_reassociate(&col, &ptr, &length, &alloc_length);
+      mariadb_dyncol_reassociate(&col, &ptr, &length, &alloc_length);
       str_value.reassociate(ptr, (uint32) length, (uint32) alloc_length,
                             &my_charset_bin);
       res= &str_value;
@@ -4032,9 +4063,6 @@ String *Item_func_dyncol_create::val_str(String *str)
     }
   }
 
-  /* cleanup */
-  cleanup_arguments();
-
   return res;
 }
 
@@ -4060,6 +4088,7 @@ void Item_func_dyncol_create::print_arguments(String *str,
     case DYN_COL_DOUBLE:
       str->append(STRING_WITH_LEN(" AS double"));
       break;
+    case DYN_COL_DYNCOL:
     case DYN_COL_STRING:
       str->append(STRING_WITH_LEN(" AS char"));
       if (defs[i].cs)
@@ -4109,7 +4138,7 @@ String *Item_func_dyncol_json::val_str(String *str)
 
   col.str= (char *)res->ptr();
   col.length= res->length();
-  if ((rc= dynamic_column_json(&col, &json)))
+  if ((rc= mariadb_dyncol_json(&col, &json)))
   {
     dynamic_column_error_message(rc);
     goto null;
@@ -4150,15 +4179,17 @@ String *Item_func_dyncol_add::val_str(String *str)
   col.length= res->length();
   memcpy(col.str, res->ptr(), col.length);
 
-  if (prepare_arguments(dynamic_column_has_names(&col)))
+  if (prepare_arguments(mariadb_dyncol_has_names(&col)))
     goto null;
 
-  if ((rc= dynamic_column_update_many_fmt(&col, column_count, keys,
-                                          vals, names || force_names)))
+  if ((rc= ((names || force_names) ?
+            mariadb_dyncol_update_many_named(&col, column_count,
+                                             keys_str, vals) :
+            mariadb_dyncol_update_many(&col, column_count,
+                                       keys_num, vals))))
   {
     dynamic_column_error_message(rc);
     dynamic_column_column_free(&col);
-    cleanup_arguments();
     goto null;
   }
 
@@ -4166,15 +4197,12 @@ String *Item_func_dyncol_add::val_str(String *str)
     /* Move result from DYNAMIC_COLUMN to str */
     char *ptr;
     size_t length, alloc_length;
-    dynamic_column_reassociate(&col, &ptr, &length, &alloc_length);
+    mariadb_dyncol_reassociate(&col, &ptr, &length, &alloc_length);
     str->reassociate(ptr, (uint32) length, (uint32) alloc_length,
                      &my_charset_bin);
     null_value= FALSE;
   }
 
-  /* cleanup */
-  cleanup_arguments();
-
   return str;
 
 null:
@@ -4264,8 +4292,8 @@ bool Item_dyncol_get::get_dyn_value(DYNAMIC_COLUMN_VALUE *val, String *tmp)
   dyn_str.str=   (char*) res->ptr();
   dyn_str.length= res->length();
   if ((rc= ((name == NULL) ?
-            dynamic_column_get(&dyn_str, (uint) num, val) :
-            dynamic_column_get_str(&dyn_str, name, val))))
+            mariadb_dyncol_get(&dyn_str, (uint) num, val) :
+            mariadb_dyncol_get_named(&dyn_str, name, val))))
   {
     dynamic_column_error_message(rc);
     null_value= 1;
@@ -4297,6 +4325,7 @@ String *Item_dyncol_get::val_str(String *str_result)
   case DYN_COL_DOUBLE:
     str_result->set_real(val.x.double_value, NOT_FIXED_DEC, &my_charset_latin1);
     break;
+  case DYN_COL_DYNCOL:
   case DYN_COL_STRING:
     if ((char*) tmp.ptr() <= val.x.string.value.str &&
         (char*) tmp.ptr() + tmp.length() >= val.x.string.value.str)
@@ -4373,6 +4402,7 @@ longlong Item_dyncol_get::val_int()
     return 0;
 
   switch (val.type) {
+  case DYN_COL_DYNCOL:
   case DYN_COL_NULL:
     goto null;
   case DYN_COL_UINT:
@@ -4453,6 +4483,7 @@ double Item_dyncol_get::val_real()
     return 0.0;
 
   switch (val.type) {
+  case DYN_COL_DYNCOL:
   case DYN_COL_NULL:
     goto null;
   case DYN_COL_UINT:
@@ -4510,6 +4541,7 @@ my_decimal *Item_dyncol_get::val_decimal(my_decimal *decimal_value)
     return NULL;
 
   switch (val.type) {
+  case DYN_COL_DYNCOL:
   case DYN_COL_NULL:
     goto null;
   case DYN_COL_UINT:
@@ -4569,6 +4601,7 @@ bool Item_dyncol_get::get_date(MYSQL_TIME *ltime, ulonglong fuzzy_date)
     return 1;                                   // Error
 
   switch (val.type) {
+  case DYN_COL_DYNCOL:
   case DYN_COL_NULL:
     goto null;
   case DYN_COL_INT:
@@ -4631,7 +4664,8 @@ String *Item_func_dyncol_list::val_str(String *str)
 {
   uint i;
   enum enum_dyncol_func_result rc;
-  DYNAMIC_ARRAY arr;
+  LEX_STRING *names= 0;
+  uint count;
   DYNAMIC_COLUMN col;
   String *res= args[0]->val_str(str);
 
@@ -4640,37 +4674,37 @@ String *Item_func_dyncol_list::val_str(String *str)
   col.length= res->length();
   /* We do not change the string, so could do this trick */
   col.str= (char *)res->ptr();
-  if ((rc= dynamic_column_list_str(&col, &arr)))
+  if ((rc= mariadb_dyncol_list_named(&col, &count, &names)))
   {
+    bzero(&col, sizeof(col));
     dynamic_column_error_message(rc);
-    for (i= 0; i < arr.elements; i++)
-      my_free(dynamic_element(&arr, i, LEX_STRING*)->str);
-    delete_dynamic(&arr);
     goto null;
   }
+  bzero(&col, sizeof(col));
 
   /*
     We estimate average name length as 10
   */
-  if (str->alloc(arr.elements * 13))
+  if (str->alloc(count * 13))
     goto null;
 
   str->length(0);
   str->set_charset(&my_charset_utf8_general_ci);
-  for (i= 0; i < arr.elements; i++)
+  for (i= 0; i < count; i++)
   {
-    LEX_STRING *name= dynamic_element(&arr, i, LEX_STRING *);
-    append_identifier(current_thd, str, name->str, name->length);
-    if (i < arr.elements - 1)
+    append_identifier(current_thd, str, names[i].str, names[i].length);
+    if (i < count - 1)
       str->qs_append(',');
-    my_free(name->str);
   }
-
   null_value= FALSE;
-  delete_dynamic(&arr);
+  if (names)
+    my_free(names);
   return str;
 
 null:
   null_value= TRUE;
+  if (names)
+    my_free(names);
   return NULL;
 }
+
diff --git a/sql/item_strfunc.h b/sql/item_strfunc.h
index e6a7e1291b8..8ef67654b35 100644
--- a/sql/item_strfunc.h
+++ b/sql/item_strfunc.h
@@ -1001,10 +1001,10 @@ class Item_func_dyncol_create: public Item_str_func
 protected:
   DYNCALL_CREATE_DEF *defs;
   DYNAMIC_COLUMN_VALUE *vals;
-  uchar *keys;
+  uint *keys_num;
+  LEX_STRING *keys_str;
   bool names, force_names;
   bool prepare_arguments(bool force_names);
-  void cleanup_arguments();
   void print_arguments(String *str, enum_query_type query_type);
 public:
   Item_func_dyncol_create(List<Item> &args, DYNCALL_CREATE_DEF *dfs);
@@ -1013,6 +1013,7 @@ public:
   const char *func_name() const{ return "column_create"; }
   String *val_str(String *);
   virtual void print(String *str, enum_query_type query_type);
+  virtual enum Functype functype() const   { return DYNCOL; }
 };
 
 
@@ -1081,3 +1082,4 @@ public:
 extern String my_empty_string;
 
 #endif /* ITEM_STRFUNC_INCLUDED */
+
diff --git a/sql/lex.h b/sql/lex.h
index ea68ea0972a..34eda613e71 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -126,10 +126,7 @@ static SYMBOL symbols[] = {
   { "COLUMN_CHECK",     SYM(COLUMN_CHECK_SYM)},
   { "COLUMN_CREATE",    SYM(COLUMN_CREATE_SYM)},
   { "COLUMN_DELETE",    SYM(COLUMN_DELETE_SYM)},
-  { "COLUMN_EXISTS",    SYM(COLUMN_EXISTS_SYM)},
   { "COLUMN_GET",       SYM(COLUMN_GET_SYM)},
-  { "COLUMN_JSON",      SYM(COLUMN_JSON_SYM)},
-  { "COLUMN_LIST",      SYM(COLUMN_LIST_SYM)},
   { "COMMENT",		SYM(COMMENT_SYM)},
   { "COMMIT",		SYM(COMMIT_SYM)},
   { "COMMITTED",	SYM(COMMITTED_SYM)},
diff --git a/sql/sql_join_cache.cc b/sql/sql_join_cache.cc
index d785366ae69..14d4d5fca6c 100644
--- a/sql/sql_join_cache.cc
+++ b/sql/sql_join_cache.cc
@@ -3872,10 +3872,10 @@ int JOIN_TAB_SCAN_MRR::next()
   int rc= join_tab->table->file->multi_range_read_next((range_id_t*)ptr) ? -1 : 0;
   if (!rc)
   {
-    /* 
+    /*
       If a record in in an incremental cache contains no fields then the
       association for the last record in cache will be equal to cache->end_pos
-    */ 
+    */
     /* 
     psergey: this makes no sense where HA_MRR_NO_ASSOC is used.
     DBUG_ASSERT(cache->buff <= (uchar *) (*ptr) &&
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 48dd47bccea..87a7df34986 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -885,10 +885,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 %token  COLUMN_CHECK_SYM
 %token  COLUMN_CREATE_SYM
 %token  COLUMN_DELETE_SYM
-%token  COLUMN_EXISTS_SYM
 %token  COLUMN_GET_SYM
-%token  COLUMN_JSON_SYM
-%token  COLUMN_LIST_SYM
 %token  COLUMN_SYM                    /* SQL-2003-R */
 %token  COLUMN_NAME_SYM               /* SQL-2003-N */
 %token  COMMENT_SYM
@@ -8808,20 +8805,6 @@ function_call_nonkeyword:
               MYSQL_YYABORT;
           }
         |
-          COLUMN_EXISTS_SYM '(' expr ',' expr ')'
-          {
-            $$= new (YYTHD->mem_root) Item_func_dyncol_exists($3, $5);
-            if ($$ == NULL)
-              MYSQL_YYABORT;
-          }
-        |
-          COLUMN_LIST_SYM '(' expr ')'
-          {
-            $$= new (YYTHD->mem_root) Item_func_dyncol_list($3);
-            if ($$ == NULL)
-              MYSQL_YYABORT;
-          }
-        |
           COLUMN_CREATE_SYM '(' dyncall_create_list ')'
           {
             $$= create_func_dyncol_create(YYTHD, *$3);
@@ -8829,13 +8812,6 @@ function_call_nonkeyword:
               MYSQL_YYABORT;
           }
         |
-          COLUMN_JSON_SYM '(' expr ')'
-          {
-            $$= create_func_dyncol_json(YYTHD, $3);
-            if ($$ == NULL)
-              MYSQL_YYABORT;
-          }
-        |
           COLUMN_GET_SYM '(' expr ',' expr AS cast_type ')'
           {
             LEX *lex= Lex;
@@ -12927,10 +12903,7 @@ keyword:
         | COLUMN_CHECK_SYM      {}
         | COLUMN_CREATE_SYM     {}
         | COLUMN_DELETE_SYM     {}
-        | COLUMN_EXISTS_SYM     {}
         | COLUMN_GET_SYM        {}
-        | COLUMN_JSON_SYM       {}
-        | COLUMN_LIST_SYM       {}
         | COMMENT_SYM           {}
         | COMMIT_SYM            {}
         | CONTAINS_SYM          {}
diff --git a/storage/cassandra/CMakeLists.txt b/storage/cassandra/CMakeLists.txt
index 7e92d3cc0bd..7852177b9b3 100644
--- a/storage/cassandra/CMakeLists.txt
+++ b/storage/cassandra/CMakeLists.txt
@@ -13,9 +13,9 @@ SET(cassandra_sources
 
 #INCLUDE_DIRECTORIES(BEFORE ${Boost_INCLUDE_DIRS})
 
-#INCLUDE_DIRECTORIES(AFTER /usr/local/include/thrift)
+INCLUDE_DIRECTORIES(AFTER /usr/local/include/thrift)
 #INCLUDE_DIRECTORIES(AFTER /home/buildbot/build/thrift-inst/include/thrift/) 
-INCLUDE_DIRECTORIES(AFTER /home/psergey/cassandra/thrift/include/thrift/) 
+#INCLUDE_DIRECTORIES(AFTER /home/psergey/cassandra/thrift/include/thrift/) 
 
 # 
 STRING(REPLACE "-fno-exceptions" "" CMAKE_CXX_FLAGS ${CMAKE_CXX_FLAGS})
diff --git a/storage/cassandra/ha_cassandra.cc b/storage/cassandra/ha_cassandra.cc
index a54190e080c..75eff80a3cf 100644
--- a/storage/cassandra/ha_cassandra.cc
+++ b/storage/cassandra/ha_cassandra.cc
@@ -89,7 +89,7 @@ ha_create_table_option cassandra_field_option_list[]=
   /*
     Collect all other columns as dynamic here,
     the valid values are YES/NO, ON/OFF, 1/0.
-    The default is 0, that is true, yes, on.
+    The default is 0, that is false, no, off.
   */
   HA_FOPTION_BOOL("DYNAMIC_COLUMN_STORAGE", dyncol_field, 0),
   HA_FOPTION_END
@@ -862,9 +862,37 @@ public:
 /**
   Converting dynamic columns types to/from casandra types
 */
+
+
+/**
+  Check and initialize (if it is needed) string MEM_ROOT
+*/
+static void alloc_strings_memroot(MEM_ROOT *mem_root)
+{
+  if (!alloc_root_inited(mem_root))
+  {
+    /*
+      The mem_root used to allocate UUID (of length 36 + \0) so make
+      appropriate allocated size
+    */
+    init_alloc_root(mem_root,
+                    (36 + 1 + ALIGN_SIZE(sizeof(USED_MEM))) * 10 +
+                    ALLOC_ROOT_MIN_BLOCK_SIZE,
+                    (36 + 1 + ALIGN_SIZE(sizeof(USED_MEM))) * 10 +
+                    ALLOC_ROOT_MIN_BLOCK_SIZE);
+  }
+}
+
+static void free_strings_memroot(MEM_ROOT *mem_root)
+{
+  if (alloc_root_inited(mem_root))
+    free_root(mem_root, MYF(0));
+}
+
 bool cassandra_to_dyncol_intLong(const char *cass_data,
                                  int cass_data_len __attribute__((unused)),
-                                 DYNAMIC_COLUMN_VALUE *value)
+                                 DYNAMIC_COLUMN_VALUE *value,
+                                 MEM_ROOT *mem_root __attribute__((unused)))
 {
   value->type= DYN_COL_INT;
 #ifdef WORDS_BIGENDIAN
@@ -881,7 +909,7 @@ bool dyncol_to_cassandraLong(DYNAMIC_COLUMN_VALUE *value,
 {
   longlong *tmp= (longlong *) buff;
   enum enum_dyncol_func_result rc=
-    dynamic_column_val_long(tmp, value);
+    mariadb_dyncol_val_long(tmp, value);
   if (rc < 0)
     return true;
   *cass_data_len= sizeof(longlong);
@@ -897,7 +925,8 @@ bool dyncol_to_cassandraLong(DYNAMIC_COLUMN_VALUE *value,
 
 bool cassandra_to_dyncol_intInt32(const char *cass_data,
                                   int cass_data_len __attribute__((unused)),
-                                  DYNAMIC_COLUMN_VALUE *value)
+                                  DYNAMIC_COLUMN_VALUE *value,
+                                  MEM_ROOT *mem_root __attribute__((unused)))
 {
   int32 tmp;
   value->type= DYN_COL_INT;
@@ -917,7 +946,7 @@ bool dyncol_to_cassandraInt32(DYNAMIC_COLUMN_VALUE *value,
 {
   longlong *tmp= (longlong *) ((char *)buff + sizeof(longlong));
   enum enum_dyncol_func_result rc=
-    dynamic_column_val_long(tmp, value);
+    mariadb_dyncol_val_long(tmp, value);
   if (rc < 0)
     return true;
   *cass_data_len= sizeof(int32);
@@ -937,7 +966,8 @@ bool dyncol_to_cassandraInt32(DYNAMIC_COLUMN_VALUE *value,
 
 bool cassandra_to_dyncol_intCounter(const char *cass_data,
                                     int cass_data_len __attribute__((unused)),
-                                    DYNAMIC_COLUMN_VALUE *value)
+                                    DYNAMIC_COLUMN_VALUE *value,
+                                    MEM_ROOT *mem_root __attribute__((unused)))
 {
   value->type= DYN_COL_INT;
   value->x.long_value= *((longlong *)cass_data);
@@ -951,7 +981,7 @@ bool dyncol_to_cassandraCounter(DYNAMIC_COLUMN_VALUE *value,
 {
   longlong *tmp= (longlong *)buff;
   enum enum_dyncol_func_result rc=
-    dynamic_column_val_long(tmp, value);
+    mariadb_dyncol_val_long(tmp, value);
   if (rc < 0)
     return true;
   *cass_data_len= sizeof(longlong);
@@ -962,7 +992,8 @@ bool dyncol_to_cassandraCounter(DYNAMIC_COLUMN_VALUE *value,
 
 bool cassandra_to_dyncol_doubleFloat(const char *cass_data,
                                      int cass_data_len __attribute__((unused)),
-                                     DYNAMIC_COLUMN_VALUE *value)
+                                     DYNAMIC_COLUMN_VALUE *value,
+                                     MEM_ROOT *mem_root __attribute__((unused)))
 {
   value->type= DYN_COL_DOUBLE;
   value->x.double_value= *((float *)cass_data);
@@ -975,7 +1006,7 @@ bool dyncol_to_cassandraFloat(DYNAMIC_COLUMN_VALUE *value,
 {
   double tmp;
   enum enum_dyncol_func_result rc=
-    dynamic_column_val_double(&tmp, value);
+    mariadb_dyncol_val_double(&tmp, value);
   if (rc < 0)
     return true;
   *((float *)buff)= (float) tmp;
@@ -987,7 +1018,9 @@ bool dyncol_to_cassandraFloat(DYNAMIC_COLUMN_VALUE *value,
 
 bool cassandra_to_dyncol_doubleDouble(const char *cass_data,
                                       int cass_data_len __attribute__((unused)),
-                                      DYNAMIC_COLUMN_VALUE *value)
+                                      DYNAMIC_COLUMN_VALUE *value,
+                                      MEM_ROOT *mem_root
+                                      __attribute__((unused)))
 {
   value->type= DYN_COL_DOUBLE;
   value->x.double_value= *((double *)cass_data);
@@ -1000,7 +1033,7 @@ bool dyncol_to_cassandraDouble(DYNAMIC_COLUMN_VALUE *value,
 {
   double *tmp= (double *)buff;
   enum enum_dyncol_func_result rc=
-    dynamic_column_val_double(tmp, value);
+    mariadb_dyncol_val_double(tmp, value);
   if (rc < 0)
     return true;
   *cass_data_len= sizeof(double);
@@ -1018,7 +1051,6 @@ bool cassandra_to_dyncol_strStr(const char *cass_data,
   value->x.string.charset= cs;
   value->x.string.value.str= (char *)cass_data;
   value->x.string.value.length= cass_data_len;
-  value->x.string.nonfreeable= TRUE; // do not try to free
   return 0;
 }
 
@@ -1030,7 +1062,7 @@ bool dyncol_to_cassandraStr(DYNAMIC_COLUMN_VALUE *value,
   if (init_dynamic_string(&tmp, NULL, 1024, 1024))
     return 1;
   enum enum_dyncol_func_result rc=
-    dynamic_column_val_str(&tmp, value, cs, FALSE);
+    mariadb_dyncol_val_str(&tmp, value, cs, '\0');
   if (rc < 0)
   {
     dynstr_free(&tmp);
@@ -1044,7 +1076,8 @@ bool dyncol_to_cassandraStr(DYNAMIC_COLUMN_VALUE *value,
 
 bool cassandra_to_dyncol_strBytes(const char *cass_data,
                                   int cass_data_len,
-                                  DYNAMIC_COLUMN_VALUE *value)
+                                  DYNAMIC_COLUMN_VALUE *value,
+                                  MEM_ROOT *mem_root __attribute__((unused)))
 {
   return cassandra_to_dyncol_strStr(cass_data, cass_data_len, value,
                                     &my_charset_bin);
@@ -1060,7 +1093,8 @@ bool dyncol_to_cassandraBytes(DYNAMIC_COLUMN_VALUE *value,
 
 bool cassandra_to_dyncol_strAscii(const char *cass_data,
                                   int cass_data_len,
-                                  DYNAMIC_COLUMN_VALUE *value)
+                                  DYNAMIC_COLUMN_VALUE *value,
+                                  MEM_ROOT *mem_root __attribute__((unused)))
 {
   return cassandra_to_dyncol_strStr(cass_data, cass_data_len, value,
                                     &my_charset_latin1_bin);
@@ -1076,7 +1110,8 @@ bool dyncol_to_cassandraAscii(DYNAMIC_COLUMN_VALUE *value,
 
 bool cassandra_to_dyncol_strUTF8(const char *cass_data,
                                  int cass_data_len,
-                                 DYNAMIC_COLUMN_VALUE *value)
+                                 DYNAMIC_COLUMN_VALUE *value,
+                                 MEM_ROOT *mem_root __attribute__((unused)))
 {
   return cassandra_to_dyncol_strStr(cass_data, cass_data_len, value,
                                     &my_charset_utf8_unicode_ci);
@@ -1092,20 +1127,20 @@ bool dyncol_to_cassandraUTF8(DYNAMIC_COLUMN_VALUE *value,
 
 bool cassandra_to_dyncol_strUUID(const char *cass_data,
                                  int cass_data_len,
-                                 DYNAMIC_COLUMN_VALUE *value)
+                                 DYNAMIC_COLUMN_VALUE *value,
+                                 MEM_ROOT *mem_root)
 {
   value->type= DYN_COL_STRING;
   value->x.string.charset= &my_charset_bin;
-  value->x.string.value.str= (char *)my_malloc(37, MYF(0));
+  alloc_strings_memroot(mem_root);
+  value->x.string.value.str= (char *)alloc_root(mem_root, 37);
   if (!value->x.string.value.str)
   {
     value->x.string.value.length= 0;
-    value->x.string.nonfreeable= TRUE;
     return 1;
   }
   convert_uuid2string(value->x.string.value.str, cass_data);
   value->x.string.value.length= 36;
-  value->x.string.nonfreeable= FALSE;
   return 0;
 }
 
@@ -1117,7 +1152,7 @@ bool dyncol_to_cassandraUUID(DYNAMIC_COLUMN_VALUE *value,
   if (init_dynamic_string(&tmp, NULL, 1024, 1024))
     return true;
   enum enum_dyncol_func_result rc=
-    dynamic_column_val_str(&tmp, value, &my_charset_latin1_bin, FALSE);
+    mariadb_dyncol_val_str(&tmp, value, &my_charset_latin1_bin, '\0');
   if (rc < 0 || tmp.length != 36 || convert_string2uuid((char *)buff, tmp.str))
   {
     dynstr_free(&tmp);
@@ -1132,7 +1167,8 @@ bool dyncol_to_cassandraUUID(DYNAMIC_COLUMN_VALUE *value,
 
 bool cassandra_to_dyncol_intBool(const char *cass_data,
                                  int cass_data_len,
-                                 DYNAMIC_COLUMN_VALUE *value)
+                                 DYNAMIC_COLUMN_VALUE *value,
+                                 MEM_ROOT *mem_root __attribute__((unused)))
 {
   value->type= DYN_COL_INT;
   value->x.long_value= (cass_data[0] ? 1 : 0);
@@ -1145,7 +1181,7 @@ bool dyncol_to_cassandraBool(DYNAMIC_COLUMN_VALUE *value,
 {
   longlong tmp;
   enum enum_dyncol_func_result rc=
-    dynamic_column_val_long(&tmp, value);
+    mariadb_dyncol_val_long(&tmp, value);
   if (rc < 0)
     return true;
   ((char *)buff)[0]= (tmp ? 1 : 0);
@@ -1430,10 +1466,7 @@ bool ha_cassandra::setup_field_converters(Field **field_arg, uint n_fields)
       (CASSANDRA_TYPE_DEF *)(field_converters + n_fields);
     special_type_field_names=
       ((LEX_STRING*)(special_type_field_converters + max_non_default_fields));
-  }
 
-  if (dyncol_set)
-  {
     if (init_dynamic_array(&dynamic_values,
                            sizeof(DYNAMIC_COLUMN_VALUE),
                            DYNCOL_USUAL, DYNCOL_DELTA))
@@ -1667,14 +1700,6 @@ void ha_cassandra::print_conversion_error(const char *field_name,
 }
 
 
-void free_strings(DYNAMIC_COLUMN_VALUE *vals, uint num)
-{
-  for (uint i= 0; i < num; i++)
-    if (vals[i].type == DYN_COL_STRING &&
-        !vals[i].x.string.nonfreeable)
-      my_free(vals[i].x.string.value.str);
-}
-
 
 CASSANDRA_TYPE_DEF * ha_cassandra::get_cassandra_field_def(char *cass_name,
                                                            int cass_name_len)
@@ -1695,6 +1720,7 @@ CASSANDRA_TYPE_DEF * ha_cassandra::get_cassandra_field_def(char *cass_name,
 
 int ha_cassandra::read_cassandra_columns(bool unpack_pk)
 {
+  MEM_ROOT strings_root;
   char *cass_name;
   char *cass_value;
   int cass_value_len, cass_name_len;
@@ -1702,6 +1728,7 @@ int ha_cassandra::read_cassandra_columns(bool unpack_pk)
   int res= 0;
   ulong total_name_len= 0;
 
+  clear_alloc_root(&strings_root);
   /*
     cassandra_to_mariadb() calls will use field->store(...) methods, which
     require that the column is in the table->write_set
@@ -1770,7 +1797,8 @@ int ha_cassandra::read_cassandra_columns(bool unpack_pk)
       }
 
       if ((res= (*(type->cassandra_to_dynamic))(cass_value,
-                                                cass_value_len, &val)) ||
+                                                cass_value_len, &val,
+                                                &strings_root)) ||
           insert_dynamic(&dynamic_names, (uchar *) &nm) ||
           insert_dynamic(&dynamic_values, (uchar *) &val))
       {
@@ -1778,10 +1806,9 @@ int ha_cassandra::read_cassandra_columns(bool unpack_pk)
         {
           print_conversion_error(cass_name, cass_value, cass_value_len);
         }
-        free_strings((DYNAMIC_COLUMN_VALUE *)dynamic_values.buffer,
-                     dynamic_values.elements);
+        free_strings_memroot(&strings_root);
         // EOM shouldm be already reported if happened
-        res=1;
+        res= 1;
         goto err;
       }
     }
@@ -1790,21 +1817,17 @@ int ha_cassandra::read_cassandra_columns(bool unpack_pk)
   dynamic_rec.length= 0;
   if (dyncol_set)
   {
-    if (dynamic_column_create_many_internal_fmt(&dynamic_rec,
-                                                dynamic_names.elements,
-                                                dynamic_names.buffer,
-                                                (DYNAMIC_COLUMN_VALUE *)
-                                                dynamic_values.buffer,
-                                                FALSE,
-                                                TRUE) < 0)
+    if (mariadb_dyncol_create_many_named(&dynamic_rec,
+                                         dynamic_names.elements,
+                                         (LEX_STRING *)dynamic_names.buffer,
+                                         (DYNAMIC_COLUMN_VALUE *)
+                                         dynamic_values.buffer,
+                                         FALSE) < 0)
       dynamic_rec.length= 0;
 
-    free_strings((DYNAMIC_COLUMN_VALUE *)dynamic_values.buffer,
-                 dynamic_values.elements);
+    free_strings_memroot(&strings_root);
     dynamic_values.elements= dynamic_names.elements= 0;
-  }
-  if (dyncol_set)
-  {
+
     if (dynamic_rec.length == 0)
       table->field[dyncol_field]->set_null();
     else
@@ -1836,11 +1859,14 @@ err:
   return res;
 }
 
-int ha_cassandra::read_dyncol(DYNAMIC_ARRAY *vals, DYNAMIC_ARRAY *names,
-                              String *valcol, char **freenames)
+int ha_cassandra::read_dyncol(uint *count,
+                              DYNAMIC_COLUMN_VALUE **vals,
+                              LEX_STRING **names,
+                              String *valcol)
 {
   String *strcol;
   DYNAMIC_COLUMN col;
+
   enum enum_dyncol_func_result rc;
   DBUG_ENTER("ha_cassandra::read_dyncol");
 
@@ -1850,8 +1876,9 @@ int ha_cassandra::read_dyncol(DYNAMIC_ARRAY *vals, DYNAMIC_ARRAY *names,
   strcol= field->val_str(NULL, valcol);
   if (field->is_null())
   {
-    bzero(vals, sizeof(DYNAMIC_ARRAY));
-    bzero(names, sizeof(DYNAMIC_ARRAY));
+    *count= 0;
+    *names= 0;
+    *vals= 0;
     DBUG_RETURN(0); // nothing to write
   }
   /*
@@ -1861,7 +1888,7 @@ int ha_cassandra::read_dyncol(DYNAMIC_ARRAY *vals, DYNAMIC_ARRAY *names,
   bzero(&col, sizeof(col));
   col.str= (char *)strcol->ptr();
   col.length= strcol->length();
-  if ((rc= dynamic_column_vals(&col, names, vals, freenames)) < 0)
+  if ((rc= mariadb_dyncol_unpack(&col, count, names, vals)) < 0)
   {
     dynamic_column_error_message(rc);
     DBUG_RETURN(HA_ERR_INTERNAL_ERROR);
@@ -1869,34 +1896,33 @@ int ha_cassandra::read_dyncol(DYNAMIC_ARRAY *vals, DYNAMIC_ARRAY *names,
   DBUG_RETURN(0);
 }
 
-int ha_cassandra::write_dynamic_row(DYNAMIC_ARRAY *vals, DYNAMIC_ARRAY *names)
+int ha_cassandra::write_dynamic_row(uint count,
+                                    DYNAMIC_COLUMN_VALUE *vals,
+                                    LEX_STRING *names)
 {
   uint i;
   DBUG_ENTER("ha_cassandra::write_dynamic_row");
   DBUG_ASSERT(dyncol_set);
 
 
-  DBUG_ASSERT(names->elements == vals->elements);
-  for (i= 0; i < names->elements; i++)
+  for (i= 0; i < count; i++)
   {
     char buff[16];
     CASSANDRA_TYPE_DEF *type;
     void *freemem= NULL;
     char *cass_data;
     int cass_data_len;
-    LEX_STRING *name= dynamic_element(names, i, LEX_STRING*);
-    DYNAMIC_COLUMN_VALUE *val= dynamic_element(vals, i, DYNAMIC_COLUMN_VALUE*);
 
-    DBUG_PRINT("info", ("field %*s", (int)name->length, name->str));
-    type= get_cassandra_field_def(name->str, (int) name->length);
-    if ((*type->dynamic_to_cassandra)(val, &cass_data, &cass_data_len,
+    DBUG_PRINT("info", ("field %*s", (int)names[i].length, names[i].str));
+    type= get_cassandra_field_def(names[i].str, (int) names[i].length);
+    if ((*type->dynamic_to_cassandra)(vals +i, &cass_data, &cass_data_len,
                                       buff, &freemem))
     {
       my_error(ER_WARN_DATA_OUT_OF_RANGE, MYF(0),
-               name->str, insert_lineno);
-      DBUG_RETURN(HA_ERR_AUTOINC_ERANGE);
+               names[i].str, insert_lineno);
+      DBUG_RETURN(HA_ERR_GENERIC);
     }
-    se->add_insert_column(name->str, name->length,
+    se->add_insert_column(names[i].str, names[i].length,
                           cass_data, cass_data_len);
     if (freemem)
       my_free(freemem);
@@ -1904,13 +1930,19 @@ int ha_cassandra::write_dynamic_row(DYNAMIC_ARRAY *vals, DYNAMIC_ARRAY *names)
   DBUG_RETURN(0);
 }
 
-void ha_cassandra::free_dynamic_row(DYNAMIC_ARRAY *vals, DYNAMIC_ARRAY *names,
-                                    char *free_names)
+void ha_cassandra::free_dynamic_row(DYNAMIC_COLUMN_VALUE **vals,
+                                    LEX_STRING **names)
 {
-  delete_dynamic(names);
-  delete_dynamic(vals);
-  if (free_names)
-    my_free(free_names);
+  if (*vals)
+  {
+    my_free(*vals);
+    *vals= 0;
+  }
+  if (*names)
+  {
+    my_free(*names);
+    *names= 0;
+  }
 }
 
 int ha_cassandra::write_row(uchar *buf)
@@ -1949,13 +1981,14 @@ int ha_cassandra::write_row(uchar *buf)
     if (dyncol_set && dyncol_field == i)
     {
       String valcol;
-      DYNAMIC_ARRAY vals, names;
-      char *free_names= NULL;
+      DYNAMIC_COLUMN_VALUE *vals;
+      LEX_STRING *names;
+      uint count;
       int rc;
       DBUG_ASSERT(field_converters[i] == NULL);
-      if (!(rc= read_dyncol(&vals, &names, &valcol, &free_names)))
-        rc= write_dynamic_row(&vals, &names);
-      free_dynamic_row(&vals, &names, free_names);
+      if (!(rc= read_dyncol(&count, &vals, &names, &valcol)))
+        rc= write_dynamic_row(count, vals, names);
+      free_dynamic_row(&vals, &names);
       if (rc)
       {
         dbug_tmp_restore_column_map(table->read_set, old_map);
@@ -2336,9 +2369,10 @@ public:
 
 int ha_cassandra::update_row(const uchar *old_data, uchar *new_data)
 {
-  DYNAMIC_ARRAY oldvals, oldnames, vals, names;
+  DYNAMIC_COLUMN_VALUE *oldvals, *vals;
+  LEX_STRING *oldnames, *names;
+  uint oldcount, count;
   String oldvalcol, valcol;
-  char *oldfree_names= NULL, *free_names= NULL;
   my_bitmap_map *old_map;
   int res;
   DBUG_ENTER("ha_cassandra::update_row");
@@ -2381,12 +2415,12 @@ int ha_cassandra::update_row(const uchar *old_data, uchar *new_data)
     my_ptrdiff_t diff;
     diff= (my_ptrdiff_t) (old_data - new_data);
     field->move_field_offset(diff);      // Points now at old_data
-    if ((res= read_dyncol(&oldvals, &oldnames, &oldvalcol, &oldfree_names)))
+    if ((res= read_dyncol(&oldcount, &oldvals, &oldnames, &oldvalcol)))
       DBUG_RETURN(res);
     field->move_field_offset(-diff);     // back to new_data
-    if ((res= read_dyncol(&vals, &names, &valcol, &free_names)))
+    if ((res= read_dyncol(&count, &vals, &names, &valcol)))
     {
-      free_dynamic_row(&oldnames, &oldvals, oldfree_names);
+      free_dynamic_row(&oldvals, &oldnames);
       DBUG_RETURN(res);
     }
   }
@@ -2399,9 +2433,9 @@ int ha_cassandra::update_row(const uchar *old_data, uchar *new_data)
     */
     Column_name_enumerator_impl name_enumerator(this);
     se->add_row_deletion(old_key, old_key_len, &name_enumerator,
-                         (LEX_STRING *)oldnames.buffer,
-                         (dyncol_set ? oldnames.elements : 0));
-    oldnames.elements= oldvals.elements= 0; // they will be deleted
+                         oldnames,
+                         (dyncol_set ? oldcount : 0));
+    oldcount= 0; // they will be deleted
   }
 
   se->start_row_insert(new_key, new_key_len);
@@ -2414,7 +2448,7 @@ int ha_cassandra::update_row(const uchar *old_data, uchar *new_data)
     if (dyncol_set && dyncol_field == i)
     {
       DBUG_ASSERT(field_converters[i] == NULL);
-      if ((res= write_dynamic_row(&vals, &names)))
+      if ((res= write_dynamic_row(count, vals, names)))
         goto err;
     }
     else
@@ -2434,24 +2468,19 @@ int ha_cassandra::update_row(const uchar *old_data, uchar *new_data)
   {
     /* find removed fields */
     uint i= 0, j= 0;
-    LEX_STRING *onames= (LEX_STRING *)oldnames.buffer;
-    LEX_STRING *nnames= (LEX_STRING *)names.buffer;
     /* both array are sorted */
-    for(; i < oldnames.elements; i++)
+    for(; i < oldcount; i++)
     {
       int scmp= 0;
-      while (j < names.elements &&
-             (nnames[j].length < onames[i].length ||
-             (nnames[j].length == onames[i].length &&
-              (scmp= memcmp(nnames[j].str, onames[i].str,
-                            onames[i].length)) < 0)))
+      while (j < count &&
+             (scmp = mariadb_dyncol_column_cmp_named(names + j,
+                                                     oldnames + i)) < 0)
         j++;
-      if (j < names.elements &&
-          nnames[j].length == onames[i].length &&
+      if (j < count &&
           scmp == 0)
         j++;
       else
-        se->add_insert_delete_column(onames[i].str, onames[i].length);
+        se->add_insert_delete_column(oldnames[i].str, oldnames[i].length);
     }
   }
 
@@ -2465,8 +2494,8 @@ int ha_cassandra::update_row(const uchar *old_data, uchar *new_data)
 err:
   if (dyncol_set)
   {
-    free_dynamic_row(&oldnames, &oldvals, oldfree_names);
-    free_dynamic_row(&names, &vals, free_names);
+    free_dynamic_row(&oldvals, &oldnames);
+    free_dynamic_row(&vals, &names);
   }
 
   DBUG_RETURN(res? HA_ERR_INTERNAL_ERROR: 0);
diff --git a/storage/cassandra/ha_cassandra.h b/storage/cassandra/ha_cassandra.h
index be8a49af493..ecd8bdb57e3 100644
--- a/storage/cassandra/ha_cassandra.h
+++ b/storage/cassandra/ha_cassandra.h
@@ -37,6 +37,8 @@ typedef struct st_cassandra_share {
 } CASSANDRA_SHARE;
 
 class ColumnDataConverter;
+struct st_dynamic_column_value;
+typedef struct st_dynamic_column_value DYNAMIC_COLUMN_VALUE;
 
 struct ha_table_option_struct;
 
@@ -45,7 +47,8 @@ struct st_dynamic_column_value;
 
 typedef bool (* CAS2DYN_CONVERTER)(const char *cass_data,
                                    int cass_data_len,
-                                   struct st_dynamic_column_value *value);
+                                   struct st_dynamic_column_value *value,
+                                   MEM_ROOT *mem_root);
 typedef bool (* DYN2CAS_CONVERTER)(struct st_dynamic_column_value *value,
                                    char **cass_data,
                                    int *cass_data_len,
@@ -227,11 +230,14 @@ private:
   bool source_exhausted;
   bool mrr_start_read();
   int check_field_options(Field **fields);
-  int read_dyncol(DYNAMIC_ARRAY *vals, DYNAMIC_ARRAY *names,
-                  String *valcol, char **freenames);
-  int write_dynamic_row(DYNAMIC_ARRAY *names, DYNAMIC_ARRAY *vals);
-  void static free_dynamic_row(DYNAMIC_ARRAY *vals, DYNAMIC_ARRAY *names,
-                               char *free_names);
+  int read_dyncol(uint *count,
+                  DYNAMIC_COLUMN_VALUE **vals, LEX_STRING **names,
+                  String *valcol);
+  int write_dynamic_row(uint count,
+                        DYNAMIC_COLUMN_VALUE *vals,
+                        LEX_STRING *names);
+  void static free_dynamic_row(DYNAMIC_COLUMN_VALUE **vals,
+                               LEX_STRING **names);
   CASSANDRA_TYPE_DEF * get_cassandra_field_def(char *cass_name,
                                                int cass_name_length);
 public: