MDEV-31297 Create table as select on system versioned tables do not

	   work consistently on replication

Row-based replication does not execute CREATE .. SELECT but instead
CREATE TABLE. CREATE .. SELECT creates implict system fields on
unusual place: in-between declared fields and select fields. That was
done because select_field_pos logic requires select fields go last in
create_list.

So, CREATE .. SELECT on master and CREATE TABLE on slave create system
fields on different positions and replication gets field mismatch.

To fix this we've changed CREATE .. SELECT to create implicit system
fields on usual place in the end and updated select_field_pos for
handling this case.

mysql-test/suite/versioning/r/rpl_row.result

@@ -49,4 +49,28 @@ x	y
 1	1
 connection master;
 drop table t1;
+#
+# MDEV-31297 Create table as select on system versioned tables do not work consistently on replication
+#
+connection master;
+create table t engine=innodb with system versioning as select 1 as i;
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `i` int(1) NOT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci WITH SYSTEM VERSIONING
+select * from t;
+i
+1
+connection slave;
+show create table t;
+Table	Create Table
+t	CREATE TABLE `t` (
+  `i` int(1) NOT NULL
+) ENGINE=InnoDB DEFAULT CHARSET=latin1 COLLATE=latin1_swedish_ci WITH SYSTEM VERSIONING
+select * from t;
+i
+1
+connection master;
+drop table t;
 include/rpl_end.inc

mysql-test/suite/versioning/t/rpl_row.test

@@ -56,4 +56,19 @@ drop table t1;
 --remove_files_wildcard $TMP *.txt
 --remove_files_wildcard $TMP *.sql
 
+--echo #
+--echo # MDEV-31297 Create table as select on system versioned tables do not work consistently on replication
+--echo #
+--connection master
+create table t engine=innodb with system versioning as select 1 as i;
+show create table t;
+select * from t;
+
+--sync_slave_with_master
+show create table t;
+select * from t;
+
+--connection master
+drop table t;
+
 --source include/rpl_end.inc

sql/handler.cc

@@ -8017,6 +8017,21 @@ bool Table_scope_and_contents_source_st::vers_fix_system_fields(
 }
 
 
+int get_select_field_pos(Alter_info *alter_info, int select_field_count,
+                         bool versioned)
+{
+  int select_field_pos= alter_info->create_list.elements - select_field_count;
+  if (select_field_count && versioned &&
+      /*
+        ALTER_PARSER_ADD_COLUMN indicates system fields was created implicitly,
+        select_field_count guarantees it's not ALTER TABLE
+      */
+      alter_info->flags & ALTER_PARSER_ADD_COLUMN)
+    select_field_pos-= 2;
+  return select_field_pos;
+}
+
+
 bool Table_scope_and_contents_source_st::vers_check_system_fields(
         THD *thd, Alter_info *alter_info, const Lex_table_name &table_name,
         const Lex_table_name &db, int select_count)
@@ -8030,6 +8045,8 @@ bool Table_scope_and_contents_source_st::vers_check_system_fields(
   {
     uint fieldnr= 0;
     List_iterator<Create_field> field_it(alter_info->create_list);
+    uint select_field_pos= (uint) get_select_field_pos(alter_info, select_count,
+                                                       true);
     while (Create_field *f= field_it++)
     {
       /*
@@ -8040,7 +8057,7 @@ bool Table_scope_and_contents_source_st::vers_check_system_fields(
          SELECT go last there.
        */
       bool is_dup= false;
-      if (fieldnr >= alter_info->create_list.elements - select_count)
+      if (fieldnr >= select_field_pos && f->invisible < INVISIBLE_SYSTEM)
       {
         List_iterator<Create_field> dup_it(alter_info->create_list);
         for (Create_field *dup= dup_it++; !is_dup && dup != f; dup= dup_it++)

sql/handler.h

@@ -5307,4 +5307,7 @@ int del_global_index_stat(THD *thd, TABLE* table, KEY* key_info);
 int del_global_table_stat(THD *thd, const  LEX_CSTRING *db, const LEX_CSTRING *table);
 uint ha_count_rw_all(THD *thd, Ha_trx_info **ptr_ha_info);
 bool non_existing_table_error(int error);
+
+int get_select_field_pos(Alter_info *alter_info, int select_field_count,
+                         bool versioned);
 #endif /* HANDLER_INCLUDED */

sql/sql_insert.cc

@@ -4524,9 +4524,6 @@ TABLE *select_create::create_table_from_items(THD *thd, List<Item> *items,
   if (!(thd->variables.option_bits & OPTION_EXPLICIT_DEF_TIMESTAMP))
     promote_first_timestamp_column(&alter_info->create_list);
 
-  if (create_info->fix_create_fields(thd, alter_info, *create_table))
-    DBUG_RETURN(NULL);
-
   while ((item=it++))
   {
     Field *tmp_field= item->create_field_for_create_select(thd->mem_root,
@@ -4564,6 +4561,9 @@ TABLE *select_create::create_table_from_items(THD *thd, List<Item> *items,
     alter_info->create_list.push_back(cr_field, thd->mem_root);
   }
 
+  if (create_info->fix_create_fields(thd, alter_info, *create_table))
+    DBUG_RETURN(NULL);
+
   /*
     Item*::type_handler() always returns pointers to
     type_handler_{time2|datetime2|timestamp2} no matter what

sql/sql_table.cc

@@ -3634,7 +3634,8 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     DBUG_RETURN(TRUE);
   }
 
-  select_field_pos= alter_info->create_list.elements - select_field_count;
+  select_field_pos= get_select_field_pos(alter_info, select_field_count,
+                                         create_info->versioned());
   null_fields= 0;
   create_info->varchar= 0;
   max_key_length= file->max_key_length();
@@ -3699,7 +3700,16 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 	/*
 	  If this was a CREATE ... SELECT statement, accept a field
 	  redefinition if we are changing a field in the SELECT part
+
+          The cases are:
+
+          field_no < select_field_pos: both field and dup are table fields;
+          dup_no >= select_field_pos: both field and dup are select fields or
+            field is implicit systrem field and dup is select field.
+
+          We are not allowed to put row_start/row_end into SELECT expression.
 	*/
+        DBUG_ASSERT(dup_no < field_no);
 	if (field_no < select_field_pos || dup_no >= select_field_pos ||
             dup_field->invisible >= INVISIBLE_SYSTEM)
 	{