An assertion added (transaction must be re-enabled before end of

top-level statement) and fixes for the bugs it finds.
Fix for non-serious Valgrind warning.


sql/sql_insert.cc:
  When CREATE TABLE IF NOT EXISTS finds the table already exists,
  'table' is the existing table. So if that table is temporary we don't
  re-enable transactions which is a bug.
sql/sql_parse.cc:
  verify that at the end of each top-statement transactions have
  been re-enabled. Does not apply to substatements (consider
  CREATE TABLE t1 SELECT stored_func() : the substatements inside
  stored_func() run with transaction disabled).
  I am not putting the assertion into ha_external_lock(F_UNLCK) because
  performance schema tables get closed in the middle of a statement
  sometimes while transaction is disabled.
sql/sql_table.cc:
  copy_data_between_tables() forgot to clean-up several things in error
  conditions (ha_enable_transaction(), free-ing 'copy', etc) as found
  by the assertion added to sql_parse.cc.
storage/maria/ha_maria.cc:
  Comment
storage/maria/ma_blockrec.c:
  fix for Valgrind warning: a temporary table was created, a blob
  page of its was flushed to disk and had random bytes in the checksum
  area ("write of uninitialized bytes in pwrite")
storage/maria/ma_pagecrc.c:
  typo

6 files changed, 36 insertions, 28 deletions

diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index f4084c12c1f..1dc256bb7c3 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -3670,16 +3670,16 @@ bool select_create::send_eof()
     abort();
   else
   {
+    if ((thd->lex->create_info.options & HA_LEX_CREATE_TMP_TABLE) == 0)
+      ha_enable_transaction(thd, TRUE);
     /*
       Do an implicit commit at end of statement for non-temporary
       tables.  This can fail, but we should unlock the table
       nevertheless.
     */
     if (!table->s->tmp_table)
-    {
-      ha_enable_transaction(thd, TRUE);
       ha_commit(thd);               // Can fail, but we proceed anyway
-    }
+
     table->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
     table->file->extra(HA_EXTRA_WRITE_CANNOT_REPLACE);
     if (m_plock)
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 5543ede9efe..a739ddcc194 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -5356,6 +5356,7 @@ void mysql_reset_thd_for_next_command(THD *thd)
   DBUG_ENTER("mysql_reset_thd_for_next_command");
   DBUG_ASSERT(!thd->spcont); /* not for substatements of routines */
   DBUG_ASSERT(! thd->in_sub_stmt);
+  DBUG_ASSERT(thd->transaction.on);
   thd->free_list= 0;
   thd->select_number= 1;
   /*
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 69969a582c6..06cb8ef161c 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -6926,8 +6926,8 @@ copy_data_between_tables(TABLE *from,TABLE *to,
                          enum enum_enable_or_disable keys_onoff,
                          bool error_if_not_empty)
 {
-  int error;
-  Copy_field *copy,*copy_end;
+  int error= 1, errpos= 0;
+  Copy_field *copy= NULL, *copy_end;
   ulong found_count,delete_count;
   THD *thd= current_thd;
   uint length= 0;
@@ -6938,8 +6938,10 @@ copy_data_between_tables(TABLE *from,TABLE *to,
   List<Item>   all_fields;
   ha_rows examined_rows;
   bool auto_increment_field_copied= 0;
-  ulong save_sql_mode;
+  ulong save_sql_mode= thd->variables.sql_mode;
   ulonglong prev_insert_id;
+  List_iterator<Create_field> it(create);
+  Create_field *def;
   DBUG_ENTER("copy_data_between_tables");
 
   /*
@@ -6948,15 +6950,16 @@ copy_data_between_tables(TABLE *from,TABLE *to,
     
     This needs to be done before external_lock
   */
-  error= ha_enable_transaction(thd, FALSE);
-  if (error)
-    DBUG_RETURN(-1);
-  
+  if (ha_enable_transaction(thd, FALSE))
+    goto err;
+  errpos=1;
+
   if (!(copy= new Copy_field[to->s->fields]))
-    DBUG_RETURN(-1);				/* purecov: inspected */
+    goto err;		/* purecov: inspected */
 
   if (to->file->ha_external_lock(thd, F_WRLCK))
-    DBUG_RETURN(-1);
+    goto err;
+  errpos= 2;
 
   /* We need external lock before we can disable/enable keys */
   alter_table_manage_keys(to, from->file->indexes_are_disabled(), keys_onoff);
@@ -6968,11 +6971,8 @@ copy_data_between_tables(TABLE *from,TABLE *to,
 
   from->file->info(HA_STATUS_VARIABLE);
   to->file->ha_start_bulk_insert(from->file->stats.records);
+  errpos= 3;
 
-  save_sql_mode= thd->variables.sql_mode;
-
-  List_iterator<Create_field> it(create);
-  Create_field *def;
   copy_end=copy;
   for (Field **ptr=to->field ; *ptr ; ptr++)
   {
@@ -7017,7 +7017,6 @@ copy_data_between_tables(TABLE *from,TABLE *to,
       tables.table= from;
       tables.alias= tables.table_name= from->s->table_name.str;
       tables.db= from->s->db.str;
-      error= 1;
 
       if (thd->lex->select_lex.setup_ref_array(thd, order_num) ||
           setup_order(thd, thd->lex->select_lex.ref_pointer_array,
@@ -7034,6 +7033,7 @@ copy_data_between_tables(TABLE *from,TABLE *to,
   /* Tell handler that we have values for all columns in the to table */
   to->use_all_columns();
   init_read_record(&info, thd, from, (SQL_SELECT *) 0, 1,1);
+  errpos= 4;
   if (ignore)
     to->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
   thd->row_count= 0;
@@ -7097,22 +7097,22 @@ copy_data_between_tables(TABLE *from,TABLE *to,
     else
       found_count++;
   }
-  end_read_record(&info);
+
+err:
+  if (errpos >= 4)
+    end_read_record(&info);
   free_io_cache(from);
-  delete [] copy;				// This is never 0
+  delete [] copy;
 
-  if (to->file->ha_end_bulk_insert() && error <= 0)
+  if (errpos >= 3 && to->file->ha_end_bulk_insert() && error <= 0)
   {
     to->file->print_error(my_errno,MYF(0));
     error=1;
   }
   to->file->extra(HA_EXTRA_NO_IGNORE_DUP_KEY);
 
-  if (ha_enable_transaction(thd, TRUE))
-  {
+  if (errpos >= 1 && ha_enable_transaction(thd, TRUE))
     error= 1;
-    goto err;
-  }
   
   /*
     Ensure that the new table is saved properly to disk so that we
@@ -7123,14 +7123,12 @@ copy_data_between_tables(TABLE *from,TABLE *to,
   if (ha_commit(thd))
     error=1;
 
- err:
   thd->variables.sql_mode= save_sql_mode;
   thd->abort_on_warning= 0;
-  free_io_cache(from);
   *copied= found_count;
   *deleted=delete_count;
   to->file->ha_release_auto_increment();
-  if (to->file->ha_external_lock(thd,F_UNLCK))
+  if (errpos >= 2 && to->file->ha_external_lock(thd,F_UNLCK))
     error=1;
   DBUG_RETURN(error > 0 ? -1 : 0);
 }
diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 0ad823c1755..4aa7ee7d25f 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -2212,6 +2212,11 @@ int ha_maria::external_lock(THD *thd, int lock_type)
   }
   else
   {
+    /*
+      We always re-enable, don't rely on thd->transaction.on as it is
+      sometimes reset to true after unlocking (see mysql_truncate() for a
+      partitioned table based on Maria).
+    */
     if (_ma_reenable_logging_for_table(file, TRUE))
       DBUG_RETURN(1);
     /** @todo zero file->trn also in commit and rollback */
diff --git a/storage/maria/ma_blockrec.c b/storage/maria/ma_blockrec.c
index 2263e79fe52..158de5cd451 100644
--- a/storage/maria/ma_blockrec.c
+++ b/storage/maria/ma_blockrec.c
@@ -1936,6 +1936,10 @@ static my_bool write_full_pages(MARIA_HA *info,
       bzero(buff + block_size - PAGE_SUFFIX_SIZE - (data_size - copy_length),
             (data_size - copy_length) + PAGE_SUFFIX_SIZE);
 
+#ifdef HAVE_purify /* avoid warning about writing uninitialized CRC to disk */
+    int4store_aligned(buff + block_size - CRC_SIZE,
+                      MARIA_NO_CRC_NORMAL_PAGE);
+#endif
     if (pagecache_write(share->pagecache,
                         &info->dfile, page, 0,
                         buff, share->page_type,
diff --git a/storage/maria/ma_pagecrc.c b/storage/maria/ma_pagecrc.c
index 357527c6faa..698cc8b95e0 100644
--- a/storage/maria/ma_pagecrc.c
+++ b/storage/maria/ma_pagecrc.c
@@ -148,7 +148,7 @@ my_bool maria_page_crc_set_index(uchar *page,
   MARIA_SHARE *share= (MARIA_SHARE *)data_ptr;
   int data_length= _ma_get_page_used(share, page);
   uint32 crc= maria_page_crc((uint32) page_no, page, data_length);
-  DBUG_ENTER("maria_page_crc_set");
+  DBUG_ENTER("maria_page_crc_set_index");
   DBUG_PRINT("info", ("Page %lu  crc: %lu",
                       (ulong) page_no, (ulong) crc));
   DBUG_ASSERT((uint)data_length <= share->block_size - CRC_SIZE);