~drizzle-trunk/drizzle/development

/* -*- mode: c++; c-basic-offset: 2; indent-tabs-mode: nil; -*-

 *  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:

 *

 *  Copyright (C) 2008 Sun Microsystems, Inc.

 *

 *  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., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA

 */

/**

  @file Cursor.cc

  Handler-calling-functions

*/

#include "config.h"

#include <fcntl.h>

#include "drizzled/my_hash.h"

#include "drizzled/error.h"

#include "drizzled/gettext.h"

#include "drizzled/probes.h"

#include "drizzled/sql_parse.h"

#include "drizzled/optimizer/cost_vector.h"

#include "drizzled/session.h"

#include "drizzled/sql_base.h"

#include "drizzled/transaction_services.h"

#include "drizzled/lock.h"

#include "drizzled/item/int.h"

#include "drizzled/item/empty_string.h"

#include "drizzled/field/epoch.h"

#include "drizzled/message/table.pb.h"

#include "drizzled/plugin/client.h"

#include "drizzled/internal/my_sys.h"

#include "drizzled/plugin/event_observer.h"

using namespace std;

namespace drizzled

{

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

** General Cursor functions

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

Cursor::Cursor(plugin::StorageEngine &engine_arg,

               Table &arg)

  : table(arg),

    engine(engine_arg),

    estimation_rows_to_insert(0),

    ref(0),

    key_used_on_scan(MAX_KEY), active_index(MAX_KEY),

    ref_length(sizeof(internal::my_off_t)),

    inited(NONE),

    locked(false),

    next_insert_id(0), insert_id_for_cur_row(0)

{ }

Cursor::~Cursor(void)

{

  assert(locked == false);

  /* TODO: assert(inited == NONE); */

}

/*

 * @note this only used in

 * optimizer::QuickRangeSelect::init_ror_merged_scan(bool reuse_handler) as

 * of the writing of this comment. -Brian

 */

Cursor *Cursor::clone(memory::Root *mem_root)

{

  Cursor *new_handler= getTable()->getMutableShare()->db_type()->getCursor(*getTable());

  /*

    Allocate Cursor->ref here because otherwise ha_open will allocate it

    on this->table->mem_root and we will not be able to reclaim that memory

    when the clone Cursor object is destroyed.

  */

  if (!(new_handler->ref= (unsigned char*) mem_root->alloc_root(ALIGN_SIZE(ref_length)*2)))

    return NULL;

  identifier::Table identifier(getTable()->getShare()->getSchemaName(),

                             getTable()->getShare()->getTableName(),

                             getTable()->getShare()->getType());

  if (new_handler && !new_handler->ha_open(identifier,

                                           getTable()->getDBStat(),

                                           HA_OPEN_IGNORE_IF_LOCKED))

    return new_handler;

  return NULL;

}

/*

  DESCRIPTION

    given a buffer with a key value, and a map of keyparts

    that are present in this value, returns the length of the value

*/

uint32_t Cursor::calculate_key_len(uint32_t key_position, key_part_map keypart_map_arg)

{

  /* works only with key prefixes */

  assert(((keypart_map_arg + 1) & keypart_map_arg) == 0);

  const KeyPartInfo *key_part_found= getTable()->getShare()->getKeyInfo(key_position).key_part;

  const KeyPartInfo *end_key_part_found= key_part_found + getTable()->getShare()->getKeyInfo(key_position).key_parts;

  uint32_t length= 0;

  while (key_part_found < end_key_part_found && keypart_map_arg)

  {

    length+= key_part_found->store_length;

    keypart_map_arg >>= 1;

    key_part_found++;

  }

  return length;

}

int Cursor::startIndexScan(uint32_t idx, bool sorted)

{

  int result;

  assert(inited == NONE);

  if (!(result= doStartIndexScan(idx, sorted)))

    inited=INDEX;

  end_range= NULL;

  return result;

}

int Cursor::endIndexScan()

{

  assert(inited==INDEX);

  inited=NONE;

  end_range= NULL;

  return(doEndIndexScan());

}

int Cursor::startTableScan(bool scan)

{

  int result;

  assert(inited==NONE || (inited==RND && scan));

  inited= (result= doStartTableScan(scan)) ? NONE: RND;

  return result;

}

int Cursor::endTableScan()

{

  assert(inited==RND);

  inited=NONE;

  return(doEndTableScan());

}

int Cursor::ha_index_or_rnd_end()

{

  return inited == INDEX ? endIndexScan() : inited == RND ? endTableScan() : 0;

}

void Cursor::ha_start_bulk_insert(ha_rows rows)

{

  estimation_rows_to_insert= rows;

  start_bulk_insert(rows);

}

int Cursor::ha_end_bulk_insert()

{

  estimation_rows_to_insert= 0;

  return end_bulk_insert();

}

const key_map *Cursor::keys_to_use_for_scanning()

{

  return &key_map_empty;

}

bool Cursor::has_transactions()

{

  return (getTable()->getShare()->db_type()->check_flag(HTON_BIT_DOES_TRANSACTIONS));

}

void Cursor::ha_statistic_increment(uint64_t system_status_var::*offset) const

{

  (getTable()->in_use->status_var.*offset)++;

}

void **Cursor::ha_data(Session *session) const

{

  return session->getEngineData(getEngine());

}

bool Cursor::is_fatal_error(int error, uint32_t flags)

{

  if (!error ||

      ((flags & HA_CHECK_DUP_KEY) &&

       (error == HA_ERR_FOUND_DUPP_KEY ||

        error == HA_ERR_FOUND_DUPP_UNIQUE)))

    return false;

  return true;

}

ha_rows Cursor::records() { return stats.records; }

uint64_t Cursor::tableSize() { return stats.index_file_length + stats.data_file_length; }

uint64_t Cursor::rowSize() { return getTable()->getRecordLength() + getTable()->sizeFields(); }

int Cursor::doOpen(const identifier::Table &identifier, int mode, uint32_t test_if_locked)

{

  return open(identifier.getPath().c_str(), mode, test_if_locked);

}

/**

  Open database-Cursor.

  Try O_RDONLY if cannot open as O_RDWR

  Don't wait for locks if not HA_OPEN_WAIT_IF_LOCKED is set

*/

int Cursor::ha_open(const identifier::Table &identifier,

                    int mode,

                    int test_if_locked)

{

  int error;

  if ((error= doOpen(identifier, mode, test_if_locked)))

  {

    if ((error == EACCES || error == EROFS) && mode == O_RDWR &&

        (getTable()->db_stat & HA_TRY_READ_ONLY))

    {

      getTable()->db_stat|=HA_READ_ONLY;

      error= doOpen(identifier, O_RDONLY,test_if_locked);

    }

  }

  if (error)

  {

    errno= error;                            /* Safeguard */

  }

  else

  {

    if (getTable()->getShare()->db_options_in_use & HA_OPTION_READ_ONLY_DATA)

      getTable()->db_stat|=HA_READ_ONLY;

    (void) extra(HA_EXTRA_NO_READCHECK);	// Not needed in SQL

    /* ref is already allocated for us if we're called from Cursor::clone() */

    if (!ref && !(ref= (unsigned char*) getTable()->alloc_root(ALIGN_SIZE(ref_length)*2)))

    {

      close();

      error=HA_ERR_OUT_OF_MEM;

    }

    else

      dup_ref=ref+ALIGN_SIZE(ref_length);

  }

  return error;

}

/**

  Read first row (only) from a table.

  This is never called for InnoDB tables, as these table types

  has the HA_STATS_RECORDS_IS_EXACT set.

*/

int Cursor::read_first_row(unsigned char * buf, uint32_t primary_key)

{

  register int error;

  ha_statistic_increment(&system_status_var::ha_read_first_count);

  /*

    If there is very few deleted rows in the table, find the first row by

    scanning the table.

    TODO remove the test for HA_READ_ORDER

  */

  if (stats.deleted < 10 || primary_key >= MAX_KEY ||

      !(getTable()->index_flags(primary_key) & HA_READ_ORDER))

  {

    error= startTableScan(1);

    if (error == 0)

    {

      while ((error= rnd_next(buf)) == HA_ERR_RECORD_DELETED) ;

      (void) endTableScan();

    }

  }

  else

  {

    /* Find the first row through the primary key */

    error= startIndexScan(primary_key, 0);

    if (error == 0)

    {

      error=index_first(buf);

      (void) endIndexScan();

    }

  }

  return error;

}

/**

  Generate the next auto-increment number based on increment and offset.

  computes the lowest number

  - strictly greater than "nr"

  - of the form: auto_increment_offset + N * auto_increment_increment

  In most cases increment= offset= 1, in which case we get:

  @verbatim 1,2,3,4,5,... @endverbatim

    If increment=10 and offset=5 and previous number is 1, we get:

  @verbatim 1,5,15,25,35,... @endverbatim

*/

inline uint64_t

compute_next_insert_id(uint64_t nr, drizzle_system_variables *variables)

{

  if (variables->auto_increment_increment == 1)

    return (nr+1); // optimization of the formula below

  nr= (((nr+ variables->auto_increment_increment -

         variables->auto_increment_offset)) /

       (uint64_t) variables->auto_increment_increment);

  return (nr* (uint64_t) variables->auto_increment_increment +

          variables->auto_increment_offset);

}

void Cursor::adjust_next_insert_id_after_explicit_value(uint64_t nr)

{

  /*

    If we have set Session::next_insert_id previously and plan to insert an

    explicitely-specified value larger than this, we need to increase

    Session::next_insert_id to be greater than the explicit value.

  */

  if ((next_insert_id > 0) && (nr >= next_insert_id))

    set_next_insert_id(compute_next_insert_id(nr, &getTable()->in_use->variables));

}

/**

  Compute a previous insert id

  Computes the largest number X:

  - smaller than or equal to "nr"

  - of the form: auto_increment_offset + N * auto_increment_increment

    where N>=0.

  @param nr            Number to "round down"

  @param variables     variables struct containing auto_increment_increment and

                       auto_increment_offset

  @return

    The number X if it exists, "nr" otherwise.

*/

inline uint64_t

prev_insert_id(uint64_t nr, drizzle_system_variables *variables)

{

  if (unlikely(nr < variables->auto_increment_offset))

  {

    /*

      There's nothing good we can do here. That is a pathological case, where

      the offset is larger than the column's max possible value, i.e. not even

      the first sequence value may be inserted. User will receive warning.

    */

    return nr;

  }

  if (variables->auto_increment_increment == 1)

    return nr; // optimization of the formula below

  nr= (((nr - variables->auto_increment_offset)) /

       (uint64_t) variables->auto_increment_increment);

  return (nr * (uint64_t) variables->auto_increment_increment +

          variables->auto_increment_offset);

}

/**

  Update the auto_increment field if necessary.

  Updates columns with type NEXT_NUMBER if:

  - If column value is set to NULL (in which case auto_increment_field_not_null is false)

  - If column is set to 0 and (sql_mode & MODE_NO_AUTO_VALUE_ON_ZERO) is not

    set. In the future we will only set NEXT_NUMBER fields if one sets them

    to NULL (or they are not included in the insert list).

    In those cases, we check if the currently reserved interval still has

    values we have not used. If yes, we pick the smallest one and use it.

    Otherwise:

  - If a list of intervals has been provided to the statement via SET

    INSERT_ID or via an Intvar_log_event (in a replication slave), we pick the

    first unused interval from this list, consider it as reserved.

  - Otherwise we set the column for the first row to the value

    next_insert_id(get_auto_increment(column))) which is usually

    max-used-column-value+1.

    We call get_auto_increment() for the first row in a multi-row

    statement. get_auto_increment() will tell us the interval of values it

    reserved for us.

  - In both cases, for the following rows we use those reserved values without

    calling the Cursor again (we just progress in the interval, computing

    each new value from the previous one). Until we have exhausted them, then

    we either take the next provided interval or call get_auto_increment()

    again to reserve a new interval.

  - In both cases, the reserved intervals are remembered in

    session->auto_inc_intervals_in_cur_stmt_for_binlog if statement-based

    binlogging; the last reserved interval is remembered in

    auto_inc_interval_for_cur_row.

    The idea is that generated auto_increment values are predictable and

    independent of the column values in the table.  This is needed to be

    able to replicate into a table that already has rows with a higher

    auto-increment value than the one that is inserted.

    After we have already generated an auto-increment number and the user

    inserts a column with a higher value than the last used one, we will

    start counting from the inserted value.

    This function's "outputs" are: the table's auto_increment field is filled

    with a value, session->next_insert_id is filled with the value to use for the

    next row, if a value was autogenerated for the current row it is stored in

    session->insert_id_for_cur_row, if get_auto_increment() was called

    session->auto_inc_interval_for_cur_row is modified, if that interval is not

    present in session->auto_inc_intervals_in_cur_stmt_for_binlog it is added to

    this list.

  @todo

    Replace all references to "next number" or NEXT_NUMBER to

    "auto_increment", everywhere (see below: there is

    table->auto_increment_field_not_null, and there also exists

    table->next_number_field, it's not consistent).

  @retval

    0	ok

  @retval

    HA_ERR_AUTOINC_READ_FAILED  get_auto_increment() was called and

    returned ~(uint64_t) 0

  @retval

    HA_ERR_AUTOINC_ERANGE storing value in field caused strict mode

    failure.

*/

#define AUTO_INC_DEFAULT_NB_ROWS 1 // Some prefer 1024 here

#define AUTO_INC_DEFAULT_NB_MAX_BITS 16

#define AUTO_INC_DEFAULT_NB_MAX ((1 << AUTO_INC_DEFAULT_NB_MAX_BITS) - 1)

int Cursor::update_auto_increment()

{

  uint64_t nr, nb_reserved_values;

  bool append= false;

  Session *session= getTable()->in_use;

  drizzle_system_variables *variables= &session->variables;

  /*

    next_insert_id is a "cursor" into the reserved interval, it may go greater

    than the interval, but not smaller.

  */

  assert(next_insert_id >= auto_inc_interval_for_cur_row.minimum());

  /* We check if auto_increment_field_not_null is false

     for an auto increment column, not a magic value like NULL is.

     same as sql_mode=NO_AUTO_VALUE_ON_ZERO */

  if ((nr= getTable()->next_number_field->val_int()) != 0

      || getTable()->auto_increment_field_not_null)

  {

    /*

      Update next_insert_id if we had already generated a value in this

      statement (case of INSERT VALUES(null),(3763),(null):

      the last NULL needs to insert 3764, not the value of the first NULL plus

      1).

    */

    adjust_next_insert_id_after_explicit_value(nr);

    insert_id_for_cur_row= 0; // didn't generate anything

    return 0;

  }

  if ((nr= next_insert_id) >= auto_inc_interval_for_cur_row.maximum())

  {

    /* next_insert_id is beyond what is reserved, so we reserve more. */

    const Discrete_interval *forced=

      session->auto_inc_intervals_forced.get_next();

    if (forced != NULL)

    {

      nr= forced->minimum();

      nb_reserved_values= forced->values();

    }

    else

    {

      /*

        Cursor::estimation_rows_to_insert was set by

        Cursor::ha_start_bulk_insert(); if 0 it means "unknown".

      */

      uint32_t nb_already_reserved_intervals=

        session->auto_inc_intervals_in_cur_stmt_for_binlog.nb_elements();

      uint64_t nb_desired_values;

      /*

        If an estimation was given to the engine:

        - use it.

        - if we already reserved numbers, it means the estimation was

        not accurate, then we'll reserve 2*AUTO_INC_DEFAULT_NB_ROWS the 2nd

        time, twice that the 3rd time etc.

        If no estimation was given, use those increasing defaults from the

        start, starting from AUTO_INC_DEFAULT_NB_ROWS.

        Don't go beyond a max to not reserve "way too much" (because

        reservation means potentially losing unused values).

      */

      if (nb_already_reserved_intervals == 0 &&

          (estimation_rows_to_insert > 0))

        nb_desired_values= estimation_rows_to_insert;

      else /* go with the increasing defaults */

      {

        /* avoid overflow in formula, with this if() */

        if (nb_already_reserved_intervals <= AUTO_INC_DEFAULT_NB_MAX_BITS)

        {

          nb_desired_values= AUTO_INC_DEFAULT_NB_ROWS *

            (1 << nb_already_reserved_intervals);

          set_if_smaller(nb_desired_values, (uint64_t)AUTO_INC_DEFAULT_NB_MAX);

        }

        else

          nb_desired_values= AUTO_INC_DEFAULT_NB_MAX;

      }

      /* This call ignores all its parameters but nr, currently */

      get_auto_increment(variables->auto_increment_offset,

                         variables->auto_increment_increment,

                         nb_desired_values, &nr,

                         &nb_reserved_values);

      if (nr == ~(uint64_t) 0)

        return HA_ERR_AUTOINC_READ_FAILED;  // Mark failure

      /*

        That rounding below should not be needed when all engines actually

        respect offset and increment in get_auto_increment(). But they don't

        so we still do it. Wonder if for the not-first-in-index we should do

        it. Hope that this rounding didn't push us out of the interval; even

        if it did we cannot do anything about it (calling the engine again

        will not help as we inserted no row).

      */

      nr= compute_next_insert_id(nr-1, variables);

    }

    if (getTable()->getShare()->next_number_keypart == 0)

    {

      /* We must defer the appending until "nr" has been possibly truncated */

      append= true;

    }

  }

  if (unlikely(getTable()->next_number_field->store((int64_t) nr, true)))

  {

    /*

      first test if the query was aborted due to strict mode constraints

    */

    if (session->getKilled() == Session::KILL_BAD_DATA)

      return HA_ERR_AUTOINC_ERANGE;

    /*

      field refused this value (overflow) and truncated it, use the result of

      the truncation (which is going to be inserted); however we try to

      decrease it to honour auto_increment_* variables.

      That will shift the left bound of the reserved interval, we don't

      bother shifting the right bound (anyway any other value from this

      interval will cause a duplicate key).

    */

    nr= prev_insert_id(getTable()->next_number_field->val_int(), variables);

    if (unlikely(getTable()->next_number_field->store((int64_t) nr, true)))

      nr= getTable()->next_number_field->val_int();

  }

  if (append)

  {

    auto_inc_interval_for_cur_row.replace(nr, nb_reserved_values,

                                          variables->auto_increment_increment);

  }

  /*

    Record this autogenerated value. If the caller then

    succeeds to insert this value, it will call

    record_first_successful_insert_id_in_cur_stmt()

    which will set first_successful_insert_id_in_cur_stmt if it's not

    already set.

  */

  insert_id_for_cur_row= nr;

  /*

    Set next insert id to point to next auto-increment value to be able to

    handle multi-row statements.

  */

  set_next_insert_id(compute_next_insert_id(nr, variables));

  return 0;

}

/**

  Reserves an interval of auto_increment values from the Cursor.

  offset and increment means that we want values to be of the form

  offset + N * increment, where N>=0 is integer.

  If the function sets *first_value to ~(uint64_t)0 it means an error.

  If the function sets *nb_reserved_values to UINT64_MAX it means it has

  reserved to "positive infinite".

  @param offset

  @param increment

  @param nb_desired_values   how many values we want

  @param first_value         (OUT) the first value reserved by the Cursor

  @param nb_reserved_values  (OUT) how many values the Cursor reserved

*/

void Cursor::ha_release_auto_increment()

{

  release_auto_increment();

  insert_id_for_cur_row= 0;

  auto_inc_interval_for_cur_row.replace(0, 0, 0);

  if (next_insert_id > 0)

  {

    next_insert_id= 0;

    /*

      this statement used forced auto_increment values if there were some,

      wipe them away for other statements.

    */

    getTable()->in_use->auto_inc_intervals_forced.empty();

  }

}

void Cursor::drop_table(const char *)

{

  close();

}

/**

  Performs checks upon the table.

  @param session                thread doing CHECK Table operation

  @param check_opt          options from the parser

  @retval

    HA_ADMIN_OK               Successful upgrade

  @retval

    HA_ADMIN_NEEDS_UPGRADE    Table has structures requiring upgrade

  @retval

    HA_ADMIN_NEEDS_ALTER      Table has structures requiring ALTER Table

  @retval

    HA_ADMIN_NOT_IMPLEMENTED

*/

int Cursor::ha_check(Session *, HA_CHECK_OPT *)

{

  return HA_ADMIN_OK;

}

/**

  A helper function to mark a transaction read-write,

  if it is started.

*/

inline

void

Cursor::setTransactionReadWrite()

{

  ResourceContext *resource_context;

  /*

   * If the cursor has not context for execution then there should be no

   * possible resource to gain (and if there is... then there is a bug such

   * that in_use should have been set.

 */

  if (not getTable()->in_use)

    return;

  resource_context= getTable()->in_use->getResourceContext(getEngine());

  /*

    When a storage engine method is called, the transaction must

    have been started, unless it's a DDL call, for which the

    storage engine starts the transaction internally, and commits

    it internally, without registering in the ha_list.

    Unfortunately here we can't know know for sure if the engine

    has registered the transaction or not, so we must check.

  */

  if (resource_context->isStarted())

  {

    resource_context->markModifiedData();

  }

}

/**

  Delete all rows: public interface.

  @sa Cursor::delete_all_rows()

  @note

  This is now equalivalent to TRUNCATE TABLE.

*/

int

Cursor::ha_delete_all_rows()

{

  setTransactionReadWrite();

  int result= delete_all_rows();

  if (result == 0)

  {

    /** 

     * Trigger post-truncate notification to plugins... 

     *

     * @todo Make TransactionServices generic to AfterTriggerServices

     * or similar...

     */

    Session *const session= getTable()->in_use;

    TransactionServices &transaction_services= TransactionServices::singleton();

    transaction_services.truncateTable(*session, *getTable());

  }

  return result;

}

/**

  Reset auto increment: public interface.

  @sa Cursor::reset_auto_increment()

*/

int

Cursor::ha_reset_auto_increment(uint64_t value)

{

  setTransactionReadWrite();

  return reset_auto_increment(value);

}

/**

  Analyze table: public interface.

  @sa Cursor::analyze()

*/

int

Cursor::ha_analyze(Session* session, HA_CHECK_OPT*)

{

  setTransactionReadWrite();

  return analyze(session);

}

/**

  Disable indexes: public interface.

  @sa Cursor::disable_indexes()

*/

int

Cursor::ha_disable_indexes(uint32_t mode)

{

  setTransactionReadWrite();