MDEV-4280: Assertion `empty_size == empty_size_on_page' failure in ma_blockrec.c or ER_NOT_KEYFILE on query with DISTINCT and GROUP BY

This could happen when using Aria for internal temporary files (default case) and using DISTINCT.
_ma_scan_restore_block_record() didn't work correctly if there was rows inserted, updated or deleted on the handler
between calls to _ma_scan_remember_block_record() and _ma_scan_restore_block_record().
The effect was that some DISTINCT queries that used remove_dup_with_compare() could fail.

.bzrignore:
  Ignore sql_yacc.hh
mysql-test/suite/maria/r/distinct.result:
  Test case for MDEV-4280
mysql-test/suite/maria/t/distinct.test:
  Test case for MDEV-4280
mysql-test/t/mysql.test:
  Fixed test suite (we could get error -1 in some cases)
sql/sql_select.cc:
  Break loop if restart_rnd_next() gives an error
storage/maria/ha_maria.cc:
  scan_restore_pos() can return disk fault error.
storage/maria/ma_blockrec.c:
  _ma_scan_remember_block_record() did incorrectly update scan.dir instead of scan_save.dir .
  _ma_scan_restore_block_record() didn't work correctly if there was rows inserted,updated or deleted on the handler
  between calls to _ma_scan_remember_block_record() and _ma_scan_restore_block_record().
  Fixed by adding counters for row changes and reading the current scan page if changes had been made.
storage/maria/ma_blockrec.h:
  scan_restore_pos() can return disk fault error.
storage/maria/ma_delete.c:
  Increment row_changes
storage/maria/ma_scan.c:
  scan_restore_pos() can return disk fault error.
storage/maria/ma_update.c:
  Increment row_changes
storage/maria/ma_write.c:
  Increment row_changes
storage/maria/maria_def.h:
  scan_restore_pos() can return disk fault error.

8 files changed, 67 insertions, 20 deletions

diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 8b1fe4d05b3..ce308887516 100644
--- a/storage/maria/ha_maria.cc
+++ b/storage/maria/ha_maria.cc
@@ -2255,7 +2255,9 @@ int ha_maria::remember_rnd_pos()
 
 int ha_maria::restart_rnd_next(uchar *buf)
 {
-  (*file->s->scan_restore_pos)(file, remember_pos);
+  int error;
+  if ((error= (*file->s->scan_restore_pos)(file, remember_pos)))
+    return error;
   return rnd_next(buf);
 }
 
diff --git a/storage/maria/ma_blockrec.c b/storage/maria/ma_blockrec.c
index 71bbb389ca2..73b3326fd28 100644
--- a/storage/maria/ma_blockrec.c
+++ b/storage/maria/ma_blockrec.c
@@ -4008,6 +4008,8 @@ static int delete_dir_entry(uchar *buff, uint block_size, uint record_number,
   uint length, empty_space;
   uchar *dir;
   DBUG_ENTER("delete_dir_entry");
+  DBUG_PRINT("enter", ("record_number: %u  number_of_records: %u",
+                       record_number, number_of_records));
 
 #ifdef SANITY_CHECKS
   if (record_number >= number_of_records ||
@@ -4024,7 +4026,8 @@ static int delete_dir_entry(uchar *buff, uint block_size, uint record_number,
   check_directory(buff, block_size, 0, (uint) -1);
   empty_space= uint2korr(buff + EMPTY_SPACE_OFFSET);
   dir= dir_entry_pos(buff, block_size, record_number);
-  length= uint2korr(dir + 2);
+  length= uint2korr(dir + 2);  /* Length of entry we just deleted */
+  DBUG_ASSERT(uint2korr(dir) != 0 && length < block_size);
 
   if (record_number == number_of_records - 1)
   {
@@ -5230,11 +5233,6 @@ void _ma_scan_end_block_record(MARIA_HA *info)
   For the moment we can only remember one position, but this is
   good enough for MySQL usage
 
-  @Warning
-    When this function is called, we assume that the thread is not deleting
-    or updating the current row before ma_scan_restore_block_record()
-    is called!
-
   @return
   @retval 0			  ok
   @retval HA_ERR_WRONG_IN_RECORD  Could not allocate memory to hold position
@@ -5254,15 +5252,18 @@ int _ma_scan_remember_block_record(MARIA_HA *info,
     info->scan_save->bitmap_buff= ((uchar*) info->scan_save +
                                    ALIGN_SIZE(sizeof(*info->scan_save)));
   }
-  /* Point to the last read row */
-  *lastpos= info->cur_row.nextpos - 1;
-  info->scan.dir+= DIR_ENTRY_SIZE;
+  /* For checking if pages have changed since we last read it */
+  info->scan.row_changes= info->row_changes;
 
   /* Remember used bitmap and used head page */
   bitmap_buff= info->scan_save->bitmap_buff;
   memcpy(info->scan_save, &info->scan, sizeof(*info->scan_save));
   info->scan_save->bitmap_buff= bitmap_buff;
   memcpy(bitmap_buff, info->scan.bitmap_buff, info->s->block_size * 2);
+
+  /* Point to the last read row */
+  *lastpos= info->cur_row.nextpos - 1;
+  info->scan_save->dir+= DIR_ENTRY_SIZE;
   DBUG_RETURN(0);
 }
 
@@ -5270,15 +5271,22 @@ int _ma_scan_remember_block_record(MARIA_HA *info,
 /**
    @brief restore scan block it's original values
 
+   @return
+   0 ok
+   # error
+
    @note
    In theory we could swap bitmap buffers instead of copy them.
    For the moment we don't do that because there are variables pointing
    inside the buffers and it's a bit of hassle to either make them relative
    or repoint them.
+
+   If the data file has changed, we will re-read the new block record
+   to ensure that when we continue scanning we can ignore any deleted rows.
 */
 
-void _ma_scan_restore_block_record(MARIA_HA *info,
-                                   MARIA_RECORD_POS lastpos)
+int _ma_scan_restore_block_record(MARIA_HA *info,
+                                  MARIA_RECORD_POS lastpos)
 {
   uchar *bitmap_buff;
   DBUG_ENTER("_ma_scan_restore_block_record");
@@ -5289,7 +5297,26 @@ void _ma_scan_restore_block_record(MARIA_HA *info,
   info->scan.bitmap_buff= bitmap_buff;
   memcpy(bitmap_buff, info->scan_save->bitmap_buff, info->s->block_size * 2);
 
-  DBUG_VOID_RETURN;
+  if (info->scan.row_changes != info->row_changes)
+  {
+    /*
+      Table has been changed. We have to re-read the current page block as
+      data may have changed on it that we have to see.
+    */
+    if (!(pagecache_read(info->s->pagecache,
+                         &info->dfile,
+                         ma_recordpos_to_page(info->scan.row_base_page),
+                         0, info->scan.page_buff,
+                         info->s->page_type,
+                         PAGECACHE_LOCK_LEFT_UNLOCKED, 0)))
+      DBUG_RETURN(my_errno);
+    info->scan.number_of_rows=
+      (uint) (uchar) info->scan.page_buff[DIR_COUNT_OFFSET];
+    info->scan.dir_end= (info->scan.page_buff + info->s->block_size -
+                         PAGE_SUFFIX_SIZE -
+                         info->scan.number_of_rows * DIR_ENTRY_SIZE);
+  }
+  DBUG_RETURN(0);
 }
 
 
@@ -5316,7 +5343,7 @@ void _ma_scan_restore_block_record(MARIA_HA *info,
 
   RETURN
     0   ok
-    #   Error code
+    #   Error code  (Normally HA_ERR_END_OF_FILE)
 */
 
 int _ma_scan_block_record(MARIA_HA *info, uchar *record,
@@ -5336,6 +5363,12 @@ restart_record_read:
     uchar *data, *end_of_data;
     int error;
 
+    /* Ensure that scan.dir and record_pos are in sync */
+    DBUG_ASSERT(info->scan.dir == dir_entry_pos(info->scan.page_buff,
+                                                share->block_size,
+                                                record_pos));
+
+    /* Search for a valid directory entry (not 0) */
     while (!(offset= uint2korr(info->scan.dir)))
     {
       info->scan.dir-= DIR_ENTRY_SIZE;
@@ -5348,13 +5381,19 @@ restart_record_read:
       }
 #endif
     }
+    /*
+      This should always be true as the directory should always start with
+      a valid entry.
+    */
+    DBUG_ASSERT(info->scan.dir >= info->scan.dir_end);
+
     /* found row */
     info->cur_row.lastpos= info->scan.row_base_page + record_pos;
     info->cur_row.nextpos= record_pos + 1;
     data= info->scan.page_buff + offset;
     length= uint2korr(info->scan.dir + 2);
     end_of_data= data + length;
-    info->scan.dir-= DIR_ENTRY_SIZE;          /* Point to previous row */
+    info->scan.dir-= DIR_ENTRY_SIZE;      /* Point to next row to process */
 #ifdef SANITY_CHECKS
     if (end_of_data > info->scan.dir_end ||
         offset < PAGE_HEADER_SIZE || length < share->base.min_block_length)
diff --git a/storage/maria/ma_blockrec.h b/storage/maria/ma_blockrec.h
index 2a323d16ffa..f6aae8d4759 100644
--- a/storage/maria/ma_blockrec.h
+++ b/storage/maria/ma_blockrec.h
@@ -168,8 +168,8 @@ my_bool _ma_scan_init_block_record(MARIA_HA *info);
 void _ma_scan_end_block_record(MARIA_HA *info);
 int _ma_scan_remember_block_record(MARIA_HA *info,
                                    MARIA_RECORD_POS *lastpos);
-void _ma_scan_restore_block_record(MARIA_HA *info,
-                                   MARIA_RECORD_POS lastpos);
+int _ma_scan_restore_block_record(MARIA_HA *info,
+                                  MARIA_RECORD_POS lastpos);
 
 MARIA_RECORD_POS _ma_write_init_block_record(MARIA_HA *info,
                                              const uchar *record);
diff --git a/storage/maria/ma_delete.c b/storage/maria/ma_delete.c
index ab66499ecfe..977c7c82cb7 100644
--- a/storage/maria/ma_delete.c
+++ b/storage/maria/ma_delete.c
@@ -112,6 +112,7 @@ int maria_delete(MARIA_HA *info,const uchar *record)
   info->state->checksum-= info->cur_row.checksum;
   info->state->records--;
   info->update= HA_STATE_CHANGED+HA_STATE_DELETED+HA_STATE_ROW_CHANGED;
+  info->row_changes++;
   share->state.changed|= (STATE_NOT_OPTIMIZED_ROWS | STATE_NOT_MOVABLE |
                           STATE_NOT_ZEROFILLED);
   info->state->changed=1;
diff --git a/storage/maria/ma_scan.c b/storage/maria/ma_scan.c
index cbac463a2c8..ad526211615 100644
--- a/storage/maria/ma_scan.c
+++ b/storage/maria/ma_scan.c
@@ -68,7 +68,8 @@ int _ma_def_scan_remember_pos(MARIA_HA *info, MARIA_RECORD_POS *lastpos)
 }
 
 
-void _ma_def_scan_restore_pos(MARIA_HA *info, MARIA_RECORD_POS lastpos)
+int _ma_def_scan_restore_pos(MARIA_HA *info, MARIA_RECORD_POS lastpos)
 {
   info->cur_row.nextpos= lastpos;
+  return 0;
 }
diff --git a/storage/maria/ma_update.c b/storage/maria/ma_update.c
index abd9116a711..80b1fd86782 100644
--- a/storage/maria/ma_update.c
+++ b/storage/maria/ma_update.c
@@ -173,6 +173,7 @@ int maria_update(register MARIA_HA *info, const uchar *oldrec, uchar *newrec)
     We can't yet have HA_STATE_AKTIV here, as block_record dosn't support it
   */
   info->update= (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED | key_changed);
+  info->row_changes++;
   share->state.changed|= STATE_NOT_MOVABLE | STATE_NOT_ZEROFILLED;
   info->state->changed= 1;
 
diff --git a/storage/maria/ma_write.c b/storage/maria/ma_write.c
index b4bf5086e0b..3c5efdc2ef0 100644
--- a/storage/maria/ma_write.c
+++ b/storage/maria/ma_write.c
@@ -288,6 +288,7 @@ int maria_write(MARIA_HA *info, uchar *record)
   info->state->records++;
   info->update= (HA_STATE_CHANGED | HA_STATE_AKTIV | HA_STATE_WRITTEN |
 		 HA_STATE_ROW_CHANGED);
+  info->row_changes++;
   share->state.changed|= STATE_NOT_MOVABLE | STATE_NOT_ZEROFILLED;
   info->state->changed= 1;
 
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index 41501b8eb74..fe9bf10fe24 100644
--- a/storage/maria/maria_def.h
+++ b/storage/maria/maria_def.h
@@ -317,7 +317,7 @@ typedef struct st_maria_share
   /* End scan */
   void (*scan_end)(MARIA_HA *);
   int (*scan_remember_pos)(MARIA_HA *, MARIA_RECORD_POS*);
-  void (*scan_restore_pos)(MARIA_HA *, MARIA_RECORD_POS);
+  int (*scan_restore_pos)(MARIA_HA *, MARIA_RECORD_POS);
   /* Pre-write of row (some handlers may do the actual write here) */
   MARIA_RECORD_POS (*write_record_init)(MARIA_HA *, const uchar *);
   /* Write record (or accept write_record_init) */
@@ -492,6 +492,7 @@ typedef struct st_maria_block_scan
   ulonglong bits;
   uint number_of_rows, bit_pos;
   MARIA_RECORD_POS row_base_page;
+  ulonglong row_changes;
 } MARIA_BLOCK_SCAN;
 
 
@@ -533,6 +534,7 @@ struct st_maria_handler
   int (*read_record)(MARIA_HA *, uchar*, MARIA_RECORD_POS);
   invalidator_by_filename invalidator;	/* query cache invalidator */
   ulonglong last_auto_increment;        /* auto value at start of statement */
+  ulonglong row_changes;                /* Incremented for each change */
   ulong this_unique;			/* uniq filenumber or thread */
   ulong last_unique;			/* last unique number */
   ulong this_loop;			/* counter for this open */
@@ -1175,7 +1177,7 @@ my_bool _ma_check_status(void *param);
 void _ma_restore_status(void *param);
 void _ma_reset_status(MARIA_HA *maria);
 int _ma_def_scan_remember_pos(MARIA_HA *info, MARIA_RECORD_POS *lastpos);
-void _ma_def_scan_restore_pos(MARIA_HA *info, MARIA_RECORD_POS lastpos);
+int _ma_def_scan_restore_pos(MARIA_HA *info, MARIA_RECORD_POS lastpos);
 
 #include "ma_commit.h"