Bitmap post-review fixes

include/my_bitmap.h

@@ -38,23 +38,23 @@ typedef struct st_bitmap
 #ifdef	__cplusplus
 extern "C" {
 #endif
-extern my_bool bitmap_cmp(MY_BITMAP *map1, MY_BITMAP *map2);
+extern my_bool bitmap_cmp(const MY_BITMAP *map1, const MY_BITMAP *map2);
 extern my_bool bitmap_init(MY_BITMAP *map, uchar *buf, uint bitmap_size, my_bool thread_safe);
-extern my_bool bitmap_is_clear_all(MY_BITMAP *map);
-extern my_bool bitmap_is_prefix(MY_BITMAP *map, uint prefix_size);
-extern my_bool bitmap_is_set(MY_BITMAP *map, uint bitmap_bit);
-extern my_bool bitmap_is_set_all(MY_BITMAP *map);
-extern my_bool bitmap_is_subset(MY_BITMAP *map1, MY_BITMAP *map2);
+extern my_bool bitmap_is_clear_all(const MY_BITMAP *map);
+extern my_bool bitmap_is_prefix(const MY_BITMAP *map, uint prefix_size);
+extern my_bool bitmap_is_set(const MY_BITMAP *map, uint bitmap_bit);
+extern my_bool bitmap_is_set_all(const MY_BITMAP *map);
+extern my_bool bitmap_is_subset(const MY_BITMAP *map1, const MY_BITMAP *map2);
 extern uint bitmap_set_next(MY_BITMAP *map);
 extern void bitmap_clear_all(MY_BITMAP *map);
 extern void bitmap_clear_bit(MY_BITMAP *map, uint bitmap_bit);
 extern void bitmap_free(MY_BITMAP *map);
-extern void bitmap_intersect(MY_BITMAP *map, MY_BITMAP *map2);
+extern void bitmap_intersect(MY_BITMAP *map, const MY_BITMAP *map2);
 extern void bitmap_set_all(MY_BITMAP *map);
 extern void bitmap_set_bit(MY_BITMAP *map, uint bitmap_bit);
 extern void bitmap_set_prefix(MY_BITMAP *map, uint prefix_size);
-extern void bitmap_subtract(MY_BITMAP *map, MY_BITMAP *map2);
-extern void bitmap_union(MY_BITMAP *map, MY_BITMAP *map2);
+extern void bitmap_subtract(MY_BITMAP *map, const MY_BITMAP *map2);
+extern void bitmap_union(MY_BITMAP *map, const MY_BITMAP *map2);
 #ifdef	__cplusplus
 }
 #endif

mysys/my_bitmap.c

@@ -16,7 +16,16 @@
 
 /*
   Handling of uchar arrays as large bitmaps.
-  We assume that the size of the used bitmap is less than ~(uint) 0
+
+  API limitations (or, rather asserted safety assumptions,
+  to encourage correct programming)
+
+    * the size of the used bitmap is less than ~(uint) 0
+    * it's a multiple of 8 (for efficiency reasons)
+    * when arguments are a bitmap and a bit number, the number
+      must be within bitmap size
+    * bitmap_set_prefix() is an exception - one can use ~0 to set all bits
+    * when both arguments are bitmaps, they must be of the same size
 
   TODO:
   Make assembler THREAD safe versions of these using test-and-set instructions
@@ -45,11 +54,11 @@ inline void bitmap_unlock(MY_BITMAP *map)
 
 my_bool bitmap_init(MY_BITMAP *map, uchar *buf, uint bitmap_size, my_bool thread_safe)
 {
-  // for efficiency reasons - MY_BITMAP is heavily used
   DBUG_ASSERT((bitmap_size & 7) == 0);
   bitmap_size/=8;
   if (!(map->bitmap=buf) &&
-       !(map->bitmap=(uchar*)my_malloc(bitmap_size + sizeof(pthread_mutex_t),
+       !(map->bitmap=(uchar*)my_malloc(bitmap_size +
+                                       (thread_safe ? sizeof(pthread_mutex_t) : 0),
                                        MYF(MY_WME | MY_ZEROFILL))))
     return 1;
   map->bitmap_size=bitmap_size;
@@ -128,7 +137,8 @@ void bitmap_set_prefix(MY_BITMAP *map, uint prefix_size)
 {
   uint prefix_bytes, prefix_bits;
 
-  DBUG_ASSERT(map->bitmap);
+  DBUG_ASSERT(map->bitmap &&
+      (prefix_size <= map->bitmap_size*8 || prefix_size == ~0));
   bitmap_lock(map);
   set_if_smaller(prefix_size, map->bitmap_size*8);
   if ((prefix_bytes= prefix_size / 8))
@@ -150,7 +160,7 @@ void bitmap_set_all(MY_BITMAP *map)
   bitmap_set_prefix(map, ~0);
 }
 
-my_bool bitmap_is_prefix(MY_BITMAP *map, uint prefix_size)
+my_bool bitmap_is_prefix(const MY_BITMAP *map, uint prefix_size)
 {
   uint prefix_bits= prefix_size & 7, res= 0;
   uchar *m= map->bitmap, *end_prefix= map->bitmap+prefix_size/8,
@@ -167,7 +177,7 @@ my_bool bitmap_is_prefix(MY_BITMAP *map, uint prefix_size)
     goto ret;
 
   while (m < end)
-    if (m++ != 0)
+    if (*m++ != 0)
       goto ret;
 
   res=1;
@@ -176,23 +186,23 @@ ret:
   return res;
 }
 
-my_bool bitmap_is_clear_all(MY_BITMAP *map)
+my_bool bitmap_is_clear_all(const MY_BITMAP *map)
 {
   return bitmap_is_prefix(map, 0);
 }
 
-my_bool bitmap_is_set_all(MY_BITMAP *map)
+my_bool bitmap_is_set_all(const MY_BITMAP *map)
 {
   return bitmap_is_prefix(map, map->bitmap_size*8);
 }
 
-my_bool bitmap_is_set(MY_BITMAP *map, uint bitmap_bit)
+my_bool bitmap_is_set(const MY_BITMAP *map, uint bitmap_bit)
 {
   DBUG_ASSERT(map->bitmap && bitmap_bit < map->bitmap_size*8);
   return map->bitmap[bitmap_bit / 8] & (1 << (bitmap_bit & 7));
 }
 
-my_bool bitmap_is_subset(MY_BITMAP *map1, MY_BITMAP *map2)
+my_bool bitmap_is_subset(const MY_BITMAP *map1, const MY_BITMAP *map2)
 {
   uint length, res=0;
   uchar *m1=map1->bitmap, *m2=map2->bitmap, *end;
@@ -215,7 +225,7 @@ ret:
   return res;
 }
 
-my_bool bitmap_cmp(MY_BITMAP *map1, MY_BITMAP *map2)
+my_bool bitmap_cmp(const MY_BITMAP *map1, const MY_BITMAP *map2)
 {
   uint res;
 
@@ -231,7 +241,7 @@ my_bool bitmap_cmp(MY_BITMAP *map1, MY_BITMAP *map2)
   return res;
 }
 
-void bitmap_intersect(MY_BITMAP *map, MY_BITMAP *map2)
+void bitmap_intersect(MY_BITMAP *map, const MY_BITMAP *map2)
 {
   uchar *to=map->bitmap, *from=map2->bitmap, *end;
 
@@ -249,7 +259,7 @@ void bitmap_intersect(MY_BITMAP *map, MY_BITMAP *map2)
   bitmap_unlock(map);
 }
 
-void bitmap_subtract(MY_BITMAP *map, MY_BITMAP *map2)
+void bitmap_subtract(MY_BITMAP *map, const MY_BITMAP *map2)
 {
   uchar *to=map->bitmap, *from=map2->bitmap, *end;
 
@@ -267,7 +277,7 @@ void bitmap_subtract(MY_BITMAP *map, MY_BITMAP *map2)
   bitmap_unlock(map);
 }
 
-void bitmap_union(MY_BITMAP *map, MY_BITMAP *map2)
+void bitmap_union(MY_BITMAP *map, const MY_BITMAP *map2)
 {
   uchar *to=map->bitmap, *from=map2->bitmap, *end;
 

sql/ha_berkeley.cc

@@ -844,7 +844,7 @@ int ha_berkeley::write_row(byte * record)
     ulong thd_options = table->in_use ? table->in_use->options : 0;
     for (uint retry=0 ; retry < berkeley_trans_retry ; retry++)
     {
-      key_map changed_keys;
+      key_map changed_keys(0);
       if (using_ignore && (thd_options & OPTION_INTERNAL_SUBTRANSACTIONS))
       {
 	if ((error=txn_begin(db_env, transaction, &sub_trans, 0))) /* purecov: deadcode */
@@ -888,7 +888,8 @@ int ha_berkeley::write_row(byte * record)
 	  else if (!changed_keys.is_clear_all())
 	  {
 	    new_error = 0;
-	    for (uint keynr=0; keynr < changed_keys.length(); keynr++)
+	    for (uint keynr=0 ; keynr < table->keys+test(hidden_primary_key) ;
+                 keynr++)
 	    {
 	      if (changed_keys.is_set(keynr))
 	      {
@@ -1012,7 +1013,7 @@ int ha_berkeley::update_primary_key(DB_TXN *trans, bool primary_key_changed,
   Clobbers keybuff2
 */
 
-int ha_berkeley::restore_keys(DB_TXN *trans, key_map changed_keys,
+int ha_berkeley::restore_keys(DB_TXN *trans, key_map *changed_keys,
 			      uint primary_key,
 			      const byte *old_row, DBT *old_key,
 			      const byte *new_row, DBT *new_key,
@@ -1034,14 +1035,13 @@ int ha_berkeley::restore_keys(DB_TXN *trans, key_map changed_keys,
      rolled back.  The last key set in changed_keys is the one that
      triggered the duplicate key error (it wasn't inserted), so for
      that one just put back the old value. */
-  if (!changed_keys.is_clear_all())
+  if (!changed_keys->is_clear_all())
   {
-    key_map map1(1);
-    for (keynr=0; keynr < changed_keys.length(); keynr++)
+    for (keynr=0 ; keynr < table->keys+test(hidden_primary_key) ; keynr++)
     {
-      if (changed_keys.is_set(keynr))
+      if (changed_keys->is_set(keynr))
       {
-        if (changed_keys.is_subset(map1) &&
+        if (changed_keys->is_prefix(1) &&
             (error = remove_key(trans, keynr, new_row, new_key)))
           break; /* purecov: inspected */
         if ((error = key_file[keynr]->put(key_file[keynr], trans,
@@ -1094,7 +1094,7 @@ int ha_berkeley::update_row(const byte * old_row, byte * new_row)
   sub_trans = transaction;
   for (uint retry=0 ; retry < berkeley_trans_retry ; retry++)
   {
-    key_map changed_keys;
+    key_map changed_keys(0);
     if (using_ignore &&	(thd_options & OPTION_INTERNAL_SUBTRANSACTIONS))
     {
       if ((error=txn_begin(db_env, transaction, &sub_trans, 0))) /* purecov: deadcode */
@@ -1152,7 +1152,7 @@ int ha_berkeley::update_row(const byte * old_row, byte * new_row)
 	  new_error=txn_abort(sub_trans); /* purecov: deadcode */
 	}
 	else if (!changed_keys.is_clear_all())
-	  new_error=restore_keys(transaction, changed_keys, primary_key,
+	  new_error=restore_keys(transaction, &changed_keys, primary_key,
 				 old_row, &old_prim_key, new_row, &prim_key,
 				 thd_options);
 	if (new_error)
@@ -1233,12 +1233,12 @@ int ha_berkeley::remove_key(DB_TXN *trans, uint keynr, const byte *record,
 /* Delete all keys for new_record */
 
 int ha_berkeley::remove_keys(DB_TXN *trans, const byte *record,
-			     DBT *new_record, DBT *prim_key, key_map keys)
+			     DBT *new_record, DBT *prim_key, key_map *keys)
 {
   int result = 0;
-  for (uint keynr=0; keynr < keys.length(); keynr++)
+  for (uint keynr=0 ; keynr < table->keys+test(hidden_primary_key) ; keynr++)
   {
-    if (keys.is_set(keynr))
+    if (keys->is_set(keynr))
     {
       int new_error=remove_key(trans, keynr, record, prim_key);
       if (new_error)
@@ -1278,7 +1278,7 @@ int ha_berkeley::delete_row(const byte * record)
 	break; /* purecov: deadcode */
       DBUG_PRINT("trans",("starting sub transaction")); /* purecov: deadcode */
     }
-    error=remove_keys(sub_trans, record, &row, &prim_key, keys);
+    error=remove_keys(sub_trans, record, &row, &prim_key, &keys);
     if (!error && (thd_options & OPTION_INTERNAL_SUBTRANSACTIONS))
     {
       DBUG_PRINT("trans",("ending sub transaction")); /* purecov: deadcode */

sql/ha_berkeley.h

@@ -72,8 +72,8 @@ class ha_berkeley: public handler
 		uint key_length);
   int remove_key(DB_TXN *trans, uint keynr, const byte *record, DBT *prim_key);
   int remove_keys(DB_TXN *trans,const byte *record, DBT *new_record,
-		  DBT *prim_key, key_map keys);
-  int restore_keys(DB_TXN *trans, key_map changed_keys, uint primary_key,
+		  DBT *prim_key, key_map *keys);
+  int restore_keys(DB_TXN *trans, key_map *changed_keys, uint primary_key,
 		   const byte *old_row, DBT *old_key,
 		   const byte *new_row, DBT *new_key,
 		   ulong thd_options);

sql/item_func.h

@@ -981,8 +981,8 @@ public:
   String value;              // value of concat
   String search_value;       // key_item()'s value converted to cmp_collation
 
-  Item_func_match(List<Item> &a, uint b): Item_real_func(a), flags(b),
-       table(0), master(0), ft_handler(0), concat(0), key(0), join_key(0) { }
+  Item_func_match(List<Item> &a, uint b): Item_real_func(a), key(0), flags(b),
+       join_key(0), ft_handler(0), table(0), master(0), concat(0) { }
   ~Item_func_match()
   {
     if (!master && ft_handler)

sql/opt_range.cc

@@ -567,19 +567,29 @@ SEL_ARG *SEL_ARG::clone_tree()
   return root;
 }
 
-/*****************************************************************************
-**	Test if a key can be used in different ranges
-**	Returns:
-**	-1 if impossible select
-**	0 if can't use quick_select
-**	1 if found usable range
-**	Updates the following in the select parameter:
-**	needed_reg ; Bits for keys with may be used if all prev regs are read
-**	quick	   ; Parameter to use when reading records.
-**	In the table struct the following information is updated:
-**	quick_keys ; Which keys can be used
-**	quick_rows ; How many rows the key matches
-*****************************************************************************/
+/*
+  Test if a key can be used in different ranges
+
+  SYNOPSIS
+   SQL_SELECT::test_quick_select(thd,keys_to_use, prev_tables,
+                                 limit, force_quick_range)
+
+   Updates the following in the select parameter:
+    needed_reg - Bits for keys with may be used if all prev regs are read
+    quick      - Parameter to use when reading records.
+   In the table struct the following information is updated:
+    quick_keys - Which keys can be used
+    quick_rows - How many rows the key matches
+
+ RETURN VALUES
+  -1 if impossible select
+   0 if can't use quick_select
+   1 if found usable range
+
+ TODO
+   check if the function really needs to modify keys_to_use, and change the
+   code to pass it by reference if not
+*/
 
 int SQL_SELECT::test_quick_select(THD *thd, key_map keys_to_use,
 				  table_map prev_tables,

sql/sql_bitmap.h

@@ -16,6 +16,7 @@ template <uint default_width> class Bitmap
   uchar buffer[(default_width+7)/8];
 public:
   Bitmap() { init(); }
+  Bitmap(Bitmap& from) { *this=from; }
   Bitmap(uint prefix_to_set) { init(prefix_to_set); }
   void init() { bitmap_init(&map, buffer, default_width, 0); }
   void init(uint prefix_to_set) { init(); set_prefix(prefix_to_set); }
@@ -24,6 +25,7 @@ public:
   {
     init();
     memcpy(buffer, map2.buffer, sizeof(buffer));
+    return *this;
   }
   void set_bit(uint n) { bitmap_set_bit(&map, n); }
   void clear_bit(uint n) { bitmap_clear_bit(&map, n); }

sql/sql_select.cc

@@ -1579,7 +1579,7 @@ err:
 
 static ha_rows get_quick_record_count(THD *thd, SQL_SELECT *select,
 				      TABLE *table,
-				      const key_map& keys,ha_rows limit)
+				      const key_map *keys,ha_rows limit)
 {
   int error;
   DBUG_ENTER("get_quick_record_count");
@@ -1587,7 +1587,7 @@ static ha_rows get_quick_record_count(THD *thd, SQL_SELECT *select,
   {
     select->head=table;
     table->reginfo.impossible_range=0;
-    if ((error=select->test_quick_select(thd, keys,(table_map) 0,limit))
+    if ((error=select->test_quick_select(thd, *(key_map *)keys,(table_map) 0,limit))
 	== 1)
       DBUG_RETURN(select->quick->records);
     if (error == -1)
@@ -1876,7 +1876,7 @@ make_join_statistics(JOIN *join,TABLE_LIST *tables,COND *conds,
 			  s->on_expr ? s->on_expr : conds,
 			  &error);
       records= get_quick_record_count(join->thd, select, s->table,
-				      s->const_keys, join->row_limit);
+				      &s->const_keys, join->row_limit);
       s->quick=select->quick;
       s->needed_reg=select->needed_reg;
       select->quick=0;
@@ -2104,8 +2104,7 @@ add_key_field(KEY_FIELD **key_fields,uint and_level,
     else
     {
       JOIN_TAB *stat=field->table->reginfo.join_tab;
-      key_map possible_keys;
-      possible_keys=field->key_start;
+      key_map possible_keys=field->key_start;
       possible_keys.intersect(field->table->keys_in_use_for_query);
       stat[0].keys.merge(possible_keys);             // Add possible keys
 
@@ -8777,7 +8776,7 @@ static void select_describe(JOIN *join, bool need_tmp_table, bool need_order,
       uint j;
       if (!tab->keys.is_clear_all())
       {
-        for (j=0 ; j < tab->keys.length() ; j++)
+        for (j=0 ; j < table->keys ; j++)
         {
           if (tab->keys.is_set(j))
           {

sql/table.cc

@@ -156,9 +156,9 @@ int openfrm(const char *name, const char *alias, uint db_stat, uint prgflag,
   VOID(my_seek(file,(ulong) uint2korr(head+6),MY_SEEK_SET,MYF(0)));
   if (read_string(file,(gptr*) &disk_buff,key_info_length))
     goto err_not_open; /* purecov: inspected */
-  if (disk_buff[1] & 0x80)
+  if (disk_buff[0] & 0x80)
   {
-    outparam->keys=      keys=      uint2korr(disk_buff) & 0x7fff;
+    outparam->keys=      keys=      (disk_buff[1] << 7) | (disk_buff[0] & 0x7f);
     outparam->key_parts= key_parts= uint2korr(disk_buff+2);
   }
   else
@@ -279,7 +279,7 @@ int openfrm(const char *name, const char *alias, uint db_stat, uint prgflag,
   if (my_pread(file,(byte*) record,(uint) outparam->reclength,
 	       (ulong) (uint2korr(head+6)+
                         ((uint2korr(head+14) == 0xffff ?
-                            uint4korr(head+10) : uint2korr(head+14)))),
+                            uint4korr(head+47) : uint2korr(head+14)))),
 	       MYF(MY_NABP)))
     goto err_not_open; /* purecov: inspected */
   /* HACK: table->record[2] is used instead of table->default_values here */

sql/unireg.cc

@@ -100,7 +100,8 @@ int rea_create_table(THD *thd, my_string file_name,
     goto err;
   maxlength=(uint) next_io_size((ulong) (uint2korr(forminfo)+1000));
   int2store(forminfo+2,maxlength);
-  int4store(fileinfo+10,key_buff_length);
+  int4store(fileinfo+10,(ulong) (filepos+maxlength));
+  int4store(fileinfo+47,key_buff_length);
   fileinfo[26]= (uchar) test((create_info->max_rows == 1) &&
 			     (create_info->min_rows == 1) && (keys == 0));
   int2store(fileinfo+28,key_info_length);
@@ -293,8 +294,8 @@ static uint pack_keys(uchar *keybuff,uint key_count,KEY *keyinfo)
 
   if (key_count > 127 || key_parts > 127)
   {
-    key_count|=0x8000;
-    int2store(keybuff,key_count);
+    keybuff[0]= (key_count & 0x7f) | 0x80;
+    keybuff[1]= key_count >> 7;
     int2store(keybuff+2,key_parts);
   }
   else