MDEV-14943 Alter table ORDER BY bug

Problem was that if copy_data_between_tables() didn't do proper
clean up in case of failures:
- copy object was not properly freed
- end_bulk_insert() was not called
- mysql_trans_prepare_alter_copy_data() set THD->transaction.on to
  false which was not properly restored

The last part caused a crash in Aria as Aria depends on that THD
is correct.

Other things:
- Reset info->switched_transactional after usage (safety)
- Reset bulk_insert_single_undo (safety)

5 files changed, 61 insertions, 8 deletions

diff --git a/mysql-test/suite/maria/alter.result b/mysql-test/suite/maria/alter.result
index 1a7daf5a1ee..c63688dddd6 100644
--- a/mysql-test/suite/maria/alter.result
+++ b/mysql-test/suite/maria/alter.result
@@ -31,3 +31,19 @@ pk	i
 8	88
 9	99
 DROP TABLE t1;
+CREATE TABLE t1 (f INT) ENGINE=Aria transactional=1;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `f` int(11) DEFAULT NULL
+) ENGINE=Aria DEFAULT CHARSET=latin1 PAGE_CHECKSUM=1 TRANSACTIONAL=1
+INSERT INTO t1 VALUES (1),(2);
+ALTER TABLE t1 ORDER BY unknown_column;
+ERROR 42S22: Unknown column 'unknown_column' in 'order clause'
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `f` int(11) DEFAULT NULL
+) ENGINE=Aria DEFAULT CHARSET=latin1 PAGE_CHECKSUM=1 TRANSACTIONAL=1
+CREATE TABLE t2 SELECT * FROM t1;
+DROP TABLE t1, t2;
diff --git a/mysql-test/suite/maria/alter.test b/mysql-test/suite/maria/alter.test
index abca4865688..09672cdfa3b 100644
--- a/mysql-test/suite/maria/alter.test
+++ b/mysql-test/suite/maria/alter.test
@@ -25,3 +25,20 @@ INSERT INTO t1 VALUES (2,0),(3,33),(4,0),(5,55),(6,66),(7,0),(8,88),(9,99);
 ALTER TABLE t1 ENABLE KEYS;
 SELECT * FROM t1 WHERE i = 0 OR pk BETWEEN 6 AND 10;
 DROP TABLE t1;
+
+#
+# MDEV-14943
+# Assertion `block->type == PAGECACHE_EMPTY_PAGE || block->type == type ||
+# type == PAGECACHE_LSN_PAGE || type == PAGECACHE_READ_UNKNOWN_PAGE ||
+# block->type == PAGECACHE_READ_UNKNOWN_PAGE' failed in pagecache_read upon
+# CREATE ... SELECT from Aria table
+#
+
+CREATE TABLE t1 (f INT) ENGINE=Aria transactional=1;
+SHOW CREATE TABLE t1;
+INSERT INTO t1 VALUES (1),(2);
+--error ER_BAD_FIELD_ERROR
+ALTER TABLE t1 ORDER BY unknown_column;
+SHOW CREATE TABLE t1;
+CREATE TABLE t2 SELECT * FROM t1;
+DROP TABLE t1, t2;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 8f3468a44db..27d579a6b19 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -9356,9 +9356,7 @@ bool mysql_trans_prepare_alter_copy_data(THD *thd)
     
     This needs to be done before external_lock.
   */
-  if (ha_enable_transaction(thd, FALSE))
-    DBUG_RETURN(TRUE);
-  DBUG_RETURN(FALSE);
+  DBUG_RETURN(ha_enable_transaction(thd, FALSE) != 0);
 }
 
 
@@ -9409,6 +9407,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   ha_rows examined_rows;
   ha_rows found_rows;
   bool auto_increment_field_copied= 0;
+  bool cleanup_done= 0;
   ulonglong save_sql_mode= thd->variables.sql_mode;
   ulonglong prev_insert_id, time_to_report_progress;
   Field **dfield_ptr= to->default_field;
@@ -9417,15 +9416,23 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   /* Two or 3 stages; Sorting, copying data and update indexes */
   thd_progress_init(thd, 2 + MY_TEST(order));
 
-  if (mysql_trans_prepare_alter_copy_data(thd))
-    DBUG_RETURN(-1);
-
   if (!(copy= new Copy_field[to->s->fields]))
     DBUG_RETURN(-1);				/* purecov: inspected */
 
+  if (mysql_trans_prepare_alter_copy_data(thd))
+  {
+    delete [] copy;
+    DBUG_RETURN(-1);
+  }
+
   /* We need external lock before we can disable/enable keys */
   if (to->file->ha_external_lock(thd, F_WRLCK))
+  {
+    /* Undo call to mysql_trans_prepare_alter_copy_data() */
+    ha_enable_transaction(thd, TRUE);
+    delete [] copy;
     DBUG_RETURN(-1);
+  }
 
   alter_table_manage_keys(to, from->file->indexes_are_disabled(), keys_onoff);
 
@@ -9435,7 +9442,6 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   from->file->info(HA_STATUS_VARIABLE);
   to->file->ha_start_bulk_insert(from->file->stats.records,
                                  ignore ? 0 : HA_CREATE_UNIQUE_INDEX_BY_SORT);
-
   List_iterator<Create_field> it(create);
   Create_field *def;
   copy_end=copy;
@@ -9637,7 +9643,6 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   }
   end_read_record(&info);
   free_io_cache(from);
-  delete [] copy;
 
   THD_STAGE_INFO(thd, stage_enabling_keys);
   thd_progress_next_stage(thd);
@@ -9652,6 +9657,7 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
     to->file->print_error(my_errno,MYF(0));
     error= 1;
   }
+  cleanup_done= 1;
   to->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
 
   if (mysql_trans_commit_alter_copy_data(thd))
@@ -9663,6 +9669,16 @@ copy_data_between_tables(THD *thd, TABLE *from, TABLE *to,
   *copied= found_count;
   *deleted=delete_count;
   to->file->ha_release_auto_increment();
+  delete [] copy;
+
+  if (!cleanup_done)
+  {
+    /* This happens if we get an error during initialzation of data */
+    DBUG_ASSERT(error);
+    to->file->ha_end_bulk_insert();
+    ha_enable_transaction(thd, TRUE);
+  }
+
   if (to->file->ha_external_lock(thd,F_UNLCK))
     error=1;
   if (error < 0 && to->file->extra(HA_EXTRA_PREPARE_FOR_RENAME))
diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index f52336de9d0..80f278ab2eb 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -2224,6 +2224,7 @@ end:
       _ma_reenable_logging_for_table(file,
                                      bulk_insert_single_undo ==
                                      BULK_INSERT_SINGLE_UNDO_AND_NO_REPAIR);
+    bulk_insert_single_undo= BULK_INSERT_NONE;  // Safety
   }
   DBUG_RETURN(err);
 }
diff --git a/storage/maria/ma_recovery.c b/storage/maria/ma_recovery.c
index 4c3310c68c5..b0c38e9f24d 100644
--- a/storage/maria/ma_recovery.c
+++ b/storage/maria/ma_recovery.c
@@ -3583,7 +3583,10 @@ my_bool _ma_reenable_logging_for_table(MARIA_HA *info, my_bool flush_pages)
 
   if (share->now_transactional == share->base.born_transactional ||
       !info->switched_transactional)
+  {
+    info->switched_transactional= FALSE;
     DBUG_RETURN(0);
+  }
   info->switched_transactional= FALSE;
 
   if ((share->now_transactional= share->base.born_transactional))