~drizzle-trunk/drizzle/development

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

Copyright (C) 2000, 2010, MySQL AB & Innobase Oy. All Rights Reserved.

Copyright (C) 2008, 2009 Google Inc.

Copyright (C) 2009, Percona Inc.

Portions of this file contain modifications contributed and copyrighted by

Google, Inc. Those modifications are gratefully acknowledged and are described

briefly in the InnoDB documentation. The contributions by Google are

incorporated with their permission, and subject to the conditions contained in

the file COPYING.Google.

Portions of this file contain modifications contributed and copyrighted

by Percona Inc.. Those modifications are

gratefully acknowledged and are described briefly in the InnoDB

documentation. The contributions by Percona Inc. are incorporated with

their permission, and subject to the conditions contained in the file

COPYING.Percona.

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

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

/* TODO list for the InnoDB Cursor in 5.0:

  - fix savepoint functions to use savepoint storage area

  - Find out what kind of problems the OS X case-insensitivity causes to

    table and database names; should we 'normalize' the names like we do

    in Windows?

*/

#include "config.h"

#include <limits.h>

#include <fcntl.h>

#include "drizzled/error.h"

#include "drizzled/errmsg_print.h"

#include "drizzled/charset_info.h"

#include "drizzled/internal/m_string.h"

#include "drizzled/internal/my_sys.h"

#include "drizzled/my_hash.h"

#include "drizzled/plugin.h"

#include "drizzled/show.h"

#include "drizzled/data_home.h"

#include "drizzled/error.h"

#include "drizzled/field.h"

#include "drizzled/charset.h"

#include "drizzled/session.h"

#include "drizzled/current_session.h"

#include "drizzled/table.h"

#include "drizzled/field/blob.h"

#include "drizzled/field/varstring.h"

#include "drizzled/plugin/xa_storage_engine.h"

#include "drizzled/plugin/daemon.h"

#include "drizzled/memory/multi_malloc.h"

#include "drizzled/pthread_globals.h"

#include "drizzled/named_savepoint.h"

#include <drizzled/transaction_services.h>

#include "drizzled/message/statement_transform.h"

#include <boost/algorithm/string.hpp>

#include <boost/program_options.hpp>

#include <boost/scoped_array.hpp>

#include <boost/filesystem.hpp>

#include <drizzled/module/option_map.h>

#include <iostream>

namespace po= boost::program_options;

namespace fs=boost::filesystem;

using namespace std;

/** @file ha_innodb.cc */

/* Include necessary InnoDB headers */

#include "univ.i"

#include "buf0lru.h"

#include "btr0sea.h"

#include "os0file.h"

#include "os0thread.h"

#include "srv0start.h"

#include "srv0srv.h"

#include "trx0roll.h"

#include "trx0trx.h"

#include "trx0sys.h"

#include "mtr0mtr.h"

#include "row0ins.h"

#include "row0mysql.h"

#include "row0sel.h"

#include "row0upd.h"

#include "log0log.h"

#include "lock0lock.h"

#include "dict0crea.h"

#include "create_replication.h"

#include "btr0cur.h"

#include "btr0btr.h"

#include "fsp0fsp.h"

#include "sync0sync.h"

#include "fil0fil.h"

#include "trx0xa.h"

#include "row0merge.h"

#include "thr0loc.h"

#include "dict0boot.h"

#include "ha_prototypes.h"

#include "ut0mem.h"

#include "ibuf0ibuf.h"

#include "ha_innodb.h"

#include "data_dictionary.h"

#include "replication_dictionary.h"

#include "internal_dictionary.h"

#include "handler0vars.h"

#include <iostream>

#include <sstream>

#include <string>

#include "plugin/innobase/handler/status_function.h"

#include "plugin/innobase/handler/replication_log.h"

#include <google/protobuf/io/zero_copy_stream.h>

#include <google/protobuf/io/zero_copy_stream_impl.h>

#include <google/protobuf/io/coded_stream.h>

#include <google/protobuf/text_format.h>

#include <boost/thread/mutex.hpp>

using namespace std;

using namespace drizzled;

/** to protect innobase_open_files */

static boost::mutex innobase_share_mutex;

/** to force correct commit order in binlog */

static ulong commit_threads = 0;

static boost::condition_variable commit_cond;

static boost::mutex commit_cond_m;

static bool innodb_inited = 0;

#define INSIDE_HA_INNOBASE_CC

/* In the Windows plugin, the return value of current_session is

undefined.  Map it to NULL. */

#if defined MYSQL_DYNAMIC_PLUGIN && defined __WIN__

# undef current_session

# define current_session NULL

# define EQ_CURRENT_SESSION(session) TRUE

#else /* MYSQL_DYNAMIC_PLUGIN && __WIN__ */

# define EQ_CURRENT_SESSION(session) ((session) == current_session)

#endif /* MYSQL_DYNAMIC_PLUGIN && __WIN__ */

static plugin::XaStorageEngine* innodb_engine_ptr= NULL;

typedef constrained_check<uint32_t, UINT32_MAX, 10> open_files_constraint;

static open_files_constraint innobase_open_files;

typedef constrained_check<uint32_t, 10, 1> mirrored_log_groups_constraint;

static mirrored_log_groups_constraint innobase_mirrored_log_groups;

typedef constrained_check<uint32_t, 100, 2> log_files_in_group_constraint;

static log_files_in_group_constraint innobase_log_files_in_group;

typedef constrained_check<uint32_t, 6, 0> force_recovery_constraint;

force_recovery_constraint innobase_force_recovery;

typedef constrained_check<size_t, SIZE_MAX, 256*1024, 1024> log_buffer_constraint;

static log_buffer_constraint innobase_log_buffer_size;

typedef constrained_check<size_t, SIZE_MAX, 512*1024, 1024> additional_mem_pool_constraint;

static additional_mem_pool_constraint innobase_additional_mem_pool_size;

typedef constrained_check<unsigned int, 1000, 1> autoextend_constraint;

static autoextend_constraint innodb_auto_extend_increment;

typedef constrained_check<size_t, SIZE_MAX, 5242880, 1048576> buffer_pool_constraint;

static buffer_pool_constraint innobase_buffer_pool_size;

typedef constrained_check<uint32_t, MAX_BUFFER_POOLS, 1> buffer_pool_instances_constraint;

static buffer_pool_instances_constraint innobase_buffer_pool_instances;

typedef constrained_check<uint32_t, UINT32_MAX, 100> io_capacity_constraint;

static io_capacity_constraint innodb_io_capacity;

typedef constrained_check<uint32_t, 5000, 1> purge_batch_constraint;

static purge_batch_constraint innodb_purge_batch_size;

typedef constrained_check<uint32_t, 1, 0> purge_threads_constraint;

static purge_threads_constraint innodb_n_purge_threads;

typedef constrained_check<uint32_t, 2, 0> trinary_constraint;

static trinary_constraint innodb_flush_log_at_trx_commit;

typedef constrained_check<unsigned int, 99, 0> max_dirty_pages_constraint;

static max_dirty_pages_constraint innodb_max_dirty_pages_pct;

static uint64_constraint innodb_max_purge_lag;

static uint64_nonzero_constraint innodb_stats_sample_pages;

typedef constrained_check<uint32_t, 64, 1> io_threads_constraint;

static io_threads_constraint innobase_read_io_threads;

static io_threads_constraint innobase_write_io_threads;

typedef constrained_check<uint32_t, 1000, 0> concurrency_constraint;

static concurrency_constraint innobase_commit_concurrency;

static concurrency_constraint innobase_thread_concurrency;

static uint32_nonzero_constraint innodb_concurrency_tickets;

typedef constrained_check<int64_t, INT64_MAX, 1024*1024, 1024*1024> log_file_constraint;

static log_file_constraint innobase_log_file_size;

static uint64_constraint innodb_replication_delay;

/** Percentage of the buffer pool to reserve for 'old' blocks.

Connected to buf_LRU_old_ratio. */

typedef constrained_check<uint32_t, 95, 5> old_blocks_constraint;

static old_blocks_constraint innobase_old_blocks_pct;

static uint32_constraint innodb_sync_spin_loops;

static uint32_constraint innodb_spin_wait_delay;

static uint32_constraint innodb_thread_sleep_delay;

typedef constrained_check<uint32_t, 64, 0> read_ahead_threshold_constraint;

static read_ahead_threshold_constraint innodb_read_ahead_threshold;

/* The default values for the following char* start-up parameters

are determined in innobase_init below: */

std::string innobase_data_home_dir;

std::string innobase_data_file_path;

std::string innobase_log_group_home_dir;

static string innobase_file_format_name;

static string innobase_change_buffering;

/* The highest file format being used in the database. The value can be

set by user, however, it will be adjusted to the newer file format if

a table of such format is created/opened. */

static string innobase_file_format_max;

/* Below we have boolean-valued start-up parameters, and their default

values */

typedef constrained_check<uint32_t, 2, 0> trinary_constraint;

static trinary_constraint innobase_fast_shutdown;

/* "innobase_file_format_check" decides whether we would continue

booting the server if the file format stamped on the system

table space exceeds the maximum file format supported

by the server. Can be set during server startup at command

line or configure file, and a read only variable after

server startup */

/* If a new file format is introduced, the file format

name needs to be updated accordingly. Please refer to

file_format_name_map[] defined in trx0sys.c for the next

file format name. */

static my_bool  innobase_file_format_check = TRUE;

static my_bool  innobase_use_doublewrite    = TRUE;

static my_bool  innobase_use_checksums      = TRUE;

static my_bool  innobase_rollback_on_timeout    = FALSE;

static my_bool  innobase_create_status_file   = FALSE;

static bool innobase_use_replication_log;

static bool support_xa;

static bool strict_mode;

typedef constrained_check<uint32_t, 1024*1024*1024, 1> lock_wait_constraint;

static lock_wait_constraint lock_wait_timeout;

static char*  internal_innobase_data_file_path  = NULL;

/* The following counter is used to convey information to InnoDB

about server activity: in selects it is not sensible to call

srv_active_wake_master_thread after each fetch or search, we only do

it every INNOBASE_WAKE_INTERVAL'th step. */

#define INNOBASE_WAKE_INTERVAL  32

static ulong  innobase_active_counter = 0;

static hash_table_t*  innobase_open_tables;

#ifdef __NETWARE__  /* some special cleanup for NetWare */

bool nw_panic = FALSE;

#endif

/** Allowed values of innodb_change_buffering */

static const char* innobase_change_buffering_values[IBUF_USE_COUNT] = {

  "none",   /* IBUF_USE_NONE */

  "inserts",	/* IBUF_USE_INSERT */

  "deletes",	/* IBUF_USE_DELETE_MARK */

  "changes",	/* IBUF_USE_INSERT_DELETE_MARK */

  "purges",	/* IBUF_USE_DELETE */

  "all"		/* IBUF_USE_ALL */

};

/* "GEN_CLUST_INDEX" is the name reserved for Innodb default

system primary index. */

static const char innobase_index_reserve_name[]= "GEN_CLUST_INDEX";

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

Gives the file extension of an InnoDB single-table tablespace. */

static const char* ha_innobase_exts[] = {

  ".ibd",

  NULL

};

#define DEFAULT_FILE_EXTENSION ".dfe" // Deep Fried Elephant

static INNOBASE_SHARE *get_share(const char *table_name);

static void free_share(INNOBASE_SHARE *share);

class InnobaseEngine : public plugin::XaStorageEngine

{

public:

  explicit InnobaseEngine(string name_arg) :

    plugin::XaStorageEngine(name_arg,

                            HTON_NULL_IN_KEY |

                            HTON_CAN_INDEX_BLOBS |

                            HTON_PRIMARY_KEY_IN_READ_INDEX |

                            HTON_PARTIAL_COLUMN_READ |

                            HTON_TABLE_SCAN_ON_INDEX |

                            HTON_HAS_FOREIGN_KEYS |

                            HTON_HAS_DOES_TRANSACTIONS)

  {

    table_definition_ext= plugin::DEFAULT_DEFINITION_FILE_EXT;

    addAlias("INNOBASE");

  }

  virtual ~InnobaseEngine()

  {

    int err= 0;

    if (innodb_inited) {

      srv_fast_shutdown = (ulint) innobase_fast_shutdown;

      innodb_inited = 0;

      hash_table_free(innobase_open_tables);

      innobase_open_tables = NULL;

      if (innobase_shutdown_for_mysql() != DB_SUCCESS) {

        err = 1;

      }

      srv_free_paths_and_sizes();

      if (internal_innobase_data_file_path)

        free(internal_innobase_data_file_path);

    }

    /* These get strdup'd from vm variables */

  }

private:

  virtual int doStartTransaction(Session *session, start_transaction_option_t options);

  virtual void doStartStatement(Session *session);

  virtual void doEndStatement(Session *session);

public:

  virtual

  int

  close_connection(

/*======================*/

      /* out: 0 or error number */

  Session*  session); /* in: handle to the MySQL thread of the user

      whose resources should be free'd */

  virtual int doSetSavepoint(Session* session,

                                 drizzled::NamedSavepoint &savepoint);

  virtual int doRollbackToSavepoint(Session* session,

                                     drizzled::NamedSavepoint &savepoint);

  virtual int doReleaseSavepoint(Session* session,

                                     drizzled::NamedSavepoint &savepoint);

  virtual int doXaCommit(Session* session, bool all)

  {

    return doCommit(session, all); /* XA commit just does a SQL COMMIT */

  }

  virtual int doXaRollback(Session *session, bool all)

  {

    return doRollback(session, all); /* XA rollback just does a SQL ROLLBACK */

  }

  virtual uint64_t doGetCurrentTransactionId(Session *session);

  virtual uint64_t doGetNewTransactionId(Session *session);

  virtual int doCommit(Session* session, bool all);

  virtual int doRollback(Session* session, bool all);

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

  This function is used to prepare X/Open XA distributed transaction   */

  virtual

  int

  doXaPrepare(

  /*================*/

        /* out: 0 or error number */

    Session*  session,  /* in: handle to the MySQL thread of the user

        whose XA transaction should be prepared */

    bool  all); /* in: TRUE - commit transaction

        FALSE - the current SQL statement ended */

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

  This function is used to recover X/Open XA distributed transactions   */

  virtual

  int

  doXaRecover(

  /*================*/

          /* out: number of prepared transactions

          stored in xid_list */

    ::drizzled::XID*  xid_list, /* in/out: prepared transactions */

    size_t len);    /* in: number of slots in xid_list */

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

  This function is used to commit one X/Open XA distributed transaction

  which is in the prepared state */

  virtual

  int

  doXaCommitXid(

  /*===================*/

        /* out: 0 or error number */

    ::drizzled::XID*  xid); /* in: X/Open XA transaction identification */

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

  This function is used to rollback one X/Open XA distributed transaction

  which is in the prepared state */

  virtual

  int

  doXaRollbackXid(

  /*=====================*/

        /* out: 0 or error number */

    ::drizzled::XID *xid);  /* in: X/Open XA transaction identification */

  virtual Cursor *create(Table &table)

  {

    return new ha_innobase(*this, table);

  }

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

  Removes all tables in the named database inside InnoDB. */

  bool

  doDropSchema(

  /*===================*/

        /* out: error number */

    const identifier::Schema  &identifier); /* in: database path; inside InnoDB the name

        of the last directory in the path is used as

        the database name: for example, in 'mysql/data/test'

        the database name is 'test' */

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

  Flushes InnoDB logs to disk and makes a checkpoint. Really, a commit flushes

  the logs, and the name of this function should be innobase_checkpoint. */

  virtual

  bool

  flush_logs();

  /*================*/

          /* out: TRUE if error */

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

  Implements the SHOW INNODB STATUS command. Sends the output of the InnoDB

  Monitor to the client. */

  virtual

  bool

  show_status(

  /*===============*/

    Session*  session,  /* in: the MySQL query thread of the caller */

    stat_print_fn *stat_print,

  enum ha_stat_type stat_type);

  virtual

  int

  doReleaseTemporaryLatches(

  /*===============================*/

        /* out: 0 */

  Session*    session); /* in: MySQL thread */

  const char** bas_ext() const {

  return(ha_innobase_exts);

  }

  UNIV_INTERN int doCreateTable(Session &session,

                                Table &form,

                                const identifier::Table &identifier,

                                message::Table&);

  UNIV_INTERN int doRenameTable(Session&, const identifier::Table &from, const identifier::Table &to);

  UNIV_INTERN int doDropTable(Session &session, const identifier::Table &identifier);

  UNIV_INTERN virtual bool get_error_message(int error, String *buf) const;

  UNIV_INTERN uint32_t max_supported_keys() const;

  UNIV_INTERN uint32_t max_supported_key_length() const;

  UNIV_INTERN uint32_t max_supported_key_part_length() const;

  UNIV_INTERN uint32_t index_flags(enum  ha_key_alg) const

  {

    return (HA_READ_NEXT |

            HA_READ_PREV |

            HA_READ_ORDER |

            HA_READ_RANGE |

            HA_KEYREAD_ONLY);

  }

  int doGetTableDefinition(drizzled::Session& session,

                           const identifier::Table &identifier,

                           drizzled::message::Table &table_proto);

  bool doDoesTableExist(drizzled::Session& session, const identifier::Table &identifier);

  void doGetTableIdentifiers(drizzled::CachedDirectory &directory,

                             const drizzled::identifier::Schema &schema_identifier,

                             drizzled::identifier::Table::vector &set_of_identifiers);

  bool validateCreateTableOption(const std::string &key, const std::string &state);

  void dropTemporarySchema();

};

bool InnobaseEngine::validateCreateTableOption(const std::string &key, const std::string &state)

{

  if (boost::iequals(key, "ROW_FORMAT"))

  {

    if (boost::iequals(state, "COMPRESSED"))

      return true;

    if (boost::iequals(state, "COMPACT"))

      return true;

    if (boost::iequals(state, "DYNAMIC"))

      return true;

    if (boost::iequals(state, "REDUNDANT"))

      return true;

  }

  return false;

}

void InnobaseEngine::doGetTableIdentifiers(drizzled::CachedDirectory &directory,

                                           const drizzled::identifier::Schema &schema_identifier,

                                           drizzled::identifier::Table::vector &set_of_identifiers)

{

  CachedDirectory::Entries entries= directory.getEntries();

  std::string search_string(schema_identifier.getSchemaName());

  boost::algorithm::to_lower(search_string);

  if (search_string.compare("data_dictionary") == 0)

  {

    set_of_identifiers.push_back(identifier::Table(schema_identifier.getSchemaName(), "SYS_REPLICATION_LOG"));

  }

  for (CachedDirectory::Entries::iterator entry_iter= entries.begin(); 

       entry_iter != entries.end(); ++entry_iter)

  {

    CachedDirectory::Entry *entry= *entry_iter;

    const string *filename= &entry->filename;

    assert(filename->size());

    const char *ext= strchr(filename->c_str(), '.');

    if (ext == NULL || my_strcasecmp(system_charset_info, ext, DEFAULT_FILE_EXTENSION) ||

        (filename->compare(0, strlen(TMP_FILE_PREFIX), TMP_FILE_PREFIX) == 0))

    { }

    else

    {

      std::string path;

      path+= directory.getPath();

      path+= FN_LIBCHAR;

      path+= entry->filename;

      message::Table definition;

      if (StorageEngine::readTableFile(path, definition))

      {

        /* 

           Using schema_identifier here to stop unused warning, could use

           definition.schema() instead

        */

        identifier::Table identifier(schema_identifier.getSchemaName(), definition.name());

        set_of_identifiers.push_back(identifier);

      }

    }

  }

}

bool InnobaseEngine::doDoesTableExist(Session &session, const identifier::Table &identifier)

{

  string proto_path(identifier.getPath());

  proto_path.append(DEFAULT_FILE_EXTENSION);

  if (session.getMessageCache().doesTableMessageExist(identifier))

    return true;

  std::string search_string(identifier.getPath());

  boost::algorithm::to_lower(search_string);

  if (search_string.compare("data_dictionary/sys_replication_log") == 0)

    return true;

  if (access(proto_path.c_str(), F_OK))

  {

    return false;

  }

  return true;

}

int InnobaseEngine::doGetTableDefinition(Session &session,

                                         const identifier::Table &identifier,

                                         message::Table &table_proto)

{

  string proto_path(identifier.getPath());

  proto_path.append(DEFAULT_FILE_EXTENSION);

  // First we check the temporary tables.

  if (session.getMessageCache().getTableMessage(identifier, table_proto))

    return EEXIST;

  if (read_replication_log_table_message(identifier.getTableName().c_str(), &table_proto) == 0)

    return EEXIST;

  if (access(proto_path.c_str(), F_OK))

  {

    return errno;

  }

  if (StorageEngine::readTableFile(proto_path, table_proto))

    return EEXIST;

  return ENOENT;

}

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

Validate the file format name and return its corresponding id.

@return valid file format id */

static

uint

innobase_file_format_name_lookup(

/*=============================*/

  const char* format_name);   /*!< in: pointer to file format

            name */

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

Validate the file format check config parameters, as a side effect it

sets the srv_max_file_format_at_startup variable.

@return	the format_id if valid config value, otherwise, return -1 */

static

int

innobase_file_format_validate_and_set(

/*================================*/

  const char* format_max);    /*!< in: parameter value */

static const char innobase_engine_name[]= "InnoDB";

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

Commits a transaction in an InnoDB database. */

static

void

innobase_commit_low(

/*================*/

  trx_t*  trx); /*!< in: transaction handle */

static drizzle_show_var innodb_status_variables[]= {

  {"buffer_pool_pages_data",

  (char*) &export_vars.innodb_buffer_pool_pages_data,   SHOW_LONG},

  {"buffer_pool_pages_dirty",

  (char*) &export_vars.innodb_buffer_pool_pages_dirty,    SHOW_LONG},

  {"buffer_pool_pages_flushed",

  (char*) &export_vars.innodb_buffer_pool_pages_flushed,  SHOW_LONG},

  {"buffer_pool_pages_free",

  (char*) &export_vars.innodb_buffer_pool_pages_free,   SHOW_LONG},

#ifdef UNIV_DEBUG

  {"buffer_pool_pages_latched",

  (char*) &export_vars.innodb_buffer_pool_pages_latched,  SHOW_LONG},

#endif /* UNIV_DEBUG */

  {"buffer_pool_pages_misc",

  (char*) &export_vars.innodb_buffer_pool_pages_misc,   SHOW_LONG},

  {"buffer_pool_pages_total",

  (char*) &export_vars.innodb_buffer_pool_pages_total,    SHOW_LONG},

  {"buffer_pool_read_ahead",

  (char*) &export_vars.innodb_buffer_pool_read_ahead, SHOW_LONG},

  {"buffer_pool_read_ahead_evicted",

  (char*) &export_vars.innodb_buffer_pool_read_ahead_evicted, SHOW_LONG},

  {"buffer_pool_read_requests",

  (char*) &export_vars.innodb_buffer_pool_read_requests,  SHOW_LONG},

  {"buffer_pool_reads",

  (char*) &export_vars.innodb_buffer_pool_reads,    SHOW_LONG},

  {"buffer_pool_wait_free",

  (char*) &export_vars.innodb_buffer_pool_wait_free,    SHOW_LONG},

  {"buffer_pool_write_requests",

  (char*) &export_vars.innodb_buffer_pool_write_requests, SHOW_LONG},

  {"data_fsyncs",

  (char*) &export_vars.innodb_data_fsyncs,      SHOW_LONG},

  {"data_pending_fsyncs",

  (char*) &export_vars.innodb_data_pending_fsyncs,    SHOW_LONG},

  {"data_pending_reads",

  (char*) &export_vars.innodb_data_pending_reads,   SHOW_LONG},

  {"data_pending_writes",

  (char*) &export_vars.innodb_data_pending_writes,    SHOW_LONG},

  {"data_read",

  (char*) &export_vars.innodb_data_read,      SHOW_LONG},

  {"data_reads",