WL#1562 (Improving spatial code)

A set of changes improving our RTree indexes and fixed few bugs
found during the tests


myisam/rt_index.c:
  Algorythm for picking the branch to insert was fixed.
  pick_by_perimeter version of the algorythm added (mostly for testing
  purposes)
myisam/rt_index.h:
  minimal size of the page set to 1/3
  It noticeable increases searching performance
myisam/rt_key.c:
  counting of the size of the filled part of the page fixed
  rtree_choose_key moved to rt_index.c
myisam/rt_key.h:
  no need to make rtree_choose_key global
myisam/rt_mbr.c:
  operations for counting the perimeter of MBR added
myisam/rt_mbr.h:
  interface for rtree_perimeter_increase
myisam/rt_split.c:
  my_multi_malloc changed with my_alloca
sql/spatial.cc:
  LINESTRING object can consist of single point

myisam/rt_index.c

@@ -22,6 +22,8 @@
 #include "rt_mbr.h"
 
 #define REINSERT_BUFFER_INC 10
+#define PICK_BY_AREA
+/*#define PICK_BY_PERIMETER*/
 
 typedef struct st_page_level
 {
@@ -436,6 +438,92 @@ int rtree_get_next(MI_INFO *info, uint keynr, uint key_length)
 }
 
 
+/*
+  Choose non-leaf better key for insertion
+*/
+
+#ifdef PICK_BY_PERIMETER
+static uchar *rtree_pick_key(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key, 
+			     uint key_length, uchar *page_buf, uint nod_flag)
+{
+  double increase;
+  double best_incr = DBL_MAX;
+  double perimeter;
+  double best_perimeter;
+  uchar *best_key;
+  uchar *k = rt_PAGE_FIRST_KEY(page_buf, nod_flag);
+  uchar *last = rt_PAGE_END(page_buf);
+
+  LINT_INIT(best_perimeter);
+  LINT_INIT(best_key);
+
+  for (; k < last; k = rt_PAGE_NEXT_KEY(k, key_length, nod_flag))
+  {
+    if ((increase = rtree_perimeter_increase(keyinfo->seg, k, key, key_length,
+					     &perimeter)) == -1)
+      return NULL;
+    if (increase < best_incr)
+    {
+      best_key = k;
+      best_perimeter= perimeter;
+      best_incr = increase;
+    }
+    else
+    {
+      if ((increase == best_incr) && (perimeter < best_perimeter))
+      {
+        best_key = k;
+        best_perimeter= perimeter;
+        best_incr = increase;
+      }
+    }
+  }
+  return best_key;
+}
+
+#endif /*PICK_BY_PERIMETER*/
+
+#ifdef PICK_BY_AREA
+static uchar *rtree_pick_key(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key, 
+			     uint key_length, uchar *page_buf, uint nod_flag)
+{
+  double increase;
+  double best_incr = DBL_MAX;
+  double area;
+  double best_area;
+  uchar *best_key;
+  uchar *k = rt_PAGE_FIRST_KEY(page_buf, nod_flag);
+  uchar *last = rt_PAGE_END(page_buf);
+
+  LINT_INIT(best_area);
+  LINT_INIT(best_key);
+
+  for (; k < last; k = rt_PAGE_NEXT_KEY(k, key_length, nod_flag))
+  {
+    if ((increase = rtree_area_increase(keyinfo->seg, k, key, key_length, 
+                                        &area)) == -1)
+      return NULL;
+    if (increase < best_incr)
+    {
+      best_key = k;
+      best_area = area;
+      best_incr = increase;
+    }
+    else
+    {
+      if ((increase == best_incr) && (area < best_area))
+      {
+        best_key = k;
+        best_area = area;
+        best_incr = increase;
+      }
+    }
+  }
+  return best_key;
+}
+
+#endif /*PICK_BY_AREA*/
+
 /*
   Go down and insert key into tree
 
@@ -467,7 +555,7 @@ static int rtree_insert_req(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
   if ((ins_level == -1 && nod_flag) ||       /* key: go down to leaf */
       (ins_level > -1 && ins_level > level)) /* branch: go down to ins_level */
   { 
-    if ((k = rtree_choose_key(info, keyinfo, key, key_length, page_buf, 
+    if ((k = rtree_pick_key(info, keyinfo, key, key_length, page_buf, 
                              nod_flag)) == NULL)
       goto err1;
     switch ((res = rtree_insert_req(info, keyinfo, key, key_length, 

myisam/rt_index.h

@@ -23,7 +23,7 @@
               (nod_flag ? nod_flag : info->s->base.rec_reflength))
 #define rt_PAGE_END(page) (page + mi_getint(page))
 
-#define rt_PAGE_MIN_SIZE(block_length) ((uint)(block_length) / 2)
+#define rt_PAGE_MIN_SIZE(block_length) ((uint)(block_length) / 3)
 
 int rtree_insert(MI_INFO *info, uint keynr, uchar *key, uint key_length);
 int rtree_delete(MI_INFO *info, uint keynr, uchar *key, uint key_length);

myisam/rt_key.c

@@ -35,7 +35,8 @@ 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);
 
-  if (page_size + key_length + nod_flag <= keyinfo->block_length)  
+  if (page_size + key_length + info->s->base.rec_reflength <=
+      keyinfo->block_length)
   {
     /* split won't be necessary */
     if (nod_flag)
@@ -94,46 +95,3 @@ int rtree_set_key_mbr(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
 
   return rtree_page_mbr(info, keyinfo->seg, info->buff, key, key_length);
 }
-
-
-/*
-  Choose non-leaf better key for insertion
-*/
-
-uchar *rtree_choose_key(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key, 
-			uint key_length, uchar *page_buf, uint nod_flag)
-{
-  double increase;
-  double best_incr = DBL_MAX;
-  double area;
-  double best_area;
-  uchar *best_key;
-  uchar *k = rt_PAGE_FIRST_KEY(page_buf, nod_flag);
-  uchar *last = rt_PAGE_END(page_buf);
-
-  LINT_INIT(best_area);
-  LINT_INIT(best_key);
-
-  for (; k < last; k = rt_PAGE_NEXT_KEY(k, key_length, nod_flag))
-  {
-    if ((increase = rtree_area_increase(keyinfo->seg, key, k, key_length, 
-                                        &area)) == -1)
-      return NULL;
-    if (increase < best_incr)
-    {
-      best_key = k;
-      best_area = area;
-      best_incr = increase;
-    }
-    else
-    {
-      if ((increase == best_incr) && (area < best_area))
-      {
-        best_key = k;
-        best_area = area;
-        best_incr = increase;
-      }
-    }
-  }
-  return best_key;
-}

myisam/rt_key.h

@@ -26,6 +26,4 @@ int rtree_delete_key(MI_INFO *info, uchar *page, uchar *key,
                     uint key_length, uint nod_flag);
 int rtree_set_key_mbr(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key, 
                     uint key_length, my_off_t child_page);
-uchar *rtree_choose_key(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key, 
-                       uint key_length, uchar *page_buf, uint nod_flag);
 #endif /* _rt_key_h */

myisam/rt_mbr.c

@@ -563,9 +563,9 @@ Calculates MBR_AREA(a+b) - MBR_AREA(a)
 double rtree_area_increase(HA_KEYSEG *keyseg, uchar* a, uchar* b, 
                           uint key_length, double *ab_area)
 {
-  double a_area = 1;
+  double a_area= 1.0;
   
-  *ab_area = 1;
+  *ab_area= 1.0;
   for (; (int)key_length > 0; keyseg += 2)
   {
     key_length -= keyseg->length * 2;
@@ -627,6 +627,105 @@ double rtree_area_increase(HA_KEYSEG *keyseg, uchar* a, uchar* b,
   return *ab_area - a_area;
 }
 
+#define RT_PERIM_INC_KORR(type, korr_func, len) \
+{ \
+   type amin, amax, bmin, bmax; \
+   amin = korr_func(a); \
+   bmin = korr_func(b); \
+   p_inc(a, b, len); \
+   amax = korr_func(a); \
+   bmax = korr_func(b); \
+   p_inc(a, b, len); \
+   a_perim+= (((double)amax) - ((double)amin)); \
+   *ab_perim+= ((double)max(amax, bmax) - (double)min(amin, bmin)); \
+   break; \
+}
+
+#define RT_PERIM_INC_GET(type, get_func, len)\
+{\
+   type amin, amax, bmin, bmax; \
+   get_func(amin, a); \
+   get_func(bmin, b); \
+   p_inc(a, b, len); \
+   get_func(amax, a); \
+   get_func(bmax, b); \
+   p_inc(a, b, len); \
+   a_perim+= (((double)amax) - ((double)amin)); \
+   *ab_perim+= ((double)max(amax, bmax) - (double)min(amin, bmin)); \
+   break; \
+}
+
+/*
+Calculates MBR_PERIMETER(a+b) - MBR_PERIMETER(a)
+*/
+double rtree_perimeter_increase(HA_KEYSEG *keyseg, uchar* a, uchar* b, 
+				uint key_length, double *ab_perim)
+{
+  double a_perim = 0.0;
+  
+  *ab_perim= 0.0;
+  for (; (int)key_length > 0; keyseg += 2)
+  {
+    key_length -= keyseg->length * 2;
+    
+    /* Handle NULL part */
+    if (keyseg->null_bit)
+    {
+      return -1;
+    }
+
+    switch ((enum ha_base_keytype) keyseg->type) {
+    case HA_KEYTYPE_TEXT:
+    case HA_KEYTYPE_BINARY:
+    case HA_KEYTYPE_VARTEXT:
+    case HA_KEYTYPE_VARBINARY:
+    case HA_KEYTYPE_NUM:
+    default:
+      return 1;
+      break;
+    case HA_KEYTYPE_INT8:
+      {
+        int amin, amax, bmin, bmax;
+        amin = (int)*((signed char *)a);
+        bmin = (int)*((signed char *)b);
+        p_inc(a, b, 1);
+        amax = (int)*((signed char *)a);
+        bmax = (int)*((signed char *)b);
+        p_inc(a, b, 1);
+        a_perim+= (((double)amax) - ((double)amin));
+        *ab_perim+= ((double)max(amax, bmax) - (double)min(amin, bmin));
+        break;
+      }
+    case HA_KEYTYPE_SHORT_INT:
+      RT_PERIM_INC_KORR(int16, mi_sint2korr, 2);
+    case HA_KEYTYPE_USHORT_INT:
+      RT_PERIM_INC_KORR(uint16, mi_uint2korr, 2);
+    case HA_KEYTYPE_INT24:
+      RT_PERIM_INC_KORR(int32, mi_sint3korr, 3);
+    case HA_KEYTYPE_UINT24:
+      RT_PERIM_INC_KORR(int32, mi_uint3korr, 3);
+    case HA_KEYTYPE_LONG_INT:
+      RT_PERIM_INC_KORR(int32, mi_sint4korr, 4);
+    case HA_KEYTYPE_ULONG_INT:
+      RT_PERIM_INC_KORR(uint32, mi_uint4korr, 4);
+#ifdef HAVE_LONG_LONG
+    case HA_KEYTYPE_LONGLONG:
+      RT_PERIM_INC_KORR(longlong, mi_sint8korr, 8);
+    case HA_KEYTYPE_ULONGLONG:
+      RT_PERIM_INC_KORR(longlong, mi_sint8korr, 8);
+#endif
+    case HA_KEYTYPE_FLOAT:
+      RT_PERIM_INC_GET(float, mi_float4get, 4);
+    case HA_KEYTYPE_DOUBLE:
+      RT_PERIM_INC_GET(double, mi_float8get, 8);
+    case HA_KEYTYPE_END:
+      return *ab_perim - a_perim;
+    }
+  }
+  return *ab_perim - a_perim;
+}
+
+
 #define RT_PAGE_MBR_KORR(type, korr_func, store_func, len) \
 { \
   type amin, amax, bmin, bmax; \

myisam/rt_mbr.h

@@ -28,6 +28,8 @@ double rtree_overlapping_area(HA_KEYSEG *keyseg, uchar *a, uchar *b,
                               uint key_length);
 double rtree_area_increase(HA_KEYSEG *keyseg, uchar *a, uchar *b, 
                            uint key_length, double *ab_area);
+double rtree_perimeter_increase(HA_KEYSEG *keyseg, uchar* a, uchar* b, 
+				uint key_length, double *ab_perim);
 int rtree_page_mbr(MI_INFO *info, HA_KEYSEG *keyseg, uchar *page_buf, 
                    uchar* c, uint key_length);
 #endif /* _rt_mbr_h */

myisam/rt_split.c

@@ -265,12 +265,12 @@ int rtree_split_page(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *page, uchar *key,
 
   n_dim = keyinfo->keysegs / 2;
   
-  if (!my_multi_malloc(MYF(0),
-		       &coord_buf, n_dim * 2 * sizeof(double) * (max_keys + 1 + 4),
-		       &task, sizeof(SplitStruct) * (max_keys + 1),
-		       NullS))
+  if (!(coord_buf= my_alloca(n_dim * 2 * sizeof(double) * (max_keys + 1 + 4) +
+			     sizeof(SplitStruct) * (max_keys + 1))))
     return -1;
 
+  task= (SplitStruct *)(coord_buf + n_dim * 2 * (max_keys + 1 + 4));
+
   next_coord = coord_buf;
  
   stop = task + max_keys;
@@ -343,6 +343,6 @@ int rtree_split_page(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *page, uchar *key,
   my_afree((byte*)new_page);
 
 split_err:
-  my_free((gptr) coord_buf, MYF(0));
+  my_afree((byte*) coord_buf);
   return err_code;
 }

sql/spatial.cc

@@ -395,7 +395,7 @@ bool Gis_line_string::init_from_wkt(Gis_read_stream *trs, String *wkb)
     if (trs->skip_char(','))			// Didn't find ','
       break;
   }
-  if (n_points < 2)
+  if (n_points < 1)
   {
     trs->set_error_msg("Too few points in LINESTRING");
     return 1;
@@ -484,6 +484,11 @@ int Gis_line_string::is_closed(int *closed) const
   if (no_data(data, 4))
     return 1;
   n_points= uint4korr(data);
+  if (n_points == 1)
+  {
+    *closed=1;
+    return 0;
+  }
   data+= 4;
   if (no_data(data, SIZEOF_STORED_DOUBLE * 2 * n_points))
     return 1;