Bug#25673 - spatial index corruption, error 126

          incorrect key file for table

In certain cases it could happen that deleting a row could
corrupt an RTREE index.

According to Guttman's algorithm, page underflow is handled
by storing the page in a list for later re-insertion. The
keys from the stored pages have to be inserted into the
remaining pages of the same level of the tree. Hence the
level number is stored in the re-insertion list together
with the page.

In the MySQL RTree implementation the level counts from zero
at the root page, increasing numbers for levels down the tree.

If during re-insertion of the keys the tree height grows, all
level numbers become invalid. The remaining keys will be
inserted at the wrong level.

The fix is to increment the level numbers stored in the
reinsert list after a split of the root block during reinsertion.


myisam/rt_index.c:
  Bug#25673 - spatial index corruption, error 126
              incorrect key file for table
  Added a loop in rtree_delete() to increment the level numbers
  stored in the reinsert list after a split of the root block
  during reinsertion.
  Added comments and DBUG statements.
myisam/rt_key.c:
  Bug#25673 - spatial index corruption, error 126
              incorrect key file for table
  Added DBUG statements.
myisam/rt_split.c:
  Bug#25673 - spatial index corruption, error 126
              incorrect key file for table
  Added DBUG statements.
mysql-test/r/gis-rtree.result:
  Bug#25673 - spatial index corruption, error 126
              incorrect key file for table
  Added the test result.
mysql-test/t/gis-rtree.test:
  Bug#25673 - spatial index corruption, error 126
              incorrect key file for table
  Added a test.

3 files changed, 92 insertions, 32 deletions

diff --git a/myisam/rt_index.c b/myisam/rt_index.c
index 1806476dc39..9c58f4ba5d2 100644
--- a/myisam/rt_index.c
+++ b/myisam/rt_index.c
@@ -186,6 +186,7 @@ int rtree_find_first(MI_INFO *info, uint keynr, uchar *key, uint key_length,
   /*
     Save searched key, include data pointer.
     The data pointer is required if the search_flag contains MBR_DATA.
+    (minimum bounding rectangle)
   */
   memcpy(info->first_mbr_key, key, keyinfo->keylength);
   info->last_rkey_length = key_length;
@@ -540,16 +541,19 @@ static int rtree_insert_req(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
   uint nod_flag;
   uchar *page_buf;
   int res;
+  DBUG_ENTER("rtree_insert_req");
 
   if (!(page_buf = (uchar*)my_alloca((uint)keyinfo->block_length + 
                                      MI_MAX_KEY_BUFF)))
   {
     my_errno = HA_ERR_OUT_OF_MEM;
-    return -1;
+    DBUG_RETURN(-1); /* purecov: inspected */
   }
   if (!_mi_fetch_keypage(info, keyinfo, page, DFLT_INIT_HITS, page_buf, 0))
     goto err1;
   nod_flag = mi_test_if_nod(page_buf);
+  DBUG_PRINT("rtree", ("page: %lu  level: %d  ins_level: %d  nod_flag: %u",
+                       (ulong) page, level, ins_level, nod_flag));
 
   if ((ins_level == -1 && nod_flag) ||       /* key: go down to leaf */
       (ins_level > -1 && ins_level > level)) /* branch: go down to ins_level */
@@ -601,11 +605,11 @@ static int rtree_insert_req(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
 
 ok:
   my_afree((byte*)page_buf);
-  return res;
+  DBUG_RETURN(res);
 
 err1:
   my_afree((byte*)page_buf);
-  return -1;
+  DBUG_RETURN(-1); /* purecov: inspected */
 }
 
 
@@ -625,7 +629,8 @@ static int rtree_insert_level(MI_INFO *info, uint keynr, uchar *key,
   MI_KEYDEF *keyinfo = info->s->keyinfo + keynr;
   int res;
   my_off_t new_page;
-  
+  DBUG_ENTER("rtree_insert_level");
+
   if ((old_root = info->s->state.key_root[keynr]) == HA_OFFSET_ERROR)
   {
     int res;
@@ -655,11 +660,12 @@ static int rtree_insert_level(MI_INFO *info, uint keynr, uchar *key,
       uchar *new_key;
       uint nod_flag = info->s->base.key_reflength;
 
+      DBUG_PRINT("rtree", ("root was split, grow a new root"));
       if (!(new_root_buf = (uchar*)my_alloca((uint)keyinfo->block_length + 
                                              MI_MAX_KEY_BUFF)))
       {
         my_errno = HA_ERR_OUT_OF_MEM;
-        return -1;
+        DBUG_RETURN(-1); /* purecov: inspected */
       }
 
       mi_putint(new_root_buf, 2, nod_flag);
@@ -685,12 +691,14 @@ static int rtree_insert_level(MI_INFO *info, uint keynr, uchar *key,
                             DFLT_INIT_HITS, new_root_buf))
         goto err1;
       info->s->state.key_root[keynr] = new_root;
+      DBUG_PRINT("rtree", ("new root page: %lu  level: %d  nod_flag: %u",
+                           (ulong) new_root, 0, mi_test_if_nod(new_root_buf)));
 
       my_afree((byte*)new_root_buf);
       break;
 err1:
       my_afree((byte*)new_root_buf);
-      return -1;
+      DBUG_RETURN(-1); /* purecov: inspected */
     }
     default:
     case -1: /* error */
@@ -698,7 +706,7 @@ err1:
       break;
     }
   }
-  return res;
+  DBUG_RETURN(res);
 }
 
 
@@ -712,8 +720,10 @@ err1:
 
 int rtree_insert(MI_INFO *info, uint keynr, uchar *key, uint key_length)
 {
-  return (!key_length ||
-	  (rtree_insert_level(info, keynr, key, key_length, -1) == -1)) ? -1 : 0;
+  DBUG_ENTER("rtree_insert");
+  DBUG_RETURN((!key_length ||
+               (rtree_insert_level(info, keynr, key, key_length, -1) == -1)) ?
+              -1 : 0);
 }
 
 
@@ -728,6 +738,8 @@ int rtree_insert(MI_INFO *info, uint keynr, uchar *key, uint key_length)
 static int rtree_fill_reinsert_list(stPageList *ReinsertList, my_off_t page, 
                                     int level)
 {
+  DBUG_ENTER("rtree_fill_reinsert_list");
+  DBUG_PRINT("rtree", ("page: %lu  level: %d", (ulong) page, level));
   if (ReinsertList->n_pages == ReinsertList->m_pages)
   {
     ReinsertList->m_pages += REINSERT_BUFFER_INC;
@@ -739,10 +751,10 @@ static int rtree_fill_reinsert_list(stPageList *ReinsertList, my_off_t page,
   ReinsertList->pages[ReinsertList->n_pages].offs = page;
   ReinsertList->pages[ReinsertList->n_pages].level = level;
   ReinsertList->n_pages++;
-  return 0;
+  DBUG_RETURN(0);
 
 err1:
-  return -1;
+  DBUG_RETURN(-1); /* purecov: inspected */
 }
 
 
@@ -766,15 +778,18 @@ static int rtree_delete_req(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
   uint nod_flag;
   uchar *page_buf;
   int res;
+  DBUG_ENTER("rtree_delete_req");
 
   if (!(page_buf = (uchar*)my_alloca((uint)keyinfo->block_length)))
   {
     my_errno = HA_ERR_OUT_OF_MEM;
-    return -1;
+    DBUG_RETURN(-1); /* purecov: inspected */
   }
   if (!_mi_fetch_keypage(info, keyinfo, page, DFLT_INIT_HITS, page_buf, 0))
     goto err1;
   nod_flag = mi_test_if_nod(page_buf);
+  DBUG_PRINT("rtree", ("page: %lu  level: %d  nod_flag: %u",
+                       (ulong) page, level, nod_flag));
 
   k = rt_PAGE_FIRST_KEY(page_buf, nod_flag);
   last = rt_PAGE_END(page_buf);
@@ -795,6 +810,7 @@ static int rtree_delete_req(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
             if (*page_size + key_length >= rt_PAGE_MIN_SIZE(keyinfo->block_length)) 
             { 
               /* OK */
+              /* Calculate a new key value (MBR) for the shrinked block. */
               if (rtree_set_key_mbr(info, keyinfo, k, key_length, 
                                   _mi_kpos(nod_flag, k)))
                 goto err1;
@@ -804,10 +820,23 @@ static int rtree_delete_req(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
             }
             else
             { 
-              /* too small: delete key & add it descendant to reinsert list */
+              /*
+                Too small: delete key & add it descendant to reinsert list.
+                Store position and level of the block so that it can be
+                accessed later for inserting the remaining keys.
+              */
+              DBUG_PRINT("rtree", ("too small. move block to reinsert list"));
               if (rtree_fill_reinsert_list(ReinsertList, _mi_kpos(nod_flag, k),
                                            level + 1))
                 goto err1;
+              /*
+                Delete the key that references the block. This makes the
+                block disappear from the index. Hence we need to insert
+                its remaining keys later. Note: if the block is a branch
+                block, we do not only remove this block, but the whole
+                subtree. So we need to re-insert its keys on the same
+                level later to reintegrate the subtrees.
+              */
               rtree_delete_key(info, page_buf, k, key_length, nod_flag);
               if (_mi_write_keypage(info, keyinfo, page,
                                     DFLT_INIT_HITS, page_buf))
@@ -867,11 +896,11 @@ static int rtree_delete_req(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
 
 ok:
   my_afree((byte*)page_buf);
-  return res;
+  DBUG_RETURN(res);
 
 err1:
   my_afree((byte*)page_buf);
-  return -1;
+  DBUG_RETURN(-1); /* purecov: inspected */
 }
 
 
@@ -889,12 +918,15 @@ int rtree_delete(MI_INFO *info, uint keynr, uchar *key, uint key_length)
   stPageList ReinsertList;
   my_off_t old_root;
   MI_KEYDEF *keyinfo = info->s->keyinfo + keynr;
+  DBUG_ENTER("rtree_delete");
 
   if ((old_root = info->s->state.key_root[keynr]) == HA_OFFSET_ERROR)
   {
     my_errno= HA_ERR_END_OF_FILE;
-    return -1;
+    DBUG_RETURN(-1); /* purecov: inspected */
   }
+  DBUG_PRINT("rtree", ("starting deletion at root page: %lu",
+                       (ulong) old_root));
 
   ReinsertList.pages = NULL;
   ReinsertList.n_pages = 0;
@@ -903,12 +935,12 @@ int rtree_delete(MI_INFO *info, uint keynr, uchar *key, uint key_length)
   switch (rtree_delete_req(info, keyinfo, key, key_length, old_root, 
                                  &page_size, &ReinsertList, 0))
   {
-    case 2:
+    case 2: /* empty */
     {
       info->s->state.key_root[keynr] = HA_OFFSET_ERROR;
-      return 0;
+      DBUG_RETURN(0);
     }
-    case 0:
+    case 0: /* deleted */
     {
       uint nod_flag;
       ulong i;
@@ -928,16 +960,34 @@ int rtree_delete(MI_INFO *info, uint keynr, uchar *key, uint key_length)
                                DFLT_INIT_HITS, page_buf, 0))
           goto err1;
         nod_flag = mi_test_if_nod(page_buf);
+        DBUG_PRINT("rtree", ("reinserting keys from "
+                             "page: %lu  level: %d  nod_flag: %u",
+                             (ulong) ReinsertList.pages[i].offs,
+                             ReinsertList.pages[i].level, nod_flag));
+
         k = rt_PAGE_FIRST_KEY(page_buf, nod_flag);
         last = rt_PAGE_END(page_buf);
         for (; k < last; k = rt_PAGE_NEXT_KEY(k, key_length, nod_flag))
         {
-          if (rtree_insert_level(info, keynr, k, key_length, 
-                                 ReinsertList.pages[i].level) == -1)
+          int res;
+          if ((res= rtree_insert_level(info, keynr, k, key_length,
+                                       ReinsertList.pages[i].level)) == -1)
           {
             my_afree((byte*)page_buf);
             goto err1;
           }
+          if (res)
+          {
+            int j;
+            DBUG_PRINT("rtree", ("root has been split, adjust levels"));
+            for (j= i; j < ReinsertList.n_pages; j++)
+            {
+              ReinsertList.pages[j].level++;
+              DBUG_PRINT("rtree", ("keys from page: %lu  now level: %d",
+                                   (ulong) ReinsertList.pages[i].offs,
+                                   ReinsertList.pages[i].level));
+            }
+          }
         }
         my_afree((byte*)page_buf);
         if (_mi_dispose(info, keyinfo, ReinsertList.pages[i].offs,
@@ -964,20 +1014,20 @@ int rtree_delete(MI_INFO *info, uint keynr, uchar *key, uint key_length)
         info->s->state.key_root[keynr] = new_root;
       }
       info->update= HA_STATE_DELETED;
-      return 0;
+      DBUG_RETURN(0);
 
 err1:
-      return -1;
+      DBUG_RETURN(-1); /* purecov: inspected */
     }
     case 1: /* not found */
     {
       my_errno = HA_ERR_KEY_NOT_FOUND;
-      return -1;
+      DBUG_RETURN(-1); /* purecov: inspected */
     }
     default:
     case -1: /* error */
     {
-      return -1;
+      DBUG_RETURN(-1); /* purecov: inspected */
     }
   }
 }
diff --git a/myisam/rt_key.c b/myisam/rt_key.c
index e2a402fbefd..b969ac30569 100644
--- a/myisam/rt_key.c
+++ b/myisam/rt_key.c
@@ -35,6 +35,7 @@ int rtree_add_key(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
 {
   uint page_size = mi_getint(page_buf);
   uint nod_flag = mi_test_if_nod(page_buf);
+  DBUG_ENTER("rtree_add_key");
 
   if (page_size + key_length + info->s->base.rec_reflength <=
       keyinfo->block_length)
@@ -43,22 +44,26 @@ int rtree_add_key(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
     if (nod_flag)
     {
       /* save key */
+      DBUG_ASSERT(_mi_kpos(nod_flag, key) < info->state->key_file_length);
       memcpy(rt_PAGE_END(page_buf), key - nod_flag, key_length + nod_flag); 
       page_size += key_length + nod_flag;
     }
     else
     {
       /* save key */
+      DBUG_ASSERT(_mi_dpos(info, nod_flag, key + key_length +
+                           info->s->base.rec_reflength) <
+                  info->state->data_file_length + info->s->base.pack_reclength);
       memcpy(rt_PAGE_END(page_buf), key, key_length + 
                                          info->s->base.rec_reflength);
       page_size += key_length + info->s->base.rec_reflength;
     }
     mi_putint(page_buf, page_size, nod_flag);
-    return 0;
+    DBUG_RETURN(0);
   }
 
-  return (rtree_split_page(info, keyinfo, page_buf, key, key_length,
-			   new_page) ? -1 : 1);
+  DBUG_RETURN((rtree_split_page(info, keyinfo, page_buf, key, key_length,
+                                new_page) ? -1 : 1));
 }
 
 /*
@@ -90,11 +95,13 @@ int rtree_delete_key(MI_INFO *info, uchar *page_buf, uchar *key,
 int rtree_set_key_mbr(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key, 
 		      uint key_length, my_off_t child_page)
 {
+  DBUG_ENTER("rtree_set_key_mbr");
+
   if (!_mi_fetch_keypage(info, keyinfo, child_page,
                          DFLT_INIT_HITS, info->buff, 0))
-    return -1;
+    DBUG_RETURN(-1); /* purecov: inspected */
 
-  return rtree_page_mbr(info, keyinfo->seg, info->buff, key, key_length);
+  DBUG_RETURN(rtree_page_mbr(info, keyinfo->seg, info->buff, key, key_length));
 }
 
 #endif /*HAVE_RTREE_KEYS*/
diff --git a/myisam/rt_split.c b/myisam/rt_split.c
index 87da22a93c7..d400274064b 100644
--- a/myisam/rt_split.c
+++ b/myisam/rt_split.c
@@ -268,13 +268,15 @@ int rtree_split_page(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *page, uchar *key,
                                    info->s->base.rec_reflength);
 
   int max_keys = (mi_getint(page)-2) / (full_length);
+  DBUG_ENTER("rtree_split_page");
+  DBUG_PRINT("rtree", ("splitting block"));
 
   n_dim = keyinfo->keysegs / 2;
   
   if (!(coord_buf= (double*) my_alloca(n_dim * 2 * sizeof(double) *
 				       (max_keys + 1 + 4) +
 				       sizeof(SplitStruct) * (max_keys + 1))))
-    return -1;
+    DBUG_RETURN(-1); /* purecov: inspected */
 
   task= (SplitStruct *)(coord_buf + n_dim * 2 * (max_keys + 1 + 4));
 
@@ -346,12 +348,13 @@ int rtree_split_page(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *page, uchar *key,
   else
     err_code= _mi_write_keypage(info, keyinfo, *new_page_offs,
                                 DFLT_INIT_HITS, new_page);
+  DBUG_PRINT("rtree", ("split new block: %lu", (ulong) *new_page_offs));
 
   my_afree((byte*)new_page);
 
 split_err:
   my_afree((byte*) coord_buf);
-  return err_code;
+  DBUG_RETURN(err_code);
 }
 
 #endif /*HAVE_RTREE_KEYS*/