MDEV-15575 different results when using CTE and big_tables=1.

This bug happened due to a defect of the implementation of the handler
function ha_delete_all_rows() for the ARIA engine.
The function maria_delete_all_rows() truncated the table, but it didn't
touch the write cache, so the cache's write offset was not reset.
In the scenario like in the function st_select_lex_unit::exec_recursive
when first all records were deleted from the table and then several new
records were added some metadata became inconsistent with the state of
the cache. As a result the table scan function could not read records
at the end of the table.
The same defect could be found in the implementation of ha_delete_all_rows()
for the MYISAM engine mi_delete_all_rows().

Additionally made late instantiation for the temporary table used to store
rows that were used for each new iteration when executing a recursive CTE.

mysql-test/r/cte_recursive.result

@@ -3063,3 +3063,20 @@ SELECT * FROM cte;
 5
 2
 3
+#
+# MDEV-15575: using recursive cte with big_tables enabled
+#
+set big_tables=1;
+with recursive qn as
+(select 123 as a union all select 1+a from qn where a<130)
+select * from qn;
+a
+123
+124
+125
+126
+127
+128
+129
+130
+set big_tables=default;

mysql-test/t/cte_recursive.test

@@ -2097,3 +2097,15 @@ WITH RECURSIVE cte AS
 	UNION
 	SELECT @c:=@c+1 FROM cte WHERE @c<3)
 SELECT * FROM cte;
+
+--echo #
+--echo # MDEV-15575: using recursive cte with big_tables enabled
+--echo #
+
+set big_tables=1;
+
+with recursive qn as
+(select 123 as a union all select 1+a from qn where a<130)
+select * from qn;
+
+set big_tables=default;

sql/sql_union.cc

@@ -214,19 +214,13 @@ select_union_recursive::create_result_table(THD *thd_arg,
   if (! (incr_table= create_tmp_table(thd_arg, &tmp_table_param, *column_types,
                                       (ORDER*) 0, false, 1,
                                       options, HA_POS_ERROR, "",
-                                      !create_table, keep_row_order)))
+                                      true, keep_row_order)))
     return true;
 
   incr_table->keys_in_use_for_query.clear_all();
   for (uint i=0; i < table->s->fields; i++)
     incr_table->field[i]->flags &= ~PART_KEY_FLAG;
 
-  if (create_table)
-  {
-    incr_table->file->extra(HA_EXTRA_WRITE_CACHE);
-    incr_table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
-  }
-
   TABLE *rec_table= 0;
   if (! (rec_table= create_tmp_table(thd_arg, &tmp_table_param, *column_types,
                                      (ORDER*) 0, false, 1,
@@ -269,9 +263,12 @@ void select_union_recursive::cleanup()
   }
 
   if (incr_table)
+  {
+    if (incr_table->is_created())
     {
       incr_table->file->extra(HA_EXTRA_RESET_STATE);
       incr_table->file->ha_delete_all_rows();
+    }
     free_tmp_table(thd, incr_table);
   }
 
@@ -1228,15 +1225,23 @@ bool st_select_lex_unit::exec_recursive()
   if (!was_executed)
     save_union_explain(thd->lex->explain);
 
-  if ((saved_error= incr_table->file->ha_delete_all_rows()))
-    goto err;
-
   if (with_element->level == 0)
   {
+    if (!incr_table->is_created() &&
+        instantiate_tmp_table(incr_table,
+                              tmp_table_param->keyinfo,
+                              tmp_table_param->start_recinfo,
+                              &tmp_table_param->recinfo,
+                              0))
+      DBUG_RETURN(1);
+    incr_table->file->extra(HA_EXTRA_WRITE_CACHE);
+    incr_table->file->extra(HA_EXTRA_IGNORE_DUP_KEY);
     start= first_select();
     if (with_element->with_anchor)
       end= with_element->first_recursive;
   }
+  else if ((saved_error= incr_table->file->ha_delete_all_rows()))
+    goto err;
 
   for (st_select_lex *sl= start ; sl != end; sl= sl->next_select())
   {

storage/maria/ma_delete_all.c

@@ -135,6 +135,9 @@ int maria_delete_all_rows(MARIA_HA *info)
       goto err;
   }
 
+  if (info->opt_flag & WRITE_CACHE_USED)
+    reinit_io_cache(&info->rec_cache, WRITE_CACHE, 0, 1, 1);
+
   _ma_writeinfo(info, WRITEINFO_UPDATE_KEYFILE);
 #ifdef HAVE_MMAP
   /* Map again */

storage/myisam/mi_delete_all.c

@@ -62,6 +62,10 @@ int mi_delete_all_rows(MI_INFO *info)
   if (mysql_file_chsize(info->dfile, 0, 0, MYF(MY_WME)) ||
       mysql_file_chsize(share->kfile, share->base.keystart, 0, MYF(MY_WME)))
     goto err;
+
+  if (info->opt_flag & WRITE_CACHE_USED)
+    reinit_io_cache(&info->rec_cache, WRITE_CACHE, 0, 1, 1);
+
   (void) _mi_writeinfo(info,WRITEINFO_UPDATE_KEYFILE);
   DBUG_RETURN(0);