From 9a5f85dcbe12dd9e3228d77834aed8ab53885f20 Mon Sep 17 00:00:00 2001
From: Anel Husakovic <anel@mariadb.org>
Date: Mon, 13 Nov 2023 15:40:13 +0100
Subject: [PATCH 1/9] MDEV-32790: Output result in show create table for
 mysql_json type should be longtext

- We don't test `json` MySQL tables from `std_data` since the error `ER_TABLE_NEEDS_REBUILD
` is invoked. However MDEV-32235 will override this test after merge,
but leave it to show behavior and historical changes.

- Closes PR #2833
Reviewer: <cvicentiu@mariadb.org>
          <serg@mariadb.com>
---
 .../main/mysql_json_table_recreate.result     | 44 +++++++++++++++++++
 .../main/mysql_json_table_recreate.test       | 36 +++++++++++++++
 plugin/type_mysql_json/type.cc                |  2 +-
 3 files changed, 81 insertions(+), 1 deletion(-)

diff --git a/mysql-test/main/mysql_json_table_recreate.result b/mysql-test/main/mysql_json_table_recreate.result
index 207dde9d8ad..b2ab19f4fbf 100644
--- a/mysql-test/main/mysql_json_table_recreate.result
+++ b/mysql-test/main/mysql_json_table_recreate.result
@@ -169,3 +169,47 @@ 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);
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `j` mysql_json /* JSON from MySQL 5.7 */ CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+drop table t1;
+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;
+#
+# 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..7e9895a1f10 100644
--- a/mysql-test/main/mysql_json_table_recreate.test
+++ b/mysql-test/main/mysql_json_table_recreate.test
@@ -88,3 +88,39 @@ 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;
+create table t1(j mysql_json);
+show create table t1;
+drop table t1;
+# `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 # End of 10.5 tests
+--echo #
diff --git a/plugin/type_mysql_json/type.cc b/plugin/type_mysql_json/type.cc
index c5168969b67..c915e0312bb 100644
--- a/plugin/type_mysql_json/type.cc
+++ b/plugin/type_mysql_json/type.cc
@@ -62,7 +62,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; }
 };

From 22f3ebe4bf11acb7b5e854bc4e6b3439af6982f1 Mon Sep 17 00:00:00 2001
From: Anel Husakovic <anel@mariadb.org>
Date: Thu, 11 Jan 2024 15:15:46 +0100
Subject: [PATCH 2/9] MDEV-32235: mysql_json cannot be used on newly created
 table

Closes PR #2839
Reviewer: cvicentiu@mariadb.org
---
 .../main/mysql_json_table_recreate.result      | 18 ++++++++++++------
 mysql-test/main/mysql_json_table_recreate.test | 17 +++++++++++++++--
 sql/sql_table.cc                               | 13 +++++++++++++
 3 files changed, 40 insertions(+), 8 deletions(-)

diff --git a/mysql-test/main/mysql_json_table_recreate.result b/mysql-test/main/mysql_json_table_recreate.result
index b2ab19f4fbf..a0c2483d3d9 100644
--- a/mysql-test/main/mysql_json_table_recreate.result
+++ b/mysql-test/main/mysql_json_table_recreate.result
@@ -181,12 +181,7 @@ t1	CREATE TABLE `t1` (
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
 drop table t1;
 create table t1(j mysql_json);
-show create table t1;
-Table	Create Table
-t1	CREATE TABLE `t1` (
-  `j` mysql_json /* JSON from MySQL 5.7 */ CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL
-) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
-drop table t1;
+ERROR HY000: Cannot create table `test`.`t1`: Run mariadb-upgrade, to upgrade table with mysql_json type.
 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;
@@ -211,5 +206,16 @@ testjson	CREATE TABLE `testjson` (
 ) 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: Cannot create table `test`.`t`: Run mariadb-upgrade, to upgrade table with mysql_json type.
+CREATE TABLE IF NOT EXISTS t(j mysql_json);
+ERROR HY000: Cannot create table `test`.`t`: Run mariadb-upgrade, to upgrade table with mysql_json type.
+CREATE OR REPLACE TABLE t(j mysql_json);
+ERROR HY000: Cannot create table `test`.`t`: Run mariadb-upgrade, to upgrade table with mysql_json type.
+CREATE TEMPORARY TABLE t(j mysql_json);
+ERROR HY000: Cannot create table `test`.`t`: Run mariadb-upgrade, to upgrade table with mysql_json type.
+#
 # 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 7e9895a1f10..bf4cac0f909 100644
--- a/mysql-test/main/mysql_json_table_recreate.test
+++ b/mysql-test/main/mysql_json_table_recreate.test
@@ -97,9 +97,8 @@ drop table mysql_json_test_big;
 create table t1(j json);
 show create table t1;
 drop table t1;
+--error ER_CANT_CREATE_TABLE
 create table t1(j mysql_json);
-show create table t1;
-drop table t1;
 # `json` type should not have character set and collation other than utf8mb4_bin
 --error ER_PARSE_ERROR
 create table `testjson` (
@@ -121,6 +120,20 @@ create table `testjson` (
 show create table testjson;
 drop table testjson;
 
+
+--echo #
+--echo # MDEV-32235: mysql_json cannot be used on newly created table
+--echo #
+
+--error ER_CANT_CREATE_TABLE
+CREATE TABLE t(j mysql_json);
+--error ER_CANT_CREATE_TABLE
+CREATE TABLE IF NOT EXISTS t(j mysql_json);
+--error ER_CANT_CREATE_TABLE
+CREATE OR REPLACE TABLE t(j mysql_json);
+--error ER_CANT_CREATE_TABLE
+CREATE TEMPORARY TABLE t(j mysql_json);
+
 --echo #
 --echo # End of 10.5 tests
 --echo #
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index d68bbd1ac4f..8088d3e0d70 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -3657,6 +3657,19 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     if (sql_field->vcol_info)
       sql_field->flags&= ~NOT_NULL_FLAG;
 
+    if (sql_field->real_field_type() == MYSQL_TYPE_BLOB &&
+        thd->lex->sql_command == SQLCOM_CREATE_TABLE)
+    {
+      if (!strcmp(sql_field->type_handler()->name().ptr(), "MYSQL_JSON"))
+      {
+        my_printf_error(ER_CANT_CREATE_TABLE,
+                        "Cannot create table %`s.%`s: "
+                        "Run mariadb-upgrade, "
+                        "to upgrade table with mysql_json type.", MYF(0),
+                        alter_info->db.str, alter_info->table_name.str);
+        DBUG_RETURN(TRUE);
+      }
+    }
     /*
       Initialize length from its original value (number of characters),
       which was set in the parser. This is necessary if we're

From 8b5c1d5afa33521d7a5024b58a7b8197e60bfd33 Mon Sep 17 00:00:00 2001
From: Anel Husakovic <anel@mariadb.org>
Date: Fri, 12 Jan 2024 15:20:25 +0100
Subject: [PATCH 3/9] Revert "MDEV-32235: mysql_json cannot be used on newly
 created table"

This reverts commit 22f3ebe4bf11acb7b5e854bc4e6b3439af6982f1.
---
 .../main/mysql_json_table_recreate.result      | 18 ++++++------------
 mysql-test/main/mysql_json_table_recreate.test | 17 ++---------------
 sql/sql_table.cc                               | 13 -------------
 3 files changed, 8 insertions(+), 40 deletions(-)

diff --git a/mysql-test/main/mysql_json_table_recreate.result b/mysql-test/main/mysql_json_table_recreate.result
index a0c2483d3d9..b2ab19f4fbf 100644
--- a/mysql-test/main/mysql_json_table_recreate.result
+++ b/mysql-test/main/mysql_json_table_recreate.result
@@ -181,7 +181,12 @@ t1	CREATE TABLE `t1` (
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
 drop table t1;
 create table t1(j mysql_json);
-ERROR HY000: Cannot create table `test`.`t1`: Run mariadb-upgrade, to upgrade table with mysql_json type.
+show create table t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `j` mysql_json /* JSON from MySQL 5.7 */ CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL
+) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
+drop table t1;
 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;
@@ -206,16 +211,5 @@ testjson	CREATE TABLE `testjson` (
 ) 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: Cannot create table `test`.`t`: Run mariadb-upgrade, to upgrade table with mysql_json type.
-CREATE TABLE IF NOT EXISTS t(j mysql_json);
-ERROR HY000: Cannot create table `test`.`t`: Run mariadb-upgrade, to upgrade table with mysql_json type.
-CREATE OR REPLACE TABLE t(j mysql_json);
-ERROR HY000: Cannot create table `test`.`t`: Run mariadb-upgrade, to upgrade table with mysql_json type.
-CREATE TEMPORARY TABLE t(j mysql_json);
-ERROR HY000: Cannot create table `test`.`t`: Run mariadb-upgrade, to upgrade table with mysql_json type.
-#
 # 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 bf4cac0f909..7e9895a1f10 100644
--- a/mysql-test/main/mysql_json_table_recreate.test
+++ b/mysql-test/main/mysql_json_table_recreate.test
@@ -97,8 +97,9 @@ drop table mysql_json_test_big;
 create table t1(j json);
 show create table t1;
 drop table t1;
---error ER_CANT_CREATE_TABLE
 create table t1(j mysql_json);
+show create table t1;
+drop table t1;
 # `json` type should not have character set and collation other than utf8mb4_bin
 --error ER_PARSE_ERROR
 create table `testjson` (
@@ -120,20 +121,6 @@ create table `testjson` (
 show create table testjson;
 drop table testjson;
 
-
---echo #
---echo # MDEV-32235: mysql_json cannot be used on newly created table
---echo #
-
---error ER_CANT_CREATE_TABLE
-CREATE TABLE t(j mysql_json);
---error ER_CANT_CREATE_TABLE
-CREATE TABLE IF NOT EXISTS t(j mysql_json);
---error ER_CANT_CREATE_TABLE
-CREATE OR REPLACE TABLE t(j mysql_json);
---error ER_CANT_CREATE_TABLE
-CREATE TEMPORARY TABLE t(j mysql_json);
-
 --echo #
 --echo # End of 10.5 tests
 --echo #
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 8088d3e0d70..d68bbd1ac4f 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -3657,19 +3657,6 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     if (sql_field->vcol_info)
       sql_field->flags&= ~NOT_NULL_FLAG;
 
-    if (sql_field->real_field_type() == MYSQL_TYPE_BLOB &&
-        thd->lex->sql_command == SQLCOM_CREATE_TABLE)
-    {
-      if (!strcmp(sql_field->type_handler()->name().ptr(), "MYSQL_JSON"))
-      {
-        my_printf_error(ER_CANT_CREATE_TABLE,
-                        "Cannot create table %`s.%`s: "
-                        "Run mariadb-upgrade, "
-                        "to upgrade table with mysql_json type.", MYF(0),
-                        alter_info->db.str, alter_info->table_name.str);
-        DBUG_RETURN(TRUE);
-      }
-    }
     /*
       Initialize length from its original value (number of characters),
       which was set in the parser. This is necessary if we're

From 8a763c014ede4adad9c84852269b5af61845c0d9 Mon Sep 17 00:00:00 2001
From: Anel Husakovic <anel@mariadb.org>
Date: Fri, 12 Jan 2024 15:39:38 +0100
Subject: [PATCH 4/9] MDEV-32235: mysql_json cannot be used on newly created
 table

- Closes PR #2839
- Usage of `Column_definition_fix_attributes()` suggested by Alexandar
  Barkov - thanks bar, that is better than hook in server code
  (reverted 22f3ebe4bf11)
  - This method is called after parsing the data type:
    * in `CREATE/ALTER TABLE`
    * in SP: return data type, parameter data type, variable data type
  - We want to disallow all these use cases of MYSQL_JSON.

- Reviewer: bar@mariadb.com
            cvicentiu@mariadb.org
---
 .../main/mysql_json_table_recreate.result     | 38 ++++++++++++++---
 .../main/mysql_json_table_recreate.test       | 41 ++++++++++++++++++-
 plugin/type_mysql_json/type.cc                |  5 +++
 3 files changed, 76 insertions(+), 8 deletions(-)

diff --git a/mysql-test/main/mysql_json_table_recreate.result b/mysql-test/main/mysql_json_table_recreate.result
index b2ab19f4fbf..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;
@@ -181,12 +187,7 @@ t1	CREATE TABLE `t1` (
 ) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
 drop table t1;
 create table t1(j mysql_json);
-show create table t1;
-Table	Create Table
-t1	CREATE TABLE `t1` (
-  `j` mysql_json /* JSON from MySQL 5.7 */ CHARACTER SET utf8mb4 COLLATE utf8mb4_bin DEFAULT NULL
-) ENGINE=MyISAM DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci
-drop table t1;
+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;
@@ -211,5 +212,30 @@ testjson	CREATE TABLE `testjson` (
 ) 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 7e9895a1f10..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;
 
@@ -97,9 +104,8 @@ drop table mysql_json_test_big;
 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);
-show create table t1;
-drop table t1;
 # `json` type should not have character set and collation other than utf8mb4_bin
 --error ER_PARSE_ERROR
 create table `testjson` (
@@ -121,6 +127,37 @@ create table `testjson` (
 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/plugin/type_mysql_json/type.cc b/plugin/type_mysql_json/type.cc
index c915e0312bb..f84f76381a0 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;

From 5b0a4159ef1717545ccd37c08db0cba0eef00e5a Mon Sep 17 00:00:00 2001
From: Kristian Nielsen <knielsen@knielsen-hq.org>
Date: Fri, 5 Jan 2024 13:44:49 +0100
Subject: [PATCH 5/9] Fix test failures on s390x in test following
 main.column_compression_rpl

The problem is the test is skipped after sourcing include/master-slave.inc.
This leaves the slave threads running after the test is skipped, causing a
following test to fail during rpl setup.

Also rename have_normal_bzip.inc to the more appropriate _zlib.

Signed-off-by: Kristian Nielsen <knielsen@knielsen-hq.org>
---
 .../include/{have_normal_bzip.inc => have_normal_zlib.inc}    | 4 ++--
 mysql-test/main/column_compression.test                       | 2 +-
 mysql-test/main/column_compression_rpl.test                   | 2 +-
 mysql-test/main/func_compress.test                            | 2 +-
 mysql-test/main/mysqlbinlog_row_compressed.test               | 2 +-
 mysql-test/main/mysqlbinlog_stmt_compressed.test              | 2 +-
 mysql-test/suite/compat/oracle/t/column_compression.test      | 2 +-
 mysql-test/suite/innodb/t/row_size_error_log_warnings_3.test  | 2 +-
 8 files changed, 9 insertions(+), 9 deletions(-)
 rename mysql-test/include/{have_normal_bzip.inc => have_normal_zlib.inc} (67%)

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 c628e8f9cca..de59cf5c7c8 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/mysqlbinlog_row_compressed.test b/mysql-test/main/mysqlbinlog_row_compressed.test
index f2806861830..f493c4db2cd 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 400f21f5ab9..1f1591e7a00 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/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/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.");
 

From 48e4962c44fa5cbdafb6d1ed9564f308a57eeedc Mon Sep 17 00:00:00 2001
From: Oleg Smirnov <olernov@gmail.com>
Date: Thu, 6 Jul 2023 11:55:40 +0700
Subject: [PATCH 6/9] MDEV-29298 INSERT ... SELECT Does not produce an
 optimizer trace

Add INSERT ... SELECT to the list of commands that can be traced

Approved by Sergei Petrunia (sergey@mariadb.com)
---
 mysql-test/main/opt_trace.result | 12 ++++++++++++
 mysql-test/main/opt_trace.test   | 14 ++++++++++++++
 sql/opt_trace.cc                 |  3 ++-
 3 files changed, 28 insertions(+), 1 deletion(-)

diff --git a/mysql-test/main/opt_trace.result b/mysql-test/main/opt_trace.result
index 5766b8ead42..66f081c6590 100644
--- a/mysql-test/main/opt_trace.result
+++ b/mysql-test/main/opt_trace.result
@@ -9231,5 +9231,17 @@ 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
 set optimizer_trace='enabled=off';
diff --git a/mysql-test/main/opt_trace.test b/mysql-test/main/opt_trace.test
index 43539fc764b..e7851fc62c5 100644
--- a/mysql-test/main/opt_trace.test
+++ b/mysql-test/main/opt_trace.test
@@ -861,5 +861,19 @@ 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 # End of 10.5 tests
 set optimizer_trace='enabled=off';
diff --git a/sql/opt_trace.cc b/sql/opt_trace.cc
index ddec6d5ed2d..4bd54549532 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,

From ee30491e508e6c8c19c899f713662ec2e5042aaa Mon Sep 17 00:00:00 2001
From: Tuukka Pasanen <tuukka.pasanen@ilmi.fi>
Date: Wed, 15 Nov 2023 11:09:26 +0200
Subject: [PATCH 7/9] MDEV-32111: Debian Sid/Trixie will not have libncurses 5
 anymore

Upstream Debian Sid which will become Debian Trixie (13)
have dropped NCurses version 5 and changed dev package name just
libncurses-dev
---
 debian/control | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/debian/control b/debian/control
index 2bbd71d2ac9..eb6a5cba16c 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,

From caad34df549b91b654c69bf3b5644277be783d31 Mon Sep 17 00:00:00 2001
From: Thirunarayanan Balathandayuthapani <thiru@mariadb.com>
Date: Mon, 15 Jan 2024 14:08:27 +0530
Subject: [PATCH 8/9] MDEV-32968  InnoDB fails to restore tablespace first page
 from doublewrite buffer when page is empty

- InnoDB fails to find the space id from the page0 of
the tablespace. In that case, InnoDB can use
doublewrite buffer to recover the page0 and write
into the file.

- buf_dblwr_t::init_or_load_pages(): Loads only the pages
which are valid.(page lsn >= checkpoint). To do that,
InnoDB has to open the redo log before system
tablespace, read the latest checkpoint information.

recv_dblwr_t::find_first_page():
1) Iterate the doublewrite buffer pages and find the 0th page
2) Read the tablespace flags, space id from the 0th page.
3) Read the 1st, 2nd and 3rd page from tablespace file and
compare the space id with the space id which is stored
in doublewrite buffer.
4) If it matches then we can write into the file.
5) Return space which matches the pages from the file.

SysTablespace::read_lsn_and_check_flags(): Remove the
retry logic for validating the first page. After
restoring the first page from doublewrite buffer,
assign tablespace flags by reading the first page.

recv_recovery_read_max_checkpoint(): Reads the maximum
checkpoint information from log file

recv_recovery_from_checkpoint_start(): Avoid reading
the checkpoint header information from log file

Datafile::validate_first_page(): Throw error in case
of first page validation fails.
---
 mysql-test/suite/innodb/r/doublewrite.result  |  25 +++-
 .../suite/innodb/r/log_file_name.result       |   7 +-
 mysql-test/suite/innodb/t/doublewrite.test    |  38 ++++-
 .../suite/innodb/t/doublewrite_debug.test     |   1 +
 mysql-test/suite/innodb/t/log_file_name.test  |   9 +-
 storage/innobase/buf/buf0dblwr.cc             |  11 +-
 storage/innobase/fsp/fsp0file.cc              |  21 ++-
 storage/innobase/fsp/fsp0sysspace.cc          |  17 +--
 storage/innobase/include/buf0dblwr.h          |   3 +-
 storage/innobase/include/log0recv.h           |  18 ++-
 storage/innobase/log/log0recv.cc              | 125 ++++++++++++----
 storage/innobase/srv/srv0start.cc             | 141 +++++++++---------
 12 files changed, 278 insertions(+), 138 deletions(-)

diff --git a/mysql-test/suite/innodb/r/doublewrite.result b/mysql-test/suite/innodb/r/doublewrite.result
index 7763aa99322..ceb01ac4807 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;
+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 2 /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 42b988ed3ca..3d48a24d217 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;
@@ -89,7 +90,7 @@ SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND 1 /\[Note\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd/ in mysqld.1.err
+FOUND 1 /\[ERROR\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd/ in mysqld.1.err
 FOUND 1 /\[ERROR\] InnoDB: Datafile .*u1.*\. Cannot determine the space ID from the first 64 pages/ in mysqld.1.err
 NOT FOUND /\[Note\] InnoDB: Cannot read first page of .*u2.ibd/ in mysqld.1.err
 # Fault 7: Missing or wrong data file and innodb_force_recovery
@@ -98,11 +99,11 @@ SELECT * FROM INFORMATION_SCHEMA.ENGINES
 WHERE engine = 'innodb'
 AND support IN ('YES', 'DEFAULT', 'ENABLED');
 ENGINE	SUPPORT	COMMENT	TRANSACTIONS	XA	SAVEPOINTS
-FOUND 1 /\[Note\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd/ in mysqld.1.err
+FOUND 1 /\[ERROR\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd/ in mysqld.1.err
 FOUND 1 /InnoDB: At LSN: \d+: unable to open file .*u[1-5].ibd for tablespace/ in mysqld.1.err
 FOUND 1 /\[ERROR\] InnoDB: Cannot replay rename of tablespace \d+ from '.*u4.ibd' to '.*u6.ibd' because the target file exists/ in mysqld.1.err
 # restart: --innodb-force-recovery=1
-FOUND 1 /\[Note\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd/ in mysqld.1.err
+FOUND 1 /\[ERROR\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd/ in mysqld.1.err
 FOUND 1 /InnoDB: At LSN: \d+: unable to open file .*u[1-5].ibd for tablespace/ in mysqld.1.err
 FOUND 1 /\[Warning\] InnoDB: Tablespace \d+ was not found at .*u[1-5].ibd, and innodb_force_recovery was set. All redo log for this tablespace will be ignored!/ in mysqld.1.err
 # restart
diff --git a/mysql-test/suite/innodb/t/doublewrite.test b/mysql-test/suite/innodb/t/doublewrite.test
index 89e6577b434..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;
+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 1bff8b4e07f..3d96a74e698 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 11eca0a7a12..494d256cd11 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;
 
@@ -172,6 +174,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
 
@@ -215,7 +218,7 @@ EOF
 --source include/start_mysqld.inc
 eval $check_no_innodb;
 
-let SEARCH_PATTERN= \[Note\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd;
+let SEARCH_PATTERN= \[ERROR\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd;
 --source include/search_pattern_in_file.inc
 
 let SEARCH_PATTERN= \[ERROR\] InnoDB: Datafile .*u1.*\. Cannot determine the space ID from the first 64 pages;
@@ -240,7 +243,7 @@ let SEARCH_PATTERN= \[Note\] InnoDB: Cannot read first page of .*u2.ibd;
 --source include/start_mysqld.inc
 eval $check_no_innodb;
 
-let SEARCH_PATTERN= \[Note\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd;
+let SEARCH_PATTERN= \[ERROR\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd;
 --source include/search_pattern_in_file.inc
 
 let SEARCH_PATTERN= InnoDB: At LSN: \d+: unable to open file .*u[1-5].ibd for tablespace;
@@ -253,7 +256,7 @@ let SEARCH_PATTERN= \[ERROR\] InnoDB: Cannot replay rename of tablespace \d+ fro
 
 --source include/restart_mysqld.inc
 
-let SEARCH_PATTERN= \[Note\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd;
+let SEARCH_PATTERN= \[ERROR\] InnoDB: Header page consists of zero bytes in datafile: .*u1.ibd;
 --source include/search_pattern_in_file.inc
 
 let SEARCH_PATTERN= InnoDB: At LSN: \d+: unable to open file .*u[1-5].ibd for tablespace;
diff --git a/storage/innobase/buf/buf0dblwr.cc b/storage/innobase/buf/buf0dblwr.cc
index 57ec3553b8d..d5c954dff89 100644
--- a/storage/innobase/buf/buf0dblwr.cc
+++ b/storage/innobase/buf/buf0dblwr.cc
@@ -324,11 +324,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 f0ead3b80a3..653c848953a 100644
--- a/storage/innobase/fsp/fsp0file.cc
+++ b/storage/innobase/fsp/fsp0file.cc
@@ -475,9 +475,16 @@ Datafile::validate_for_recovery()
 
 		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.";
+
+			m_space_id = recv_sys.dblwr.find_first_page(
+				m_filepath, m_handle);
+
+			if (m_space_id) goto free_first_page;
+
+			sql_print_error(
+				"InnoDB: Datafile '%s' is corrupted."
+				" Cannot determine the space ID from"
+				" the first 64 pages.", m_filepath);
 			return(err);
 		}
 
@@ -485,7 +492,7 @@ Datafile::validate_for_recovery()
 			m_space_id, m_filepath, m_handle)) {
 			return(DB_CORRUPTION);
 		}
-
+free_first_page:
 		/* Free the previously read first page and then re-validate. */
 		free_first_page();
 		err = validate_first_page(0);
@@ -531,9 +538,9 @@ Datafile::validate_first_page(lsn_t* flush_lsn)
 
 	if (error_txt != NULL) {
 err_exit:
-		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;
 		free_first_page();
 		return(DB_CORRUPTION);
diff --git a/storage/innobase/fsp/fsp0sysspace.cc b/storage/innobase/fsp/fsp0sysspace.cc
index a7bb417c502..eb6e8d1166e 100644
--- a/storage/innobase/fsp/fsp0sysspace.cc
+++ b/storage/innobase/fsp/fsp0sysspace.cc
@@ -574,7 +574,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);
@@ -588,20 +588,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 f82baeec89a..99a41d069ef 100644
--- a/storage/innobase/include/buf0dblwr.h
+++ b/storage/innobase/include/buf0dblwr.h
@@ -103,7 +103,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 34211392ba3..aca9f994acb 100644
--- a/storage/innobase/include/log0recv.h
+++ b/storage/innobase/include/log0recv.h
@@ -49,6 +49,11 @@ recv_find_max_checkpoint(ulint* max_field)
 ATTRIBUTE_COLD void recv_recover_page(fil_space_t* space, buf_page_t* bpage)
 	MY_ATTRIBUTE((nonnull));
 
+/** 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
@@ -141,7 +146,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 05120871b0a..52649419ec7 100644
--- a/storage/innobase/log/log0recv.cc
+++ b/storage/innobase/log/log0recv.cc
@@ -3415,6 +3415,46 @@ recv_init_crash_recovery_spaces(bool rescan, bool& missing_tablespace)
 	return DB_SUCCESS;
 }
 
+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
@@ -3422,12 +3462,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
@@ -3445,27 +3481,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
@@ -3474,8 +3494,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;
@@ -3845,8 +3863,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) != 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);
@@ -3854,10 +3916,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 56e41616d1d..0760b8ca5fd 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -297,9 +297,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);
@@ -1274,40 +1271,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()) {
@@ -1317,18 +1280,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 {
@@ -1343,51 +1305,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);
 

From fa3171df08dc9b274dfe7fa03205b66554b621ac Mon Sep 17 00:00:00 2001
From: Alexander Barkov <bar@mariadb.com>
Date: Wed, 27 Dec 2023 18:57:49 +0400
Subject: [PATCH 9/9] MDEV-27666 User variable not parsed as geometry variable
 in geometry function

Adding GEOMETRY type user variables.
---
 mysql-test/main/gis.result                    | 107 ++++++++++++++++++
 mysql-test/main/gis.test                      |  63 +++++++++++
 .../binlog/r/binlog_gis_user_var_stm.result   |  12 ++
 .../binlog/t/binlog_gis_user_var_stm.test     |  15 +++
 .../suite/rpl/r/rpl_gis_user_var.result       |  21 ++++
 mysql-test/suite/rpl/t/rpl_gis_user_var.test  |  18 +++
 plugin/user_variables/user_variables.cc       |   6 +-
 sql/item_func.cc                              |  80 ++++++-------
 sql/item_func.h                               |   7 +-
 sql/log.cc                                    |  10 +-
 sql/log_event.cc                              |  54 ++++++---
 sql/log_event.h                               |  21 ++--
 sql/log_event_client.cc                       |   7 +-
 sql/log_event_data_type.h                     |  74 ++++++++++++
 sql/log_event_server.cc                       |  70 ++++++++----
 sql/sql_binlog.cc                             |   5 +-
 sql/sql_class.h                               |   7 +-
 sql/sql_repl.cc                               |   2 +-
 sql/sql_type.cc                               |  28 +++++
 sql/sql_type.h                                |  11 ++
 sql/sql_type_geom.h                           |   7 ++
 storage/spider/spd_conn.cc                    |   2 +-
 storage/spider/spd_table.cc                   |   2 +-
 23 files changed, 510 insertions(+), 119 deletions(-)
 create mode 100644 mysql-test/suite/binlog/r/binlog_gis_user_var_stm.result
 create mode 100644 mysql-test/suite/binlog/t/binlog_gis_user_var_stm.test
 create mode 100644 mysql-test/suite/rpl/r/rpl_gis_user_var.result
 create mode 100644 mysql-test/suite/rpl/t/rpl_gis_user_var.test
 create mode 100644 sql/log_event_data_type.h

diff --git a/mysql-test/main/gis.result b/mysql-test/main/gis.result
index f8816090b04..258d0c2f050 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/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/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/user_variables/user_variables.cc b/plugin/user_variables/user_variables.cc
index f820e4ad890..c76337cb084 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/item_func.cc b/sql/item_func.cc
index 5202d395580..b2f2d6550f1 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -4658,7 +4658,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:
@@ -4670,7 +4669,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))
     {
@@ -4746,9 +4745,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);
@@ -4873,9 +4875,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));
@@ -4886,7 +4888,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)
     {
@@ -4915,20 +4917,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;
@@ -4941,9 +4941,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
@@ -4957,9 +4956,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;
@@ -4975,7 +4973,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:
@@ -5000,7 +4998,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:
@@ -5029,12 +5027,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());
@@ -5061,7 +5059,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;
@@ -5200,33 +5198,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:
@@ -5618,9 +5620,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*/
@@ -5663,9 +5664,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);
@@ -5684,6 +5685,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);
@@ -5756,7 +5759,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.
@@ -5772,15 +5775,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 1b8aefe005c..b205d457f1d 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -3066,8 +3066,8 @@ public:
   String *str_result(String *str);
   my_decimal *val_decimal_result(my_decimal *);
   bool is_null_result();
-  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);
   void make_send_field(THD *thd, Send_field *tmp_field);
   bool check(bool use_result_field);
@@ -3825,7 +3825,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 3d3ac9b2b35..75199000b85 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -6794,18 +6794,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 d367575f9e8..1a3161d49a4 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -2880,6 +2880,41 @@ XA_prepare_log_event(const char* 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 char* buf, uint event_len,
                    const Format_description_log_event* description_event)
@@ -2917,11 +2952,8 @@ User_var_log_event(const char* 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;  
   }
@@ -2936,8 +2968,8 @@ User_var_log_event(const char* 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);
 
@@ -2950,20 +2982,14 @@ User_var_log_event(const char* 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) - 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 28ac4cc78df..afe4862f152 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;
@@ -3269,32 +3271,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 char *val_arg, size_t val_len_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 c1c5ddf377c..c20b41e9632 100644
--- a/sql/log_event_client.cc
+++ b/sql/log_event_client.cc
@@ -2435,7 +2435,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)];
@@ -2447,8 +2447,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;
@@ -2503,7 +2502,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 13a981d4381..2a22021cc95 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -4170,11 +4170,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("/*") ||
+      buf->append(data_type_name.str, data_type_name.length) ||
+      buf->append("*/")));
 }
 
 void User_var_log_event::pack_info(Protocol* protocol)
@@ -4184,14 +4189,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"))
       return;
     protocol->store(buf.ptr(), buf.length(), &my_charset_bin);
   }
   else
   {
-    switch (type) {
+    switch (m_type) {
     case REAL_RESULT:
     {
       double real_val;
@@ -4200,7 +4206,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;
@@ -4213,10 +4220,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;
@@ -4229,7 +4237,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(buf2))
         return;
       protocol->store(buf.ptr(), buf.length(), &my_charset_bin);
@@ -4242,7 +4251,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("???"))
           return;
@@ -4251,7 +4260,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->csname) ||
             buf.append(" "))
@@ -4299,10 +4309,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;
@@ -4332,15 +4342,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();
 }
 
@@ -4364,7 +4387,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),
@@ -4383,7 +4406,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)
       {
@@ -4447,13 +4470,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/sql_binlog.cc b/sql/sql_binlog.cc
index 4dd5f16f351..05ae01b542f 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 06abab99be7..a8ef8648210 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -275,9 +275,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;
 
 /*
@@ -6808,7 +6807,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:
@@ -6817,8 +6816,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 1a83b96b61b..270f051b9dd 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 13e61a40be2..77931b6cf16 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -2248,6 +2248,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 edce7f45abe..c8c92e56253 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
@@ -3652,6 +3654,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);
@@ -3936,6 +3941,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 ff00beea598..858c2da2318 100644
--- a/sql/sql_type_geom.h
+++ b/sql/sql_type_geom.h
@@ -81,6 +81,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/spider/spd_conn.cc b/storage/spider/spd_conn.cc
index 9897e495c83..befd9bfd2db 100644
--- a/storage/spider/spd_conn.cc
+++ b/storage/spider/spd_conn.cc
@@ -1886,7 +1886,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 4ff54b591ba..717313df2af 100644
--- a/storage/spider/spd_table.cc
+++ b/storage/spider/spd_table.cc
@@ -4742,7 +4742,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 -