~drizzle-trunk/drizzle/development

/* Copyright (C) 2000-2006 MySQL AB

   This program is free software; you can redistribute it and/or modify

   it under the terms of the GNU General Public License as published by

   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,

   but WITHOUT ANY WARRANTY; without even the implied warranty of

   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License

   along with this program; if not, write to the Free Software

   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/* Some general useful functions */

#include "config.h"

#include <float.h>

#include <fcntl.h>

#include <string>

#include <vector>

#include <algorithm>

#include <drizzled/error.h>

#include <drizzled/gettext.h>

#include "drizzled/plugin/transactional_storage_engine.h"

#include "drizzled/plugin/authorization.h"

#include <drizzled/nested_join.h>

#include <drizzled/sql_parse.h>

#include <drizzled/item/sum.h>

#include <drizzled/table_list.h>

#include <drizzled/session.h>

#include <drizzled/sql_base.h>

#include <drizzled/sql_select.h>

#include <drizzled/field/blob.h>

#include <drizzled/field/varstring.h>

#include <drizzled/field/double.h>

#include <drizzled/unireg.h>

#include <drizzled/message/table.pb.h>

#include "drizzled/sql_table.h"

#include "drizzled/charset.h"

#include "drizzled/internal/m_string.h"

#include "plugin/myisam/myisam.h"

#include <drizzled/item/string.h>

#include <drizzled/item/int.h>

#include <drizzled/item/decimal.h>

#include <drizzled/item/float.h>

#include <drizzled/item/null.h>

#include <drizzled/temporal.h>

#include "drizzled/table_share_instance.h"

#include "drizzled/table_proto.h"

using namespace std;

namespace drizzled

{

extern pid_t current_pid;

extern plugin::StorageEngine *heap_engine;

extern plugin::StorageEngine *myisam_engine;

/* Functions defined in this cursor */

void open_table_error(TableShare *share, int error, int db_errno,

                      myf errortype, int errarg);

/*************************************************************************/

// @note this should all be the destructor

int Table::delete_table(bool free_share)

{

  int error= 0;

  if (db_stat)

    error= cursor->close();

  free((char*) alias);

  alias= NULL;

  if (field)

  {

    for (Field **ptr=field ; *ptr ; ptr++)

    {

      delete *ptr;

    }

    field= 0;

  }

  delete cursor;

  cursor= 0;				/* For easier errorchecking */

  if (free_share)

  {

    if (s->getType() == message::Table::STANDARD)

    {

      TableShare::release(s);

    }

    else

    {

      delete s;

    }

    s= NULL;

  }

  mem_root.free_root(MYF(0));

  return error;

}

void Table::resetTable(Session *session,

                       TableShare *share,

                       uint32_t db_stat_arg)

{

  s= share;

  field= NULL;

  cursor= NULL;

  next= NULL;

  prev= NULL;

  read_set= NULL;

  write_set= NULL;

  tablenr= 0;

  db_stat= db_stat_arg;

  in_use= session;

  record[0]= (unsigned char *) NULL;

  record[1]= (unsigned char *) NULL;

  insert_values.clear();

  key_info= NULL;

  next_number_field= NULL;

  found_next_number_field= NULL;

  timestamp_field= NULL;

  pos_in_table_list= NULL;

  group= NULL;

  alias= NULL;

  null_flags= NULL;

  lock_position= 0;

  lock_data_start= 0;

  lock_count= 0;

  used_fields= 0;

  status= 0;

  derived_select_number= 0;

  current_lock= F_UNLCK;

  copy_blobs= false;

  maybe_null= false;

  null_row= false;

  force_index= false;

  distinct= false;

  const_table= false;

  no_rows= false;

  key_read= false;

  no_keyread= false;

  open_placeholder= false;

  locked_by_name= false;

  no_cache= false;

  auto_increment_field_not_null= false;

  alias_name_used= false;

  query_id= 0;

  quick_condition_rows= 0;

  timestamp_field_type= TIMESTAMP_NO_AUTO_SET;

  map= 0;

  reginfo.reset();

  covering_keys.reset();

  quick_keys.reset();

  merge_keys.reset();

  keys_in_use_for_query.reset();

  keys_in_use_for_group_by.reset();

  keys_in_use_for_order_by.reset();

  memset(quick_rows, 0, sizeof(ha_rows) * MAX_KEY);

  memset(const_key_parts, 0, sizeof(ha_rows) * MAX_KEY);

  memset(quick_key_parts, 0, sizeof(unsigned int) * MAX_KEY);

  memset(quick_n_ranges, 0, sizeof(unsigned int) * MAX_KEY);

  memory::init_sql_alloc(&mem_root, TABLE_ALLOC_BLOCK_SIZE, 0);

}

/* Deallocate temporary blob storage */

void free_blobs(register Table *table)

{

  uint32_t *ptr, *end;

  for (ptr= table->getBlobField(), end=ptr + table->sizeBlobFields();

       ptr != end ;

       ptr++)

  {

    ((Field_blob*) table->getField(*ptr))->free();

  }

}

TYPELIB *typelib(memory::Root *mem_root, List<String> &strings)

{

  TYPELIB *result= (TYPELIB*) mem_root->alloc_root(sizeof(TYPELIB));

  if (!result)

    return 0;

  result->count= strings.elements;

  result->name= "";

  uint32_t nbytes= (sizeof(char*) + sizeof(uint32_t)) * (result->count + 1);

  if (!(result->type_names= (const char**) mem_root->alloc_root(nbytes)))

    return 0;

  result->type_lengths= (uint*) (result->type_names + result->count + 1);

  List_iterator<String> it(strings);

  String *tmp;

  for (uint32_t i= 0; (tmp= it++); i++)

  {

    result->type_names[i]= tmp->ptr();

    result->type_lengths[i]= tmp->length();

  }

  result->type_names[result->count]= 0;   // End marker

  result->type_lengths[result->count]= 0;

  return result;

}

	/* Check that the integer is in the internal */

int set_zone(register int nr, int min_zone, int max_zone)

{

  if (nr<=min_zone)

    return (min_zone);

  if (nr>=max_zone)

    return (max_zone);

  return (nr);

} /* set_zone */

	/* Adjust number to next larger disk buffer */

ulong next_io_size(register ulong pos)

{

  register ulong offset;

  if ((offset= pos & (IO_SIZE-1)))

    return pos-offset+IO_SIZE;

  return pos;

} /* next_io_size */

/*

  Store an SQL quoted string.

  SYNOPSIS

    append_unescaped()

    res		result String

    pos		string to be quoted

    length	it's length

  NOTE

    This function works correctly with utf8 or single-byte charset strings.

    May fail with some multibyte charsets though.

*/

void append_unescaped(String *res, const char *pos, uint32_t length)

{

  const char *end= pos+length;

  res->append('\'');

  for (; pos != end ; pos++)

  {

    uint32_t mblen;

    if (use_mb(default_charset_info) &&

        (mblen= my_ismbchar(default_charset_info, pos, end)))

    {

      res->append(pos, mblen);

      pos+= mblen - 1;

      if (pos >= end)

        break;

      continue;

    }

    switch (*pos) {

    case 0:				/* Must be escaped for 'mysql' */

      res->append('\\');

      res->append('0');

      break;

    case '\n':				/* Must be escaped for logs */

      res->append('\\');

      res->append('n');

      break;

    case '\r':

      res->append('\\');		/* This gives better readability */

      res->append('r');

      break;

    case '\\':

      res->append('\\');		/* Because of the sql syntax */

      res->append('\\');

      break;

    case '\'':

      res->append('\'');		/* Because of the sql syntax */

      res->append('\'');

      break;

    default:

      res->append(*pos);

      break;

    }

  }

  res->append('\'');

}

/*

  Set up column usage bitmaps for a temporary table

  IMPLEMENTATION

    For temporary tables, we need one bitmap with all columns set and

    a tmp_set bitmap to be used by things like filesort.

*/

void Table::setup_tmp_table_column_bitmaps(unsigned char *bitmaps)

{

  uint32_t field_count= s->sizeFields();

  this->def_read_set.init((my_bitmap_map*) bitmaps, field_count);

  this->tmp_set.init((my_bitmap_map*) (bitmaps+ bitmap_buffer_size(field_count)), field_count);

  /* write_set and all_set are copies of read_set */

  def_write_set= def_read_set;

  s->all_set= def_read_set;

  this->getMutableShare()->all_set.setAll();

  default_column_bitmaps();

}

int rename_file_ext(const char * from,const char * to,const char * ext)

{

  string from_s, to_s;

  from_s.append(from);

  from_s.append(ext);

  to_s.append(to);

  to_s.append(ext);

  return (internal::my_rename(from_s.c_str(),to_s.c_str(),MYF(MY_WME)));

}

/*

  Check if database name is valid

  SYNPOSIS

    check_db_name()

    org_name		Name of database and length

  RETURN

    false error

    true ok

*/

bool check_db_name(Session *session, SchemaIdentifier &schema_identifier)

{

  if (not plugin::Authorization::isAuthorized(session->getSecurityContext(), schema_identifier))

  {

    return false;

  }

  return schema_identifier.isValid();

}

/*

  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, uint32_t length)

{

  if (!length || length > NAME_LEN || name[length - 1] == ' ')

    return 1;

  LEX_STRING ident;

  ident.str= (char*) name;

  ident.length= length;

  return check_identifier_name(&ident);

}

/*

  Eventually, a "length" argument should be added

  to this function, and the inner loop changed to

  check_identifier_name() call.

*/

bool check_column_name(const char *name)

{

  uint32_t name_length= 0;  // name length in symbols

  bool last_char_is_space= true;

  while (*name)

  {

    last_char_is_space= my_isspace(system_charset_info, *name);

    if (use_mb(system_charset_info))

    {

      int len=my_ismbchar(system_charset_info, name,

                          name+system_charset_info->mbmaxlen);

      if (len)

      {

        if (len > 3) /* Disallow non-BMP characters */

          return 1;

        name += len;

        name_length++;

        continue;

      }

    }

    /*

      NAMES_SEP_CHAR is used in FRM format to separate SET and ENUM values.

      It is defined as 0xFF, which is a not valid byte in utf8.

      This assert is to catch use of this byte if we decide to

      use non-utf8 as system_character_set.

    */

    assert(*name != NAMES_SEP_CHAR);

    name++;

    name_length++;

  }

  /* Error if empty or too long column name */

  return last_char_is_space || (uint32_t) name_length > NAME_CHAR_LEN;

}

/*****************************************************************************

  Functions to handle column usage bitmaps (read_set, write_set etc...)

*****************************************************************************/

/* Reset all columns bitmaps */

void 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);

  */

  def_read_set.clearAll();

  def_write_set.clearAll();

  column_bitmaps_set(&def_read_set, &def_write_set);

}

/*

  Tell Cursor we are going to call position() and rnd_pos() later.

  NOTES:

  This is needed for handlers that uses the primary key to find the

  row. In this case we have to extend the read bitmap with the primary

  key fields.

*/

void Table::prepare_for_position()

{

  if ((cursor->getEngine()->check_flag(HTON_BIT_PRIMARY_KEY_IN_READ_INDEX)) &&

      s->hasPrimaryKey())

  {

    mark_columns_used_by_index_no_reset(s->getPrimaryKey());

  }

  return;

}

/*

  Mark that only fields from one key is used

  NOTE:

    This changes the bitmap to use the tmp bitmap

    After this, you can't access any other columns in the table until

    bitmaps are reset, for example with Table::clear_column_bitmaps()

    or Table::restore_column_maps_after_mark_index()

*/

void Table::mark_columns_used_by_index(uint32_t index)

{

  MyBitmap *bitmap= &tmp_set;

  (void) cursor->extra(HA_EXTRA_KEYREAD);

  bitmap->clearAll();

  mark_columns_used_by_index_no_reset(index, bitmap);

  column_bitmaps_set(bitmap, bitmap);

  return;

}

/*

  Restore to use normal column maps after key read

  NOTES

    This reverse the change done by mark_columns_used_by_index

  WARNING

    For this to work, one must have the normal table maps in place

    when calling mark_columns_used_by_index

*/

void Table::restore_column_maps_after_mark_index()

{

  key_read= 0;

  (void) cursor->extra(HA_EXTRA_NO_KEYREAD);

  default_column_bitmaps();

  return;

}

/*

  mark columns used by key, but don't reset other fields

*/

void Table::mark_columns_used_by_index_no_reset(uint32_t index)

{

    mark_columns_used_by_index_no_reset(index, read_set);

}

void Table::mark_columns_used_by_index_no_reset(uint32_t index,

                                                MyBitmap *bitmap)

{

  KeyPartInfo *key_part= key_info[index].key_part;

  KeyPartInfo *key_part_end= (key_part +

                                key_info[index].key_parts);

  for (;key_part != key_part_end; key_part++)

    bitmap->setBit(key_part->fieldnr-1);

}

/*

  Mark auto-increment fields as used fields in both read and write maps

  NOTES

    This is needed in insert & update as the auto-increment field is

    always set and sometimes read.

*/

void Table::mark_auto_increment_column()

{

  assert(found_next_number_field);

  /*

    We must set bit in read set as update_auto_increment() is using the

    store() to check overflow of auto_increment values

  */

  setReadSet(found_next_number_field->field_index);

  setWriteSet(found_next_number_field->field_index);

  if (s->next_number_keypart)

    mark_columns_used_by_index_no_reset(s->next_number_index);

}

/*

  Mark columns needed for doing an delete of a row

  DESCRIPTON

    Some table engines don't have a cursor on the retrieve rows

    so they need either to use the primary key or all columns to

    be able to delete a row.

    If the engine needs this, the function works as follows:

    - If primary key exits, mark the primary key columns to be read.

    - If not, mark all columns to be read

    If the engine has HA_REQUIRES_KEY_COLUMNS_FOR_DELETE, we will

    mark all key columns as 'to-be-read'. This allows the engine to

    loop over the given record to find all keys and doesn't have to

    retrieve the row again.

*/

void Table::mark_columns_needed_for_delete()

{

  /*

    If the Cursor has no cursor capabilites, or we have row-based

    replication active for the current statement, we have to read

    either the primary key, the hidden primary key or all columns to

    be able to do an delete

  */

  if (not s->hasPrimaryKey())

  {

    /* fallback to use all columns in the table to identify row */

    use_all_columns();

    return;

  }

  else

    mark_columns_used_by_index_no_reset(s->getPrimaryKey());

  /* If we the engine wants all predicates we mark all keys */

  if (cursor->getEngine()->check_flag(HTON_BIT_REQUIRES_KEY_COLUMNS_FOR_DELETE))

  {

    Field **reg_field;

    for (reg_field= field ; *reg_field ; reg_field++)

    {

      if ((*reg_field)->flags & PART_KEY_FLAG)

        setReadSet((*reg_field)->field_index);

    }

  }

}

/*

  Mark columns needed for doing an update of a row

  DESCRIPTON

    Some engines needs to have all columns in an update (to be able to

    build a complete row). If this is the case, we mark all not

    updated columns to be read.

    If this is no the case, we do like in the delete case and mark

    if neeed, either the primary key column or all columns to be read.

    (see mark_columns_needed_for_delete() for details)

    If the engine has HTON_BIT_REQUIRES_KEY_COLUMNS_FOR_DELETE, we will

    mark all USED key columns as 'to-be-read'. This allows the engine to

    loop over the given record to find all changed keys and doesn't have to

    retrieve the row again.

*/

void Table::mark_columns_needed_for_update()

{

  /*

    If the Cursor has no cursor capabilites, or we have row-based

    logging active for the current statement, we have to read either

    the primary key, the hidden primary key or all columns to be

    able to do an update

  */

  if (not s->hasPrimaryKey())

  {

    /* fallback to use all columns in the table to identify row */

    use_all_columns();

    return;

  }

  else

    mark_columns_used_by_index_no_reset(s->getPrimaryKey());

  if (cursor->getEngine()->check_flag(HTON_BIT_REQUIRES_KEY_COLUMNS_FOR_DELETE))

  {

    /* Mark all used key columns for read */

    Field **reg_field;

    for (reg_field= field ; *reg_field ; reg_field++)

    {

      /* Merge keys is all keys that had a column refered to in the query */

      if (is_overlapping(merge_keys, (*reg_field)->part_of_key))

        setReadSet((*reg_field)->field_index);

    }

  }

}

/*

  Mark columns the Cursor needs for doing an insert

  For now, this is used to mark fields used by the trigger

  as changed.

*/

void Table::mark_columns_needed_for_insert()

{

  if (found_next_number_field)

    mark_auto_increment_column();

}

size_t Table::max_row_length(const unsigned char *data)

{

  size_t length= getRecordLength() + 2 * sizeFields();

  uint32_t *const beg= getBlobField();

  uint32_t *const end= beg + sizeBlobFields();

  for (uint32_t *ptr= beg ; ptr != end ; ++ptr)

  {

    Field_blob* const blob= (Field_blob*) field[*ptr];

    length+= blob->get_length((const unsigned char*)

                              (data + blob->offset(getInsertRecord()))) +

      HA_KEY_BLOB_LENGTH;

  }

  return length;

}

/****************************************************************************

 Functions for creating temporary tables.

****************************************************************************/

/**

  Create field for temporary table from given field.

  @param session	       Thread Cursor

  @param org_field    field from which new field will be created

  @param name         New field name

  @param table	       Temporary table

  @param item	       !=NULL if item->result_field should point to new field.

                      This is relevant for how fill_record() is going to work:

                      If item != NULL then fill_record() will update

                      the record in the original table.

                      If item == NULL then fill_record() will update

                      the temporary table

  @param convert_blob_length   If >0 create a varstring(convert_blob_length)

                               field instead of blob.

  @retval

    NULL		on error

  @retval

    new_created field

*/

Field *create_tmp_field_from_field(Session *session, Field *org_field,

                                   const char *name, Table *table,

                                   Item_field *item, uint32_t convert_blob_length)

{

  Field *new_field;

  /*

    Make sure that the blob fits into a Field_varstring which has

    2-byte lenght.

  */

  if (convert_blob_length && convert_blob_length <= Field_varstring::MAX_SIZE &&

      (org_field->flags & BLOB_FLAG))

    new_field= new Field_varstring(convert_blob_length,

                                   org_field->maybe_null(),

                                   org_field->field_name, table->getMutableShare(),

                                   org_field->charset());

  else

    new_field= org_field->new_field(session->mem_root, table,