MDEV-11371 - column compression

Storage engine independent support for column compression.

TINYBLOB, BLOB, MEDIUMBLOB, LONGBLOB, TINYTEXT, TEXT, MEDIUMTEXT, LONGTEXT,
VARCHAR and VARBINARY columns can be compressed.

New COMPRESSED column attribute added:
COMPRESSED[=<compression_method>]

System variables added:
column_compression_threshold
column_compression_zlib_level
column_compression_zlib_strategy
column_compression_zlib_wrap

Status variables added:
Column_compressions
Column_decompressions

Limitations:
- the only supported method currently is zlib
- CSV storage engine stores data uncompressed on-disk even if COMPRESSED
  attribute is present
- it is not possible to create indexes over compressed columns.

32 files changed, 2647 insertions, 150 deletions

diff --git a/include/mysql.h.pp b/include/mysql.h.pp
index d061df71944..4a196b3c772 100644
--- a/include/mysql.h.pp
+++ b/include/mysql.h.pp
@@ -66,6 +66,8 @@ enum enum_field_types { MYSQL_TYPE_DECIMAL, MYSQL_TYPE_TINY,
                         MYSQL_TYPE_TIMESTAMP2,
                         MYSQL_TYPE_DATETIME2,
                         MYSQL_TYPE_TIME2,
+                        MYSQL_TYPE_BLOB_COMPRESSED= 140,
+                        MYSQL_TYPE_VARCHAR_COMPRESSED= 141,
                         MYSQL_TYPE_NEWDECIMAL=246,
    MYSQL_TYPE_ENUM=247,
    MYSQL_TYPE_SET=248,
diff --git a/include/mysql_com.h b/include/mysql_com.h
index 1deaa4c771c..71858c93046 100644
--- a/include/mysql_com.h
+++ b/include/mysql_com.h
@@ -485,7 +485,10 @@ enum enum_field_types { MYSQL_TYPE_DECIMAL, MYSQL_TYPE_TINY,
                         MYSQL_TYPE_TIMESTAMP2,
                         MYSQL_TYPE_DATETIME2,
                         MYSQL_TYPE_TIME2,
-                        
+                        /* Compressed types are only used internally for RBR. */
+                        MYSQL_TYPE_BLOB_COMPRESSED= 140,
+                        MYSQL_TYPE_VARCHAR_COMPRESSED= 141,
+
                         MYSQL_TYPE_NEWDECIMAL=246,
 			MYSQL_TYPE_ENUM=247,
 			MYSQL_TYPE_SET=248,
diff --git a/libmariadb b/libmariadb
-Subproject 11321f16bfcd92e210d5736af7b7d5073a89c2e
+Subproject e069fb8e76eeab096b8255805244f73048e3575
diff --git a/libmysqld/CMakeLists.txt b/libmysqld/CMakeLists.txt
index bd0ff8dcc9a..d85a9cb7c83 100644
--- a/libmysqld/CMakeLists.txt
+++ b/libmysqld/CMakeLists.txt
@@ -43,7 +43,7 @@ SET(SQL_EMBEDDED_SOURCES emb_qcache.cc libmysqld.c lib_sql.cc
            ../sql-common/my_user.c ../sql-common/pack.c
            ../sql-common/client_plugin.c ../sql-common/mysql_async.c
            ../sql/password.c ../sql/discover.cc ../sql/derror.cc 
-           ../sql/field.cc ../sql/field_conv.cc
+           ../sql/field.cc ../sql/field_conv.cc ../sql/field_comp.cc
            ../sql/filesort_utils.cc ../sql/sql_digest.cc
            ../sql/filesort.cc ../sql/gstream.cc ../sql/slave.cc
            ../sql/signal_handler.cc
diff --git a/mysql-test/r/column_compression.result b/mysql-test/r/column_compression.result
new file mode 100644
index 00000000000..243a118a706
--- /dev/null
+++ b/mysql-test/r/column_compression.result
@@ -0,0 +1,1362 @@
+SET column_compression_zlib_wrap=true;
+FLUSH STATUS;
+CREATE TABLE t1(a BLOB COMPRESSED, KEY(a(10)));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+CREATE TABLE t1(a BLOB COMPRESSED);
+ALTER TABLE t1 ADD KEY(a(10));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` blob /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+# Make sure column was actually compressed
+INSERT INTO t1 VALUES(REPEAT('a', 1000));
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	2
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+28
+# Make sure ALTER TABLE rebuilds table
+ALTER TABLE t1 MODIFY COLUMN a BLOB;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` blob DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	3
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+1008
+# Rebuild back
+ALTER TABLE t1 MODIFY COLUMN a BLOB COMPRESSED;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` blob /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	5
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+28
+# Make sure CREATE TABLE ... LIKE inherits compression
+CREATE TABLE t2 LIKE t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` blob /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TABLE t2;
+# Make sure implicit CREATE TABLE ... SELECT inherits compression
+CREATE TABLE t2 SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` blob /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	7
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+28
+DROP TABLE t2;
+# Make sure explicit CREATE TABLE ... SELECT doesn't inherit compression
+CREATE TABLE t2(a BLOB) SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` blob DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	8
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+1008
+DROP TABLE t2;
+# Make sure engine change works
+ALTER TABLE t1 ENGINE=InnoDB;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` blob /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	10
+# Make sure online add column works (requires InnoDB)
+ALTER TABLE t1 ADD COLUMN b BLOB COMPRESSED DEFAULT "must be visible";
+SELECT LEFT(a, 10), LENGTH(a), b FROM t1;
+LEFT(a, 10)	LENGTH(a)	b
+aaaaaaaaaa	1000	must be visible
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	12
+ALTER TABLE t1 DROP COLUMN b;
+ALTER TABLE t1 ENGINE=MyISAM;
+TRUNCATE TABLE t1;
+# Make sure column_compression_zlib_level works
+SET column_compression_zlib_level= 1;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	3
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+40
+TRUNCATE TABLE t1;
+SET column_compression_zlib_level= 9;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+36
+SET column_compression_zlib_level= DEFAULT;
+TRUNCATE TABLE t1;
+# No compression, original data shorter than compressed
+INSERT INTO t1 VALUES('a');
+SELECT a, LENGTH(a) FROM t1;
+a	LENGTH(a)
+a	1
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+20
+# Coverage for store(double) and store(longlong)
+INSERT INTO t1 VALUES(3.14),(CAST(9.8 AS DOUBLE)),(1),(''),(NULL);
+# and for sort_string()
+SELECT * FROM t1 ORDER BY a;
+a
+NULL
+
+1
+3.14
+9.8
+a
+# Coverage for val_real() and val_int()
+SELECT a+1 FROM t1 ORDER BY 1;
+a+1
+NULL
+1
+1
+2
+4.140000000000001
+10.8
+Warnings:
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+DROP TABLE t1;
+#
+# MDEV-13540 - Server crashes in copy or Assertion `0' failed in virtual
+#              Field* Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c1 BLOB COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t1 VALUES ('foo'),('bar');
+CREATE TABLE t2 (c2 BLOB COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t2 VALUES ('qux'),('abc');
+SELECT * FROM t1 WHERE c1 NOT IN ( SELECT c2 FROM t2 WHERE c2 = c1 );
+c1
+foo
+bar
+DROP TABLE t1, t2;
+#
+# MDEV-13541 - Server crashes in next_breadth_first_tab or Assertion `0'
+#              failed in Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c BLOB COMPRESSED) ENGINE=InnoDB;
+INSERT INTO t1 VALUES ('foo'),('bar');
+SELECT DISTINCT c FROM t1;
+c
+foo
+bar
+DROP TABLE t1;
+FLUSH STATUS;
+CREATE TABLE t1(a TEXT COMPRESSED, KEY(a(10)));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+CREATE TABLE t1(a TEXT COMPRESSED);
+ALTER TABLE t1 ADD KEY(a(10));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` text /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+# Make sure column was actually compressed
+INSERT INTO t1 VALUES(REPEAT('a', 1000));
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	2
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+28
+# Make sure ALTER TABLE rebuilds table
+ALTER TABLE t1 MODIFY COLUMN a TEXT;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` text DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	3
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+1008
+# Rebuild back
+ALTER TABLE t1 MODIFY COLUMN a TEXT COMPRESSED;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` text /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	5
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+28
+# Make sure CREATE TABLE ... LIKE inherits compression
+CREATE TABLE t2 LIKE t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` text /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TABLE t2;
+# Make sure implicit CREATE TABLE ... SELECT inherits compression
+CREATE TABLE t2 SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` text /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	7
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+28
+DROP TABLE t2;
+# Make sure explicit CREATE TABLE ... SELECT doesn't inherit compression
+CREATE TABLE t2(a TEXT) SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` text DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	8
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+1008
+DROP TABLE t2;
+# Make sure engine change works
+ALTER TABLE t1 ENGINE=InnoDB;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` text /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	10
+# Make sure online add column works (requires InnoDB)
+ALTER TABLE t1 ADD COLUMN b TEXT COMPRESSED DEFAULT "must be visible";
+SELECT LEFT(a, 10), LENGTH(a), b FROM t1;
+LEFT(a, 10)	LENGTH(a)	b
+aaaaaaaaaa	1000	must be visible
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	12
+ALTER TABLE t1 DROP COLUMN b;
+ALTER TABLE t1 ENGINE=MyISAM;
+TRUNCATE TABLE t1;
+# Make sure column_compression_zlib_level works
+SET column_compression_zlib_level= 1;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	3
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+40
+TRUNCATE TABLE t1;
+SET column_compression_zlib_level= 9;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+36
+SET column_compression_zlib_level= DEFAULT;
+TRUNCATE TABLE t1;
+# No compression, original data shorter than compressed
+INSERT INTO t1 VALUES('a');
+SELECT a, LENGTH(a) FROM t1;
+a	LENGTH(a)
+a	1
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+20
+# Coverage for store(double) and store(longlong)
+INSERT INTO t1 VALUES(3.14),(CAST(9.8 AS DOUBLE)),(1),(''),(NULL);
+# and for sort_string()
+SELECT * FROM t1 ORDER BY a;
+a
+NULL
+
+1
+3.14
+9.8
+a
+# Coverage for val_real() and val_int()
+SELECT a+1 FROM t1 ORDER BY 1;
+a+1
+NULL
+1
+1
+2
+4.140000000000001
+10.8
+Warnings:
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+DROP TABLE t1;
+#
+# MDEV-13540 - Server crashes in copy or Assertion `0' failed in virtual
+#              Field* Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c1 TEXT COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t1 VALUES ('foo'),('bar');
+CREATE TABLE t2 (c2 TEXT COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t2 VALUES ('qux'),('abc');
+SELECT * FROM t1 WHERE c1 NOT IN ( SELECT c2 FROM t2 WHERE c2 = c1 );
+c1
+foo
+bar
+DROP TABLE t1, t2;
+#
+# MDEV-13541 - Server crashes in next_breadth_first_tab or Assertion `0'
+#              failed in Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c TEXT COMPRESSED) ENGINE=InnoDB;
+INSERT INTO t1 VALUES ('foo'),('bar');
+SELECT DISTINCT c FROM t1;
+c
+foo
+bar
+DROP TABLE t1;
+FLUSH STATUS;
+CREATE TABLE t1(a VARBINARY(10000) COMPRESSED, KEY(a(10)));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+CREATE TABLE t1(a VARBINARY(10000) COMPRESSED);
+ALTER TABLE t1 ADD KEY(a(10));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` varbinary(10000) /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+# Make sure column was actually compressed
+INSERT INTO t1 VALUES(REPEAT('a', 1000));
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	2
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+28
+# Make sure ALTER TABLE rebuilds table
+ALTER TABLE t1 MODIFY COLUMN a VARBINARY(10000);
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` varbinary(10000) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	3
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+1008
+# Rebuild back
+ALTER TABLE t1 MODIFY COLUMN a VARBINARY(10000) COMPRESSED;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` varbinary(10000) /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	5
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+28
+# Make sure CREATE TABLE ... LIKE inherits compression
+CREATE TABLE t2 LIKE t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` varbinary(10000) /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TABLE t2;
+# Make sure implicit CREATE TABLE ... SELECT inherits compression
+CREATE TABLE t2 SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` varbinary(10000) /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	7
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+28
+DROP TABLE t2;
+# Make sure explicit CREATE TABLE ... SELECT doesn't inherit compression
+CREATE TABLE t2(a VARBINARY(10000)) SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` varbinary(10000) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	8
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+1008
+DROP TABLE t2;
+# Make sure engine change works
+ALTER TABLE t1 ENGINE=InnoDB;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` varbinary(10000) /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	10
+# Make sure online add column works (requires InnoDB)
+ALTER TABLE t1 ADD COLUMN b VARBINARY(10000) COMPRESSED DEFAULT "must be visible";
+SELECT LEFT(a, 10), LENGTH(a), b FROM t1;
+LEFT(a, 10)	LENGTH(a)	b
+aaaaaaaaaa	1000	must be visible
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	12
+ALTER TABLE t1 DROP COLUMN b;
+ALTER TABLE t1 ENGINE=MyISAM;
+TRUNCATE TABLE t1;
+# Make sure column_compression_zlib_level works
+SET column_compression_zlib_level= 1;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	3
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+40
+TRUNCATE TABLE t1;
+SET column_compression_zlib_level= 9;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+32
+SET column_compression_zlib_level= DEFAULT;
+TRUNCATE TABLE t1;
+# No compression, original data shorter than compressed
+INSERT INTO t1 VALUES('a');
+SELECT a, LENGTH(a) FROM t1;
+a	LENGTH(a)
+a	1
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+20
+# Coverage for store(double) and store(longlong)
+INSERT INTO t1 VALUES(3.14),(CAST(9.8 AS DOUBLE)),(1),(''),(NULL);
+# and for sort_string()
+SELECT * FROM t1 ORDER BY a;
+a
+NULL
+
+1
+3.14
+9.8
+a
+# Coverage for val_real() and val_int()
+SELECT a+1 FROM t1 ORDER BY 1;
+a+1
+NULL
+1
+1
+2
+4.140000000000001
+10.8
+Warnings:
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+DROP TABLE t1;
+#
+# MDEV-13540 - Server crashes in copy or Assertion `0' failed in virtual
+#              Field* Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c1 VARBINARY(10000) COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t1 VALUES ('foo'),('bar');
+CREATE TABLE t2 (c2 VARBINARY(10000) COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t2 VALUES ('qux'),('abc');
+SELECT * FROM t1 WHERE c1 NOT IN ( SELECT c2 FROM t2 WHERE c2 = c1 );
+c1
+foo
+bar
+DROP TABLE t1, t2;
+#
+# MDEV-13541 - Server crashes in next_breadth_first_tab or Assertion `0'
+#              failed in Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c VARBINARY(10000) COMPRESSED) ENGINE=InnoDB;
+INSERT INTO t1 VALUES ('foo'),('bar');
+SELECT DISTINCT c FROM t1;
+c
+foo
+bar
+DROP TABLE t1;
+FLUSH STATUS;
+CREATE TABLE t1(a VARCHAR(10000) COMPRESSED, KEY(a(10)));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+CREATE TABLE t1(a VARCHAR(10000) COMPRESSED);
+ALTER TABLE t1 ADD KEY(a(10));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` varchar(10000) /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+# Make sure column was actually compressed
+INSERT INTO t1 VALUES(REPEAT('a', 1000));
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	2
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+28
+# Make sure ALTER TABLE rebuilds table
+ALTER TABLE t1 MODIFY COLUMN a VARCHAR(10000);
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` varchar(10000) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	3
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+1008
+# Rebuild back
+ALTER TABLE t1 MODIFY COLUMN a VARCHAR(10000) COMPRESSED;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` varchar(10000) /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	5
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+28
+# Make sure CREATE TABLE ... LIKE inherits compression
+CREATE TABLE t2 LIKE t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` varchar(10000) /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TABLE t2;
+# Make sure implicit CREATE TABLE ... SELECT inherits compression
+CREATE TABLE t2 SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` varchar(10000) /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	7
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+28
+DROP TABLE t2;
+# Make sure explicit CREATE TABLE ... SELECT doesn't inherit compression
+CREATE TABLE t2(a VARCHAR(10000)) SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` varchar(10000) DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	8
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+1008
+DROP TABLE t2;
+# Make sure engine change works
+ALTER TABLE t1 ENGINE=InnoDB;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` varchar(10000) /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	10
+# Make sure online add column works (requires InnoDB)
+ALTER TABLE t1 ADD COLUMN b VARCHAR(10000) COMPRESSED DEFAULT "must be visible";
+SELECT LEFT(a, 10), LENGTH(a), b FROM t1;
+LEFT(a, 10)	LENGTH(a)	b
+aaaaaaaaaa	1000	must be visible
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	12
+ALTER TABLE t1 DROP COLUMN b;
+ALTER TABLE t1 ENGINE=MyISAM;
+TRUNCATE TABLE t1;
+# Make sure column_compression_zlib_level works
+SET column_compression_zlib_level= 1;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	3
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+40
+TRUNCATE TABLE t1;
+SET column_compression_zlib_level= 9;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+32
+SET column_compression_zlib_level= DEFAULT;
+TRUNCATE TABLE t1;
+# No compression, original data shorter than compressed
+INSERT INTO t1 VALUES('a');
+SELECT a, LENGTH(a) FROM t1;
+a	LENGTH(a)
+a	1
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+20
+# Coverage for store(double) and store(longlong)
+INSERT INTO t1 VALUES(3.14),(CAST(9.8 AS DOUBLE)),(1),(''),(NULL);
+# and for sort_string()
+SELECT * FROM t1 ORDER BY a;
+a
+NULL
+
+1
+3.14
+9.8
+a
+# Coverage for val_real() and val_int()
+SELECT a+1 FROM t1 ORDER BY 1;
+a+1
+NULL
+1
+1
+2
+4.140000000000001
+10.8
+Warnings:
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+DROP TABLE t1;
+#
+# MDEV-13540 - Server crashes in copy or Assertion `0' failed in virtual
+#              Field* Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c1 VARCHAR(10000) COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t1 VALUES ('foo'),('bar');
+CREATE TABLE t2 (c2 VARCHAR(10000) COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t2 VALUES ('qux'),('abc');
+SELECT * FROM t1 WHERE c1 NOT IN ( SELECT c2 FROM t2 WHERE c2 = c1 );
+c1
+foo
+bar
+DROP TABLE t1, t2;
+#
+# MDEV-13541 - Server crashes in next_breadth_first_tab or Assertion `0'
+#              failed in Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c VARCHAR(10000) COMPRESSED) ENGINE=InnoDB;
+INSERT INTO t1 VALUES ('foo'),('bar');
+SELECT DISTINCT c FROM t1;
+c
+foo
+bar
+DROP TABLE t1;
+FLUSH STATUS;
+CREATE TABLE t1(a TEXT CHARSET ucs2 COMPRESSED, KEY(a(10)));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+CREATE TABLE t1(a TEXT CHARSET ucs2 COMPRESSED);
+ALTER TABLE t1 ADD KEY(a(10));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` text /*!100301 COMPRESSED*/ CHARACTER SET ucs2 DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+# Make sure column was actually compressed
+INSERT INTO t1 VALUES(REPEAT('a', 1000));
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	2000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	2
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+36
+# Make sure ALTER TABLE rebuilds table
+ALTER TABLE t1 MODIFY COLUMN a TEXT;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` text DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	3
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+1008
+# Rebuild back
+ALTER TABLE t1 MODIFY COLUMN a TEXT CHARSET ucs2 COMPRESSED;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` text /*!100301 COMPRESSED*/ CHARACTER SET ucs2 DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	2000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	5
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+36
+# Make sure CREATE TABLE ... LIKE inherits compression
+CREATE TABLE t2 LIKE t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` text /*!100301 COMPRESSED*/ CHARACTER SET ucs2 DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TABLE t2;
+# Make sure implicit CREATE TABLE ... SELECT inherits compression
+CREATE TABLE t2 SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` text /*!100301 COMPRESSED*/ CHARACTER SET ucs2 DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	2000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	7
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+36
+DROP TABLE t2;
+# Make sure explicit CREATE TABLE ... SELECT doesn't inherit compression
+CREATE TABLE t2(a TEXT) SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` text DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	8
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+1008
+DROP TABLE t2;
+# Make sure engine change works
+ALTER TABLE t1 ENGINE=InnoDB;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` text /*!100301 COMPRESSED*/ CHARACTER SET ucs2 DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	2000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	10
+# Make sure online add column works (requires InnoDB)
+ALTER TABLE t1 ADD COLUMN b TEXT CHARSET ucs2 COMPRESSED DEFAULT "must be visible";
+SELECT LEFT(a, 10), LENGTH(a), b FROM t1;
+LEFT(a, 10)	LENGTH(a)	b
+aaaaaaaaaa	2000	must be visible
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	12
+ALTER TABLE t1 DROP COLUMN b;
+ALTER TABLE t1 ENGINE=MyISAM;
+TRUNCATE TABLE t1;
+# Make sure column_compression_zlib_level works
+SET column_compression_zlib_level= 1;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	3
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+60
+TRUNCATE TABLE t1;
+SET column_compression_zlib_level= 9;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+44
+SET column_compression_zlib_level= DEFAULT;
+TRUNCATE TABLE t1;
+# No compression, original data shorter than compressed
+INSERT INTO t1 VALUES('a');
+SELECT a, LENGTH(a) FROM t1;
+a	LENGTH(a)
+a	2
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+20
+# Coverage for store(double) and store(longlong)
+INSERT INTO t1 VALUES(3.14),(CAST(9.8 AS DOUBLE)),(1),(''),(NULL);
+# and for sort_string()
+SELECT * FROM t1 ORDER BY a;
+a
+NULL
+
+1
+3.14
+9.8
+a
+# Coverage for val_real() and val_int()
+SELECT a+1 FROM t1 ORDER BY 1;
+a+1
+NULL
+1
+1
+2
+4.140000000000001
+10.8
+Warnings:
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+DROP TABLE t1;
+#
+# MDEV-13540 - Server crashes in copy or Assertion `0' failed in virtual
+#              Field* Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c1 TEXT CHARSET ucs2 COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t1 VALUES ('foo'),('bar');
+CREATE TABLE t2 (c2 TEXT CHARSET ucs2 COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t2 VALUES ('qux'),('abc');
+SELECT * FROM t1 WHERE c1 NOT IN ( SELECT c2 FROM t2 WHERE c2 = c1 );
+c1
+foo
+bar
+DROP TABLE t1, t2;
+#
+# MDEV-13541 - Server crashes in next_breadth_first_tab or Assertion `0'
+#              failed in Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c TEXT CHARSET ucs2 COMPRESSED) ENGINE=InnoDB;
+INSERT INTO t1 VALUES ('foo'),('bar');
+SELECT DISTINCT c FROM t1;
+c
+foo
+bar
+DROP TABLE t1;
+SET column_compression_zlib_wrap=DEFAULT;
+FLUSH STATUS;
+CREATE TABLE t1(a BLOB COMPRESSED, KEY(a(10)));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+CREATE TABLE t1(a BLOB COMPRESSED);
+ALTER TABLE t1 ADD KEY(a(10));
+ERROR HY000: Compressed column 'a' can't be used in key specification
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` blob /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+# Make sure column was actually compressed
+INSERT INTO t1 VALUES(REPEAT('a', 1000));
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	2
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+24
+# Make sure ALTER TABLE rebuilds table
+ALTER TABLE t1 MODIFY COLUMN a BLOB;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` blob DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	1
+COLUMN_DECOMPRESSIONS	3
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+1008
+# Rebuild back
+ALTER TABLE t1 MODIFY COLUMN a BLOB COMPRESSED;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` blob /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	5
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+24
+# Make sure CREATE TABLE ... LIKE inherits compression
+CREATE TABLE t2 LIKE t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` blob /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+DROP TABLE t2;
+# Make sure implicit CREATE TABLE ... SELECT inherits compression
+CREATE TABLE t2 SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` blob /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	7
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+24
+DROP TABLE t2;
+# Make sure explicit CREATE TABLE ... SELECT doesn't inherit compression
+CREATE TABLE t2(a BLOB) SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+Table	Create Table
+t2	CREATE TABLE `t2` (
+  `a` blob DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	8
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DATA_LENGTH
+1008
+DROP TABLE t2;
+# Make sure engine change works
+ALTER TABLE t1 ENGINE=InnoDB;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` blob /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+LEFT(a, 10)	LENGTH(a)
+aaaaaaaaaa	1000
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	10
+# Make sure online add column works (requires InnoDB)
+ALTER TABLE t1 ADD COLUMN b BLOB COMPRESSED DEFAULT "must be visible";
+SELECT LEFT(a, 10), LENGTH(a), b FROM t1;
+LEFT(a, 10)	LENGTH(a)	b
+aaaaaaaaaa	1000	must be visible
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	2
+COLUMN_DECOMPRESSIONS	12
+ALTER TABLE t1 DROP COLUMN b;
+ALTER TABLE t1 ENGINE=MyISAM;
+TRUNCATE TABLE t1;
+# Make sure column_compression_zlib_level works
+SET column_compression_zlib_level= 1;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	3
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+36
+TRUNCATE TABLE t1;
+SET column_compression_zlib_level= 9;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+28
+SET column_compression_zlib_level= DEFAULT;
+TRUNCATE TABLE t1;
+# No compression, original data shorter than compressed
+INSERT INTO t1 VALUES('a');
+SELECT a, LENGTH(a) FROM t1;
+a	LENGTH(a)
+a	1
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+DATA_LENGTH
+20
+# Coverage for store(double) and store(longlong)
+INSERT INTO t1 VALUES(3.14),(CAST(9.8 AS DOUBLE)),(1),(''),(NULL);
+# and for sort_string()
+SELECT * FROM t1 ORDER BY a;
+a
+NULL
+
+1
+3.14
+9.8
+a
+# Coverage for val_real() and val_int()
+SELECT a+1 FROM t1 ORDER BY 1;
+a+1
+NULL
+1
+1
+2
+4.140000000000001
+10.8
+Warnings:
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+Warning	1292	Truncated incorrect DOUBLE value: 'a'
+Warning	1292	Truncated incorrect DOUBLE value: ''
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+VARIABLE_NAME	VARIABLE_VALUE
+COLUMN_COMPRESSIONS	4
+COLUMN_DECOMPRESSIONS	12
+DROP TABLE t1;
+#
+# MDEV-13540 - Server crashes in copy or Assertion `0' failed in virtual
+#              Field* Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c1 BLOB COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t1 VALUES ('foo'),('bar');
+CREATE TABLE t2 (c2 BLOB COMPRESSED) ENGINE=MyISAM;
+INSERT IGNORE INTO t2 VALUES ('qux'),('abc');
+SELECT * FROM t1 WHERE c1 NOT IN ( SELECT c2 FROM t2 WHERE c2 = c1 );
+c1
+foo
+bar
+DROP TABLE t1, t2;
+#
+# MDEV-13541 - Server crashes in next_breadth_first_tab or Assertion `0'
+#              failed in Field_varstring_compressed::new_key_field
+#
+CREATE TABLE t1 (c BLOB COMPRESSED) ENGINE=InnoDB;
+INSERT INTO t1 VALUES ('foo'),('bar');
+SELECT DISTINCT c FROM t1;
+c
+foo
+bar
+DROP TABLE t1;
+CREATE TABLE t1(a CHAR(100) COMPRESSED);
+ERROR 42000: Incorrect column specifier for column 'a'
+CREATE TABLE t1(a INT COMPRESSED);
+ERROR 42000: Incorrect column specifier for column 'a'
+CREATE TABLE t1(a BLOB COMPRESSED=unknown);
+ERROR HY000: Unknown compression method: unknown
+CREATE TABLE t1(a BLOB COMPRESSED COMPRESSED);
+DROP TABLE t1;
+CREATE TABLE t1(a INT);
+ALTER TABLE t1 MODIFY a INT COMPRESSED;
+ERROR 42000: Incorrect column specifier for column 'a'
+DROP TABLE t1;
+# Test CSV
+CREATE TABLE t1(a BLOB NOT NULL COMPRESSED) ENGINE=CSV;
+INSERT INTO t1 VALUES(REPEAT('a', 110));
+SELECT LENGTH(a) FROM t1;
+LENGTH(a)
+110
+ALTER TABLE t1 ENGINE=MyISAM;
+SELECT LENGTH(a) FROM t1;
+LENGTH(a)
+110
+ALTER TABLE t1 ENGINE=CSV;
+SELECT LENGTH(a) FROM t1;
+LENGTH(a)
+110
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` blob /*!100301 COMPRESSED*/ NOT NULL DEFAULT ''
+) ENGINE=CSV DEFAULT CHARSET=latin1
+"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+DROP TABLE t1;
+# Test fields that don't fit data
+CREATE TABLE t1(a VARCHAR(9) COMPRESSED);
+INSERT INTO t1 VALUES(REPEAT('a', 10));
+ERROR 22001: Data too long for column 'a' at row 1
+INSERT INTO t1 VALUES(REPEAT(' ', 10));
+Warnings:
+Note	1265	Data truncated for column 'a' at row 1
+SELECT a, LENGTH(a) FROM t1;
+a	LENGTH(a)
+         	9
+DROP TABLE t1;
+CREATE TABLE t1(a TINYTEXT COMPRESSED);
+SET column_compression_threshold=300;
+INSERT INTO t1 VALUES(REPEAT('a', 255));
+ERROR 22001: Data too long for column 'a' at row 1
+INSERT INTO t1 VALUES(REPEAT(' ', 255));
+Warnings:
+Note	1265	Data truncated for column 'a' at row 1
+SET column_compression_threshold=DEFAULT;
+SELECT a, LENGTH(a) FROM t1;
+a	LENGTH(a)
+                                                                                                                                                                                                                                                              	254
+DROP TABLE t1;
+# Corner case: VARCHAR(255) COMPRESSED must have 2 bytes pack length
+CREATE TABLE t1(a VARCHAR(255) COMPRESSED);
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `a` varchar(255) /*!100301 COMPRESSED*/ DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1
+SET column_compression_threshold=300;
+INSERT INTO t1 VALUES(REPEAT('a', 255));
+SET column_compression_threshold=DEFAULT;
+SELECT a, LENGTH(a) FROM t1;
+a	LENGTH(a)
+aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa	255
+DROP TABLE t1;
diff --git a/mysql-test/r/mysqld--help.result b/mysql-test/r/mysqld--help.result
index b858c34bbc8..1953d990add 100644
--- a/mysql-test/r/mysqld--help.result
+++ b/mysql-test/r/mysqld--help.result
@@ -125,6 +125,35 @@ The following options may be given as the first argument:
  -r, --chroot=name   Chroot mysqld daemon during startup.
  --collation-server=name 
  Set the default collation.
+ --column-compression-threshold=# 
+ Minimum column data length eligible for compression
+ --column-compression-zlib-level=# 
+ zlib compression level (1 gives best speed, 9 gives best
+ compression)
+ --column-compression-zlib-strategy=name 
+ The strategy parameter is used to tune the compression
+ algorithm. Use the value DEFAULT_STRATEGY for normal
+ data, FILTERED for data produced by a filter (or
+ predictor), HUFFMAN_ONLY to force Huffman encoding only
+ (no string match), or RLE to limit match distances to one
+ (run-length encoding). Filtered data consists mostly of
+ small values with a somewhat random distribution. In this
+ case, the compression algorithm is tuned to compress them
+ better. The effect of FILTERED is to force more Huffman
+ coding and less string matching; it is somewhat
+ intermediate between DEFAULT_STRATEGY and HUFFMAN_ONLY.
+ RLE is designed to be almost as fast as HUFFMAN_ONLY, but
+ give better compression for PNG image data. The strategy
+ parameter only affects the compression ratio but not the
+ correctness of the compressed output even if it is not
+ set appropriately. FIXED prevents the use of dynamic
+ Huffman codes, allowing for a simpler decoder for special
+ applications.
+ --column-compression-zlib-wrap 
+ Generate zlib header and trailer and compute adler32
+ check value. It can be used with storage engines that
+ don't provide data integrity verification to detect data
+ corruption.
  --completion-type=name 
  The transaction completion type. One of: NO_CHAIN, CHAIN,
  RELEASE
@@ -1264,6 +1293,10 @@ character-set-client-handshake TRUE
 character-set-filesystem binary
 character-sets-dir MYSQL_CHARSETSDIR/
 chroot (No default value)
+column-compression-threshold 100
+column-compression-zlib-level 6
+column-compression-zlib-strategy DEFAULT_STRATEGY
+column-compression-zlib-wrap FALSE
 completion-type NO_CHAIN
 concurrent-insert AUTO
 console FALSE
diff --git a/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result b/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result
index ad05a862d3f..2f31dfca148 100644
--- a/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result
+++ b/mysql-test/suite/sys_vars/r/sysvars_server_embedded.result
@@ -443,6 +443,62 @@ NUMERIC_BLOCK_SIZE	NULL
 ENUM_VALUE_LIST	NULL
 READ_ONLY	NO
 COMMAND_LINE_ARGUMENT	NULL
+VARIABLE_NAME	COLUMN_COMPRESSION_THRESHOLD
+SESSION_VALUE	100
+GLOBAL_VALUE	100
+GLOBAL_VALUE_ORIGIN	COMPILE-TIME
+DEFAULT_VALUE	100
+VARIABLE_SCOPE	SESSION
+VARIABLE_TYPE	INT UNSIGNED
+VARIABLE_COMMENT	Minimum column data length eligible for compression
+NUMERIC_MIN_VALUE	0
+NUMERIC_MAX_VALUE	4294967295
+NUMERIC_BLOCK_SIZE	1
+ENUM_VALUE_LIST	NULL
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	COLUMN_COMPRESSION_ZLIB_LEVEL
+SESSION_VALUE	6
+GLOBAL_VALUE	6
+GLOBAL_VALUE_ORIGIN	COMPILE-TIME
+DEFAULT_VALUE	6
+VARIABLE_SCOPE	SESSION
+VARIABLE_TYPE	INT UNSIGNED
+VARIABLE_COMMENT	zlib compression level (1 gives best speed, 9 gives best compression)
+NUMERIC_MIN_VALUE	0
+NUMERIC_MAX_VALUE	9
+NUMERIC_BLOCK_SIZE	1
+ENUM_VALUE_LIST	NULL
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	COLUMN_COMPRESSION_ZLIB_STRATEGY
+SESSION_VALUE	DEFAULT_STRATEGY
+GLOBAL_VALUE	DEFAULT_STRATEGY
+GLOBAL_VALUE_ORIGIN	COMPILE-TIME
+DEFAULT_VALUE	DEFAULT_STRATEGY
+VARIABLE_SCOPE	SESSION
+VARIABLE_TYPE	ENUM
+VARIABLE_COMMENT	The strategy parameter is used to tune the compression algorithm. Use the value DEFAULT_STRATEGY for normal data, FILTERED for data produced by a filter (or predictor), HUFFMAN_ONLY to force Huffman encoding only (no string match), or RLE to limit match distances to one (run-length encoding). Filtered data consists mostly of small values with a somewhat random distribution. In this case, the compression algorithm is tuned to compress them better. The effect of FILTERED is to force more Huffman coding and less string matching; it is somewhat intermediate between DEFAULT_STRATEGY and HUFFMAN_ONLY. RLE is designed to be almost as fast as HUFFMAN_ONLY, but give better compression for PNG image data. The strategy parameter only affects the compression ratio but not the correctness of the compressed output even if it is not set appropriately. FIXED prevents the use of dynamic Huffman codes, allowing for a simpler decoder for special applications.
+NUMERIC_MIN_VALUE	NULL
+NUMERIC_MAX_VALUE	NULL
+NUMERIC_BLOCK_SIZE	NULL
+ENUM_VALUE_LIST	DEFAULT_STRATEGY,FILTERED,HUFFMAN_ONLY,RLE,FIXED
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	COLUMN_COMPRESSION_ZLIB_WRAP
+SESSION_VALUE	OFF
+GLOBAL_VALUE	OFF
+GLOBAL_VALUE_ORIGIN	COMPILE-TIME
+DEFAULT_VALUE	OFF
+VARIABLE_SCOPE	SESSION
+VARIABLE_TYPE	BOOLEAN
+VARIABLE_COMMENT	Generate zlib header and trailer and compute adler32 check value. It can be used with storage engines that don't provide data integrity verification to detect data corruption.
+NUMERIC_MIN_VALUE	NULL
+NUMERIC_MAX_VALUE	NULL
+NUMERIC_BLOCK_SIZE	NULL
+ENUM_VALUE_LIST	OFF,ON
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	OPTIONAL
 VARIABLE_NAME	COMPLETION_TYPE
 SESSION_VALUE	NO_CHAIN
 GLOBAL_VALUE	NO_CHAIN
diff --git a/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result b/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
index 895dff73678..7209ac1dda3 100644
--- a/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
+++ b/mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result
@@ -443,6 +443,62 @@ NUMERIC_BLOCK_SIZE	NULL
 ENUM_VALUE_LIST	NULL
 READ_ONLY	NO
 COMMAND_LINE_ARGUMENT	NULL
+VARIABLE_NAME	COLUMN_COMPRESSION_THRESHOLD
+SESSION_VALUE	100
+GLOBAL_VALUE	100
+GLOBAL_VALUE_ORIGIN	COMPILE-TIME
+DEFAULT_VALUE	100
+VARIABLE_SCOPE	SESSION
+VARIABLE_TYPE	INT UNSIGNED
+VARIABLE_COMMENT	Minimum column data length eligible for compression
+NUMERIC_MIN_VALUE	0
+NUMERIC_MAX_VALUE	4294967295
+NUMERIC_BLOCK_SIZE	1
+ENUM_VALUE_LIST	NULL
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	COLUMN_COMPRESSION_ZLIB_LEVEL
+SESSION_VALUE	6
+GLOBAL_VALUE	6
+GLOBAL_VALUE_ORIGIN	COMPILE-TIME
+DEFAULT_VALUE	6
+VARIABLE_SCOPE	SESSION
+VARIABLE_TYPE	INT UNSIGNED
+VARIABLE_COMMENT	zlib compression level (1 gives best speed, 9 gives best compression)
+NUMERIC_MIN_VALUE	0
+NUMERIC_MAX_VALUE	9
+NUMERIC_BLOCK_SIZE	1
+ENUM_VALUE_LIST	NULL
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	COLUMN_COMPRESSION_ZLIB_STRATEGY
+SESSION_VALUE	DEFAULT_STRATEGY
+GLOBAL_VALUE	DEFAULT_STRATEGY
+GLOBAL_VALUE_ORIGIN	COMPILE-TIME
+DEFAULT_VALUE	DEFAULT_STRATEGY
+VARIABLE_SCOPE	SESSION
+VARIABLE_TYPE	ENUM
+VARIABLE_COMMENT	The strategy parameter is used to tune the compression algorithm. Use the value DEFAULT_STRATEGY for normal data, FILTERED for data produced by a filter (or predictor), HUFFMAN_ONLY to force Huffman encoding only (no string match), or RLE to limit match distances to one (run-length encoding). Filtered data consists mostly of small values with a somewhat random distribution. In this case, the compression algorithm is tuned to compress them better. The effect of FILTERED is to force more Huffman coding and less string matching; it is somewhat intermediate between DEFAULT_STRATEGY and HUFFMAN_ONLY. RLE is designed to be almost as fast as HUFFMAN_ONLY, but give better compression for PNG image data. The strategy parameter only affects the compression ratio but not the correctness of the compressed output even if it is not set appropriately. FIXED prevents the use of dynamic Huffman codes, allowing for a simpler decoder for special applications.
+NUMERIC_MIN_VALUE	NULL
+NUMERIC_MAX_VALUE	NULL
+NUMERIC_BLOCK_SIZE	NULL
+ENUM_VALUE_LIST	DEFAULT_STRATEGY,FILTERED,HUFFMAN_ONLY,RLE,FIXED
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	COLUMN_COMPRESSION_ZLIB_WRAP
+SESSION_VALUE	OFF
+GLOBAL_VALUE	OFF
+GLOBAL_VALUE_ORIGIN	COMPILE-TIME
+DEFAULT_VALUE	OFF
+VARIABLE_SCOPE	SESSION
+VARIABLE_TYPE	BOOLEAN
+VARIABLE_COMMENT	Generate zlib header and trailer and compute adler32 check value. It can be used with storage engines that don't provide data integrity verification to detect data corruption.
+NUMERIC_MIN_VALUE	NULL
+NUMERIC_MAX_VALUE	NULL
+NUMERIC_BLOCK_SIZE	NULL
+ENUM_VALUE_LIST	OFF,ON
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	OPTIONAL
 VARIABLE_NAME	COMPLETION_TYPE
 SESSION_VALUE	NO_CHAIN
 GLOBAL_VALUE	NO_CHAIN
diff --git a/mysql-test/t/column_compression.inc b/mysql-test/t/column_compression.inc
new file mode 100644
index 00000000000..89a96ef67ad
--- /dev/null
+++ b/mysql-test/t/column_compression.inc
@@ -0,0 +1,125 @@
+FLUSH STATUS;
+
+--error ER_COMPRESSED_COLUMN_USED_AS_KEY
+eval CREATE TABLE t1(a $typec, KEY(a(10)));
+
+eval CREATE TABLE t1(a $typec);
+--error ER_COMPRESSED_COLUMN_USED_AS_KEY
+ALTER TABLE t1 ADD KEY(a(10));
+SHOW CREATE TABLE t1;
+
+--echo # Make sure column was actually compressed
+INSERT INTO t1 VALUES(REPEAT('a', 1000));
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+
+--echo # Make sure ALTER TABLE rebuilds table
+eval ALTER TABLE t1 MODIFY COLUMN a $typeu;
+SHOW CREATE TABLE t1;
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+
+--echo # Rebuild back
+eval ALTER TABLE t1 MODIFY COLUMN a $typec;
+SHOW CREATE TABLE t1;
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+
+--echo # Make sure CREATE TABLE ... LIKE inherits compression
+CREATE TABLE t2 LIKE t1;
+SHOW CREATE TABLE t2;
+DROP TABLE t2;
+
+--echo # Make sure implicit CREATE TABLE ... SELECT inherits compression
+CREATE TABLE t2 SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DROP TABLE t2;
+
+--echo # Make sure explicit CREATE TABLE ... SELECT doesn't inherit compression
+eval CREATE TABLE t2(a $typeu) SELECT * FROM t1;
+SHOW CREATE TABLE t2;
+SELECT LEFT(a, 10), LENGTH(a) FROM t2;
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t2';
+DROP TABLE t2;
+
+--echo # Make sure engine change works
+ALTER TABLE t1 ENGINE=InnoDB;
+SHOW CREATE TABLE t1;
+SELECT LEFT(a, 10), LENGTH(a) FROM t1;
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+
+--echo # Make sure online add column works (requires InnoDB)
+eval ALTER TABLE t1 ADD COLUMN b $typec DEFAULT "must be visible";
+SELECT LEFT(a, 10), LENGTH(a), b FROM t1;
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+ALTER TABLE t1 DROP COLUMN b;
+
+ALTER TABLE t1 ENGINE=MyISAM;
+TRUNCATE TABLE t1;
+
+--echo # Make sure column_compression_zlib_level works
+SET column_compression_zlib_level= 1;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+TRUNCATE TABLE t1;
+
+SET column_compression_zlib_level= 9;
+INSERT INTO t1 VALUES(REPEAT('ab', 1000));
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+
+SET column_compression_zlib_level= DEFAULT;
+TRUNCATE TABLE t1;
+
+--echo # No compression, original data shorter than compressed
+INSERT INTO t1 VALUES('a');
+SELECT a, LENGTH(a) FROM t1;
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+SELECT DATA_LENGTH FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA='test' AND TABLE_NAME='t1';
+
+--echo # Coverage for store(double) and store(longlong)
+INSERT INTO t1 VALUES(3.14),(CAST(9.8 AS DOUBLE)),(1),(''),(NULL);
+--echo # and for sort_string()
+SELECT * FROM t1 ORDER BY a;
+
+--echo # Coverage for val_real() and val_int()
+SELECT a+1 FROM t1 ORDER BY 1;
+SELECT * FROM INFORMATION_SCHEMA.SESSION_STATUS WHERE VARIABLE_NAME IN('Column_compressions', 'Column_decompressions');
+
+DROP TABLE t1;
+
+--echo #
+--echo # MDEV-13540 - Server crashes in copy or Assertion `0' failed in virtual
+--echo #              Field* Field_varstring_compressed::new_key_field
+--echo #
+eval CREATE TABLE t1 (c1 $typec) ENGINE=MyISAM;
+INSERT IGNORE INTO t1 VALUES ('foo'),('bar');
+
+eval CREATE TABLE t2 (c2 $typec) ENGINE=MyISAM; # Compression is optional
+INSERT IGNORE INTO t2 VALUES ('qux'),('abc');  # Optional
+
+SELECT * FROM t1 WHERE c1 NOT IN ( SELECT c2 FROM t2 WHERE c2 = c1 );
+
+# Cleanup
+DROP TABLE t1, t2;
+
+
+--echo #
+--echo # MDEV-13541 - Server crashes in next_breadth_first_tab or Assertion `0'
+--echo #              failed in Field_varstring_compressed::new_key_field
+--echo #
+eval CREATE TABLE t1 (c $typec) ENGINE=InnoDB;
+INSERT INTO t1 VALUES ('foo'),('bar');
+
+SELECT DISTINCT c FROM t1;
+
+# Cleanup
+DROP TABLE t1;
diff --git a/mysql-test/t/column_compression.test b/mysql-test/t/column_compression.test
new file mode 100644
index 00000000000..d3f848144b6
--- /dev/null
+++ b/mysql-test/t/column_compression.test
@@ -0,0 +1,81 @@
+--source include/have_innodb.inc
+--source include/have_csv.inc
+
+let $MYSQLD_DATADIR= `select @@datadir`;
+
+SET column_compression_zlib_wrap=true;
+let $typec= BLOB COMPRESSED;
+let $typeu= BLOB;
+--source column_compression.inc
+
+let $typec= TEXT COMPRESSED;
+let $typeu= TEXT;
+--source column_compression.inc
+
+let $typec= VARBINARY(10000) COMPRESSED;
+let $typeu= VARBINARY(10000);
+--source column_compression.inc
+
+let $typec= VARCHAR(10000) COMPRESSED;
+let $typeu= VARCHAR(10000);
+--source column_compression.inc
+
+let $typec= TEXT CHARSET ucs2 COMPRESSED;
+let $typeu= TEXT;
+--source column_compression.inc
+
+SET column_compression_zlib_wrap=DEFAULT;
+let $typec= BLOB COMPRESSED;
+let $typeu= BLOB;
+--source column_compression.inc
+
+--error ER_WRONG_FIELD_SPEC
+CREATE TABLE t1(a CHAR(100) COMPRESSED);
+--error ER_WRONG_FIELD_SPEC
+CREATE TABLE t1(a INT COMPRESSED);
+--error ER_UNKNOWN_COMPRESSION_METHOD
+CREATE TABLE t1(a BLOB COMPRESSED=unknown);
+CREATE TABLE t1(a BLOB COMPRESSED COMPRESSED);
+DROP TABLE t1;
+CREATE TABLE t1(a INT);
+--error ER_WRONG_FIELD_SPEC
+ALTER TABLE t1 MODIFY a INT COMPRESSED;
+DROP TABLE t1;
+
+--echo # Test CSV
+CREATE TABLE t1(a BLOB NOT NULL COMPRESSED) ENGINE=CSV;
+INSERT INTO t1 VALUES(REPEAT('a', 110));
+SELECT LENGTH(a) FROM t1;
+ALTER TABLE t1 ENGINE=MyISAM;
+SELECT LENGTH(a) FROM t1;
+ALTER TABLE t1 ENGINE=CSV;
+SELECT LENGTH(a) FROM t1;
+SHOW CREATE TABLE t1;
+--cat_file $MYSQLD_DATADIR/test/t1.CSV
+DROP TABLE t1;
+
+--echo # Test fields that don't fit data
+CREATE TABLE t1(a VARCHAR(9) COMPRESSED);
+--error ER_DATA_TOO_LONG
+INSERT INTO t1 VALUES(REPEAT('a', 10));
+INSERT INTO t1 VALUES(REPEAT(' ', 10));
+SELECT a, LENGTH(a) FROM t1;
+DROP TABLE t1;
+
+CREATE TABLE t1(a TINYTEXT COMPRESSED);
+SET column_compression_threshold=300;
+--error ER_DATA_TOO_LONG
+INSERT INTO t1 VALUES(REPEAT('a', 255));
+INSERT INTO t1 VALUES(REPEAT(' ', 255));
+SET column_compression_threshold=DEFAULT;
+SELECT a, LENGTH(a) FROM t1;
+DROP TABLE t1;
+
+--echo # Corner case: VARCHAR(255) COMPRESSED must have 2 bytes pack length
+CREATE TABLE t1(a VARCHAR(255) COMPRESSED);
+SHOW CREATE TABLE t1;
+SET column_compression_threshold=300;
+INSERT INTO t1 VALUES(REPEAT('a', 255));
+SET column_compression_threshold=DEFAULT;
+SELECT a, LENGTH(a) FROM t1;
+DROP TABLE t1;
diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt
index 3ad123f0e51..b6cd663ae1e 100644
--- a/sql/CMakeLists.txt
+++ b/sql/CMakeLists.txt
@@ -88,7 +88,8 @@ ENDIF()
 
 SET (SQL_SOURCE
               ../sql-common/client.c compat56.cc derror.cc des_key_file.cc
-               discover.cc ../sql-common/errmsg.c field.cc field_conv.cc
+               discover.cc ../sql-common/errmsg.c
+               field.cc field_conv.cc field_comp.cc
                filesort_utils.cc
                filesort.cc gstream.cc
                signal_handler.cc
diff --git a/sql/field.cc b/sql/field.cc
index db6d79bd2f8..4449d0ecf31 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -5184,36 +5184,6 @@ static longlong read_lowendian(const uchar *from, uint bytes)
   }
 }
 
-static void store_bigendian(ulonglong num, uchar *to, uint bytes)
-{
-  switch(bytes) {
-  case 1: mi_int1store(to, num); break;
-  case 2: mi_int2store(to, num); break;
-  case 3: mi_int3store(to, num); break;
-  case 4: mi_int4store(to, num); break;
-  case 5: mi_int5store(to, num); break;
-  case 6: mi_int6store(to, num); break;
-  case 7: mi_int7store(to, num); break;
-  case 8: mi_int8store(to, num); break;
-  default: DBUG_ASSERT(0);
-  }
-}
-
-static longlong read_bigendian(const uchar *from, uint bytes)
-{
-  switch(bytes) {
-  case 1: return mi_uint1korr(from);
-  case 2: return mi_uint2korr(from);
-  case 3: return mi_uint3korr(from);
-  case 4: return mi_uint4korr(from);
-  case 5: return mi_uint5korr(from);
-  case 6: return mi_uint6korr(from);
-  case 7: return mi_uint7korr(from);
-  case 8: return mi_sint8korr(from);
-  default: DBUG_ASSERT(0); return 0;
-  }
-}
-
 void Field_timestamp_hires::store_TIME(my_time_t timestamp, ulong sec_part)
 {
   mi_int4store(ptr, timestamp);
@@ -6797,6 +6767,20 @@ int Field_string::store(const char *from,uint length,CHARSET_INFO *cs)
 }
 
 
+int Field_str::store(longlong nr, bool unsigned_val)
+{
+  char buff[64];
+  uint  length;
+  length= (uint) (field_charset->cset->longlong10_to_str)(field_charset,
+                                                          buff,
+                                                          sizeof(buff),
+                                                          (unsigned_val ? 10:
+                                                           -10),
+                                                           nr);
+  return store(buff, length, field_charset);
+}
+
+
 /**
   Store double value in Field_string or Field_varstring.
 
@@ -6809,7 +6793,8 @@ int Field_str::store(double nr)
 {
   ASSERT_COLUMN_MARKED_FOR_WRITE_OR_COMPUTED;
   char buff[DOUBLE_TO_STRING_CONVERSION_BUFFER_SIZE];
-  uint local_char_length= field_length / charset()->mbmaxlen;
+  uint local_char_length= MY_MIN(sizeof(buff),
+                                 field_length / field_charset->mbmaxlen);
   size_t length= 0;
   my_bool error= (local_char_length == 0);
 
@@ -6841,17 +6826,6 @@ uint Field_str::is_equal(Create_field *new_field)
 }
 
 
-int Field_string::store(longlong nr, bool unsigned_val)
-{
-  char buff[64];
-  int  l;
-  CHARSET_INFO *cs=charset();
-  l= (cs->cset->longlong10_to_str)(cs,buff,sizeof(buff),
-                                   unsigned_val ? 10 : -10, nr);
-  return Field_string::store(buff,(uint)l,cs);
-}
-
-
 int Field_longstr::store_decimal(const my_decimal *d)
 {
   char buff[DECIMAL_MAX_STR_LENGTH+1];
@@ -7338,20 +7312,6 @@ int Field_varstring::store(const char *from,uint length,CHARSET_INFO *cs)
 }
 
 
-int Field_varstring::store(longlong nr, bool unsigned_val)
-{
-  char buff[64];
-  uint  length;
-  length= (uint) (field_charset->cset->longlong10_to_str)(field_charset,
-                                                          buff,
-                                                          sizeof(buff),
-                                                          (unsigned_val ? 10:
-                                                           -10),
-                                                           nr);
-  return Field_varstring::store(buff, length, field_charset);
-}
-
-
 double Field_varstring::val_real(void)
 {
   ASSERT_COLUMN_MARKED_FOR_READ;
@@ -7468,26 +7428,29 @@ int Field_varstring::key_cmp(const uchar *a,const uchar *b)
 
 void Field_varstring::sort_string(uchar *to,uint length)
 {
-  uint tot_length=  length_bytes == 1 ? (uint) *ptr : uint2korr(ptr);
+  String buf;
+
+  val_str(&buf, &buf);
 
   if (field_charset == &my_charset_bin)
   {
     /* Store length last in high-byte order to sort longer strings first */
     if (length_bytes == 1)
-      to[length-1]= tot_length;
+      to[length - 1]= buf.length();
     else
-      mi_int2store(to+length-2, tot_length);
+      mi_int2store(to + length - 2, buf.length());
     length-= length_bytes;
   }
- 
-  tot_length= field_charset->coll->strnxfrm(field_charset,
-                                            to, length,
-                                            char_length() *
-                                            field_charset->strxfrm_multiply,
-                                            ptr + length_bytes, tot_length,
-                                            MY_STRXFRM_PAD_WITH_SPACE |
-                                            MY_STRXFRM_PAD_TO_MAXLEN);
-  DBUG_ASSERT(tot_length == length);
+
+#ifndef DBUG_OFF
+    uint rc=
+#endif
+  field_charset->coll->strnxfrm(field_charset, to, length,
+                                char_length() * field_charset->strxfrm_multiply,
+                                (const uchar*) buf.ptr(), buf.length(),
+                                MY_STRXFRM_PAD_WITH_SPACE |
+                                MY_STRXFRM_PAD_TO_MAXLEN);
+  DBUG_ASSERT(rc == length);
 }
 
 
@@ -7503,6 +7466,11 @@ enum ha_base_keytype Field_varstring::key_type() const
 }
 
 
+/*
+  Compressed columns need one extra byte to store the compression method.
+  This byte is invisible to the end user, but not for the storage engine.
+*/
+
 void Field_varstring::sql_type(String &res) const
 {
   THD *thd= table->in_use;
@@ -7512,7 +7480,8 @@ void Field_varstring::sql_type(String &res) const
   length= cs->cset->snprintf(cs,(char*) res.ptr(),
                              res.alloced_length(), "%s(%d)",
                               (has_charset() ? "varchar" : "varbinary"),
-                             (int) field_length / charset()->mbmaxlen);
+                             (int) field_length / charset()->mbmaxlen -
+                             MY_TEST(compression_method()));
   res.length(length);
   if ((thd->variables.sql_mode & (MODE_MYSQL323 | MODE_MYSQL40)) &&
       has_charset() && (charset()->state & MY_CS_BINSORT))
@@ -7614,32 +7583,36 @@ uint Field_varstring::max_packed_col_length(uint max_length)
 uint Field_varstring::get_key_image(uchar *buff, uint length,
                                     imagetype type_arg)
 {
-  uint f_length=  length_bytes == 1 ? (uint) *ptr : uint2korr(ptr);
-  uint local_char_length= length / field_charset->mbmaxlen;
-  uchar *pos= ptr+length_bytes;
-  local_char_length= my_charpos(field_charset, pos, pos + f_length,
-                                local_char_length);
-  set_if_smaller(f_length, local_char_length);
+  String val;
+  uint local_char_length;
+  my_bitmap_map *old_map;
+
+  old_map= dbug_tmp_use_all_columns(table, table->read_set);
+  val_str(&val, &val);
+  dbug_tmp_restore_column_map(table->read_set, old_map);
+
+  local_char_length= val.charpos(length / field_charset->mbmaxlen);
+  if (local_char_length < val.length())
+    val.length(local_char_length);
   /* Key is always stored with 2 bytes */
-  int2store(buff,f_length);
-  memcpy(buff+HA_KEY_BLOB_LENGTH, pos, f_length);
-  if (f_length < length)
+  int2store(buff, val.length());
+  memcpy(buff + HA_KEY_BLOB_LENGTH, val.ptr(), val.length());
+  if (val.length() < length)
   {
     /*
       Must clear this as we do a memcmp in opt_range.cc to detect
       identical keys
     */
-    bzero(buff+HA_KEY_BLOB_LENGTH+f_length, (length-f_length));
+    memset(buff + HA_KEY_BLOB_LENGTH + val.length(), 0, length - val.length());
   }
-  return HA_KEY_BLOB_LENGTH+f_length;
+  return HA_KEY_BLOB_LENGTH + val.length();
 }
 
 
 void Field_varstring::set_key_image(const uchar *buff,uint length)
 {
   length= uint2korr(buff);			// Real length is here
-  (void) Field_varstring::store((const char*) buff+HA_KEY_BLOB_LENGTH, length,
-                                field_charset);
+  (void) store((const char*) buff + HA_KEY_BLOB_LENGTH, length, field_charset);
 }
 
 
@@ -7696,13 +7669,14 @@ Field *Field_varstring::new_key_field(MEM_ROOT *root, TABLE *new_table,
 uint Field_varstring::is_equal(Create_field *new_field)
 {
   if (new_field->type_handler() == type_handler() &&
-      new_field->charset == field_charset)
+      new_field->charset == field_charset &&
+      !new_field->compression_method() == !compression_method())
   {
-    if (new_field->length == max_display_length())
+    if (new_field->length == field_length)
       return IS_EQUAL_YES;
-    if (new_field->length > max_display_length() &&
-	((new_field->length <= 255 && max_display_length() <= 255) ||
-	 (new_field->length > 255 && max_display_length() > 255)))
+    if (new_field->length > field_length &&
+	((new_field->length <= 255 && field_length <= 255) ||
+	 (new_field->length > 255 && field_length > 255)))
       return IS_EQUAL_PACK_LENGTH; // VARCHAR, longer variable length
   }
   return IS_EQUAL_NO;
@@ -7724,6 +7698,201 @@ void Field_varstring::hash(ulong *nr, ulong *nr2)
 }
 
 
+/**
+  Compress field
+
+  @param[out]    to         destination buffer for compressed data
+  @param[in,out] to_length  in: size of to, out: compressed data length
+  @param[in]     from       data to compress
+  @param[in]     length     from length
+  @param[in]     cs         from character set
+
+  In worst case (no compression performed) storage requirement is increased by
+  1 byte to store header. If it exceeds field length, normal data truncation is
+  performed.
+
+  Generic compressed header format (1 byte):
+
+  Bits 1-4: method specific bits
+  Bits 5-8: compression method
+
+  If compression method is 0 then header is immediately followed by
+  uncompressed data.
+
+  If compression method is zlib:
+
+  Bits 1-3: number of bytes occupied by original data length
+  Bits   4: true if zlib wrapper not present
+  Bits 5-8: store 8 (zlib)
+
+  Header is immediately followed by original data length,
+  followed by compressed data.
+*/
+
+int Field_longstr::compress(char *to, uint *to_length,
+                            const char *from, uint length,
+                            CHARSET_INFO *cs)
+{
+  THD *thd= get_thd();
+  char *buf= 0;
+  int rc= 0;
+
+  if (length == 0)
+  {
+    *to_length= 0;
+    return 0;
+  }
+
+  if (String::needs_conversion_on_storage(length, cs, field_charset) ||
+      *to_length <= length)
+  {
+    String_copier copier;
+    const char *end= from + length;
+
+    if (!(buf= (char*) my_malloc(*to_length - 1, MYF(MY_WME))))
+    {
+      *to_length= 0;
+      return -1;
+    }
+
+    length= copier.well_formed_copy(field_charset, buf, *to_length - 1,
+                                    cs, from, length,
+                                    (*to_length - 1) / field_charset->mbmaxlen);
+    rc= check_conversion_status(&copier, end, cs, true);
+    from= buf;
+    DBUG_ASSERT(length > 0);
+  }
+
+  if (length >= thd->variables.column_compression_threshold &&
+      (*to_length= compression_method()->compress(thd, to, from, length)))
+    status_var_increment(thd->status_var.column_compressions);
+  else
+  {
+    /* Store uncompressed */
+    to[0]= 0;
+    memcpy(to + 1, from, length);
+    *to_length= length + 1;
+  }
+
+  if (buf)
+    my_free(buf);
+  return rc;
+}
+
+
+/*
+  Memory is allocated only when original data was actually compressed.
+  Otherwise val_ptr points at data located immediately after header.
+
+  Data can be stored uncompressed if data was shorter than threshold
+  or compressed data was longer than original data.
+*/
+
+String *Field_longstr::uncompress(String *val_buffer, String *val_ptr,
+                                  const uchar *from, uint from_length)
+{
+  if (from_length)
+  {
+    uchar method= (*from & 0xF0) >> 4;
+
+    /* Uncompressed data */
+    if (!method)
+    {
+      val_ptr->set((const char*) from + 1, from_length - 1, field_charset);
+      return val_ptr;
+    }
+
+    if (compression_methods[method].uncompress)
+    {
+      if (!compression_methods[method].uncompress(val_buffer, from, from_length,
+                                                  field_length))
+      {
+        val_buffer->set_charset(field_charset);
+        status_var_increment(get_thd()->status_var.column_decompressions);
+        return val_buffer;
+      }
+    }
+  }
+
+  /*
+    It would be better to return 0 in case of errors, but to take the
+    safer route, let's return a zero string and let the general
+    handler catch the error.
+  */
+  val_ptr->set("", 0, field_charset);
+  return val_ptr;
+}
+
+
+int Field_varstring_compressed::store(const char *from, uint length,
+                                      CHARSET_INFO *cs)
+{
+  ASSERT_COLUMN_MARKED_FOR_WRITE_OR_COMPUTED;
+  uint to_length= MY_MIN(field_length, field_charset->mbmaxlen * length + 1);
+  int rc= compress((char*) get_data(), &to_length, from, length, cs);
+  store_length(to_length);
+  return rc;
+}
+
+
+String *Field_varstring_compressed::val_str(String *val_buffer, String *val_ptr)
+{
+  ASSERT_COLUMN_MARKED_FOR_READ;
+  return uncompress(val_buffer, val_ptr, get_data(), get_length());
+}
+
+
+double Field_varstring_compressed::val_real(void)
+{
+  ASSERT_COLUMN_MARKED_FOR_READ;
+  THD *thd= get_thd();
+  String buf;
+  val_str(&buf, &buf);
+  return Converter_strntod_with_warn(thd, Warn_filter(thd), field_charset,
+                                     buf.ptr(), buf.length()).result();
+}
+
+
+longlong Field_varstring_compressed::val_int(void)
+{
+  ASSERT_COLUMN_MARKED_FOR_READ;
+  THD *thd= get_thd();
+  String buf;
+  val_str(&buf, &buf);
+  return Converter_strntoll_with_warn(thd, Warn_filter(thd), field_charset,
+                                      buf.ptr(), buf.length()).result();
+}
+
+
+int Field_varstring_compressed::cmp_max(const uchar *a_ptr, const uchar *b_ptr,
+                                        uint max_len)
+{
+  String a, b;
+  uint a_length, b_length;
+
+  if (length_bytes == 1)
+  {
+    a_length= (uint) *a_ptr;
+    b_length= (uint) *b_ptr;
+  }
+  else
+  {
+    a_length= uint2korr(a_ptr);
+    b_length= uint2korr(b_ptr);
+  }
+
+  uncompress(&a, &a, a_ptr + length_bytes, a_length);
+  uncompress(&b, &b, b_ptr + length_bytes, b_length);
+
+  if (a.length() > max_len)
+    a.length(max_len);
+  if (b.length() > max_len)
+    b.length(max_len);
+
+  return sortcmp(&a, &b, field_charset);
+}
+
+
 /****************************************************************************
 ** blob type
 ** A blob is saved as a length and a pointer. The length is stored in the
@@ -7766,6 +7935,7 @@ uint32 Field_blob::get_length(const uchar *pos, uint packlength_arg) const
 int Field_blob::copy_value(Field_blob *from)
 {
   DBUG_ASSERT(field_charset == from->charset());
+  DBUG_ASSERT(!compression_method() == !from->compression_method());
   int rc= 0;
   uint32 length= from->get_length();
   uchar *data= from->get_ptr();
@@ -7873,22 +8043,6 @@ oom_error:
 }
 
 
-int Field_blob::store(double nr)
-{
-  CHARSET_INFO *cs=charset();
-  value.set_real(nr, NOT_FIXED_DEC, cs);
-  return Field_blob::store(value.ptr(),(uint) value.length(), cs);
-}
-
-
-int Field_blob::store(longlong nr, bool unsigned_val)
-{
-  CHARSET_INFO *cs=charset();
-  value.set_int(nr, unsigned_val, cs);
-  return Field_blob::store(value.ptr(), (uint) value.length(), cs);
-}
-
-
 double Field_blob::val_real(void)
 {
   ASSERT_COLUMN_MARKED_FOR_READ;
@@ -8123,33 +8277,30 @@ uint32 Field_blob::sort_length() const
 
 void Field_blob::sort_string(uchar *to,uint length)
 {
-  uchar *blob;
-  uint blob_length=get_length();
+  String buf;
 
-  if (!blob_length && field_charset->pad_char == 0)
+  val_str(&buf, &buf);
+  if (!buf.length() && field_charset->pad_char == 0)
     bzero(to,length);
   else
   {
     if (field_charset == &my_charset_bin)
     {
-      uchar *pos;
-
       /*
         Store length of blob last in blob to shorter blobs before longer blobs
       */
       length-= packlength;
-      pos= to+length;
-
-      store_bigendian(blob_length, pos, packlength);
+      store_bigendian(buf.length(), to + length, packlength);
     }
-    memcpy(&blob, ptr+packlength, sizeof(char*));
-    
-    blob_length= field_charset->coll->strnxfrm(field_charset,
-                                               to, length, length,
-                                               blob, blob_length,
-                                               MY_STRXFRM_PAD_WITH_SPACE |
-                                               MY_STRXFRM_PAD_TO_MAXLEN);
-    DBUG_ASSERT(blob_length == length);
+
+#ifndef DBUG_OFF
+    uint rc=
+#endif
+    field_charset->coll->strnxfrm(field_charset, to, length, length,
+                                  (const uchar*) buf.ptr(), buf.length(),
+                                  MY_STRXFRM_PAD_WITH_SPACE |
+                                  MY_STRXFRM_PAD_TO_MAXLEN);
+    DBUG_ASSERT(rc == length);
   }
 }
 
@@ -8244,11 +8395,9 @@ const uchar *Field_blob::unpack(uchar *to, const uchar *from,
     DBUG_RETURN(0);                             // Error in data
   uint32 const length= get_length(from, master_packlength);
   DBUG_DUMP("packed", from, length + master_packlength);
-  bitmap_set_bit(table->write_set, field_index);
   if (from + master_packlength + length > from_end)
     DBUG_RETURN(0);
-  store(reinterpret_cast<const char*>(from) + master_packlength,
-        length, field_charset);
+  set_ptr(length, const_cast<uchar*> (from) + master_packlength);
   DBUG_DUMP("record", to, table->s->reclength);
   DBUG_RETURN(from + master_packlength + length);
 }
@@ -8268,11 +8417,68 @@ uint Field_blob::max_packed_col_length(uint max_length)
 }
 
 
+/*
+  Blob fields are regarded equal if they have same character set,
+  same blob store length and if either both are compressed or both are
+  uncompressed.
+  The logic for compression is that we don't have to uncompress and compress
+  again an already compressed field just because compression method changes.
+*/
+
 uint Field_blob::is_equal(Create_field *new_field)
 {
   return new_field->type_handler() == type_handler() &&
          new_field->charset == field_charset &&
-         new_field->pack_length == pack_length();
+         new_field->pack_length == pack_length() &&
+         !new_field->compression_method() == !compression_method();
+}
+
+
+int Field_blob_compressed::store(const char *from, uint length,
+                                 CHARSET_INFO *cs)
+{
+  ASSERT_COLUMN_MARKED_FOR_WRITE_OR_COMPUTED;
+  uint to_length= MY_MIN(max_data_length(), field_charset->mbmaxlen * length + 1);
+  int rc;
+
+  if (value.alloc(to_length))
+  {
+    set_ptr((uint32) 0, NULL);
+    return -1;
+  }
+
+  rc= compress((char*) value.ptr(), &to_length, from, length, cs);
+  set_ptr(to_length, (uchar*) value.ptr());
+  return rc;
+}
+
+
+String *Field_blob_compressed::val_str(String *val_buffer, String *val_ptr)
+{
+  ASSERT_COLUMN_MARKED_FOR_READ;
+  return uncompress(val_buffer, val_ptr, get_ptr(), get_length());
+}
+
+
+double Field_blob_compressed::val_real(void)
+{
+  ASSERT_COLUMN_MARKED_FOR_READ;
+  THD *thd= get_thd();
+  String buf;
+  val_str(&buf, &buf);
+  return Converter_strntod_with_warn(thd, Warn_filter(thd), field_charset,
+                                     buf.ptr(), buf.length()).result();
+}
+
+
+longlong Field_blob_compressed::val_int(void)
+{
+  ASSERT_COLUMN_MARKED_FOR_READ;
+  THD *thd= get_thd();
+  String buf;
+  val_str(&buf, &buf);
+  return Converter_strntoll_with_warn(thd, Warn_filter(thd), field_charset,
+                                      buf.ptr(), buf.length()).result();
 }
 
 
@@ -10130,6 +10336,16 @@ Field *make_field(TABLE_SHARE *share,
                        unireg_check, field_name,
                        field_charset);
       if (field_type == MYSQL_TYPE_VARCHAR)
+      {
+        if (unireg_check == Field::TMYSQL_COMPRESSED)
+          return new (mem_root)
+            Field_varstring_compressed(
+                            ptr, field_length,
+                            HA_VARCHAR_PACKLENGTH(field_length),
+                            null_pos, null_bit,
+                            unireg_check, field_name,
+                            share, field_charset, zlib_compression_method);
+
         return new (mem_root)
           Field_varstring(ptr,field_length,
                           HA_VARCHAR_PACKLENGTH(field_length),
@@ -10137,6 +10353,7 @@ Field *make_field(TABLE_SHARE *share,
                           unireg_check, field_name,
                           share,
                           field_charset);
+      }
       return 0;                                 // Error
     }
 
@@ -10158,10 +10375,18 @@ Field *make_field(TABLE_SHARE *share,
     }
 #endif
     if (f_is_blob(pack_flag))
+    {
+      if (unireg_check == Field::TMYSQL_COMPRESSED)
+        return new (mem_root)
+          Field_blob_compressed(ptr, null_pos, null_bit,
+                     unireg_check, field_name, share,
+                     pack_length, field_charset, zlib_compression_method);
+
       return new (mem_root)
         Field_blob(ptr,null_pos,null_bit,
                    unireg_check, field_name, share,
                    pack_length, field_charset);
+    }
     if (interval)
     {
       if (f_is_enum(pack_flag))
@@ -10340,10 +10565,25 @@ Column_definition::Column_definition(THD *thd, Field *old_field,
   comment=    old_field->comment;
   decimals=   old_field->decimals();
   vcol_info=  old_field->vcol_info;
-  default_value= orig_field ? orig_field->default_value : 0;
-  check_constraint= orig_field ? orig_field->check_constraint : 0;
   option_list= old_field->option_list;
   pack_flag= 0;
+  compression_method_ptr= 0;
+
+  if (orig_field)
+  {
+    default_value= orig_field->default_value;
+    check_constraint= orig_field->check_constraint;
+    if (orig_field->unireg_check == Field::TMYSQL_COMPRESSED)
+    {
+      unireg_check= Field::TMYSQL_COMPRESSED;
+      compression_method_ptr= zlib_compression_method;
+    }
+  }
+  else
+  {
+    default_value= 0;
+    check_constraint= 0;
+  }
 
   switch (real_field_type()) {
   case MYSQL_TYPE_TINY_BLOB:
@@ -10364,7 +10604,8 @@ Column_definition::Column_definition(THD *thd, Field *old_field,
   case MYSQL_TYPE_VARCHAR:
   case MYSQL_TYPE_VAR_STRING:
     /* This is corrected in create_length_to_internal_length */
-    length= (length+charset->mbmaxlen-1) / charset->mbmaxlen;
+    length= (length+charset->mbmaxlen-1) / charset->mbmaxlen -
+            MY_TEST(old_field->compression_method());
     break;
 #ifdef HAVE_SPATIAL
   case MYSQL_TYPE_GEOMETRY:
@@ -10528,6 +10769,29 @@ bool Column_definition::has_default_expression()
            (flags & BLOB_FLAG)));
 }
 
+
+bool Column_definition::set_compressed(const char *method)
+{
+  enum enum_field_types sql_type= real_field_type();
+  /* We can't use f_is_blob here as pack_flag is not yet set */
+  if (sql_type == MYSQL_TYPE_VARCHAR || sql_type == MYSQL_TYPE_TINY_BLOB ||
+      sql_type == MYSQL_TYPE_BLOB || sql_type == MYSQL_TYPE_MEDIUM_BLOB ||
+      sql_type == MYSQL_TYPE_LONG_BLOB)
+  {
+    if (!method || !strcmp(method, zlib_compression_method->name))
+    {
+      unireg_check= Field::TMYSQL_COMPRESSED;
+      compression_method_ptr= zlib_compression_method;
+      return false;
+    }
+    my_error(ER_UNKNOWN_COMPRESSION_METHOD, MYF(0), method);
+  }
+  else
+    my_error(ER_WRONG_FIELD_SPEC, MYF(0), field_name.str);
+  return true;
+}
+
+
 /**
   maximum possible display length for blob.
 
diff --git a/sql/field.h b/sql/field.h
index 5792ef3d18b..139b292c96a 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -32,6 +32,7 @@
 #include "sql_error.h"                          /* Sql_condition */
 #include "compat56.h"
 #include "sql_type.h"                           /* Type_std_attributes */
+#include "field_comp.h"
 
 class Send_field;
 class Copy_field;
@@ -707,7 +708,8 @@ public:
     TIMESTAMP_OLD_FIELD=18,     // TIMESTAMP created before 4.1.3
     TIMESTAMP_DN_FIELD=21,      // TIMESTAMP DEFAULT NOW()
     TIMESTAMP_UN_FIELD=22,      // TIMESTAMP ON UPDATE NOW()
-    TIMESTAMP_DNUN_FIELD=23     // TIMESTAMP DEFAULT NOW() ON UPDATE NOW()
+    TIMESTAMP_DNUN_FIELD=23,    // TIMESTAMP DEFAULT NOW() ON UPDATE NOW()
+    TMYSQL_COMPRESSED= 24,      // Compatibility with TMySQL
     };
   enum geometry_type
   {
@@ -1508,6 +1510,8 @@ public:
   /* Mark field in read map. Updates also virtual fields */
   void register_field_in_read_map();
 
+  virtual Compression_method *compression_method() const { return 0; }
+
   friend int cre_myisam(char * name, register TABLE *form, uint options,
 			ulonglong auto_increment_value);
   friend class Copy_field;
@@ -1676,7 +1680,7 @@ public:
            charset() == from->charset();
   }
   int  store(double nr);
-  int  store(longlong nr, bool unsigned_val)=0;
+  int  store(longlong nr, bool unsigned_val);
   int  store_decimal(const my_decimal *);
   int  store(const char *to,uint length,CHARSET_INFO *cs)=0;
   int  store_hex_hybrid(const char *str, uint length)
@@ -1724,6 +1728,11 @@ protected:
                                          const Item *item) const;
   bool cmp_to_string_with_stricter_collation(const Item_bool_func *cond,
                                              const Item *item) const;
+  int compress(char *to, uint *to_length,
+               const char *from, uint length,
+               CHARSET_INFO *cs);
+  String *uncompress(String *val_buffer, String *val_ptr,
+                     const uchar *from, uint from_length);
 public:
   Field_longstr(uchar *ptr_arg, uint32 len_arg, uchar *null_ptr_arg,
                 uchar null_bit_arg, utype unireg_check_arg,
@@ -3059,8 +3068,7 @@ public:
     return 0;
   }
   int store(const char *to,uint length,CHARSET_INFO *charset);
-  int store(longlong nr, bool unsigned_val);
-  int store(double nr) { return Field_str::store(nr); } /* QQ: To be deleted */
+  using Field_str::store;
   double val_real(void);
   longlong val_int(void);
   String *val_str(String*,String *);
@@ -3098,6 +3106,7 @@ private:
 
 
 class Field_varstring :public Field_longstr {
+protected:
   uchar *get_data() const
   {
     return ptr + length_bytes;
@@ -3106,6 +3115,13 @@ class Field_varstring :public Field_longstr {
   {
     return length_bytes == 1 ? (uint) *ptr : uint2korr(ptr);
   }
+  void store_length(uint32 number)
+  {
+    if (length_bytes == 1)
+      *ptr= (uchar) number;
+    else
+      int2store(ptr, number);
+  }
 public:
   /*
     The maximum space available in a Field_varstring, in bytes. See
@@ -3151,11 +3167,11 @@ public:
   bool memcpy_field_possible(const Field *from) const
   {
     return Field_str::memcpy_field_possible(from) &&
+           !compression_method() == !from->compression_method() &&
            length_bytes == ((Field_varstring*) from)->length_bytes;
   }
   int  store(const char *to,uint length,CHARSET_INFO *charset);
-  int  store(longlong nr, bool unsigned_val);
-  int  store(double nr) { return Field_str::store(nr); } /* QQ: To be deleted */
+  using Field_str::store;
   double val_real(void);
   longlong val_int(void);
   String *val_str(String*,String *);
@@ -3193,6 +3209,87 @@ private:
 };
 
 
+class Field_varstring_compressed: public Field_varstring {
+public:
+  Field_varstring_compressed(uchar *ptr_arg,
+                             uint32 len_arg, uint length_bytes_arg,
+                             uchar *null_ptr_arg, uchar null_bit_arg,
+                             enum utype unireg_check_arg,
+                             const LEX_CSTRING *field_name_arg,
+                             TABLE_SHARE *share, const DTCollation &collation,
+                             Compression_method *compression_method_arg):
+    Field_varstring(ptr_arg, len_arg, length_bytes_arg, null_ptr_arg,
+                    null_bit_arg, unireg_check_arg, field_name_arg,
+                    share, collation),
+    compression_method_ptr(compression_method_arg) { DBUG_ASSERT(len_arg > 0); }
+  Compression_method *compression_method() const
+  { return compression_method_ptr; }
+private:
+  Compression_method *compression_method_ptr;
+  int store(const char *to, uint length, CHARSET_INFO *charset);
+  using Field_str::store;
+  String *val_str(String *, String *);
+  double val_real(void);
+  longlong val_int(void);
+  uint size_of() const { return sizeof(*this); }
+  enum_field_types binlog_type() const { return MYSQL_TYPE_VARCHAR_COMPRESSED; }
+  void sql_type(String &str) const
+  {
+    Field_varstring::sql_type(str);
+    str.append(STRING_WITH_LEN(" /*!100301 COMPRESSED*/"));
+  }
+  uint32 max_display_length() { return field_length - 1; }
+  int cmp_max(const uchar *a_ptr, const uchar *b_ptr, uint max_len);
+
+  /*
+    Compressed fields can't have keys as two rows may have different
+    compression methods or compression levels.
+  */
+
+  int key_cmp(const uchar *str, uint length)
+  { DBUG_ASSERT(0); return 0; }
+  using Field_varstring::key_cmp;
+};
+
+
+static inline uint8 number_storage_requirement(uint32 n)
+{
+  return n < 256 ? 1 : n < 65536 ? 2 : n < 16777216 ? 3 : 4;
+}
+
+
+static inline void store_bigendian(ulonglong num, uchar *to, uint bytes)
+{
+  switch(bytes) {
+  case 1: mi_int1store(to, num); break;
+  case 2: mi_int2store(to, num); break;
+  case 3: mi_int3store(to, num); break;
+  case 4: mi_int4store(to, num); break;
+  case 5: mi_int5store(to, num); break;
+  case 6: mi_int6store(to, num); break;
+  case 7: mi_int7store(to, num); break;
+  case 8: mi_int8store(to, num); break;
+  default: DBUG_ASSERT(0);
+  }
+}
+
+
+static inline longlong read_bigendian(const uchar *from, uint bytes)
+{
+  switch(bytes) {
+  case 1: return mi_uint1korr(from);
+  case 2: return mi_uint2korr(from);
+  case 3: return mi_uint3korr(from);
+  case 4: return mi_uint4korr(from);
+  case 5: return mi_uint5korr(from);
+  case 6: return mi_uint6korr(from);
+  case 7: return mi_uint7korr(from);
+  case 8: return mi_sint8korr(from);
+  default: DBUG_ASSERT(0); return 0;
+  }
+}
+
+
 extern LEX_CSTRING temp_lex_str;
 
 class Field_blob :public Field_longstr {
@@ -3235,13 +3332,7 @@ public:
                    NONE, field_name_arg, collation)
   {
     flags|= BLOB_FLAG;
-    packlength= 4;
-    if (set_packlength)
-    {
-      packlength= len_arg <= 255 ? 1 :
-                  len_arg <= 65535 ? 2 :
-                  len_arg <= 16777215 ? 3 : 4;
-    }
+    packlength= set_packlength ? number_storage_requirement(len_arg) : 4;
   }
   Field_blob(uint32 packlength_arg)
     :Field_longstr((uchar*) 0, 0, (uchar*) "", 0, NONE, &temp_lex_str,
@@ -3274,7 +3365,8 @@ public:
     if (from->type() == MYSQL_TYPE_BIT)
       return do_field_int;
     */
-    if (!(from->flags & BLOB_FLAG) || from->charset() != charset())
+    if (!(from->flags & BLOB_FLAG) || from->charset() != charset() ||
+        !from->compression_method() != !compression_method())
       return do_conv_blob;
     if (from->pack_length() != Field_blob::pack_length())
       return do_copy_blob;
@@ -3291,11 +3383,11 @@ public:
   bool memcpy_field_possible(const Field *from) const
   {
     return Field_str::memcpy_field_possible(from) &&
+           !compression_method() == !from->compression_method() &&
            !table->copy_blobs;
   }
-  int  store(const char *to,uint length,CHARSET_INFO *charset);
-  int  store(double nr);
-  int  store(longlong nr, bool unsigned_val);
+  int store(const char *to, uint length, CHARSET_INFO *charset);
+  using Field_str::store;
   double val_real(void);
   longlong val_int(void);
   String *val_str(String*,String *);
@@ -3433,6 +3525,53 @@ private:
 };
 
 
+class Field_blob_compressed: public Field_blob {
+public:
+  Field_blob_compressed(uchar *ptr_arg, uchar *null_ptr_arg,
+                        uchar null_bit_arg, enum utype unireg_check_arg,
+                        const LEX_CSTRING *field_name_arg, TABLE_SHARE *share,
+                        uint blob_pack_length, const DTCollation &collation,
+                        Compression_method *compression_method_arg):
+    Field_blob(ptr_arg, null_ptr_arg, null_bit_arg, unireg_check_arg,
+               field_name_arg, share, blob_pack_length, collation),
+    compression_method_ptr(compression_method_arg) {}
+  Compression_method *compression_method() const
+  { return compression_method_ptr; }
+private:
+  Compression_method *compression_method_ptr;
+  int store(const char *to, uint length, CHARSET_INFO *charset);
+  using Field_str::store;
+  String *val_str(String *, String *);
+  double val_real(void);
+  longlong val_int(void);
+  uint size_of() const { return sizeof(*this); }
+  enum_field_types binlog_type() const { return MYSQL_TYPE_BLOB_COMPRESSED; }
+  void sql_type(String &str) const
+  {
+    Field_blob::sql_type(str);
+    str.append(STRING_WITH_LEN(" /*!100301 COMPRESSED*/"));
+  }
+
+  /*
+    Compressed fields can't have keys as two rows may have different
+    compression methods or compression levels.
+  */
+
+  uint get_key_image(uchar *buff, uint length, imagetype type_arg)
+  { DBUG_ASSERT(0); return 0; }
+  void set_key_image(const uchar *buff, uint length)
+  { DBUG_ASSERT(0); }
+  int key_cmp(const uchar *a, const uchar *b)
+  { DBUG_ASSERT(0); return 0; }
+  int key_cmp(const uchar *str, uint length)
+  { DBUG_ASSERT(0); return 0; }
+  Field *new_key_field(MEM_ROOT *root, TABLE *new_table,
+                       uchar *new_ptr, uint32 length,
+                       uchar *new_null_ptr, uint new_null_bit)
+  { DBUG_ASSERT(0); return 0; }
+};
+
+
 #ifdef HAVE_SPATIAL
 class Field_geom :public Field_blob {
 public:
@@ -3855,6 +3994,7 @@ class Column_definition: public Sql_alloc,
   bool prepare_stage1_check_typelib_default();
   bool prepare_stage1_convert_default(THD *, MEM_ROOT *, CHARSET_INFO *to);
   const Type_handler *field_type() const; // Prevent using this
+  Compression_method *compression_method_ptr;
 public:
   LEX_CSTRING field_name;
   LEX_CSTRING comment;			// Comment for field
@@ -3892,6 +4032,7 @@ public:
 
   Column_definition()
    :Type_handler_hybrid_field_type(&type_handler_null),
+    compression_method_ptr(0),
     comment(null_clex_str),
     on_update(NULL), length(0), decimals(0),
     flags(0), pack_length(0), key_length(0), unireg_check(Field::NONE),
@@ -3916,6 +4057,8 @@ public:
   void create_length_to_internal_length_string()
   {
     length*= charset->mbmaxlen;
+    if (real_field_type() == MYSQL_TYPE_VARCHAR && compression_method())
+      length++;
     key_length= length;
     pack_length= type_handler()->calc_pack_length(length);
   }
@@ -4047,6 +4190,11 @@ public:
   {
     *this= *def;
   }
+  bool set_compressed(const char *method);
+  void set_compression_method(Compression_method *compression_method_arg)
+  { compression_method_ptr= compression_method_arg; }
+  Compression_method *compression_method() const
+  { return compression_method_ptr; }
 };
 
 
diff --git a/sql/field_comp.cc b/sql/field_comp.cc
new file mode 100644
index 00000000000..473d470940d
--- /dev/null
+++ b/sql/field_comp.cc
@@ -0,0 +1,128 @@
+/* Copyright (C) 2017 MariaDB Foundation
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+
+
+#include <my_global.h>
+#include "sql_string.h"
+#include "sql_class.h"
+#include "field_comp.h"
+#include <zlib.h>
+
+
+static uint compress_zlib(THD *thd, char *to, const char *from, uint length)
+{
+  uint level= thd->variables.column_compression_zlib_level;
+
+  if (level > 0)
+  {
+    z_stream stream;
+    int wbits= thd->variables.column_compression_zlib_wrap ? MAX_WBITS :
+                                                            -MAX_WBITS;
+    uint strategy= thd->variables.column_compression_zlib_strategy;
+    /* Store only meaningful bytes of original data length. */
+    uchar original_pack_length= number_storage_requirement(length);
+
+    *to= 0x80 + original_pack_length + (wbits < 0 ? 8 : 0);
+    store_bigendian(length, (uchar*) to + 1, original_pack_length);
+
+    stream.avail_in= length;
+    stream.next_in= (Bytef*) from;
+
+    stream.avail_out= length - original_pack_length - 1;
+    stream.next_out= (Bytef*) to + original_pack_length + 1;
+
+    stream.zalloc= 0;
+    stream.zfree= 0;
+    stream.opaque= 0;
+
+    if (deflateInit2(&stream, level, Z_DEFLATED, wbits, 8, strategy) == Z_OK &&
+        deflate(&stream, Z_FINISH) == Z_STREAM_END &&
+        deflateEnd(&stream) == Z_OK)
+      return (uint) (stream.next_out - (Bytef*) to);
+  }
+  return 0;
+}
+
+
+static int uncompress_zlib(String *to, const uchar *from, uint from_length,
+                           uint field_length)
+{
+  z_stream stream;
+  uchar original_pack_length;
+  int wbits;
+
+  original_pack_length= *from & 0x07;
+  wbits= *from & 8 ? -MAX_WBITS : MAX_WBITS;
+
+  from++;
+  from_length--;
+
+  if (from_length < original_pack_length)
+  {
+    my_error(ER_ZLIB_Z_DATA_ERROR, MYF(0));
+    return 1;
+  }
+
+  stream.avail_out= read_bigendian(from, original_pack_length);
+
+  if (stream.avail_out > field_length)
+  {
+    my_error(ER_ZLIB_Z_DATA_ERROR, MYF(0));
+    return 1;
+  }
+
+  if (to->alloc(stream.avail_out))
+    return 1;
+
+  stream.next_out= (Bytef*) to->ptr();
+
+  stream.avail_in= from_length - original_pack_length;
+  stream.next_in= (Bytef*) from + original_pack_length;
+
+  stream.zalloc= 0;
+  stream.zfree= 0;
+  stream.opaque= 0;
+
+  if (inflateInit2(&stream, wbits) == Z_OK &&
+      inflate(&stream, Z_FINISH) == Z_STREAM_END &&
+      inflateEnd(&stream) == Z_OK)
+  {
+    to->length(stream.total_out);
+    return 0;
+  }
+  my_error(ER_ZLIB_Z_DATA_ERROR, MYF(0));
+  return 1;
+}
+
+
+Compression_method compression_methods[MAX_COMPRESSION_METHODS]=
+{
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { "zlib", compress_zlib, uncompress_zlib },
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { 0, 0, 0 },
+  { 0, 0, 0 }
+};
diff --git a/sql/field_comp.h b/sql/field_comp.h
new file mode 100644
index 00000000000..7eb8ab1e75e
--- /dev/null
+++ b/sql/field_comp.h
@@ -0,0 +1,33 @@
+#ifndef FIELD_COMP_H_INCLUDED
+#define FIELD_COMP_H_INCLUDED
+/* Copyright (C) 2017 MariaDB Foundation
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; version 2 of the License.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+
+
+#define MAX_COMPRESSION_METHODS 16
+
+struct Compression_method
+{
+  const char *name;
+  uint (*compress)(THD *thd, char *to, const char *from, uint length);
+  int (*uncompress)(String *to, const uchar *from, uint from_length,
+                    uint field_length);
+};
+
+
+extern Compression_method compression_methods[MAX_COMPRESSION_METHODS];
+#define zlib_compression_method (&compression_methods[8])
+
+#endif
diff --git a/sql/field_conv.cc b/sql/field_conv.cc
index fc3667c0f87..d97c9f1fd04 100644
--- a/sql/field_conv.cc
+++ b/sql/field_conv.cc
@@ -737,7 +737,8 @@ Field::Copy_func *Field_varstring::get_copy_func(const Field *from) const
     return do_field_varbinary_pre50;
   if (Field_varstring::real_type() != from->real_type() ||
       Field_varstring::charset() != from->charset() ||
-      length_bytes != ((const Field_varstring*) from)->length_bytes)
+      length_bytes != ((const Field_varstring*) from)->length_bytes ||
+      !compression_method() != !from->compression_method())
     return do_field_string;
   return length_bytes == 1 ?
          (from->charset()->mbmaxlen == 1 ? do_varstring1 : do_varstring1_mb) :
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index 19de4bc8cac..190a927ba03 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -4531,6 +4531,9 @@ bool Item_func_dyncol_create::prepare_arguments(THD *thd, bool force_names_arg)
       case MYSQL_TYPE_GEOMETRY:
         type= DYN_COL_STRING;
         break;
+      case MYSQL_TYPE_VARCHAR_COMPRESSED:
+      case MYSQL_TYPE_BLOB_COMPRESSED:
+        DBUG_ASSERT(0);
       }
     }
     if (type == DYN_COL_STRING &&
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index f3fc17f2ecd..4dc7176e968 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -8435,6 +8435,8 @@ SHOW_VAR status_vars[]= {
   {"Busy_time",                (char*) offsetof(STATUS_VAR, busy_time), SHOW_DOUBLE_STATUS},
   {"Bytes_received",           (char*) offsetof(STATUS_VAR, bytes_received), SHOW_LONGLONG_STATUS},
   {"Bytes_sent",               (char*) offsetof(STATUS_VAR, bytes_sent), SHOW_LONGLONG_STATUS},
+  {"Column_compressions",      (char*) offsetof(STATUS_VAR, column_compressions), SHOW_LONG_STATUS},
+  {"Column_decompressions",    (char*) offsetof(STATUS_VAR, column_decompressions), SHOW_LONG_STATUS},
   {"Com",                      (char*) com_status_vars, SHOW_ARRAY},
   {"Compression",              (char*) &show_net_compression, SHOW_SIMPLE_FUNC},
   {"Connections",              (char*) &global_thread_id,         SHOW_LONG_NOFLUSH},
diff --git a/sql/rpl_utility.cc b/sql/rpl_utility.cc
index bf0d98e1cf7..774f582b4b9 100644
--- a/sql/rpl_utility.cc
+++ b/sql/rpl_utility.cc
@@ -128,6 +128,8 @@ max_display_length_for_field(enum_field_types sql_type, unsigned int metadata)
   case MYSQL_TYPE_VAR_STRING:
   case MYSQL_TYPE_VARCHAR:
     return metadata;
+  case MYSQL_TYPE_VARCHAR_COMPRESSED:
+    return metadata - 1;
 
     /*
       The actual length for these types does not really matter since
@@ -145,6 +147,7 @@ max_display_length_for_field(enum_field_types sql_type, unsigned int metadata)
     return my_set_bits(3 * 8);
 
   case MYSQL_TYPE_BLOB:
+  case MYSQL_TYPE_BLOB_COMPRESSED:
     /*
       For the blob type, Field::real_type() lies and say that all
       blobs are of type MYSQL_TYPE_BLOB. In that case, we have to look
@@ -294,6 +297,7 @@ uint32 table_def::calc_field_size(uint col, uchar *master_data) const
     break;
   }
   case MYSQL_TYPE_VARCHAR:
+  case MYSQL_TYPE_VARCHAR_COMPRESSED:
   {
     length= m_field_metadata[col] > 255 ? 2 : 1; // c&p of Field_varstring::data_length()
     length+= length == 1 ? (uint32) *master_data : uint2korr(master_data);
@@ -303,6 +307,7 @@ uint32 table_def::calc_field_size(uint col, uchar *master_data) const
   case MYSQL_TYPE_MEDIUM_BLOB:
   case MYSQL_TYPE_LONG_BLOB:
   case MYSQL_TYPE_BLOB:
+  case MYSQL_TYPE_BLOB_COMPRESSED:
   case MYSQL_TYPE_GEOMETRY:
   {
     /*
@@ -406,11 +411,14 @@ void show_sql_type(enum_field_types type, uint16 metadata, String *str, CHARSET_
 
   case MYSQL_TYPE_VAR_STRING:
   case MYSQL_TYPE_VARCHAR:
+  case MYSQL_TYPE_VARCHAR_COMPRESSED:
     {
       CHARSET_INFO *cs= str->charset();
       uint32 length=
         cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
-                           "varchar(%u)", metadata);
+                           "varchar(%u)%s", metadata,
+                           type == MYSQL_TYPE_VARCHAR_COMPRESSED ? " compressed"
+                                                                 : "");
       str->length(length);
     }
     break;
@@ -455,6 +463,7 @@ void show_sql_type(enum_field_types type, uint16 metadata, String *str, CHARSET_
     break;
 
   case MYSQL_TYPE_BLOB:
+  case MYSQL_TYPE_BLOB_COMPRESSED:
     /*
       Field::real_type() lies regarding the actual type of a BLOB, so
       it is necessary to check the pack length to figure out what kind
@@ -482,6 +491,9 @@ void show_sql_type(enum_field_types type, uint16 metadata, String *str, CHARSET_
       DBUG_ASSERT(0);
       break;
     }
+
+    if (type == MYSQL_TYPE_BLOB_COMPRESSED)
+      str->append(STRING_WITH_LEN(" compressed"));
     break;
 
   case MYSQL_TYPE_STRING:
@@ -583,6 +595,7 @@ can_convert_field_to(Field *field,
                      int *order_var)
 {
   DBUG_ENTER("can_convert_field_to");
+  bool same_type;
 #ifndef DBUG_OFF
   char field_type_buf[MAX_FIELD_WIDTH];
   String field_type(field_type_buf, sizeof(field_type_buf), &my_charset_latin1);
@@ -590,11 +603,30 @@ can_convert_field_to(Field *field,
   DBUG_PRINT("enter", ("field_type: %s, target_type: %d, source_type: %d, source_metadata: 0x%x",
                        field_type.c_ptr_safe(), field->real_type(), source_type, metadata));
 #endif
+  /**
+    @todo
+      Implement Field_varstring_cmopressed::real_type() and
+      Field_blob_compressed::real_type() properly. All occurencies
+      of Field::real_type() have to be inspected and adjusted if needed.
+
+      Until it is not ready we have to compare source_type against
+      binlog_type() when replicating from or to compressed data types.
+
+      @sa Comment for Field::binlog_type()
+  */
+  if (source_type == MYSQL_TYPE_VARCHAR_COMPRESSED ||
+      source_type == MYSQL_TYPE_BLOB_COMPRESSED ||
+      field->binlog_type() == MYSQL_TYPE_VARCHAR_COMPRESSED ||
+      field->binlog_type() == MYSQL_TYPE_BLOB_COMPRESSED)
+    same_type= field->binlog_type() == source_type;
+  else
+    same_type= field->real_type() == source_type;
+
   /*
     If the real type is the same, we need to check the metadata to
     decide if conversions are allowed.
    */
-  if (field->real_type() == source_type)
+  if (same_type)
   {
     if (metadata == 0) // Metadata can only be zero if no metadata was provided
     {
@@ -731,18 +763,22 @@ can_convert_field_to(Field *field,
   case MYSQL_TYPE_MEDIUM_BLOB:
   case MYSQL_TYPE_LONG_BLOB:
   case MYSQL_TYPE_BLOB:
+  case MYSQL_TYPE_BLOB_COMPRESSED:
   case MYSQL_TYPE_STRING:
   case MYSQL_TYPE_VAR_STRING:
   case MYSQL_TYPE_VARCHAR:
+  case MYSQL_TYPE_VARCHAR_COMPRESSED:
     switch (field->real_type())
     {
     case MYSQL_TYPE_TINY_BLOB:
     case MYSQL_TYPE_MEDIUM_BLOB:
     case MYSQL_TYPE_LONG_BLOB:
     case MYSQL_TYPE_BLOB:
+    case MYSQL_TYPE_BLOB_COMPRESSED:
     case MYSQL_TYPE_STRING:
     case MYSQL_TYPE_VAR_STRING:
     case MYSQL_TYPE_VARCHAR:
+    case MYSQL_TYPE_VARCHAR_COMPRESSED:
       *order_var= compare_lengths(field, source_type, metadata);
       /*
         Here we know that the types are different, so if the order
@@ -1036,6 +1072,7 @@ table_def::table_def(unsigned char *types, ulong size,
       switch (binlog_type(i)) {
       case MYSQL_TYPE_TINY_BLOB:
       case MYSQL_TYPE_BLOB:
+      case MYSQL_TYPE_BLOB_COMPRESSED:
       case MYSQL_TYPE_MEDIUM_BLOB:
       case MYSQL_TYPE_LONG_BLOB:
       case MYSQL_TYPE_DOUBLE:
@@ -1066,6 +1103,7 @@ table_def::table_def(unsigned char *types, ulong size,
         break;
       }
       case MYSQL_TYPE_VARCHAR:
+      case MYSQL_TYPE_VARCHAR_COMPRESSED:
       {
         /*
           These types store two bytes.
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index eed2641e1a1..d0fb0ee1772 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7775,3 +7775,7 @@ ER_WRONG_INSERT_INTO_SEQUENCE
         eng "Wrong INSERT into a SEQUENCE. One can only do single table INSERT into a sequence object (like with mysqldump).  If you want to change the SEQUENCE, use ALTER SEQUENCE instead."
 ER_SP_STACK_TRACE
         eng "At line %u in %s"
+ER_COMPRESSED_COLUMN_USED_AS_KEY
+  eng "Compressed column '%-.192s' can't be used in key specification"
+ER_UNKNOWN_COMPRESSION_METHOD
+  eng "Unknown compression method: %s"
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 7c1a730baf8..20989459c5f 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -551,6 +551,7 @@ typedef struct system_variables
   ha_rows max_join_size;
   ha_rows expensive_subquery_limit;
   ulong auto_increment_increment, auto_increment_offset;
+  ulong column_compression_zlib_strategy;
   ulong lock_wait_timeout;
   ulong join_cache_level;
   ulong max_allowed_packet;
@@ -638,6 +639,7 @@ typedef struct system_variables
   my_bool sql_log_bin_off;
   my_bool binlog_annotate_row_events;
   my_bool binlog_direct_non_trans_update;
+  my_bool column_compression_zlib_wrap;
 
   plugin_ref table_plugin;
   plugin_ref tmp_table_plugin;
@@ -692,6 +694,8 @@ typedef struct system_variables
   uint idle_transaction_timeout;
   uint idle_readonly_transaction_timeout;
   uint idle_readwrite_transaction_timeout;
+  uint column_compression_threshold;
+  uint column_compression_zlib_level;
 } SV;
 
 /**
@@ -702,6 +706,8 @@ typedef struct system_variables
 
 typedef struct system_status_var
 {
+  ulong column_compressions;
+  ulong column_decompressions;
   ulong com_stat[(uint) SQLCOM_END];
   ulong com_create_tmp_table;
   ulong com_drop_tmp_table;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 9af076f4215..318378b4acc 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -3690,6 +3690,14 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 	  DBUG_RETURN(TRUE);
 	}
       }
+
+      if (sql_field->compression_method())
+      {
+        my_error(ER_COMPRESSED_COLUMN_USED_AS_KEY, MYF(0),
+                 column->field_name.str);
+        DBUG_RETURN(TRUE);
+      }
+
       cols2.rewind();
       if (key->type == Key::FULLTEXT)
       {
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index b657bb0b51f..3f6ba89fb50 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -55,11 +55,13 @@ Type_handler_set         type_handler_set;
 Type_handler_string      type_handler_string;
 Type_handler_var_string  type_handler_var_string;
 Type_handler_varchar     type_handler_varchar;
+static Type_handler_varchar_compressed type_handler_varchar_compressed;
 
 Type_handler_tiny_blob   type_handler_tiny_blob;
 Type_handler_medium_blob type_handler_medium_blob;
 Type_handler_long_blob   type_handler_long_blob;
 Type_handler_blob        type_handler_blob;
+static Type_handler_blob_compressed type_handler_blob_compressed;
 
 #ifdef HAVE_SPATIAL
 Type_handler_geometry    type_handler_geometry;
@@ -923,6 +925,9 @@ Type_handler::get_handler_by_field_type(enum_field_types type)
       in field_type() context and add DBUG_ASSERT(0) here.
     */
     return &type_handler_newdate;
+  case MYSQL_TYPE_VARCHAR_COMPRESSED:
+  case MYSQL_TYPE_BLOB_COMPRESSED:
+    break;
   };
   DBUG_ASSERT(0);
   return &type_handler_string;
@@ -946,10 +951,12 @@ Type_handler::get_handler_by_real_type(enum_field_types type)
   case MYSQL_TYPE_DOUBLE:      return &type_handler_double;
   case MYSQL_TYPE_NULL:        return &type_handler_null;
   case MYSQL_TYPE_VARCHAR:     return &type_handler_varchar;
+  case MYSQL_TYPE_VARCHAR_COMPRESSED: return &type_handler_varchar_compressed;
   case MYSQL_TYPE_TINY_BLOB:   return &type_handler_tiny_blob;
   case MYSQL_TYPE_MEDIUM_BLOB: return &type_handler_medium_blob;
   case MYSQL_TYPE_LONG_BLOB:   return &type_handler_long_blob;
   case MYSQL_TYPE_BLOB:        return &type_handler_blob;
+  case MYSQL_TYPE_BLOB_COMPRESSED: return &type_handler_blob_compressed;
   case MYSQL_TYPE_VAR_STRING:
     /*
       VAR_STRING is actually a field_type(), not a real_type(),
@@ -1302,6 +1309,21 @@ Field *Type_handler_varchar::make_conversion_table_field(TABLE *table,
 }
 
 
+Field *Type_handler_varchar_compressed::make_conversion_table_field(TABLE *table,
+                                                         uint metadata,
+                                                         const Field *target)
+                                                         const
+{
+  return new(table->in_use->mem_root)
+         Field_varstring_compressed(NULL, metadata,
+                                    HA_VARCHAR_PACKLENGTH(metadata),
+                                    (uchar *) "", 1, Field::NONE,
+                                    &empty_clex_str,
+                                    table->s, target->charset(),
+                                    zlib_compression_method);
+}
+
+
 Field *Type_handler_tiny_blob::make_conversion_table_field(TABLE *table,
                                                            uint metadata,
                                                            const Field *target)
@@ -1324,6 +1346,19 @@ Field *Type_handler_blob::make_conversion_table_field(TABLE *table,
 }
 
 
+Field *Type_handler_blob_compressed::make_conversion_table_field(TABLE *table,
+                                                      uint metadata,
+                                                      const Field *target)
+                                                      const
+{
+  return new(table->in_use->mem_root)
+         Field_blob_compressed(NULL, (uchar *) "", 1, Field::NONE,
+                               &empty_clex_str,
+                               table->s, 2, target->charset(),
+                               zlib_compression_method);
+}
+
+
 Field *Type_handler_medium_blob::make_conversion_table_field(TABLE *table,
                                                            uint metadata,
                                                            const Field *target)
@@ -1693,6 +1728,7 @@ bool Type_handler_string_result::
                                          const
 {
   def->redefine_stage1_common(dup, file, schema);
+  def->set_compression_method(dup->compression_method());
   def->create_length_to_internal_length_string();
   return false;
 }
diff --git a/sql/sql_type.h b/sql/sql_type.h
index e6a1d8f31ff..4b23daeeb11 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -2525,6 +2525,14 @@ public:
 };
 
 
+class Type_handler_varchar_compressed: public Type_handler_varchar
+{
+public:
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
+};
+
+
 class Type_handler_blob_common: public Type_handler_longstr
 {
 public:
@@ -2625,6 +2633,14 @@ public:
 };
 
 
+class Type_handler_blob_compressed: public Type_handler_blob
+{
+public:
+  Field *make_conversion_table_field(TABLE *, uint metadata,
+                                     const Field *target) const;
+};
+
+
 #ifdef HAVE_SPATIAL
 class Type_handler_geometry: public Type_handler_string_result
 {
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 508e6461f81..7e717cbccfd 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1642,6 +1642,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, ulong *yystacksize);
 
 %type <const_simple_string>
         field_length opt_field_length opt_field_length_default_1
+        opt_compression_method
 
 %type <string>
         text_string hex_or_bin_String opt_gconcat_separator
@@ -6644,9 +6645,19 @@ attribute:
                                 $2->name,Lex->charset->csname));
             Lex->last_field->charset= $2;
           }
+        | COMPRESSED_SYM opt_compression_method
+          {
+            if (Lex->last_field->set_compressed($2))
+              MYSQL_YYABORT;
+          }
         | serial_attribute
         ;
 
+opt_compression_method:
+          /* empty */ { $$= NULL; }
+        | equal ident { $$= $2.str; }
+        ;
+
 serial_attribute:
           not NULL_SYM { Lex->last_field->flags|= NOT_NULL_FLAG; }
         | opt_primary KEY_SYM
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 9f70b34100d..6c5bc4baec6 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -762,6 +762,53 @@ static Sys_var_struct Sys_collation_server(
        offsetof(CHARSET_INFO, name), DEFAULT(&default_charset_info),
        NO_MUTEX_GUARD, IN_BINLOG, ON_CHECK(check_collation_not_null));
 
+static Sys_var_uint Sys_column_compression_threshold(
+       "column_compression_threshold",
+       "Minimum column data length eligible for compression",
+       SESSION_VAR(column_compression_threshold), CMD_LINE(REQUIRED_ARG),
+       VALID_RANGE(0, UINT_MAX), DEFAULT(100), BLOCK_SIZE(1));
+
+static Sys_var_uint Sys_column_compression_zlib_level(
+       "column_compression_zlib_level",
+       "zlib compression level (1 gives best speed, 9 gives best compression)",
+       SESSION_VAR(column_compression_zlib_level), CMD_LINE(REQUIRED_ARG),
+       VALID_RANGE(0, 9), DEFAULT(6), BLOCK_SIZE(1));
+
+/*
+  Note that names must correspond to zlib strategy definition. So that we can
+  pass column_compression_zlib_strategy directly to deflateInit2().
+*/
+static const char *column_compression_zlib_strategy_names[]=
+{ "DEFAULT_STRATEGY", "FILTERED", "HUFFMAN_ONLY", "RLE", "FIXED", 0 };
+
+static Sys_var_enum Sys_column_compression_zlib_strategy(
+       "column_compression_zlib_strategy",
+       "The strategy parameter is used to tune the compression algorithm. Use "
+       "the value DEFAULT_STRATEGY for normal data, FILTERED for data produced "
+       "by a filter (or predictor), HUFFMAN_ONLY to force Huffman encoding "
+       "only (no string match), or RLE to limit match distances to one "
+       "(run-length encoding). Filtered data consists mostly of small values "
+       "with a somewhat random distribution. In this case, the compression "
+       "algorithm is tuned to compress them better. The effect of FILTERED is "
+       "to force more Huffman coding and less string matching; it is somewhat "
+       "intermediate between DEFAULT_STRATEGY and HUFFMAN_ONLY. RLE is "
+       "designed to be almost as fast as HUFFMAN_ONLY, but give better "
+       "compression for PNG image data. The strategy parameter only affects "
+       "the compression ratio but not the correctness of the compressed output "
+       "even if it is not set appropriately. FIXED prevents the use of dynamic "
+       "Huffman codes, allowing for a simpler decoder for special "
+       "applications.",
+       SESSION_VAR(column_compression_zlib_strategy), CMD_LINE(REQUIRED_ARG),
+       column_compression_zlib_strategy_names, DEFAULT(0));
+
+static Sys_var_mybool Sys_column_compression_zlib_wrap(
+       "column_compression_zlib_wrap",
+       "Generate zlib header and trailer and compute adler32 check value. "
+       "It can be used with storage engines that don't provide data integrity "
+       "verification to detect data corruption.",
+       SESSION_VAR(column_compression_zlib_wrap), CMD_LINE(OPT_ARG),
+       DEFAULT(FALSE));
+
 static const char *concurrent_insert_names[]= {"NEVER", "AUTO", "ALWAYS", 0};
 static Sys_var_enum Sys_concurrent_insert(
        "concurrent_insert", "Use concurrent insert with MyISAM",
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 2e1f315f6de..6685f73bf0c 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -727,6 +727,13 @@ ha_innobase::check_if_supported_inplace_alter(
 
 			DBUG_RETURN(HA_ALTER_INPLACE_NOT_SUPPORTED);
 		}
+
+		/* Disable online ALTER TABLE for compressed columns until
+		MDEV-13359 - "Online ALTER TABLE will be disabled for compressed columns"
+		is fixed. */
+		if (field->compression_method()) {
+			DBUG_RETURN(HA_ALTER_INPLACE_NOT_SUPPORTED);
+		}
 	}
 
 	ulint n_indexes = UT_LIST_GET_LEN((m_prebuilt->table)->indexes);
@@ -1025,6 +1032,12 @@ ha_innobase::check_if_supported_inplace_alter(
 			    || (ha_alter_info->handler_flags
 				& Alter_inplace_info::ADD_COLUMN));
 
+		/* Disable online ALTER TABLE for compressed columns until
+		MDEV-13359 - "Online ALTER TABLE will be disabled for compressed columns"
+		is fixed. */
+		if (cf->compression_method()) {
+			DBUG_RETURN(HA_ALTER_INPLACE_NOT_SUPPORTED);
+		}
 		if (const Field* f = cf->field) {
 			/* This could be changing an existing column
 			from NULL to NOT NULL. */
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index d310d014100..c75146b2b25 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -4797,7 +4797,9 @@ row_merge_build_indexes(
 	}
 
 	/* Reset the MySQL row buffer that is used when reporting
-	duplicate keys. */
+	duplicate keys.
+
+	This is likely reason for a problem described in MDEV-13359. */
 	innobase_rec_reset(table);
 
 	if (global_system_variables.log_warnings > 2) {
diff --git a/storage/mroonga/ha_mroonga.cpp b/storage/mroonga/ha_mroonga.cpp
index a7e4a9c807a..963f657d735 100644
--- a/storage/mroonga/ha_mroonga.cpp
+++ b/storage/mroonga/ha_mroonga.cpp
@@ -1410,6 +1410,9 @@ static grn_builtin_type mrn_grn_type_from_field(grn_ctx *ctx, Field *field,
   case MYSQL_TYPE_GEOMETRY:     // case-by-case
     type = GRN_DB_WGS84_GEO_POINT; // 8bytes
     break;
+  case MYSQL_TYPE_VARCHAR_COMPRESSED:
+  case MYSQL_TYPE_BLOB_COMPRESSED:
+    DBUG_ASSERT(0);
   }
   return type;
 }
@@ -10906,6 +10909,9 @@ void ha_mroonga::storage_store_field(Field *field,
   case MYSQL_TYPE_GEOMETRY:
     storage_store_field_geometry(field, value, value_length);
     break;
+  case MYSQL_TYPE_VARCHAR_COMPRESSED:
+  case MYSQL_TYPE_BLOB_COMPRESSED:
+    DBUG_ASSERT(0);
   }
 }
 
diff --git a/storage/mroonga/lib/mrn_condition_converter.cpp b/storage/mroonga/lib/mrn_condition_converter.cpp
index cd739b15e51..d1f0fe21615 100644
--- a/storage/mroonga/lib/mrn_condition_converter.cpp
+++ b/storage/mroonga/lib/mrn_condition_converter.cpp
@@ -352,6 +352,9 @@ namespace mrn {
     case MYSQL_TYPE_GEOMETRY:
       type = UNSUPPORTED_TYPE;
       break;
+    case MYSQL_TYPE_VARCHAR_COMPRESSED:
+    case MYSQL_TYPE_BLOB_COMPRESSED:
+      DBUG_ASSERT(0);
     }
 
     DBUG_RETURN(type);
diff --git a/storage/mroonga/lib/mrn_multiple_column_key_codec.cpp b/storage/mroonga/lib/mrn_multiple_column_key_codec.cpp
index c7ef9dd5851..d189a31c5de 100644
--- a/storage/mroonga/lib/mrn_multiple_column_key_codec.cpp
+++ b/storage/mroonga/lib/mrn_multiple_column_key_codec.cpp
@@ -518,6 +518,9 @@ namespace mrn {
       *data_type = TYPE_BYTE_SEQUENCE;
       *data_size = key_part->length;
       break;
+    case MYSQL_TYPE_VARCHAR_COMPRESSED:
+    case MYSQL_TYPE_BLOB_COMPRESSED:
+      DBUG_ASSERT(0);
     }
     DBUG_VOID_RETURN;
   }
diff --git a/storage/tokudb/hatoku_cmp.cc b/storage/tokudb/hatoku_cmp.cc
index ee57e064fbf..b615d4437d2 100644
--- a/storage/tokudb/hatoku_cmp.cc
+++ b/storage/tokudb/hatoku_cmp.cc
@@ -80,6 +80,8 @@ static bool field_valid_for_tokudb_table(Field* field) {
     case MYSQL_TYPE_DECIMAL:
     case MYSQL_TYPE_VAR_STRING:
     case MYSQL_TYPE_NULL:
+    case MYSQL_TYPE_VARCHAR_COMPRESSED:
+    case MYSQL_TYPE_BLOB_COMPRESSED:
         ret_val = false;
     }
 exit:
@@ -242,6 +244,8 @@ static TOKU_TYPE mysql_to_toku_type (Field* field) {
     case MYSQL_TYPE_DECIMAL:
     case MYSQL_TYPE_VAR_STRING:
     case MYSQL_TYPE_NULL:
+    case MYSQL_TYPE_VARCHAR_COMPRESSED:
+    case MYSQL_TYPE_BLOB_COMPRESSED:
         assert_unreachable();
     }
 exit:
@@ -3204,6 +3208,8 @@ static bool fields_are_same_type(Field* a, Field* b) {
     case MYSQL_TYPE_DECIMAL:
     case MYSQL_TYPE_VAR_STRING:
     case MYSQL_TYPE_NULL:
+    case MYSQL_TYPE_VARCHAR_COMPRESSED:
+    case MYSQL_TYPE_BLOB_COMPRESSED:
         assert_unreachable();
     }