1 files changed, 157 insertions, 124 deletions

diff --git a/sql/table.cc b/sql/table.cc
index 2b1a85a74ea..302c5090925 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -54,6 +54,8 @@ static uint find_field(Field **fields, uchar *record, uint start, uint length);
 
 inline bool is_system_table_name(const char *name, uint length);
 
+static ulong get_form_pos(File file, uchar *head);
+
 /**************************************************************************
   Object_creation_ctx implementation.
 **************************************************************************/
@@ -306,13 +308,6 @@ TABLE_SHARE *alloc_table_share(TABLE_LIST *table_list, char *key,
     share->version=       refresh_version;
 
     /*
-      This constant is used to mark that no table map version has been
-      assigned.  No arithmetic is done on the value: it will be
-      overwritten with a value taken from MYSQL_BIN_LOG.
-    */
-    share->table_map_version= ~(ulonglong)0;
-
-    /*
       Since alloc_table_share() can be called without any locking (for
       example, ha_create_table... functions), we do not assign a table
       map id here.  Instead we assign a value that is not used
@@ -375,11 +370,6 @@ void init_tmp_table_share(THD *thd, TABLE_SHARE *share, const char *key,
   share->path.length= share->normalized_path.length= strlen(path);
   share->frm_version= 		 FRM_VER_TRUE_VARCHAR;
 
-  /*
-    Temporary tables are not replicated, but we set up these fields
-    anyway to be able to catch errors.
-   */
-  share->table_map_version= ~(ulonglong)0;
   share->cached_row_logging_check= -1;
 
   /*
@@ -708,15 +698,20 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   const char **interval_array;
   enum legacy_db_type legacy_db_type;
   my_bitmap_map *bitmaps;
+  bool null_bits_are_used;
   DBUG_ENTER("open_binary_frm");
 
+  LINT_INIT(options);
+  LINT_INIT(options_len);
+
   new_field_pack_flag= head[27];
   new_frm_ver= (head[2] - FRM_VER);
   field_pack_length= new_frm_ver < 2 ? 11 : 17;
   disk_buff= 0;
 
   error= 3;
-  if (!(pos=get_form_pos(file,head,(TYPELIB*) 0)))
+  /* Position of the form in the form file. */
+  if (!(pos= get_form_pos(file, head)))
     goto err;                                   /* purecov: inspected */
 
   share->frm_version= head[2];
@@ -885,7 +880,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
     /* Read extra data segment */
     uchar *next_chunk, *buff_end;
     DBUG_PRINT("info", ("extra segment size is %u bytes", n_length));
-    if (!(next_chunk= buff= (uchar*) my_malloc(n_length, MYF(MY_WME))))
+    if (!(next_chunk= buff= (uchar*) my_malloc(n_length+1, MYF(MY_WME))))
       goto err;
     if (my_pread(file, buff, n_length, record_offset + share->reclength,
                  MYF(MY_NABP)))
@@ -960,6 +955,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
       {
         /* purecov: begin inspected */
         error= 8;
+        name.str[name.length]= 0;
         my_error(ER_UNKNOWN_STORAGE_ENGINE, MYF(0), name.str);
         goto free_and_err;
         /* purecov: end */
@@ -1154,6 +1150,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
     goto free_and_err;
 
   record= share->default_values-1;              /* Fieldstart = 1 */
+  null_bits_are_used= share->null_fields != 0;
   if (share->null_field_first)
   {
     null_flags= null_pos= (uchar*) record+1;
@@ -1372,6 +1369,7 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
     reg_field->stored_in_db= fld_stored_in_db;
     if (field_type == MYSQL_TYPE_BIT && !f_bit_as_char(pack_flag))
     {
+      null_bits_are_used= 1;
       if ((null_bit_pos+= field_length & 7) > 7)
       {
         null_pos++;
@@ -1505,12 +1503,6 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
           keyinfo->extra_length+=HA_KEY_BLOB_LENGTH;
           key_part->store_length+=HA_KEY_BLOB_LENGTH;
           keyinfo->key_length+= HA_KEY_BLOB_LENGTH;
-          /*
-            Mark that there may be many matching values for one key
-            combination ('a', 'a ', 'a  '...)
-          */
-          if (!(field->flags & BINARY_FLAG))
-            keyinfo->flags|= HA_END_SPACE_KEY;
         }
         if (field->type() == MYSQL_TYPE_BIT)
           key_part->key_part_flag|= HA_BIT_PART;
@@ -1694,6 +1686,9 @@ static int open_binary_frm(THD *thd, TABLE_SHARE *share, uchar *head,
   share->null_bytes= (null_pos - (uchar*) null_flags +
                       (null_bit_pos + 7) / 8);
   share->last_null_bit_pos= null_bit_pos;
+  share->null_bytes_for_compare= null_bits_are_used ? share->null_bytes : 0;
+  share->can_cmp_whole_record= (share->blob_fields == 0 &&
+                                share->varchar_fields == 0);
 
   share->db_low_byte_first= handler_file->low_byte_first();
   share->column_bitmap_size= bitmap_buffer_size(share->fields);
@@ -1851,6 +1846,11 @@ bool fix_vcol_expr(THD *thd,
     goto end;
   }
   thd->where= save_where;
+  if (unlikely(func_expr->result_type() == ROW_RESULT))
+  {
+     my_error(ER_ROW_EXPR_FOR_VCOL, MYF(0));
+     goto end;
+  }
 #ifdef PARANOID
   /*
     Walk through the Item tree checking if all items are valid
@@ -1915,10 +1915,8 @@ end:
     parses it, building an item object for it. The pointer to this item is
     placed into in field->vcol_info.expr_item. After this the function performs
     semantic analysis of the item by calling the the function fix_vcol_expr.
-    Since the defining expression is part of the table definition the item
-    for it is created in table->memroot within a separate Query_arena.
-    The free_list of this arena is saved in field->vcol_info.item_free_list
-    to be freed when the table defition is removed from the TABLE_SHARE cache.
+    Since the defining expression is part of the table definition the item for
+    it is created in table->memroot within the special arena TABLE::expr_arena.
 
   @note
     Before passing 'vcol_expr" to the parser the function embraces it in 
@@ -1935,18 +1933,25 @@ bool unpack_vcol_info_from_frm(THD *thd,
                                LEX_STRING *vcol_expr,
                                bool *error_reported)
 {
-  bool rc= FALSE;
+  bool rc;
+  char *vcol_expr_str;
+  int str_len;
+  CHARSET_INFO *old_character_set_client;
+  Query_arena *backup_stmt_arena_ptr;
+  Query_arena backup_arena;
+  Query_arena *vcol_arena= 0;
+  Parser_state parser_state;
   DBUG_ENTER("unpack_vcol_info_from_frm");
   DBUG_ASSERT(vcol_expr);
 
+  old_character_set_client= thd->variables.character_set_client;
+  backup_stmt_arena_ptr= thd->stmt_arena;
+
   /* 
     Step 1: Construct the input string for the parser.
     The string to be parsed has to be of the following format:
     "PARSE_VCOL_EXPR (<expr_string_from_frm>)".
   */
-  char *vcol_expr_str;
-  int str_len= 0;
-  CHARSET_INFO *old_character_set_client;
   
   if (!(vcol_expr_str= (char*) alloc_root(&table->mem_root,
                                           vcol_expr->length + 
@@ -1968,19 +1973,27 @@ bool unpack_vcol_info_from_frm(THD *thd,
   str_len++;
   memcpy(vcol_expr_str + str_len, "\0", 1);
   str_len++;
-  Parser_state parser_state(thd, vcol_expr_str, str_len);
+
+  if (parser_state.init(thd, vcol_expr_str, str_len))
+    goto err;
 
   /* 
     Step 2: Setup thd for parsing.
   */
-  Query_arena *backup_stmt_arena_ptr= thd->stmt_arena;
-  Query_arena backup_arena;
-  Query_arena vcol_arena(&table->mem_root, Query_arena::INITIALIZED);
-  thd->set_n_backup_active_arena(&vcol_arena, &backup_arena);
-  thd->stmt_arena= &vcol_arena;
+  vcol_arena= table->expr_arena;
+  if (!vcol_arena)
+  {
+    Query_arena expr_arena(&table->mem_root, Query_arena::INITIALIZED);
+    if (!(vcol_arena= (Query_arena *) alloc_root(&table->mem_root,
+                                                 sizeof(Query_arena))))
+      goto err;
+    *vcol_arena= expr_arena;
+    table->expr_arena= vcol_arena;
+  }
+  thd->set_n_backup_active_arena(vcol_arena, &backup_arena);
+  thd->stmt_arena= vcol_arena;
 
   thd->lex->parse_vcol_expr= TRUE;
-  old_character_set_client= thd->variables.character_set_client;
 
   /* 
     Step 3: Use the parser to build an Item object from vcol_expr_str.
@@ -1999,7 +2012,7 @@ bool unpack_vcol_info_from_frm(THD *thd,
     field->vcol_info= 0;
     goto err;
   }
-  field->vcol_info->item_free_list= thd->free_list;
+  rc= FALSE;
   goto end;
 
 err:
@@ -2008,7 +2021,8 @@ err:
   thd->free_items();
 end:
   thd->stmt_arena= backup_stmt_arena_ptr;
-  thd->restore_active_arena(&vcol_arena, &backup_arena);
+  if (vcol_arena)
+    thd->restore_active_arena(vcol_arena, &backup_arena);
   thd->variables.character_set_client= old_character_set_client;
 
   DBUG_RETURN(rc);
@@ -2323,9 +2337,9 @@ partititon_err:
               (my_bitmap_map*) bitmaps, share->fields, FALSE);
   bitmap_init(&outparam->def_write_set,
               (my_bitmap_map*) (bitmaps+bitmap_size), share->fields, FALSE);
-  bitmap_init(&outparam->tmp_set,
+  bitmap_init(&outparam->def_vcol_set,
               (my_bitmap_map*) (bitmaps+bitmap_size*2), share->fields, FALSE);
-  bitmap_init(&outparam->vcol_set,
+  bitmap_init(&outparam->tmp_set,
               (my_bitmap_map*) (bitmaps+bitmap_size*3), share->fields, FALSE);
   outparam->default_column_bitmaps();
 
@@ -2431,12 +2445,12 @@ int closefrm(register TABLE *table, bool free_share)
   }
   my_free((char*) table->alias, MYF(MY_ALLOW_ZERO_PTR));
   table->alias= 0;
+  if (table->expr_arena)
+    table->expr_arena->free_items();
   if (table->field)
   {
     for (Field **ptr=table->field ; *ptr ; ptr++)
     {
-      if ((*ptr)->vcol_info)
-        free_items((*ptr)->vcol_info->item_free_list);
       delete *ptr;
     }
     table->field= 0;
@@ -2497,55 +2511,46 @@ void free_field_buffers_larger_than(TABLE *table, uint32 size)
   }
 }
 
-	/* Find where a form starts */
-	/* if formname is NullS then only formnames is read */
+/**
+  Find where a form starts.
+
+  @param head The start of the form file.
+
+  @remark If formname is NULL then only formnames is read.
+
+  @retval The form position.
+*/
 
-ulong get_form_pos(File file, uchar *head, TYPELIB *save_names)
+static ulong get_form_pos(File file, uchar *head)
 {
-  uint a_length,names,length;
-  uchar *pos,*buf;
+  uchar *pos, *buf;
+  uint names, length;
   ulong ret_value=0;
   DBUG_ENTER("get_form_pos");
 
-  LINT_INIT(buf);
+  names= uint2korr(head+8);
 
-  names=uint2korr(head+8);
-  a_length=(names+2)*sizeof(char *);		/* Room for two extra */
+  if (!(names= uint2korr(head+8)))
+    DBUG_RETURN(0);
 
-  if (!save_names)
-    a_length=0;
-  else
-    save_names->type_names=0;			/* Clear if error */
-
-  if (names)
-  {
-    length=uint2korr(head+4);
-    VOID(my_seek(file,64L,MY_SEEK_SET,MYF(0)));
-    if (!(buf= (uchar*) my_malloc((size_t) length+a_length+names*4,
-				  MYF(MY_WME))) ||
-	my_read(file, buf+a_length, (size_t) (length+names*4),
-		MYF(MY_NABP)))
-    {						/* purecov: inspected */
-      x_free((uchar*) buf);			/* purecov: inspected */
-      DBUG_RETURN(0L);				/* purecov: inspected */
-    }
-    pos= buf+a_length+length;
-    ret_value=uint4korr(pos);
-  }
-  if (! save_names)
-  {
-    if (names)
-      my_free((uchar*) buf,MYF(0));
-  }
-  else if (!names)
-    bzero((char*) save_names,sizeof(save_names));
-  else
+  length= uint2korr(head+4);
+
+  my_seek(file, 64L, MY_SEEK_SET, MYF(0));
+
+  if (!(buf= (uchar*) my_malloc(length+names*4, MYF(MY_WME))))
+    DBUG_RETURN(0);
+
+  if (my_read(file, buf, length+names*4, MYF(MY_NABP)))
   {
-    char *str;
-    const char **tmp = (const char**) buf;
-    str=(char *) (buf+a_length);
-    fix_type_pointers(&tmp, save_names, 1, &str);
+    x_free(buf);
+    DBUG_RETURN(0);
   }
+
+  pos= buf+length;
+  ret_value= uint4korr(pos);
+
+  my_free(buf, MYF(0));
+
   DBUG_RETURN(ret_value);
 }
 
@@ -3143,6 +3148,13 @@ bool check_db_name(LEX_STRING *org_name)
 {
   char *name= org_name->str;
   uint name_length= org_name->length;
+  bool check_for_path_chars;
+
+  if ((check_for_path_chars= check_mysql50_prefix(name)))
+  {
+    name+= MYSQL50_TABLE_NAME_PREFIX_LENGTH;
+    name_length-= MYSQL50_TABLE_NAME_PREFIX_LENGTH;
+  }
 
   if (!name_length || name_length > NAME_LEN)
     return 1;
@@ -3150,41 +3162,29 @@ bool check_db_name(LEX_STRING *org_name)
   if (lower_case_table_names && name != any_db)
     my_casedn_str(files_charset_info, name);
 
-#if defined(USE_MB) && defined(USE_MB_IDENT)
-  if (use_mb(system_charset_info))
-  {
-    name_length= 0;
-    bool last_char_is_space= TRUE;
-    char *end= name + org_name->length;
-    while (name < end)
-    {
-      int len;
-      last_char_is_space= my_isspace(system_charset_info, *name);
-      len= my_ismbchar(system_charset_info, name, end);
-      if (!len)
-        len= 1;
-      name+= len;
-      name_length++;
-    }
-    return (last_char_is_space || name_length > NAME_CHAR_LEN);
-  }
-  else
-#endif
-    return ((org_name->str[org_name->length - 1] != ' ') ||
-            (name_length > NAME_CHAR_LEN)); /* purecov: inspected */
+  return check_table_name(name, name_length, check_for_path_chars);
 }
 
+
 /*
   Allow anything as a table name, as long as it doesn't contain an
   ' ' at the end
   returns 1 on error
 */
 
-
 bool check_table_name(const char *name, uint length, bool check_for_path_chars)
 {
   uint name_length= 0;  // name length in symbols
   const char *end= name+length;
+
+
+  if (!check_for_path_chars &&
+      (check_for_path_chars= check_mysql50_prefix(name)))
+  {
+    name+= MYSQL50_TABLE_NAME_PREFIX_LENGTH;
+    length-= MYSQL50_TABLE_NAME_PREFIX_LENGTH;
+  }
+
   if (!length || length > NAME_LEN)
     return 1;
 #if defined(USE_MB) && defined(USE_MB_IDENT)
@@ -3208,10 +3208,10 @@ bool check_table_name(const char *name, uint length, bool check_for_path_chars)
         continue;
       }
     }
+#endif
     if (check_for_path_chars &&
         (*name == '/' || *name == '\\' || *name == '~' || *name == FN_EXTCHAR))
       return 1;
-#endif
     name++;
     name_length++;
   }
@@ -4789,10 +4789,10 @@ void st_table::clear_column_bitmaps()
     Reset column read/write usage. It's identical to:
     bitmap_clear_all(&table->def_read_set);
     bitmap_clear_all(&table->def_write_set);
+    bitmap_clear_all(&table->def_vcol_set);
   */
-  bzero((char*) def_read_set.bitmap, s->column_bitmap_size*2);
-  bzero((char*) def_read_set.bitmap, s->column_bitmap_size*4);
-  column_bitmaps_set(&def_read_set, &def_write_set);
+  bzero((char*) def_read_set.bitmap, s->column_bitmap_size*3);
+  column_bitmaps_set(&def_read_set, &def_write_set, &def_vcol_set);
 }
 
 
@@ -4844,6 +4844,27 @@ void st_table::mark_columns_used_by_index(uint index)
 
 
 /*
+  Add fields used by a specified index to the table's read_set.
+
+  NOTE:
+    The original state can be restored with
+    restore_column_maps_after_mark_index().
+*/
+
+void st_table::add_read_columns_used_by_index(uint index)
+{
+  MY_BITMAP *bitmap= &tmp_set;
+  DBUG_ENTER("st_table::add_read_columns_used_by_index");
+
+  enable_keyread();
+  bitmap_copy(bitmap, read_set);
+  mark_columns_used_by_index_no_reset(index, bitmap);
+  column_bitmaps_set(bitmap, write_set);
+  DBUG_VOID_RETURN;
+}
+
+
+/*
   Restore to use normal column maps after key read
 
   NOTES
@@ -5011,7 +5032,7 @@ void st_table::mark_columns_needed_for_update()
     }
   }
   /* Mark all virtual columns needed for update */
-  mark_virtual_columns_for_write();
+  mark_virtual_columns_for_write(FALSE);
   DBUG_VOID_RETURN;
 }
 
@@ -5039,7 +5060,7 @@ void st_table::mark_columns_needed_for_insert()
   if (found_next_number_field)
     mark_auto_increment_column();
   /* Mark virtual columns for insert */
-  mark_virtual_columns_for_write();
+  mark_virtual_columns_for_write(TRUE);
 }
 
 
@@ -5065,7 +5086,7 @@ bool st_table::mark_virtual_col(Field *field)
 {
   bool res;
   DBUG_ASSERT(field->vcol_info);
-  if (!(res= bitmap_fast_test_and_set(&vcol_set, field->field_index)))
+  if (!(res= bitmap_fast_test_and_set(vcol_set, field->field_index)))
   {
     Item *vcol_item= field->vcol_info->expr_item;
     DBUG_ASSERT(vcol_item);
@@ -5077,10 +5098,14 @@ bool st_table::mark_virtual_col(Field *field)
 
 /* 
   @brief Mark virtual columns for update/insert commands
+    
+  @param insert_fl    <-> virtual columns are marked for insert command 
 
   @details
     The function marks virtual columns used in a update/insert commands
     in the vcol_set bitmap.
+    For an insert command a virtual column is always marked in write_set if
+    it is a stored column.
     If a virtual column is from  write_set it is always marked in vcol_set.
     If a stored virtual column is not from write_set but it is computed
     through columns from write_set it is also marked in vcol_set, and,
@@ -5099,7 +5124,7 @@ bool st_table::mark_virtual_col(Field *field)
     be added to read_set either.           
 */
 
-void st_table::mark_virtual_columns_for_write(void)
+void st_table::mark_virtual_columns_for_write(bool insert_fl)
 {
   Field **vfield_ptr, *tmp_vfield;
   bool bitmap_updated= FALSE;
@@ -5111,16 +5136,21 @@ void st_table::mark_virtual_columns_for_write(void)
       bitmap_updated= mark_virtual_col(tmp_vfield);
     else if (tmp_vfield->stored_in_db)
     {
-      MY_BITMAP *save_read_set;
-      Item *vcol_item= tmp_vfield->vcol_info->expr_item;
-      DBUG_ASSERT(vcol_item);
-      bitmap_clear_all(&tmp_set);
-      save_read_set= read_set;
-      read_set= &tmp_set;
-      vcol_item->walk(&Item::register_field_in_read_map, 1, (uchar *) 0);
-      read_set= save_read_set;
-      bitmap_intersect(&tmp_set, write_set);
-      if (!bitmap_is_clear_all(&tmp_set))
+      bool mark_fl= insert_fl;
+      if (!mark_fl)
+      {
+        MY_BITMAP *save_read_set;
+        Item *vcol_item= tmp_vfield->vcol_info->expr_item;
+        DBUG_ASSERT(vcol_item);
+        bitmap_clear_all(&tmp_set);
+        save_read_set= read_set;
+        read_set= &tmp_set;
+        vcol_item->walk(&Item::register_field_in_read_map, 1, (uchar *) 0);
+        read_set= save_read_set;
+        bitmap_intersect(&tmp_set, write_set);
+        mark_fl= !bitmap_is_clear_all(&tmp_set);
+      }
+      if (mark_fl)
       {
         bitmap_set_bit(write_set, tmp_vfield->field_index);
         mark_virtual_col(tmp_vfield);
@@ -5529,6 +5559,7 @@ size_t max_row_length(TABLE *table, const uchar *data)
 /*
   @brief Compute values for virtual columns used in query
 
+  @param  thd              Thread handle
   @param  table            The TABLE object
   @param  for_write        Requests to compute only fields needed for write   
   
@@ -5545,7 +5576,7 @@ size_t max_row_length(TABLE *table, const uchar *data)
     >0   Error occurred when storing a virtual field value
 */
 
-int update_virtual_fields(TABLE *table, bool for_write)
+int update_virtual_fields(THD *thd, TABLE *table, bool for_write)
 {
   DBUG_ENTER("update_virtual_fields");
   Field **vfield_ptr, *vfield;
@@ -5553,13 +5584,14 @@ int update_virtual_fields(TABLE *table, bool for_write)
   if (!table || !table->vfield)
     DBUG_RETURN(0);
 
+  thd->reset_arena_for_cached_items(table->expr_arena);
   /* Iterate over virtual fields in the table */
   for (vfield_ptr= table->vfield; *vfield_ptr; vfield_ptr++)
   {
     vfield= (*vfield_ptr);
     DBUG_ASSERT(vfield->vcol_info && vfield->vcol_info->expr_item);
     /* Only update those fields that are marked in the vcol_set bitmap */
-    if (bitmap_is_set(&table->vcol_set, vfield->field_index) &&
+    if (bitmap_is_set(table->vcol_set, vfield->field_index) &&
         (for_write || !vfield->stored_in_db))
     {
       /* Compute the actual value of the virtual fields */
@@ -5571,6 +5603,7 @@ int update_virtual_fields(TABLE *table, bool for_write)
       DBUG_PRINT("info", ("field '%s' - skipped", vfield->field_name));
     }
   }
+  thd->reset_arena_for_cached_items(0);
   DBUG_RETURN(0);
 }