diff --git a/debian/control b/debian/control
index 7ba4d82e9e1..30a31b6a6f9 100644
--- a/debian/control
+++ b/debian/control
@@ -26,8 +26,7 @@ Build-Depends: bison,
                libjudy-dev,
                libkrb5-dev,
                liblz4-dev,
-               libncurses5-dev (>= 5.0-6~),
-               libncurses5-dev:native (>= 5.0-6~),
+               libncurses-dev,
                libnuma-dev [linux-any],
                libpam0g-dev,
                libpcre2-dev,
diff --git a/mysql-test/include/have_normal_bzip.inc b/mysql-test/include/have_normal_zlib.inc
similarity index 67%
rename from mysql-test/include/have_normal_bzip.inc
rename to mysql-test/include/have_normal_zlib.inc
index 36c06274398..a4531e687bc 100644
--- a/mysql-test/include/have_normal_bzip.inc
+++ b/mysql-test/include/have_normal_zlib.inc
@@ -1,9 +1,9 @@
 --source include/have_compress.inc
 
-# Test that the system is using the default/standard bzip library.
+# Test that the system is using the default/standard zlib library.
 # If not, we have to skip the test as the compression lengths displayed
 # in the test will not match the results from used compression library.
 
 if (`select length(COMPRESS(space(5000))) != 33`) {
-   skip Test skipped as standard bzip is needed;
+   skip Test skipped as standard zlib is needed;
 }
diff --git a/mysql-test/main/column_compression.test b/mysql-test/main/column_compression.test
index 642399be1f6..5fa8192df3b 100644
--- a/mysql-test/main/column_compression.test
+++ b/mysql-test/main/column_compression.test
@@ -1,6 +1,6 @@
 --source include/have_innodb.inc
 --source include/have_csv.inc
---source include/have_normal_bzip.inc
+--source include/have_normal_zlib.inc
 
 let $MYSQLD_DATADIR= `select @@datadir`;
 
diff --git a/mysql-test/main/column_compression_rpl.test b/mysql-test/main/column_compression_rpl.test
index df8e889016b..18992f33f3c 100644
--- a/mysql-test/main/column_compression_rpl.test
+++ b/mysql-test/main/column_compression_rpl.test
@@ -1,6 +1,6 @@
 --source include/have_innodb.inc
+--source include/have_normal_zlib.inc
 --source include/master-slave.inc
---source include/have_normal_bzip.inc
 
 --let $engine_type= myisam
 --let $engine_type2= innodb
diff --git a/mysql-test/main/func_compress.test b/mysql-test/main/func_compress.test
index 221dddd59f5..d18af140ed3 100644
--- a/mysql-test/main/func_compress.test
+++ b/mysql-test/main/func_compress.test
@@ -1,5 +1,5 @@
 -- source include/have_compress.inc
--- source include/have_normal_bzip.inc
+-- source include/have_normal_zlib.inc
 #
 # Test for compress and uncompress functions:
 #
diff --git a/mysql-test/main/gis.result b/mysql-test/main/gis.result
index 614e0fd6f25..d38d8b8ee5b 100644
--- a/mysql-test/main/gis.result
+++ b/mysql-test/main/gis.result
@@ -5328,5 +5328,112 @@ SELECT BIT_XOR(a) FROM t1;
 ERROR HY000: Illegal parameter data type geometry for operation 'bit_xor('
 DROP TABLE t1;
 #
+# MDEV-27666 User variable not parsed as geometry variable in geometry function.
+#
+set @g= point(1, 1);
+select ST_AsWKT(GeometryCollection(Point(44, 6), @g));
+ST_AsWKT(GeometryCollection(Point(44, 6), @g))
+GEOMETRYCOLLECTION(POINT(44 6),POINT(1 1))
+set @g=  "just a string";
+select ST_AsWKT(GeometryCollection(Point(44, 6), @g));
+ERROR HY000: Illegal parameter data type longblob for operation 'geometrycollection'
+SET @g= LineString(Point(0,0), Point(0,1));
+SELECT AsText(PointN(@g, 1));
+AsText(PointN(@g, 1))
+POINT(0 0)
+SELECT AsText(PointN(@g, 2));
+AsText(PointN(@g, 2))
+POINT(0 1)
+SET @g= Point(1, 1);
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `g` point DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT AsText(g) FROM t1;
+AsText(g)
+POINT(1 1)
+DROP TABLE t1;
+SET @g= MultiPoint(Point(1, 1), Point(-1,-1));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `g` multipoint DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT AsText(g) FROM t1;
+AsText(g)
+MULTIPOINT(1 1,-1 -1)
+DROP TABLE t1;
+SET @g= LineString(Point(1, 1), Point(2,2));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `g` linestring DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT AsText(g) FROM t1;
+AsText(g)
+LINESTRING(1 1,2 2)
+DROP TABLE t1;
+SET @g= MultiLineString(LineString(Point(1, 1), Point(2,2)),
+LineString(Point(-1, -1), Point(-2,-2)));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `g` multilinestring DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT AsText(g) FROM t1;
+AsText(g)
+MULTILINESTRING((1 1,2 2),(-1 -1,-2 -2))
+DROP TABLE t1;
+SET @g= Polygon(LineString(Point(0, 0), Point(30, 0), Point(30, 30), Point(0, 0)));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `g` polygon DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT AsText(g) FROM t1;
+AsText(g)
+POLYGON((0 0,30 0,30 30,0 0))
+DROP TABLE t1;
+SET @g= MultiPolygon(Polygon(LineString(Point(0, 3), Point(3, 3),
+Point(3, 0), Point(0, 3))));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `g` multipolygon DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT AsText(g) FROM t1;
+AsText(g)
+MULTIPOLYGON(((0 3,3 3,3 0,0 3)))
+DROP TABLE t1;
+SET @g= GeometryCollection(Point(44, 6), LineString(Point(3, 6), Point(7, 9)));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `g` geometrycollection DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT AsText(g) FROM t1;
+AsText(g)
+GEOMETRYCOLLECTION(POINT(44 6),LINESTRING(3 6,7 9))
+DROP TABLE t1;
+SET @g= GeometryFromText('POINT(1 1)');
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `g` geometry DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT AsText(g) FROM t1;
+AsText(g)
+POINT(1 1)
+DROP TABLE t1;
+#
 # End of 10.5 tests
 #
diff --git a/mysql-test/main/gis.test b/mysql-test/main/gis.test
index 4d66fb2b07d..f160d8f1a10 100644
--- a/mysql-test/main/gis.test
+++ b/mysql-test/main/gis.test
@@ -3374,6 +3374,69 @@ SELECT BIT_OR(a) FROM t1;
 SELECT BIT_XOR(a) FROM t1;
 DROP TABLE t1;
 
+--echo #
+--echo # MDEV-27666 User variable not parsed as geometry variable in geometry function.
+--echo #
+
+set @g= point(1, 1);
+select ST_AsWKT(GeometryCollection(Point(44, 6), @g));
+set @g=  "just a string";
+--error ER_ILLEGAL_PARAMETER_DATA_TYPE_FOR_OPERATION
+select ST_AsWKT(GeometryCollection(Point(44, 6), @g));
+
+SET @g= LineString(Point(0,0), Point(0,1));
+SELECT AsText(PointN(@g, 1));
+SELECT AsText(PointN(@g, 2));
+
+SET @g= Point(1, 1);
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+SELECT AsText(g) FROM t1;
+DROP TABLE t1;
+
+SET @g= MultiPoint(Point(1, 1), Point(-1,-1));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+SELECT AsText(g) FROM t1;
+DROP TABLE t1;
+
+SET @g= LineString(Point(1, 1), Point(2,2));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+SELECT AsText(g) FROM t1;
+DROP TABLE t1;
+
+SET @g= MultiLineString(LineString(Point(1, 1), Point(2,2)),
+                        LineString(Point(-1, -1), Point(-2,-2)));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+SELECT AsText(g) FROM t1;
+DROP TABLE t1;
+
+SET @g= Polygon(LineString(Point(0, 0), Point(30, 0), Point(30, 30), Point(0, 0)));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+SELECT AsText(g) FROM t1;
+DROP TABLE t1;
+
+SET @g= MultiPolygon(Polygon(LineString(Point(0, 3), Point(3, 3),
+                                        Point(3, 0), Point(0, 3))));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+SELECT AsText(g) FROM t1;
+DROP TABLE t1;
+
+SET @g= GeometryCollection(Point(44, 6), LineString(Point(3, 6), Point(7, 9)));
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+SELECT AsText(g) FROM t1;
+DROP TABLE t1;
+
+SET @g= GeometryFromText('POINT(1 1)');
+CREATE TABLE t1 AS SELECT @g AS g;
+SHOW CREATE TABLE t1;
+SELECT AsText(g) FROM t1;
+DROP TABLE t1;
 
 --echo #
 --echo # End of 10.5 tests
diff --git a/mysql-test/main/mysql_json_table_recreate.result b/mysql-test/main/mysql_json_table_recreate.result
index 207dde9d8ad..a61377fe21d 100644
--- a/mysql-test/main/mysql_json_table_recreate.result
+++ b/mysql-test/main/mysql_json_table_recreate.result
@@ -30,6 +30,12 @@ show create table mysql_json_test;
 ERROR HY000: Table rebuild required. Please do "ALTER TABLE `test.mysql_json_test` FORCE" or dump/reload to fix it!
 select * from mysql_json_test;
 ERROR HY000: Table rebuild required. Please do "ALTER TABLE `test.mysql_json_test` FORCE" or dump/reload to fix it!
+CREATE TABLE t2 AS SELECT * FROM mysql_json_test;
+ERROR HY000: Table rebuild required. Please do "ALTER TABLE `test.mysql_json_test` FORCE" or dump/reload to fix it!
+CREATE TABLE t2 (a mysql_json /*new column*/) AS SELECT * FROM mysql_json_test;
+ERROR HY000: 'MYSQL_JSON' is not allowed in this context
+CREATE TABLE t2 (actual mysql_json /*existing column*/) AS SELECT * FROM mysql_json_test;
+ERROR HY000: 'MYSQL_JSON' is not allowed in this context
 LOCK TABLES mysql_json_test WRITE;
 ERROR HY000: Table rebuild required. Please do "ALTER TABLE `test.mysql_json_test` FORCE" or dump/reload to fix it!
 alter table mysql_json_test force;
@@ -169,3 +175,67 @@ Total_Number_of_Tests	Succesful_Tests	String_is_valid_JSON
 drop table tempty;
 drop table mysql_json_test;
 drop table mysql_json_test_big;
+#
+# MDEV-32790: Output result in show create table
+#             for mysql_json type should be longtext
+#
+create table t1(j json);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `j` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL CHECK (json_valid(`j`))
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+drop table t1;
+create table t1(j mysql_json);
+ERROR HY000: 'MYSQL_JSON' is not allowed in this context
+create table `testjson` (
+`t` json /* JSON from MySQL 5.7*/ CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci;
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL
+) ENGINE=InnoDB DEFAULT CH...' at line 2
+create table `testjson` (
+`t` json /* JSON from MySQL 5.7*/ COLLATE utf8mb4_bin NOT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci;
+show create table testjson;
+Table	Create Table
+testjson	CREATE TABLE `testjson` (
+  `t` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL CHECK (json_valid(`t`))
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+drop table testjson;
+create table `testjson` (
+`t` longtext /* JSON from MySQL 5.7 */ CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci;
+show create table testjson;
+Table	Create Table
+testjson	CREATE TABLE `testjson` (
+  `t` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+drop table testjson;
+#
+# MDEV-32235: mysql_json cannot be used on newly created table
+#
+CREATE TABLE t(j mysql_json);
+ERROR HY000: 'MYSQL_JSON' is not allowed in this context
+CREATE TABLE IF NOT EXISTS t(j mysql_json);
+ERROR HY000: 'MYSQL_JSON' is not allowed in this context
+CREATE OR REPLACE TABLE t(j mysql_json);
+ERROR HY000: 'MYSQL_JSON' is not allowed in this context
+CREATE TEMPORARY TABLE t(j mysql_json);
+ERROR HY000: 'MYSQL_JSON' is not allowed in this context
+CREATE TABLE t1 (a TEXT);
+ALTER TABLE t1 MODIFY a mysql_json;
+ERROR HY000: 'MYSQL_JSON' is not allowed in this context
+DROP TABLE t1;
+CREATE FUNCTION f1() RETURNS mysql_json RETURN NULL;
+ERROR HY000: 'MYSQL_JSON' is not allowed in this context
+CREATE FUNCTION f1(a mysql_json) RETURNS INT RETURN 0;
+ERROR HY000: 'MYSQL_JSON' is not allowed in this context
+CREATE PROCEDURE p1()
+BEGIN
+DECLARE a mysql_json;
+END;
+$$
+ERROR HY000: 'MYSQL_JSON' is not allowed in this context
+#
+# End of 10.5 tests
+#
diff --git a/mysql-test/main/mysql_json_table_recreate.test b/mysql-test/main/mysql_json_table_recreate.test
index a399b546591..a6f1d3194ae 100644
--- a/mysql-test/main/mysql_json_table_recreate.test
+++ b/mysql-test/main/mysql_json_table_recreate.test
@@ -51,6 +51,13 @@ show create table mysql_json_test;
 --error ER_TABLE_NEEDS_REBUILD
 select * from mysql_json_test;
 
+--error ER_TABLE_NEEDS_REBUILD
+CREATE TABLE t2 AS SELECT * FROM mysql_json_test;
+--error ER_NOT_ALLOWED_IN_THIS_CONTEXT
+CREATE TABLE t2 (a mysql_json /*new column*/) AS SELECT * FROM mysql_json_test;
+--error ER_NOT_ALLOWED_IN_THIS_CONTEXT
+CREATE TABLE t2 (actual mysql_json /*existing column*/) AS SELECT * FROM mysql_json_test;
+
 --error ER_TABLE_NEEDS_REBUILD
 LOCK TABLES mysql_json_test WRITE;
 
@@ -88,3 +95,69 @@ from mysql_json_test_big;
 drop table tempty;
 drop table mysql_json_test;
 drop table mysql_json_test_big;
+
+--echo #
+--echo # MDEV-32790: Output result in show create table
+--echo #             for mysql_json type should be longtext
+--echo #
+
+create table t1(j json);
+show create table t1;
+drop table t1;
+--error ER_NOT_ALLOWED_IN_THIS_CONTEXT
+create table t1(j mysql_json);
+# `json` type should not have character set and collation other than utf8mb4_bin
+--error ER_PARSE_ERROR
+create table `testjson` (
+  `t` json /* JSON from MySQL 5.7*/ CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci;
+
+# By removing character set from `json` field query should work and
+# expand to `longtext` with characterset
+create table `testjson` (
+  `t` json /* JSON from MySQL 5.7*/ COLLATE utf8mb4_bin NOT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci;
+show create table testjson;
+drop table testjson;
+
+# `longtext` that is alias can have character set
+create table `testjson` (
+  `t` longtext /* JSON from MySQL 5.7 */ CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci;
+show create table testjson;
+drop table testjson;
+
+--echo #
+--echo # MDEV-32235: mysql_json cannot be used on newly created table
+--echo #
+
+--error ER_NOT_ALLOWED_IN_THIS_CONTEXT
+CREATE TABLE t(j mysql_json);
+--error ER_NOT_ALLOWED_IN_THIS_CONTEXT
+CREATE TABLE IF NOT EXISTS t(j mysql_json);
+--error ER_NOT_ALLOWED_IN_THIS_CONTEXT
+CREATE OR REPLACE TABLE t(j mysql_json);
+--error ER_NOT_ALLOWED_IN_THIS_CONTEXT
+CREATE TEMPORARY TABLE t(j mysql_json);
+
+CREATE TABLE t1 (a TEXT);
+--error ER_NOT_ALLOWED_IN_THIS_CONTEXT
+ALTER TABLE t1 MODIFY a mysql_json;
+DROP TABLE t1;
+
+--error ER_NOT_ALLOWED_IN_THIS_CONTEXT
+CREATE FUNCTION f1() RETURNS mysql_json RETURN NULL;
+--error ER_NOT_ALLOWED_IN_THIS_CONTEXT
+CREATE FUNCTION f1(a mysql_json) RETURNS INT RETURN 0;
+DELIMITER $$;
+--error ER_NOT_ALLOWED_IN_THIS_CONTEXT
+CREATE PROCEDURE p1()
+BEGIN
+  DECLARE a mysql_json;
+END;
+$$
+DELIMITER ;$$
+
+--echo #
+--echo # End of 10.5 tests
+--echo #
diff --git a/mysql-test/main/mysqlbinlog_row_compressed.test b/mysql-test/main/mysqlbinlog_row_compressed.test
index be9736347c9..03868d3e6b3 100644
--- a/mysql-test/main/mysqlbinlog_row_compressed.test
+++ b/mysql-test/main/mysqlbinlog_row_compressed.test
@@ -4,7 +4,7 @@
 
 --source include/have_log_bin.inc
 --source include/have_binlog_format_row.inc
---source include/have_normal_bzip.inc
+--source include/have_normal_zlib.inc
 
 #
 #  
diff --git a/mysql-test/main/mysqlbinlog_stmt_compressed.test b/mysql-test/main/mysqlbinlog_stmt_compressed.test
index 08db2009d60..4a65124339d 100644
--- a/mysql-test/main/mysqlbinlog_stmt_compressed.test
+++ b/mysql-test/main/mysqlbinlog_stmt_compressed.test
@@ -4,7 +4,7 @@
 
 --source include/have_log_bin.inc
 --source include/have_binlog_format_statement.inc
---source include/have_normal_bzip.inc
+--source include/have_normal_zlib.inc
 #
 #  
 # mysqlbinlog: compressed query event 
diff --git a/mysql-test/main/opt_trace.result b/mysql-test/main/opt_trace.result
index e059aee9c48..f004fd4cf57 100644
--- a/mysql-test/main/opt_trace.result
+++ b/mysql-test/main/opt_trace.result
@@ -9309,6 +9309,18 @@ json_detailed(json_extract(trace, '$**.in_to_subquery_conversion'))
 set in_predicate_conversion_threshold=@tmp;
 drop table t0;
 #
+# MDEV-29298: INSERT ... SELECT Does not produce an optimizer trace
+#
+create table t1 (a int, b int);
+create table t2 (a int, b int);
+insert into t1 values (1,1), (2,2), (3,3), (4,4), (5,5);
+set optimizer_trace=1;
+insert into t2 select * from t1 where a<= b and a>4;
+select QUERY, LENGTH(trace)>1 from information_schema.optimizer_trace;
+QUERY	LENGTH(trace)>1
+insert into t2 select * from t1 where a<= b and a>4	1
+drop table t1, t2;
+#
 # End of 10.5 tests
 #
 #
diff --git a/mysql-test/main/opt_trace.test b/mysql-test/main/opt_trace.test
index 315aa9bb82a..7d6c197b33f 100644
--- a/mysql-test/main/opt_trace.test
+++ b/mysql-test/main/opt_trace.test
@@ -869,6 +869,20 @@ set in_predicate_conversion_threshold=@tmp;
 drop table t0;
 --enable_view_protocol
 
+--echo #
+--echo # MDEV-29298: INSERT ... SELECT Does not produce an optimizer trace
+--echo #
+create table t1 (a int, b int);
+create table t2 (a int, b int);
+insert into t1 values (1,1), (2,2), (3,3), (4,4), (5,5);
+set optimizer_trace=1;
+
+insert into t2 select * from t1 where a<= b and a>4;
+
+select QUERY, LENGTH(trace)>1 from information_schema.optimizer_trace;
+
+drop table t1, t2;
+
 --echo #
 --echo # End of 10.5 tests
 --echo #
diff --git a/mysql-test/suite/binlog/r/binlog_gis_user_var_stm.result b/mysql-test/suite/binlog/r/binlog_gis_user_var_stm.result
new file mode 100644
index 00000000000..e467c9c89b4
--- /dev/null
+++ b/mysql-test/suite/binlog/r/binlog_gis_user_var_stm.result
@@ -0,0 +1,12 @@
+SET @g0= POINT(1,1);
+SET @g1= Polygon(LineString(Point(0, 0), Point(30, 0), Point(30, 30), Point(0, 0)));
+CREATE TABLE t1 AS SELECT @g0 AS g0, @g1 AS g1;
+DROP TABLE t1;
+include/show_binlog_events.inc
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	User var	#	#	@`g0`=/*point*/_binary X'000000000101000000000000000000F03F000000000000F03F' COLLATE binary
+master-bin.000001	#	User var	#	#	@`g1`=/*polygon*/_binary X'0000000001030000000100000004000000000000000000000000000000000000000000000000003E4000000000000000000000000000003E400000000000003E4000000000000000000000000000000000' COLLATE binary
+master-bin.000001	#	Query	#	#	use `test`; CREATE TABLE t1 AS SELECT @g0 AS g0, @g1 AS g1
+master-bin.000001	#	Gtid	#	#	GTID #-#-#
+master-bin.000001	#	Query	#	#	use `test`; DROP TABLE `t1` /* generated by server */
diff --git a/mysql-test/suite/binlog/t/binlog_gis_user_var_stm.test b/mysql-test/suite/binlog/t/binlog_gis_user_var_stm.test
new file mode 100644
index 00000000000..7e789cd7ae4
--- /dev/null
+++ b/mysql-test/suite/binlog/t/binlog_gis_user_var_stm.test
@@ -0,0 +1,15 @@
+--source include/not_embedded.inc
+--source include/have_binlog_format_statement.inc
+--source include/have_geometry.inc
+
+--disable_query_log
+reset master; # get rid of previous tests binlog
+--enable_query_log
+
+SET @g0= POINT(1,1);
+SET @g1= Polygon(LineString(Point(0, 0), Point(30, 0), Point(30, 30), Point(0, 0)));
+CREATE TABLE t1 AS SELECT @g0 AS g0, @g1 AS g1;
+DROP TABLE t1;
+
+--let $binlog_file = LAST
+source include/show_binlog_events.inc;
diff --git a/mysql-test/suite/compat/oracle/t/column_compression.test b/mysql-test/suite/compat/oracle/t/column_compression.test
index 01d4977ba96..e8d55000bc6 100644
--- a/mysql-test/suite/compat/oracle/t/column_compression.test
+++ b/mysql-test/suite/compat/oracle/t/column_compression.test
@@ -1,6 +1,6 @@
 --source include/have_innodb.inc
 --source include/have_csv.inc
---source include/have_normal_bzip.inc
+--source include/have_normal_zlib.inc
 
 SET sql_mode=ORACLE;
 
diff --git a/mysql-test/suite/innodb/r/doublewrite.result b/mysql-test/suite/innodb/r/doublewrite.result
index 0c4f2c3fe65..65ff203a28e 100644
--- a/mysql-test/suite/innodb/r/doublewrite.result
+++ b/mysql-test/suite/innodb/r/doublewrite.result
@@ -1,7 +1,7 @@
 #
 # MDEV-32242 innodb.doublewrite test case always is skipped
 #
-create table t1 (f1 int primary key, f2 blob) engine=innodb stats_persistent=0;
+create table t1 (f1 int primary key, f2 blob) stats_persistent=0, engine=innodb;
 start transaction;
 insert into t1 values(1, repeat('#',12));
 insert into t1 values(2, repeat('+',12));
@@ -19,6 +19,7 @@ XA PREPARE 'x';
 disconnect dml;
 connection default;
 flush table t1 for export;
+# Kill the server
 # restart
 FOUND 1 /InnoDB: Restoring page \[page id: space=[1-9][0-9]*, page number=0\] of datafile/ in mysqld.1.err
 FOUND 1 /InnoDB: Recovered page \[page id: space=[1-9][0-9]*, page number=3\]/ in mysqld.1.err
@@ -33,5 +34,27 @@ f1	f2
 3	////////////
 4	------------
 5	............
+connect  dml,localhost,root,,;
+XA START 'x';
+insert into t1 values (6, repeat('%', @@innodb_page_size/2));
+XA END 'x';
+XA PREPARE 'x';
+disconnect dml;
+connection default;
+flush table t1 for export;
+# Kill the server
+# restart
+FOUND 1 /InnoDB: Restoring page \[page id: space=[1-9][0-9]*, page number=0\] of datafile/ in mysqld.1.err
+XA ROLLBACK 'x';
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+select f1, f2 from t1;
+f1	f2
+1	############
+2	++++++++++++
+3	////////////
+4	------------
+5	............
 drop table t1;
 # End of 10.5 tests
diff --git a/mysql-test/suite/innodb/r/log_file_name.result b/mysql-test/suite/innodb/r/log_file_name.result
index 766122959ab..a7f8cc8c7b2 100644
--- a/mysql-test/suite/innodb/r/log_file_name.result
+++ b/mysql-test/suite/innodb/r/log_file_name.result
@@ -1,3 +1,4 @@
+call mtr.add_suppression("InnoDB: Header page consists of zero bytes in datafile:");
 SET GLOBAL innodb_file_per_table=ON;
 FLUSH TABLES;
 CREATE TABLE t1(a INT PRIMARY KEY) ENGINE=InnoDB;
diff --git a/mysql-test/suite/innodb/t/doublewrite.test b/mysql-test/suite/innodb/t/doublewrite.test
index 33813dabdc4..541e4d23a19 100644
--- a/mysql-test/suite/innodb/t/doublewrite.test
+++ b/mysql-test/suite/innodb/t/doublewrite.test
@@ -17,6 +17,7 @@ call mtr.add_suppression("InnoDB: A bad Space ID was found in datafile");
 call mtr.add_suppression("InnoDB: Checksum mismatch in datafile: ");
 call mtr.add_suppression("InnoDB: Inconsistent tablespace ID in .*t1\\.ibd");
 call mtr.add_suppression("\\[Warning\\] Found 1 prepared XA transactions");
+call mtr.add_suppression("InnoDB: Header page consists of zero bytes in datafile:");
 --enable_query_log
 
 let INNODB_PAGE_SIZE=`select @@innodb_page_size`;
@@ -24,7 +25,7 @@ let MYSQLD_DATADIR=`select @@datadir`;
 let ALGO=`select @@innodb_checksum_algorithm`;
 let SEARCH_FILE= $MYSQLTEST_VARDIR/log/mysqld.1.err;
 
-create table t1 (f1 int primary key, f2 blob) engine=innodb stats_persistent=0;
+create table t1 (f1 int primary key, f2 blob) stats_persistent=0, engine=innodb;
 
 start transaction;
 insert into t1 values(1, repeat('#',12));
@@ -38,7 +39,7 @@ commit work;
 SET GLOBAL innodb_fast_shutdown = 0;
 let $shutdown_timeout=;
 --source include/restart_mysqld.inc
-
+--source ../include/no_checkpoint_start.inc
 connect (dml,localhost,root,,);
 XA START 'x';
 insert into t1 values (6, repeat('%', @@innodb_page_size/2));
@@ -50,8 +51,8 @@ connection default;
 flush table t1 for export;
 
 let $restart_parameters=;
-let $shutdown_timeout=0;
---source include/shutdown_mysqld.inc
+--let CLEANUP_IF_CHECKPOINT=drop table t1, unexpected_checkpoint;
+--source ../include/no_checkpoint_end.inc
 
 perl;
 use IO::Handle;
@@ -119,6 +120,35 @@ let SEARCH_PATTERN=InnoDB: Recovered page \[page id: space=[1-9][0-9]*, page num
 XA ROLLBACK 'x';
 check table t1;
 select f1, f2 from t1;
+
+--source ../include/no_checkpoint_start.inc
+connect (dml,localhost,root,,);
+XA START 'x';
+insert into t1 values (6, repeat('%', @@innodb_page_size/2));
+XA END 'x';
+XA PREPARE 'x';
+disconnect dml;
+connection default;
+
+flush table t1 for export;
+
+let $restart_parameters=;
+--source ../include/no_checkpoint_end.inc
+
+# Zero out the first page in file and try to recover from dblwr
+perl;
+use IO::Handle;
+open(FILE, "+<", "$ENV{'MYSQLD_DATADIR'}test/t1.ibd") or die;
+syswrite(FILE, chr(0) x $ENV{INNODB_PAGE_SIZE});
+close FILE;
+EOF
+
+--source include/start_mysqld.inc
+let SEARCH_PATTERN=InnoDB: Restoring page \[page id: space=[1-9][0-9]*, page number=0\] of datafile;
+--source include/search_pattern_in_file.inc
+XA ROLLBACK 'x';
+check table t1;
+select f1, f2 from t1;
 drop table t1;
 
 --echo # End of 10.5 tests
diff --git a/mysql-test/suite/innodb/t/doublewrite_debug.test b/mysql-test/suite/innodb/t/doublewrite_debug.test
index ebae84e2211..ab7fd8eba22 100644
--- a/mysql-test/suite/innodb/t/doublewrite_debug.test
+++ b/mysql-test/suite/innodb/t/doublewrite_debug.test
@@ -17,6 +17,7 @@ call mtr.add_suppression("Plugin 'InnoDB' (init function returned error|registra
 call mtr.add_suppression("InnoDB: A bad Space ID was found in datafile");
 call mtr.add_suppression("InnoDB: Checksum mismatch in datafile: ");
 call mtr.add_suppression("InnoDB: Inconsistent tablespace ID in .*t1\\.ibd");
+call mtr.add_suppression("InnoDB: Header page consists of zero bytes in datafile:");
 --enable_query_log
 
 let INNODB_PAGE_SIZE=`select @@innodb_page_size`;
diff --git a/mysql-test/suite/innodb/t/log_file_name.test b/mysql-test/suite/innodb/t/log_file_name.test
index ee9c34fb631..6b7eb3fde1d 100644
--- a/mysql-test/suite/innodb/t/log_file_name.test
+++ b/mysql-test/suite/innodb/t/log_file_name.test
@@ -7,6 +7,8 @@
 # Embedded server does not support crashing
 --source include/not_embedded.inc
 
+call mtr.add_suppression("InnoDB: Header page consists of zero bytes in datafile:");
+
 SET GLOBAL innodb_file_per_table=ON;
 FLUSH TABLES;
 
@@ -174,6 +176,7 @@ call mtr.add_suppression("InnoDB: Table test/u[123] in the InnoDB data dictionar
 call mtr.add_suppression("InnoDB: Cannot replay rename of tablespace.*");
 call mtr.add_suppression("InnoDB: Attempted to open a previously opened tablespace");
 call mtr.add_suppression("InnoDB: Recovery cannot access file");
+call mtr.add_suppression("InnoDB: Cannot read first page in datafile:");
 FLUSH TABLES;
 --enable_query_log
 
diff --git a/mysql-test/suite/innodb/t/row_size_error_log_warnings_3.test b/mysql-test/suite/innodb/t/row_size_error_log_warnings_3.test
index dab9bcfa864..24029a48aa4 100644
--- a/mysql-test/suite/innodb/t/row_size_error_log_warnings_3.test
+++ b/mysql-test/suite/innodb/t/row_size_error_log_warnings_3.test
@@ -1,7 +1,7 @@
 --source include/have_innodb.inc
 --source include/have_sequence.inc
 --source include/innodb_page_size_small.inc
---source include/have_normal_bzip.inc
+--source include/have_normal_zlib.inc
 
 call mtr.add_suppression("InnoDB: Cannot add field .* in table .* because after adding it, the row size is .* which is greater than maximum allowed size (.*) for a record on index leaf page.");
 
diff --git a/mysql-test/suite/rpl/r/rpl_gis_user_var.result b/mysql-test/suite/rpl/r/rpl_gis_user_var.result
new file mode 100644
index 00000000000..c6aab9e03f0
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_gis_user_var.result
@@ -0,0 +1,21 @@
+include/master-slave.inc
+[connection master]
+#
+#
+#
+connection master;
+SET @p=POINT(1,1);
+CREATE TABLE t1 AS SELECT @p AS p;
+connection slave;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `p` point DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+SELECT ST_AsWKT(p) FROM t1;
+ST_AsWKT(p)
+POINT(1 1)
+connection master;
+DROP TABLE t1;
+connection slave;
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_gis_user_var.test b/mysql-test/suite/rpl/t/rpl_gis_user_var.test
new file mode 100644
index 00000000000..8edd8cb9309
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_gis_user_var.test
@@ -0,0 +1,18 @@
+--source include/have_geometry.inc
+--source include/master-slave.inc
+
+--echo #
+--echo #
+--echo #
+
+connection master;
+SET @p=POINT(1,1);
+CREATE TABLE t1 AS SELECT @p AS p;
+sync_slave_with_master;
+SHOW CREATE TABLE t1;
+SELECT ST_AsWKT(p) FROM t1;
+connection master;
+DROP TABLE t1;
+sync_slave_with_master;
+
+--source include/rpl_end.inc
diff --git a/plugin/type_mysql_json/type.cc b/plugin/type_mysql_json/type.cc
index 57f252564c3..8454201cd82 100644
--- a/plugin/type_mysql_json/type.cc
+++ b/plugin/type_mysql_json/type.cc
@@ -37,6 +37,11 @@ public:
   Field *make_table_field(MEM_ROOT *, const LEX_CSTRING *,
                           const Record_addr &, const Type_all_attributes &,
                           TABLE_SHARE *) const override;
+  bool Column_definition_fix_attributes(Column_definition *c) const override
+  {
+    my_error(ER_NOT_ALLOWED_IN_THIS_CONTEXT, MYF(0), "MYSQL_JSON");
+    return true;
+  }
   void Column_definition_reuse_fix_attributes(THD *thd,
                                               Column_definition *def,
                                               const Field *field) const override;
@@ -62,7 +67,7 @@ public:
   bool parse_mysql(String *dest, const char *data, size_t length) const;
   bool send(Protocol *protocol) { return Field::send(protocol); }
   void sql_type(String &s) const
-  { s.set_ascii(STRING_WITH_LEN("json /* MySQL 5.7 */")); }
+  { s.set_ascii(STRING_WITH_LEN("mysql_json /* JSON from MySQL 5.7 */")); }
   /* this will make ALTER TABLE to consider it different from built-in field */
   Compression_method *compression_method() const { return (Compression_method*)1; }
 };
diff --git a/plugin/user_variables/user_variables.cc b/plugin/user_variables/user_variables.cc
index fe87e17f4ee..7daa5e99e87 100644
--- a/plugin/user_variables/user_variables.cc
+++ b/plugin/user_variables/user_variables.cc
@@ -79,9 +79,9 @@ static int user_variables_fill(THD *thd, TABLE_LIST *tables, COND *cond)
     else
       return 1;
 
-    const LEX_CSTRING *tmp= var->unsigned_flag ?
-                            &unsigned_result_types[var->type] :
-                            &result_types[var->type];
+    const LEX_CSTRING *tmp= var->type_handler()->is_unsigned() ?
+            &unsigned_result_types[var->type_handler()->result_type()] :
+            &result_types[var->type_handler()->result_type()];
     field[2]->store(tmp->str, tmp->length, system_charset_info);
 
     if (var->charset())
diff --git a/sql/debug.cc b/sql/debug.cc
index a0e2340e254..beb667755b8 100644
--- a/sql/debug.cc
+++ b/sql/debug.cc
@@ -40,7 +40,8 @@ static bool debug_decrement_counter(const LEX_CSTRING *name)
   THD *thd= current_thd;
   user_var_entry *entry= (user_var_entry*)
     my_hash_search(&thd->user_vars, (uchar*) name->str, name->length);
-  if (!entry || entry->type != INT_RESULT || ! entry->value)
+  if (!entry || !entry->value ||
+      entry->type_handler()->result_type() != INT_RESULT)
     return 0;
   (*(ulonglong*) entry->value)= (*(ulonglong*) entry->value)-1;
   return !*(ulonglong*) entry->value;
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 707fac4291e..1aec01abc6e 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -4645,7 +4645,6 @@ user_var_entry *get_variable(HASH *hash, LEX_CSTRING *name,
     entry->length=0;
     entry->update_query_id=0;
     entry->set_charset(NULL);
-    entry->unsigned_flag= 0;
     /*
       If we are here, we were called from a SET or a query which sets a
       variable. Imagine it is this:
@@ -4657,7 +4656,7 @@ user_var_entry *get_variable(HASH *hash, LEX_CSTRING *name,
       by Item_func_get_user_var (because that's not necessary).
     */
     entry->used_query_id=current_thd->query_id;
-    entry->type=STRING_RESULT;
+    entry->set_handler(&type_handler_long_blob);
     memcpy((char*) entry->name.str, name->str, name->length+1);
     if (my_hash_insert(hash,(uchar*) entry))
     {
@@ -4733,9 +4732,12 @@ bool Item_func_set_user_var::fix_fields(THD *thd, Item **ref)
   switch (args[0]->result_type()) {
   case STRING_RESULT:
   case TIME_RESULT:
-    set_handler(type_handler_long_blob.
-                type_handler_adjusted_to_max_octet_length(max_length,
-                                                          collation.collation));
+    if (args[0]->field_type() == MYSQL_TYPE_GEOMETRY)
+      set_handler(args[0]->type_handler());
+    else
+      set_handler(type_handler_long_blob.
+                  type_handler_adjusted_to_max_octet_length(max_length,
+                                                            collation.collation));
     break;
   case REAL_RESULT:
     set_handler(&type_handler_double);
@@ -4860,9 +4862,9 @@ bool Item_func_set_user_var::register_field_in_bitmap(void *arg)
 
 bool
 update_hash(user_var_entry *entry, bool set_null, void *ptr, size_t length,
-            Item_result type, CHARSET_INFO *cs,
-            bool unsigned_arg)
+            const Type_handler *th, CHARSET_INFO *cs)
 {
+  entry->set_handler(th);
   if (set_null)
   {
     char *pos= (char*) entry+ ALIGN_SIZE(sizeof(user_var_entry));
@@ -4873,7 +4875,7 @@ update_hash(user_var_entry *entry, bool set_null, void *ptr, size_t length,
   }
   else
   {
-    if (type == STRING_RESULT)
+    if (th->result_type() == STRING_RESULT)
       length++;					// Store strings with end \0
     if (length <= extra_size)
     {
@@ -4902,20 +4904,18 @@ update_hash(user_var_entry *entry, bool set_null, void *ptr, size_t length,
 	  return 1;
       }
     }
-    if (type == STRING_RESULT)
+    if (th->result_type() == STRING_RESULT)
     {
       length--;					// Fix length change above
       entry->value[length]= 0;			// Store end \0
     }
     if (length)
       memmove(entry->value, ptr, length);
-    if (type == DECIMAL_RESULT)
+    if (th->result_type() == DECIMAL_RESULT)
       ((my_decimal*)entry->value)->fix_buffer_pointer();
     entry->length= length;
     entry->set_charset(cs);
-    entry->unsigned_flag= unsigned_arg;
   }
-  entry->type=type;
 #ifdef USER_VAR_TRACKING
 #ifndef EMBEDDED_LIBRARY
   THD *thd= current_thd;
@@ -4928,9 +4928,8 @@ update_hash(user_var_entry *entry, bool set_null, void *ptr, size_t length,
 
 bool
 Item_func_set_user_var::update_hash(void *ptr, size_t length,
-                                    Item_result res_type,
-                                    CHARSET_INFO *cs,
-                                    bool unsigned_arg)
+                                    const Type_handler *th,
+                                    CHARSET_INFO *cs)
 {
   /*
     If we set a variable explicitly to NULL then keep the old
@@ -4944,9 +4943,8 @@ Item_func_set_user_var::update_hash(void *ptr, size_t length,
   else
     null_value= args[0]->null_value;
   if (null_value && null_item)
-    res_type= m_var_entry->type;                 // Don't change type of item
-  if (::update_hash(m_var_entry, null_value,
-                    ptr, length, res_type, cs, unsigned_arg))
+    th= m_var_entry->type_handler(); // Don't change type of item
+  if (::update_hash(m_var_entry, null_value, ptr, length, th, cs))
   {
     null_value= 1;
     return 1;
@@ -4962,7 +4960,7 @@ double user_var_entry::val_real(bool *null_value)
   if ((*null_value= (value == 0)))
     return 0.0;
 
-  switch (type) {
+  switch (type_handler()->result_type()) {
   case REAL_RESULT:
     return *(double*) value;
   case INT_RESULT:
@@ -4987,7 +4985,7 @@ longlong user_var_entry::val_int(bool *null_value) const
   if ((*null_value= (value == 0)))
     return 0;
 
-  switch (type) {
+  switch (type_handler()->result_type()) {
   case REAL_RESULT:
     return (longlong) *(double*) value;
   case INT_RESULT:
@@ -5016,12 +5014,12 @@ String *user_var_entry::val_str(bool *null_value, String *str,
   if ((*null_value= (value == 0)))
     return (String*) 0;
 
-  switch (type) {
+  switch (type_handler()->result_type()) {
   case REAL_RESULT:
     str->set_real(*(double*) value, decimals, charset());
     break;
   case INT_RESULT:
-    if (!unsigned_flag)
+    if (!type_handler()->is_unsigned())
       str->set(*(longlong*) value, charset());
     else
       str->set(*(ulonglong*) value, charset());
@@ -5048,7 +5046,7 @@ my_decimal *user_var_entry::val_decimal(bool *null_value, my_decimal *val)
   if ((*null_value= (value == 0)))
     return 0;
 
-  switch (type) {
+  switch (type_handler()->result_type()) {
   case REAL_RESULT:
     double2my_decimal(E_DEC_FATAL_ERROR, *(double*) value, val);
     break;
@@ -5187,33 +5185,37 @@ Item_func_set_user_var::update()
   case REAL_RESULT:
   {
     res= update_hash((void*) &save_result.vreal,sizeof(save_result.vreal),
-		     REAL_RESULT, &my_charset_numeric, 0);
+		     &type_handler_double, &my_charset_numeric);
     break;
   }
   case INT_RESULT:
   {
     res= update_hash((void*) &save_result.vint, sizeof(save_result.vint),
-                     INT_RESULT, &my_charset_numeric, unsigned_flag);
+                     unsigned_flag ? (Type_handler *) &type_handler_ulonglong :
+                                     (Type_handler *) &type_handler_slonglong,
+                     &my_charset_numeric);
     break;
   }
   case STRING_RESULT:
   {
     if (!save_result.vstr)					// Null value
-      res= update_hash((void*) 0, 0, STRING_RESULT, &my_charset_bin, 0);
+      res= update_hash((void*) 0, 0, &type_handler_long_blob, &my_charset_bin);
     else
       res= update_hash((void*) save_result.vstr->ptr(),
-		       save_result.vstr->length(), STRING_RESULT,
-		       save_result.vstr->charset(), 0);
+                       save_result.vstr->length(),
+                       field_type() == MYSQL_TYPE_GEOMETRY ?
+                       type_handler() : &type_handler_long_blob,
+                       save_result.vstr->charset());
     break;
   }
   case DECIMAL_RESULT:
   {
     if (!save_result.vdec)					// Null value
-      res= update_hash((void*) 0, 0, DECIMAL_RESULT, &my_charset_bin, 0);
+      res= update_hash((void*) 0, 0, &type_handler_newdecimal, &my_charset_bin);
     else
       res= update_hash((void*) save_result.vdec,
-                       sizeof(my_decimal), DECIMAL_RESULT,
-                       &my_charset_numeric, 0);
+                       sizeof(my_decimal), &type_handler_newdecimal,
+                       &my_charset_numeric);
     break;
   }
   case ROW_RESULT:
@@ -5605,9 +5607,8 @@ get_var_with_binlog(THD *thd, enum_sql_command sql_command,
   user_var_event->value= (char*) user_var_event +
     ALIGN_SIZE(sizeof(BINLOG_USER_VAR_EVENT));
   user_var_event->user_var_event= var_entry;
-  user_var_event->type= var_entry->type;
+  user_var_event->th= var_entry->type_handler();
   user_var_event->charset_number= var_entry->charset()->number;
-  user_var_event->unsigned_flag= var_entry->unsigned_flag;
   if (!var_entry->value)
   {
     /* NULL value*/
@@ -5650,9 +5651,9 @@ bool Item_func_get_user_var::fix_length_and_dec()
   */
   if (likely(!error && m_var_entry))
   {
-    unsigned_flag= m_var_entry->unsigned_flag;
+    unsigned_flag= m_var_entry->type_handler()->is_unsigned();
     max_length= (uint32)m_var_entry->length;
-    switch (m_var_entry->type) {
+    switch (m_var_entry->type_handler()->result_type()) {
     case REAL_RESULT:
       collation.set(&my_charset_numeric, DERIVATION_NUMERIC);
       fix_char_length(DBL_DIG + 8);
@@ -5671,6 +5672,8 @@ bool Item_func_get_user_var::fix_length_and_dec()
       collation.set(m_var_entry->charset(), DERIVATION_IMPLICIT);
       max_length= MAX_BLOB_WIDTH - 1;
       set_handler(&type_handler_long_blob);
+      if (m_var_entry->type_handler()->field_type() == MYSQL_TYPE_GEOMETRY)
+        set_handler(m_var_entry->type_handler());
       break;
     case DECIMAL_RESULT:
       collation.set(&my_charset_numeric, DERIVATION_NUMERIC);
@@ -5743,7 +5746,7 @@ bool Item_user_var_as_out_param::fix_fields(THD *thd, Item **ref)
   DBUG_ASSERT(thd->lex->exchange);
   if (!(entry= get_variable(&thd->user_vars, &org_name, 1)))
     return TRUE;
-  entry->type= STRING_RESULT;
+  entry->set_handler(&type_handler_long_blob);
   /*
     Let us set the same collation which is used for loading
     of fields in LOAD DATA INFILE.
@@ -5759,15 +5762,14 @@ bool Item_user_var_as_out_param::fix_fields(THD *thd, Item **ref)
 
 void Item_user_var_as_out_param::set_null_value(CHARSET_INFO* cs)
 {
-  ::update_hash(entry, TRUE, 0, 0, STRING_RESULT, cs, 0 /* unsigned_arg */);
+  ::update_hash(entry, TRUE, 0, 0, &type_handler_long_blob, cs);
 }
 
 
 void Item_user_var_as_out_param::set_value(const char *str, uint length,
                                            CHARSET_INFO* cs)
 {
-  ::update_hash(entry, FALSE, (void*)str, length, STRING_RESULT, cs,
-                0 /* unsigned_arg */);
+  ::update_hash(entry, FALSE, (void*)str, length, &type_handler_long_blob, cs);
 }
 
 
diff --git a/sql/item_func.h b/sql/item_func.h
index fc921f8d91b..170fc943681 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -3447,8 +3447,8 @@ public:
   String *str_result(String *str) override;
   my_decimal *val_decimal_result(my_decimal *) override;
   bool is_null_result() override;
-  bool update_hash(void *ptr, size_t length, enum Item_result type,
-                   CHARSET_INFO *cs, bool unsigned_arg);
+  bool update_hash(void *ptr, size_t length, const Type_handler *th,
+                   CHARSET_INFO *cs);
   bool send(Protocol *protocol, st_value *buffer) override;
   void make_send_field(THD *thd, Send_field *tmp_field) override;
   bool check(bool use_result_field);
@@ -4289,7 +4289,6 @@ double my_double_round(double value, longlong dec, bool dec_unsigned,
 extern bool volatile  mqh_used;
 
 bool update_hash(user_var_entry *entry, bool set_null, void *ptr, size_t length,
-                 Item_result type, CHARSET_INFO *cs,
-                 bool unsigned_arg);
+                 const Type_handler *th, CHARSET_INFO *cs);
 
 #endif /* ITEM_FUNC_INCLUDED */
diff --git a/sql/log.cc b/sql/log.cc
index 2e55949a722..518f4df3f94 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -6830,18 +6830,12 @@ bool MYSQL_BIN_LOG::write(Log_event *event_info, my_bool *with_annotate)
             BINLOG_USER_VAR_EVENT *user_var_event;
             get_dynamic(&thd->user_var_events,(uchar*) &user_var_event, i);
 
-            /* setting flags for user var log event */
-            uchar flags= User_var_log_event::UNDEF_F;
-            if (user_var_event->unsigned_flag)
-              flags|= User_var_log_event::UNSIGNED_F;
-
             User_var_log_event e(thd, user_var_event->user_var_event->name.str,
                                  user_var_event->user_var_event->name.length,
                                  user_var_event->value,
                                  user_var_event->length,
-                                 user_var_event->type,
-                                 user_var_event->charset_number,
-                                 flags,
+                            user_var_event->th->user_var_log_event_data_type(
+                               user_var_event->charset_number),
                                  using_trans,
                                  direct);
             if (write_event(&e, cache_data, file))
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 5c1ff3d2e68..c5d58c04054 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -2907,6 +2907,41 @@ XA_prepare_log_event(const uchar *buf,
   User_var_log_event methods
 **************************************************************************/
 
+bool Log_event_data_type::unpack_optional_attributes(const char *pos,
+                                                     const char *end)
+
+{
+  for ( ; pos < end; )
+  {
+    switch (*pos) {
+    case CHUNK_SIGNED:
+      m_is_unsigned= false;
+      pos++;
+      continue;
+    case CHUNK_UNSIGNED:
+      m_is_unsigned= true;
+      pos++;
+      continue;
+    case CHUNK_DATA_TYPE_NAME:
+      {
+        pos++;
+        if (pos >= end)
+          return true;
+        uint length= (uchar) *pos++;
+        if (pos + length > end)
+          return true;
+        m_data_type_name= {pos, length};
+        pos+= length;
+        continue;
+      }
+    default:
+      break; // Unknown chunk
+    }
+  }
+  return false;
+}
+
+
 User_var_log_event::
 User_var_log_event(const uchar *buf, uint event_len,
                    const Format_description_log_event* description_event)
@@ -2916,7 +2951,8 @@ User_var_log_event(const uchar *buf, uint event_len,
 #endif
 {
   bool error= false;
-  const uchar *buf_start= buf, *buf_end= buf + event_len;
+  const uchar *const buf_start= buf;
+  const char *buf_end= reinterpret_cast<const char*>(buf) + event_len;
 
   /* The Post-Header is empty. The Variable Data part begins immediately. */
   buf+= description_event->common_header_len +
@@ -2944,11 +2980,8 @@ User_var_log_event(const uchar *buf, uint event_len,
 
   buf+= UV_NAME_LEN_SIZE + name_len;
   is_null= (bool) *buf;
-  flags= User_var_log_event::UNDEF_F;    // defaults to UNDEF_F
   if (is_null)
   {
-    type= STRING_RESULT;
-    charset_number= my_charset_bin.number;
     val_len= 0;
     val= 0;  
   }
@@ -2963,8 +2996,8 @@ User_var_log_event(const uchar *buf, uint event_len,
       goto err;
     }
 
-    type= (Item_result) buf[UV_VAL_IS_NULL];
-    charset_number= uint4korr(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE);
+    m_type= (Item_result) buf[UV_VAL_IS_NULL];
+    m_charset_number= uint4korr(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE);
     val_len= uint4korr(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE +
                        UV_CHARSET_NUMBER_SIZE);
 
@@ -2977,20 +3010,14 @@ User_var_log_event(const uchar *buf, uint event_len,
       the flags value.
 
       Old events will not have this extra byte, thence,
-      we keep the flags set to UNDEF_F.
+      we keep m_is_unsigned==false.
     */
-    size_t bytes_read= (val + val_len) - (char*) buf_start;
-    if (bytes_read > event_len)
+    const char *pos= val + val_len;
+    if (pos > buf_end || unpack_optional_attributes(pos, buf_end))
     {
       error= true;
       goto err;
     }
-    if ((data_written - bytes_read) > 0)
-    {
-      flags= (uint) *(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE +
-                    UV_CHARSET_NUMBER_SIZE + UV_VAL_LEN_SIZE +
-                    val_len);
-    }
   }
 
 err:
diff --git a/sql/log_event.h b/sql/log_event.h
index a3b48f03184..a85dcf9d782 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -57,6 +57,8 @@
 
 #include "rpl_gtid.h"
 
+#include "log_event_data_type.h"
+
 /* Forward declarations */
 #ifndef MYSQL_CLIENT
 class String;
@@ -3284,33 +3286,27 @@ private:
   @section User_var_log_event_binary_format Binary Format  
 */
 
-class User_var_log_event: public Log_event
+
+class User_var_log_event: public Log_event, public Log_event_data_type
 {
 public:
-  enum {
-    UNDEF_F= 0,
-    UNSIGNED_F= 1
-  };
   const char *name;
   size_t name_len;
   const char *val;
   size_t val_len;
-  Item_result type;
-  uint charset_number;
   bool is_null;
-  uchar flags;
 #ifdef MYSQL_SERVER
   bool deferred;
   query_id_t query_id;
   User_var_log_event(THD* thd_arg, const char *name_arg, size_t name_len_arg,
                      const char *val_arg, size_t val_len_arg,
-                     Item_result type_arg,
-		     uint charset_number_arg, uchar flags_arg,
+                     const Log_event_data_type &data_type,
                      bool using_trans, bool direct)
     :Log_event(thd_arg, 0, using_trans),
+    Log_event_data_type(data_type),
     name(name_arg), name_len(name_len_arg), val(val_arg),
-    val_len(val_len_arg), type(type_arg), charset_number(charset_number_arg),
-    flags(flags_arg), deferred(false)
+    val_len(val_len_arg),
+    deferred(false)
     {
       is_null= !val;
       if (direct)
diff --git a/sql/log_event_client.cc b/sql/log_event_client.cc
index 51667f4fcce..cbb593d0888 100644
--- a/sql/log_event_client.cc
+++ b/sql/log_event_client.cc
@@ -2436,7 +2436,7 @@ bool User_var_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
   }
   else
   {
-    switch (type) {
+    switch (m_type) {
     case REAL_RESULT:
       double real_val;
       char real_buf[FMT_G_BUFSIZE(14)];
@@ -2448,8 +2448,7 @@ bool User_var_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
       break;
     case INT_RESULT:
       char int_buf[22];
-      longlong10_to_str(uint8korr(val), int_buf, 
-                        ((flags & User_var_log_event::UNSIGNED_F) ? 10 : -10));
+      longlong10_to_str(uint8korr(val), int_buf,  is_unsigned() ? 10 : -10);
       if (my_b_printf(&cache, ":=%s%s\n", int_buf,
                       print_event_info->delimiter))
         goto err;
@@ -2504,7 +2503,7 @@ bool User_var_log_event::print(FILE* file, PRINT_EVENT_INFO* print_event_info)
         people want to mysqlbinlog|mysql into another server not supporting the
         character set. But there's not much to do about this and it's unlikely.
       */
-      if (!(cs= get_charset(charset_number, MYF(0))))
+      if (!(cs= get_charset(m_charset_number, MYF(0))))
       {        /*
           Generate an unusable command (=> syntax error) is probably the best
           thing we can do here.
diff --git a/sql/log_event_data_type.h b/sql/log_event_data_type.h
new file mode 100644
index 00000000000..e3b2039a63c
--- /dev/null
+++ b/sql/log_event_data_type.h
@@ -0,0 +1,74 @@
+/* Copyright (c) 2024, MariaDB Corporation.
+
+   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-1335  USA */
+
+#ifndef LOG_EVENT_DATA_TYPE_H
+#define LOG_EVENT_DATA_TYPE_H
+
+class Log_event_data_type
+{
+public:
+
+  enum {
+    CHUNK_SIGNED= 0,
+    CHUNK_UNSIGNED= 1,
+    CHUNK_DATA_TYPE_NAME= 2
+  };
+
+protected:
+  LEX_CSTRING m_data_type_name;
+  Item_result m_type;
+  uint m_charset_number;
+  bool m_is_unsigned;
+
+public:
+
+  Log_event_data_type()
+   :m_data_type_name({NULL,0}),
+    m_type(STRING_RESULT),
+    m_charset_number(my_charset_bin.number),
+    m_is_unsigned(false)
+  { }
+
+  Log_event_data_type(const LEX_CSTRING &data_type_name_arg,
+                      Item_result type_arg,
+                      uint charset_number_arg,
+                      bool is_unsigned_arg)
+   :m_data_type_name(data_type_name_arg),
+    m_type(type_arg),
+    m_charset_number(charset_number_arg),
+    m_is_unsigned(is_unsigned_arg)
+  { }
+
+  const LEX_CSTRING & data_type_name() const
+  {
+    return m_data_type_name;
+  }
+  Item_result type() const
+  {
+    return m_type;
+  }
+  uint charset_number() const
+  {
+    return m_charset_number;
+  }
+  bool is_unsigned() const
+  {
+    return m_is_unsigned;
+  }
+
+  bool unpack_optional_attributes(const char *str, const char *end);
+};
+
+#endif // LOG_EVENT_DATA_TYPE_H
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index 4aa64b2e088..c4be02a5a51 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -4218,11 +4218,16 @@ bool XA_prepare_log_event::write()
 #if defined(HAVE_REPLICATION)
 static bool
 user_var_append_name_part(THD *thd, String *buf,
-                          const char *name, size_t name_len)
+                          const char *name, size_t name_len,
+                          const LEX_CSTRING &data_type_name)
 {
   return buf->append('@') ||
     append_identifier(thd, buf, name, name_len) ||
-    buf->append('=');
+    buf->append('=') ||
+    (data_type_name.length &&
+     (buf->append(STRING_WITH_LEN("/*")) ||
+      buf->append(data_type_name.str, data_type_name.length) ||
+      buf->append(STRING_WITH_LEN("*/"))));
 }
 
 void User_var_log_event::pack_info(Protocol* protocol)
@@ -4232,14 +4237,15 @@ void User_var_log_event::pack_info(Protocol* protocol)
     char buf_mem[FN_REFLEN+7];
     String buf(buf_mem, sizeof(buf_mem), system_charset_info);
     buf.length(0);
-    if (user_var_append_name_part(protocol->thd, &buf, name, name_len) ||
+    if (user_var_append_name_part(protocol->thd, &buf, name, name_len,
+                                  m_data_type_name) ||
         buf.append(NULL_clex_str))
       return;
     protocol->store(buf.ptr(), buf.length(), &my_charset_bin);
   }
   else
   {
-    switch (type) {
+    switch (m_type) {
     case REAL_RESULT:
     {
       double real_val;
@@ -4248,7 +4254,8 @@ void User_var_log_event::pack_info(Protocol* protocol)
       String buf(buf_mem, sizeof(buf_mem), system_charset_info);
       float8get(real_val, val);
       buf.length(0);
-      if (user_var_append_name_part(protocol->thd, &buf, name, name_len) ||
+      if (user_var_append_name_part(protocol->thd, &buf, name, name_len,
+                                    m_data_type_name) ||
           buf.append(buf2, my_gcvt(real_val, MY_GCVT_ARG_DOUBLE,
                                    MY_GCVT_MAX_FIELD_WIDTH, buf2, NULL)))
         return;
@@ -4261,10 +4268,11 @@ void User_var_log_event::pack_info(Protocol* protocol)
       char buf_mem[FN_REFLEN + 22];
       String buf(buf_mem, sizeof(buf_mem), system_charset_info);
       buf.length(0);
-      if (user_var_append_name_part(protocol->thd, &buf, name, name_len) ||
+      if (user_var_append_name_part(protocol->thd, &buf, name, name_len,
+                                    m_data_type_name) ||
           buf.append(buf2,
                  longlong10_to_str(uint8korr(val), buf2,
-                   ((flags & User_var_log_event::UNSIGNED_F) ? 10 : -10))-buf2))
+                   (is_unsigned() ? 10 : -10))-buf2))
         return;
       protocol->store(buf.ptr(), buf.length(), &my_charset_bin);
       break;
@@ -4277,7 +4285,8 @@ void User_var_log_event::pack_info(Protocol* protocol)
       String str(buf2, sizeof(buf2), &my_charset_bin);
       buf.length(0);
       my_decimal((const uchar *) (val + 2), val[0], val[1]).to_string(&str);
-      if (user_var_append_name_part(protocol->thd, &buf, name, name_len) ||
+      if (user_var_append_name_part(protocol->thd, &buf, name, name_len,
+                                    m_data_type_name) ||
           buf.append(str))
         return;
       protocol->store(buf.ptr(), buf.length(), &my_charset_bin);
@@ -4291,7 +4300,7 @@ void User_var_log_event::pack_info(Protocol* protocol)
       String buf(buf_mem, sizeof(buf_mem), system_charset_info);
       CHARSET_INFO *cs;
       buf.length(0);
-      if (!(cs= get_charset(charset_number, MYF(0))))
+      if (!(cs= get_charset(m_charset_number, MYF(0))))
       {
         if (buf.append(STRING_WITH_LEN("???")))
           return;
@@ -4300,7 +4309,8 @@ void User_var_log_event::pack_info(Protocol* protocol)
       {
         size_t old_len;
         char *beg, *end;
-        if (user_var_append_name_part(protocol->thd, &buf, name, name_len) ||
+        if (user_var_append_name_part(protocol->thd, &buf, name, name_len,
+                                      m_data_type_name) ||
             buf.append('_') ||
             buf.append(cs->cs_name) ||
             buf.append(' '))
@@ -4348,10 +4358,10 @@ bool User_var_log_event::write()
   }    
   else
   {
-    buf1[1]= type;
-    int4store(buf1 + 2, charset_number);
+    buf1[1]= m_type;
+    int4store(buf1 + 2, m_charset_number);
 
-    switch (type) {
+    switch (m_type) {
     case REAL_RESULT:
       float8store(buf2, *(double*) val);
       break;
@@ -4381,15 +4391,28 @@ bool User_var_log_event::write()
     buf1_length= 10;
   }
 
-  /* Length of the whole event */
-  event_length= sizeof(buf)+ name_len + buf1_length + val_len + unsigned_len;
+  uchar data_type_name_chunk_signature= (uchar) CHUNK_DATA_TYPE_NAME;
+  uint data_type_name_chunk_signature_length= m_data_type_name.length ? 1 : 0;
+  uchar data_type_name_length_length= m_data_type_name.length ? 1 : 0;
 
+  /* Length of the whole event */
+  event_length= sizeof(buf)+ name_len + buf1_length + val_len + unsigned_len +
+                data_type_name_chunk_signature_length +
+                data_type_name_length_length +
+                (uint) m_data_type_name.length;
+
+  uchar unsig= m_is_unsigned ? CHUNK_UNSIGNED : CHUNK_SIGNED;
+  uchar data_type_name_length= (uchar) m_data_type_name.length;
   return write_header(event_length) ||
          write_data(buf, sizeof(buf))   ||
          write_data(name, name_len)     ||
          write_data(buf1, buf1_length) ||
          write_data(pos, val_len) ||
-         write_data(&flags, unsigned_len) ||
+         write_data(&unsig, unsigned_len) ||
+         write_data(&data_type_name_chunk_signature,
+                    data_type_name_chunk_signature_length) ||
+         write_data(&data_type_name_length, data_type_name_length_length) ||
+         write_data(m_data_type_name.str, (uint) m_data_type_name.length) ||
          write_footer();
 }
 
@@ -4413,7 +4436,7 @@ int User_var_log_event::do_apply_event(rpl_group_info *rgi)
     current_thd->query_id= query_id; /* recreating original time context */
   }
 
-  if (!(charset= get_charset(charset_number, MYF(MY_WME))))
+  if (!(charset= get_charset(m_charset_number, MYF(MY_WME))))
   {
     rgi->rli->report(ERROR_LEVEL, ER_SLAVE_FATAL_ERROR,
                 ER_THD(thd, ER_SLAVE_FATAL_ERROR),
@@ -4432,7 +4455,7 @@ int User_var_log_event::do_apply_event(rpl_group_info *rgi)
   }
   else
   {
-    switch (type) {
+    switch (m_type) {
     case REAL_RESULT:
       if (val_len != 8)
       {
@@ -4496,13 +4519,10 @@ int User_var_log_event::do_apply_event(rpl_group_info *rgi)
   if (e->fix_fields(thd, 0))
     DBUG_RETURN(1);
 
-  /*
-    A variable can just be considered as a table with
-    a single record and with a single column. Thus, like
-    a column value, it could always have IMPLICIT derivation.
-   */
-  e->update_hash((void*) val, val_len, type, charset,
-                 (flags & User_var_log_event::UNSIGNED_F));
+  const Type_handler *th= Type_handler::handler_by_log_event_data_type(thd,
+                                                                       *this);
+  e->update_hash((void*) val, val_len, th, charset);
+
   if (!is_deferred())
     free_root(thd->mem_root, 0);
   else
diff --git a/sql/opt_trace.cc b/sql/opt_trace.cc
index 4bc493940fb..972c7da6a2a 100644
--- a/sql/opt_trace.cc
+++ b/sql/opt_trace.cc
@@ -103,7 +103,8 @@ inline bool sql_command_can_be_traced(enum enum_sql_command sql_command)
          sql_command == SQLCOM_UPDATE ||
          sql_command == SQLCOM_DELETE ||
          sql_command == SQLCOM_DELETE_MULTI ||
-         sql_command == SQLCOM_UPDATE_MULTI;
+         sql_command == SQLCOM_UPDATE_MULTI ||
+         sql_command == SQLCOM_INSERT_SELECT;
 }
 
 void opt_trace_print_expanded_query(THD *thd, SELECT_LEX *select_lex,
diff --git a/sql/sql_binlog.cc b/sql/sql_binlog.cc
index 9f61135232f..63fe62d8a19 100644
--- a/sql/sql_binlog.cc
+++ b/sql/sql_binlog.cc
@@ -134,7 +134,7 @@ int binlog_defragment(THD *thd)
     entry[k]=
       (user_var_entry*) my_hash_search(&thd->user_vars, (uchar*) name[k].str,
                                        name[k].length);
-    if (!entry[k] || entry[k]->type != STRING_RESULT)
+    if (!entry[k] || entry[k]->type_handler()->result_type() != STRING_RESULT)
     {
       my_error(ER_WRONG_TYPE_FOR_VAR, MYF(0), name[k].str);
       return -1;
@@ -159,7 +159,8 @@ int binlog_defragment(THD *thd)
     gathered_length += entry[k]->length;
   }
   for (uint k=0; k < 2; k++)
-    update_hash(entry[k], true, NULL, 0, STRING_RESULT, &my_charset_bin, 0);
+    update_hash(entry[k], true, NULL, 0,
+                &type_handler_long_blob, &my_charset_bin);
 
   DBUG_ASSERT(gathered_length == thd->lex->comment.length);
 
diff --git a/sql/sql_class.h b/sql/sql_class.h
index e34aa73017b..6bbac5647be 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -278,9 +278,8 @@ typedef struct st_user_var_events
   user_var_entry *user_var_event;
   char *value;
   size_t length;
-  Item_result type;
+  const Type_handler *th;
   uint charset_number;
-  bool unsigned_flag;
 } BINLOG_USER_VAR_EVENT;
 
 /*
@@ -7133,7 +7132,7 @@ public:
 
 
 // this is needed for user_vars hash
-class user_var_entry
+class user_var_entry: public Type_handler_hybrid_field_type
 {
   CHARSET_INFO *m_charset;
  public:
@@ -7142,8 +7141,6 @@ class user_var_entry
   char *value;
   size_t length;
   query_id_t update_query_id, used_query_id;
-  Item_result type;
-  bool unsigned_flag;
 
   double val_real(bool *null_value);
   longlong val_int(bool *null_value) const;
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index c880ff98f71..70b4b9aedf7 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -510,7 +510,7 @@ static enum enum_binlog_checksum_alg get_binlog_checksum_value_at_connect(THD *
   }
   else
   {
-    DBUG_ASSERT(entry->type == STRING_RESULT);
+    DBUG_ASSERT(entry->type_handler()->result_type() == STRING_RESULT);
     String str;
     uint dummy_errors;
     str.copy(entry->value, entry->length, &my_charset_bin, &my_charset_bin,
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index b62909cbb53..24b0b9f146c 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -2242,6 +2242,34 @@ Type_handler::get_handler_by_real_type(enum_field_types type)
 }
 
 
+const Type_handler *
+Type_handler::handler_by_log_event_data_type(THD *thd,
+                                             const Log_event_data_type &type)
+{
+  if (type.data_type_name().length)
+  {
+    const Type_handler *th= handler_by_name(thd, type.data_type_name());
+    if (th)
+      return th;
+  }
+  switch (type.type()) {
+  case STRING_RESULT:
+  case ROW_RESULT:
+  case TIME_RESULT:
+    break;
+  case REAL_RESULT:
+    return &type_handler_double;
+  case INT_RESULT:
+    if (type.is_unsigned())
+      return &type_handler_ulonglong;
+    return &type_handler_slonglong;
+  case DECIMAL_RESULT:
+    return &type_handler_newdecimal;
+  }
+  return &type_handler_long_blob;
+}
+
+
 /**
   Create a DOUBLE field by default.
 */
diff --git a/sql/sql_type.h b/sql/sql_type.h
index 732d5250b33..a3f863ffa4b 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -30,6 +30,8 @@
 #include "sql_type_string.h"
 #include "sql_type_real.h"
 #include "compat56.h"
+#include "log_event_data_type.h"
+
 C_MODE_START
 #include <ma_dyncol.h>
 C_MODE_END
@@ -3685,6 +3687,9 @@ public:
   static const Type_handler *handler_by_name(THD *thd, const LEX_CSTRING &name);
   static const Type_handler *handler_by_name_or_error(THD *thd,
                                                       const LEX_CSTRING &name);
+  static const Type_handler *handler_by_log_event_data_type(
+                                             THD *thd,
+                                             const Log_event_data_type &type);
   static const Type_handler *odbc_literal_type_handler(const LEX_CSTRING *str);
   static const Type_handler *blob_type_handler(uint max_octet_length);
   static const Type_handler *string_type_handler(uint max_octet_length);
@@ -3969,6 +3974,12 @@ public:
   {
     return false;
   }
+
+  virtual Log_event_data_type user_var_log_event_data_type(uint charset_nr) const
+  {
+    return Log_event_data_type({NULL,0}/*data type name*/, result_type(),
+                               charset_nr, is_unsigned());
+  }
   virtual uint Column_definition_gis_options_image(uchar *buff,
                                                    const Column_definition &def)
                                                    const
diff --git a/sql/sql_type_geom.h b/sql/sql_type_geom.h
index 95726aa3bf7..e6308cc6b90 100644
--- a/sql/sql_type_geom.h
+++ b/sql/sql_type_geom.h
@@ -82,6 +82,13 @@ public:
   Field *make_conversion_table_field(MEM_ROOT *root,
                                      TABLE *table, uint metadata,
                                      const Field *target) const override;
+  Log_event_data_type user_var_log_event_data_type(uint charset_nr)
+                                                              const override
+  {
+    return Log_event_data_type(name().lex_cstring(), result_type(),
+                               charset_nr, false/*unsigned*/);
+  }
+
   uint Column_definition_gis_options_image(uchar *buff,
                                            const Column_definition &def)
                                            const override;
diff --git a/storage/innobase/buf/buf0dblwr.cc b/storage/innobase/buf/buf0dblwr.cc
index 2503b1a81dd..c1b4fdd4bf3 100644
--- a/storage/innobase/buf/buf0dblwr.cc
+++ b/storage/innobase/buf/buf0dblwr.cc
@@ -338,11 +338,14 @@ func_exit:
     os_file_flush(file);
   }
   else
-    for (ulint i= 0; i < size * 2; i++, page += srv_page_size)
-      if (mach_read_from_8(my_assume_aligned<8>(page + FIL_PAGE_LSN)))
-        /* Each valid page header must contain a nonzero FIL_PAGE_LSN field. */
+  {
+    alignas(8) char checkpoint[8];
+    mach_write_to_8(checkpoint, log_sys.next_checkpoint_lsn);
+    for (auto i= size * 2; i--; page += srv_page_size)
+      if (memcmp_aligned<8>(page + FIL_PAGE_LSN, checkpoint, 8) >= 0)
+        /* Valid pages are not older than the log checkpoint. */
         recv_sys.dblwr.add(page);
-
+  }
   err= DB_SUCCESS;
   goto func_exit;
 }
diff --git a/storage/innobase/fsp/fsp0file.cc b/storage/innobase/fsp/fsp0file.cc
index 269a05751d7..e321efc7465 100644
--- a/storage/innobase/fsp/fsp0file.cc
+++ b/storage/innobase/fsp/fsp0file.cc
@@ -436,12 +436,22 @@ Datafile::validate_for_recovery()
 			return(err);
 		}
 
+		if (!m_space_id) {
+			m_space_id = recv_sys.dblwr.find_first_page(
+				m_filepath, m_handle);
+			if (m_space_id) {
+				m_defer= false;
+				goto free_first_page;
+			} else return err;
+		}
+
 		if (!m_defer) {
 			err = find_space_id();
 			if (err != DB_SUCCESS || m_space_id == 0) {
-				ib::error() << "Datafile '" << m_filepath
-					<< "' is corrupted. Cannot determine "
-					"the space ID from the first 64 pages.";
+				sql_print_error(
+					"InnoDB: Datafile '%s' is corrupted."
+					" Cannot determine the space ID from"
+					" the first 64 pages.", m_filepath);
 				return(err);
 			}
 		}
@@ -454,7 +464,7 @@ Datafile::validate_for_recovery()
 			m_space_id, m_filepath, m_handle)) {
 			return m_defer ? err : DB_CORRUPTION;
 		}
-
+free_first_page:
 		/* Free the previously read first page and then re-validate. */
 		free_first_page();
 		m_defer = false;
@@ -502,11 +512,11 @@ err_exit:
 			return DB_SUCCESS;
 		}
 
-		ib::info() << error_txt << " in datafile: " << m_filepath
-			<< ", Space ID:" << m_space_id  << ", Flags: "
-			<< m_flags;
+		sql_print_error("InnoDB: %s in datafile: %s, Space ID: %zu, "
+				"Flags: %zu", error_txt, m_filepath,
+				m_space_id, m_flags);
 		m_is_valid = false;
-		return(DB_CORRUPTION);
+		return DB_CORRUPTION;
 	}
 
 	/* Check if the whole page is blank. */
diff --git a/storage/innobase/fsp/fsp0sysspace.cc b/storage/innobase/fsp/fsp0sysspace.cc
index 3ca0f9975fd..9ede0a070f6 100644
--- a/storage/innobase/fsp/fsp0sysspace.cc
+++ b/storage/innobase/fsp/fsp0sysspace.cc
@@ -573,7 +573,7 @@ SysTablespace::read_lsn_and_check_flags(lsn_t* flushed_lsn)
 	}
 
 	err = it->read_first_page(
-		m_ignore_read_only ?  false : srv_read_only_mode);
+		m_ignore_read_only && srv_read_only_mode);
 
 	if (err != DB_SUCCESS) {
 		return(err);
@@ -587,20 +587,17 @@ SysTablespace::read_lsn_and_check_flags(lsn_t* flushed_lsn)
 
 	/* Check the contents of the first page of the
 	first datafile. */
-	for (int retry = 0; retry < 2; ++retry) {
+	err = it->validate_first_page(flushed_lsn);
 
-		err = it->validate_first_page(flushed_lsn);
-
-		if (err != DB_SUCCESS
-		    && (retry == 1
-			|| recv_sys.dblwr.restore_first_page(
+	if (err != DB_SUCCESS) {
+		if (recv_sys.dblwr.restore_first_page(
 				it->m_space_id, it->m_filepath,
-				it->handle()))) {
-
+				it->handle())) {
 			it->close();
-
 			return(err);
 		}
+		err = it->read_first_page(
+			m_ignore_read_only && srv_read_only_mode);
 	}
 
 	/* Make sure the tablespace space ID matches the
diff --git a/storage/innobase/include/buf0dblwr.h b/storage/innobase/include/buf0dblwr.h
index 9932b0e521f..6e7662d9b81 100644
--- a/storage/innobase/include/buf0dblwr.h
+++ b/storage/innobase/include/buf0dblwr.h
@@ -105,7 +105,8 @@ public:
   If we are upgrading from a version before MySQL 4.1, then this
   function performs the necessary update operations to support
   innodb_file_per_table. If we are in a crash recovery, this function
-  loads the pages from double write buffer into memory.
+  loads the pages from double write buffer which are not older than
+  the checkpoint into memory.
   @param file File handle
   @param path Path name of file
   @return DB_SUCCESS or error code */
diff --git a/storage/innobase/include/log0recv.h b/storage/innobase/include/log0recv.h
index 844d01caa36..542563997cf 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -51,6 +51,11 @@ ATTRIBUTE_COLD MY_ATTRIBUTE((nonnull, warn_unused_result))
 @return whether the page was recovered correctly */
 bool recv_recover_page(fil_space_t* space, buf_page_t* bpage);
 
+/** Read the latest checkpoint information from log file
+and store it in log_sys.next_checkpoint and recv_sys.mlog_checkpoint_lsn
+@return error code or DB_SUCCESS */
+dberr_t recv_recovery_read_max_checkpoint();
+
 /** Start recovering from a redo log checkpoint.
 @param[in]	flush_lsn	FIL_PAGE_FILE_FLUSH_LSN
 of first system tablespace page
@@ -147,7 +152,18 @@ struct recv_dblwr_t
   @param file      tablespace file handle
   @return whether the operation failed */
   bool restore_first_page(
-         ulint space_id, const char *name, os_file_t file);
+         ulint space_id, const char *name, pfs_os_file_t file);
+
+  /** Restore the first page of the given tablespace from
+  doublewrite buffer.
+  1) Find the page which has page_no as 0
+  2) Read first 3 pages from tablespace file
+  3) Compare the space_ids from the pages with page0 which
+  was retrieved from doublewrite buffer
+  @param name tablespace filepath
+  @param file tablespace file handle
+  @return space_id or 0 in case of error */
+  uint32_t find_first_page(const char *name, pfs_os_file_t file);
 
   typedef std::deque<byte*, ut_allocator<byte*> > list;
 
diff --git a/storage/innobase/log/log0recv.cc b/storage/innobase/log/log0recv.cc
index 390fec8f4fb..495f7becfc9 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -4408,6 +4408,46 @@ done:
   return err;
 }
 
+dberr_t recv_recovery_read_max_checkpoint()
+{
+  ut_ad(srv_operation <= SRV_OPERATION_EXPORT_RESTORED ||
+        srv_operation == SRV_OPERATION_RESTORE ||
+        srv_operation == SRV_OPERATION_RESTORE_EXPORT);
+  ut_d(mysql_mutex_lock(&buf_pool.mutex));
+  ut_ad(UT_LIST_GET_LEN(buf_pool.LRU) == 0);
+  ut_ad(UT_LIST_GET_LEN(buf_pool.unzip_LRU) == 0);
+  ut_d(mysql_mutex_unlock(&buf_pool.mutex));
+
+  if (srv_force_recovery >= SRV_FORCE_NO_LOG_REDO)
+  {
+    sql_print_information("InnoDB: innodb_force_recovery=6 skips redo log apply");
+    return DB_SUCCESS;
+  }
+
+  mysql_mutex_lock(&log_sys.mutex);
+  ulint max_cp;
+  dberr_t err= recv_find_max_checkpoint(&max_cp);
+
+  if (err != DB_SUCCESS)
+    recv_sys.recovered_lsn= log_sys.get_lsn();
+  else
+  {
+    byte* buf= log_sys.checkpoint_buf;
+    err= log_sys.log.read(max_cp, {buf, OS_FILE_LOG_BLOCK_SIZE});
+    if (err == DB_SUCCESS)
+    {
+      log_sys.next_checkpoint_no=
+        mach_read_from_8(buf + LOG_CHECKPOINT_NO);
+      log_sys.next_checkpoint_lsn=
+        mach_read_from_8(buf + LOG_CHECKPOINT_LSN);
+      recv_sys.mlog_checkpoint_lsn=
+        mach_read_from_8(buf + LOG_CHECKPOINT_END_LSN);
+    }
+  }
+  mysql_mutex_unlock(&log_sys.mutex);
+  return err;
+}
+
 /** Start recovering from a redo log checkpoint.
 @param[in]	flush_lsn	FIL_PAGE_FILE_FLUSH_LSN
 of first system tablespace page
@@ -4415,12 +4455,8 @@ of first system tablespace page
 dberr_t
 recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 {
-	ulint		max_cp_field;
-	lsn_t		checkpoint_lsn;
 	bool		rescan = false;
-	ib_uint64_t	checkpoint_no;
 	lsn_t		contiguous_lsn;
-	byte*		buf;
 	dberr_t		err = DB_SUCCESS;
 
 	ut_ad(srv_operation <= SRV_OPERATION_EXPORT_RESTORED
@@ -4438,27 +4474,11 @@ recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 		return(DB_SUCCESS);
 	}
 
-	recv_sys.recovery_on = true;
-
 	mysql_mutex_lock(&log_sys.mutex);
-
-	err = recv_find_max_checkpoint(&max_cp_field);
-
-	if (err != DB_SUCCESS) {
-
-		recv_sys.recovered_lsn = log_sys.get_lsn();
-		mysql_mutex_unlock(&log_sys.mutex);
-		return(err);
-	}
-
-	buf = log_sys.checkpoint_buf;
-	if ((err= log_sys.log.read(max_cp_field, {buf, OS_FILE_LOG_BLOCK_SIZE}))) {
-		mysql_mutex_unlock(&log_sys.mutex);
-		return(err);
-	}
-
-	checkpoint_lsn = mach_read_from_8(buf + LOG_CHECKPOINT_LSN);
-	checkpoint_no = mach_read_from_8(buf + LOG_CHECKPOINT_NO);
+	uint64_t checkpoint_no= log_sys.next_checkpoint_no;
+	lsn_t checkpoint_lsn= log_sys.next_checkpoint_lsn;
+	const lsn_t end_lsn= recv_sys.mlog_checkpoint_lsn;
+	recv_sys.recovery_on = true;
 
 	/* Start reading the log from the checkpoint lsn. The variable
 	contiguous_lsn contains an lsn up to which the log is known to
@@ -4467,8 +4487,6 @@ recv_recovery_from_checkpoint_start(lsn_t flush_lsn)
 
 	ut_ad(RECV_SCAN_SIZE <= srv_log_buffer_size);
 
-	const lsn_t	end_lsn = mach_read_from_8(
-		buf + LOG_CHECKPOINT_END_LSN);
 
 	ut_ad(recv_sys.pages.empty());
 	contiguous_lsn = checkpoint_lsn;
@@ -4849,8 +4867,52 @@ byte *recv_dblwr_t::find_page(const page_id_t page_id,
   return result;
 }
 
+uint32_t recv_dblwr_t::find_first_page(const char *name, pfs_os_file_t file)
+{
+  os_offset_t file_size= os_file_get_size(file);
+  if (file_size != (os_offset_t) -1)
+  {
+    for (const page_t *page : pages)
+    {
+      uint32_t space_id= page_get_space_id(page);
+      if (page_get_page_no(page) > 0 || space_id == 0)
+next_page:
+        continue;
+      uint32_t flags= mach_read_from_4(
+        FSP_HEADER_OFFSET + FSP_SPACE_FLAGS + page);
+      page_id_t page_id(space_id, 0);
+      size_t page_size= fil_space_t::physical_size(flags);
+      if (file_size < 4 * page_size)
+        goto next_page;
+      byte *read_page=
+        static_cast<byte*>(aligned_malloc(3 * page_size, page_size));
+      /* Read 3 pages from the file and match the space id
+      with the space id which is stored in
+      doublewrite buffer page. */
+      if (os_file_read(IORequestRead, file, read_page, page_size,
+                       3 * page_size, nullptr) != DB_SUCCESS)
+        goto next_page;
+      for (ulint j= 0; j <= 2; j++)
+      {
+        byte *cur_page= read_page + j * page_size;
+        if (buf_is_zeroes(span<const byte>(cur_page, page_size)))
+          return 0;
+        if (mach_read_from_4(cur_page + FIL_PAGE_OFFSET) != j + 1 ||
+            memcmp(cur_page + FIL_PAGE_SPACE_ID,
+                   page + FIL_PAGE_SPACE_ID, 4) ||
+            buf_page_is_corrupted(false, cur_page, flags))
+          goto next_page;
+      }
+      if (!restore_first_page(space_id, name, file))
+        return space_id;
+      break;
+    }
+  }
+  return 0;
+}
+
 bool recv_dblwr_t::restore_first_page(ulint space_id, const char *name,
-                                      os_file_t file)
+                                      pfs_os_file_t file)
 {
   const page_id_t page_id(space_id, 0);
   const byte* page= find_page(page_id);
@@ -4858,10 +4920,11 @@ bool recv_dblwr_t::restore_first_page(ulint space_id, const char *name,
   {
     /* If the first page of the given user tablespace is not there
     in the doublewrite buffer, then the recovery is going to fail
-    now. Hence this is treated as error. */
-    ib::error()
-            << "Corrupted page " << page_id << " of datafile '"
-            << name <<"' could not be found in the doublewrite buffer.";
+    now. Report error only when doublewrite buffer is not empty */
+    if (pages.size())
+      ib::error() << "Corrupted page " << page_id << " of datafile '"
+                  << name <<"' could not be found in the "
+                  <<"doublewrite buffer.";
     return true;
   }
 
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index 8834caeb17c..537a2904bf2 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -296,9 +296,6 @@ static dberr_t create_log_file(bool create_new_db, lsn_t lsn,
 	log_sys.log.create();
 
 	log_sys.log.open_file(logfile0);
-	if (!fil_system.sys_space->open(create_new_db)) {
-		return DB_ERROR;
-	}
 
 	/* Create a log checkpoint. */
 	mysql_mutex_lock(&log_sys.mutex);
@@ -1261,40 +1258,6 @@ dberr_t srv_start(bool create_new_db)
 
 	/* Check if undo tablespaces and redo log files exist before creating
 	a new system tablespace */
-	if (create_new_db) {
-		err = srv_check_undo_redo_logs_exists();
-		if (err != DB_SUCCESS) {
-			return(srv_init_abort(DB_ERROR));
-		}
-		recv_sys.debug_free();
-	}
-
-	/* Open or create the data files. */
-	ulint	sum_of_new_sizes;
-
-	err = srv_sys_space.open_or_create(
-		false, create_new_db, &sum_of_new_sizes, &flushed_lsn);
-
-	switch (err) {
-	case DB_SUCCESS:
-		break;
-	case DB_CANNOT_OPEN_FILE:
-		ib::error()
-			<< "Could not open or create the system tablespace. If"
-			" you tried to add new data files to the system"
-			" tablespace, and it failed here, you should now"
-			" edit innodb_data_file_path in my.cnf back to what"
-			" it was, and remove the new ibdata files InnoDB"
-			" created in this failed attempt. InnoDB only wrote"
-			" those files full of zeros, but did not yet use"
-			" them in any way. But be careful: do not remove"
-			" old data files which contain your precious data!";
-		/* fall through */
-	default:
-		/* Other errors might come from Datafile::validate_first_page() */
-		return(srv_init_abort(err));
-	}
-
 	srv_log_file_size_requested = srv_log_file_size;
 
 	if (innodb_encrypt_temporary_tables && !log_crypt_init()) {
@@ -1304,18 +1267,17 @@ dberr_t srv_start(bool create_new_db)
 	std::string logfile0;
 	bool create_new_log = create_new_db;
 	if (create_new_db) {
+		err = srv_check_undo_redo_logs_exists();
+		if (err != DB_SUCCESS) {
+			return(srv_init_abort(DB_ERROR));
+		}
+		recv_sys.debug_free();
 		flushed_lsn = log_sys.get_lsn();
 		log_sys.set_flushed_lsn(flushed_lsn);
 
 		err = create_log_file(true, flushed_lsn, logfile0);
 
 		if (err != DB_SUCCESS) {
-			for (Tablespace::const_iterator
-			       i = srv_sys_space.begin();
-			     i != srv_sys_space.end(); i++) {
-				os_file_delete(innodb_data_file_key,
-					       i->filepath());
-			}
 			return(srv_init_abort(err));
 		}
 	} else {
@@ -1330,51 +1292,84 @@ dberr_t srv_start(bool create_new_db)
 		}
 
 		create_new_log = srv_log_file_size == 0;
-		if (create_new_log) {
-			if (flushed_lsn < lsn_t(1000)) {
-				ib::error()
-					<< "Cannot create log file because"
-					" data files are corrupt or the"
-					" database was not shut down cleanly"
-					" after creating the data files.";
-				return srv_init_abort(DB_ERROR);
+		if (!create_new_log) {
+			srv_log_file_found = log_file_found;
+
+			log_sys.log.open_file(get_log_file_path());
+
+			log_sys.log.create();
+
+			if (!log_set_capacity(
+				srv_log_file_size_requested)) {
+				return(srv_init_abort(DB_ERROR));
 			}
 
-			srv_log_file_size = srv_log_file_size_requested;
+			/* Enable checkpoints in the page cleaner. */
+			recv_sys.recovery_on = false;
 
-			err = create_log_file(false, flushed_lsn, logfile0);
-
-			if (err == DB_SUCCESS) {
-				err = create_log_file_rename(flushed_lsn,
-							     logfile0);
-			}
+			err= recv_recovery_read_max_checkpoint();
 
 			if (err != DB_SUCCESS) {
-				return(srv_init_abort(err));
+				return srv_init_abort(err);
 			}
+		}
+	}
 
-			/* Suppress the message about
-			crash recovery. */
-			flushed_lsn = log_sys.get_lsn();
-			goto file_checked;
+	/* Open or create the data files. */
+	ulint	sum_of_new_sizes;
+
+	err = srv_sys_space.open_or_create(
+		false, create_new_db, &sum_of_new_sizes, &flushed_lsn);
+
+	switch (err) {
+	case DB_SUCCESS:
+		break;
+	case DB_CANNOT_OPEN_FILE:
+		sql_print_error("InnoDB: Could not open or create the system"
+			" tablespace. If you tried to add new data files"
+			" to the system tablespace, and it failed here,"
+			" you should now edit innodb_data_file_path"
+			" in my.cnf back to what it was, and remove the"
+			" new ibdata files InnoDB created in this failed"
+			" attempt. InnoDB only wrote those files full of"
+			" zeros, but did not yet use them in any way. But"
+			" be careful: do not remove old data files which"
+			" contain your precious data!");
+		/* fall through */
+	default:
+		/* Other errors might come from Datafile::validate_first_page() */
+		return(srv_init_abort(err));
+	}
+
+	if (!create_new_db && create_new_log) {
+		if (flushed_lsn < lsn_t(1000)) {
+			sql_print_error(
+				"InnoDB: Cannot create log file because"
+				" data files are corrupt or the"
+				" database was not shut down cleanly"
+				" after creating the data files.");
+			return srv_init_abort(DB_ERROR);
 		}
 
-		srv_log_file_found = log_file_found;
+		srv_log_file_size = srv_log_file_size_requested;
 
-		log_sys.log.open_file(get_log_file_path());
-
-		log_sys.log.create();
-
-		if (!log_set_capacity(srv_log_file_size_requested)) {
-			return(srv_init_abort(DB_ERROR));
+		err = create_log_file(false, flushed_lsn, logfile0);
+		if (err == DB_SUCCESS) {
+			err = create_log_file_rename(flushed_lsn, logfile0);
 		}
 
-		/* Enable checkpoints in the page cleaner. */
-		recv_sys.recovery_on = false;
+		if (err != DB_SUCCESS) {
+			return(srv_init_abort(err));
+		}
+
+		/* Suppress the message about
+		crash recovery. */
+		flushed_lsn = log_sys.get_lsn();
+		goto file_checked;
 	}
 
 file_checked:
-	/* Open log file and data files in the systemtablespace: we keep
+	/* Open data files in the systemtablespace: we keep
         them open until database shutdown */
 	ut_d(fil_system.sys_space->recv_size = srv_sys_space_size_debug);
 
diff --git a/storage/spider/spd_conn.cc b/storage/spider/spd_conn.cc
index f908fffc85b..4079b383438 100644
--- a/storage/spider/spd_conn.cc
+++ b/storage/spider/spd_conn.cc
@@ -1909,7 +1909,7 @@ int spider_conn_queue_loop_check(
   loop_check_buf[lex_str.length] = '\0';
   DBUG_PRINT("info", ("spider param name=%s", lex_str.str));
   loop_check = get_variable(&thd->user_vars, &lex_str, FALSE);
-  if (!loop_check || loop_check->type != STRING_RESULT)
+  if (!loop_check || loop_check->type_handler()->result_type() != STRING_RESULT)
   {
     DBUG_PRINT("info", ("spider client is not Spider"));
     lex_str.str = "";
diff --git a/storage/spider/spd_table.cc b/storage/spider/spd_table.cc
index 5dc489daca3..9defa33c8f1 100644
--- a/storage/spider/spd_table.cc
+++ b/storage/spider/spd_table.cc
@@ -4913,7 +4913,7 @@ SPIDER_SHARE *spider_get_share(
     ((char *) lex_str.str)[lex_str.length] = '\0';
     DBUG_PRINT("info",("spider loop check param name=%s", lex_str.str));
     loop_check = get_variable(&thd->user_vars, &lex_str, FALSE);
-    if (loop_check && loop_check->type == STRING_RESULT)
+    if (loop_check && loop_check->type_handler()->result_type() == STRING_RESULT)
     {
       lex_str.length = top_share->path.length + spider_unique_id.length + 1;
       lex_str.str = loop_check_buf + buf_sz - top_share->path.length -