~drizzle-trunk/drizzle/development

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

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

 *

 *  Copyright (C) 2009 - 2010 Toru Maesaka

 *

 *  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

 */

#include "config.h"

#include "ha_blitz.h"

using namespace std;

using namespace drizzled;

static pthread_mutex_t blitz_utility_mutex;

static const char *ha_blitz_exts[] = {

  BLITZ_DATA_EXT,

  BLITZ_INDEX_EXT,

  BLITZ_SYSTEM_EXT,

  NULL

};

/* Global Variables for Startup Options */

uint64_t blitz_estimated_rows;

class BlitzEngine : public drizzled::plugin::StorageEngine {

private:

  TCMAP *blitz_table_cache;

public:

  BlitzEngine(const std::string &name_arg) :

    drizzled::plugin::StorageEngine(name_arg,

                                    drizzled::HTON_NULL_IN_KEY |

                                    drizzled::HTON_PRIMARY_KEY_IN_READ_INDEX |

                                    drizzled::HTON_STATS_RECORDS_IS_EXACT |

                                    drizzled::HTON_SKIP_STORE_LOCK) {

    table_definition_ext = BLITZ_SYSTEM_EXT;

  }

  virtual ~BlitzEngine() {

    pthread_mutex_destroy(&blitz_utility_mutex);

    tcmapdel(blitz_table_cache);

  }

  virtual drizzled::Cursor *create(drizzled::TableShare &table) {

    return new ha_blitz(*this, table);

  }

  const char **bas_ext() const {

    return ha_blitz_exts;

  }

  int doCreateTable(drizzled::Session &session,

                    drizzled::Table &table_arg,

                    const drizzled::TableIdentifier &identifier,

                    drizzled::message::Table &table_proto);

  int doRenameTable(drizzled::Session &session,

                    const drizzled::TableIdentifier &from_identifier,

                    const drizzled::TableIdentifier &to_identifier);

  int doDropTable(drizzled::Session &session,

                  const drizzled::TableIdentifier &identifier);

  int doGetTableDefinition(drizzled::Session &session,

                           const drizzled::TableIdentifier &identifier,

                           drizzled::message::Table &table_proto);

  void doGetTableIdentifiers(drizzled::CachedDirectory &directory,

                             const drizzled::SchemaIdentifier &schema_identifier,

                             drizzled::TableIdentifiers &set_of_identifiers);

  bool doDoesTableExist(drizzled::Session &session,

                        const drizzled::TableIdentifier &identifier);

  bool validateCreateTableOption(const std::string &key,

                                 const std::string &state);

  bool doCreateTableCache(void);

  BlitzShare *getTableShare(const std::string &name);

  void cacheTableShare(const std::string &name, BlitzShare *share);

  void deleteTableShare(const std::string &name);

  uint32_t max_supported_keys() const { return BLITZ_MAX_INDEX; }

  uint32_t max_supported_key_length() const { return BLITZ_MAX_KEY_LEN; }

  uint32_t max_supported_key_part_length() const { return BLITZ_MAX_KEY_LEN; }

  uint32_t index_flags(enum drizzled::ha_key_alg) const {

    return (HA_READ_NEXT | HA_READ_PREV | HA_READ_ORDER |

            HA_READ_RANGE | HA_ONLY_WHOLE_INDEX | HA_KEYREAD_ONLY);

  }

};

/* A key stored in BlitzDB's B+Tree is a byte array that also includes

   a key to that row in the data dictionary. Two keys are merged and

   stored as a key because we want to avoid reading the leaf node and

   thus save disk IO and some computation in the tree. Note that the

   comparison function of BlitzDB's btree only takes into accound the

   actual index key. See blitzcmp.cc for details.

   With the above in mind, this helper function returns a pointer to

   the dictionary key by calculating the offset. */

static char *skip_btree_key(const char *key, const size_t skip_len,

                            int *return_klen);

static bool str_is_numeric(const std::string &str);

int BlitzEngine::doCreateTable(drizzled::Session &,

                               drizzled::Table &table,

                               const drizzled::TableIdentifier &identifier,

                               drizzled::message::Table &proto) {

  BlitzData dict;

  BlitzTree btree;

  int ecode;

  /* Temporary fix for blocking composite keys. We need to add this

     check because version 1 doesn't handle composite indexes. */

  for (uint32_t i = 0; i < table.s->keys; i++) {

    if (table.key_info[i].key_parts > 1)

      return HA_ERR_UNSUPPORTED;

  }

  /* Create relevant files for a new table and close them immediately.

     All we want to do here is somewhat like UNIX touch(1). */

  if ((ecode = dict.create_data_table(proto, table, identifier)) != 0)

    return ecode;

  if ((ecode = dict.create_system_table(identifier.getPath())) != 0)

    return ecode;

  /* Create b+tree index(es) for this table. */

  for (uint32_t i = 0; i < table.s->keys; i++) {

    if ((ecode = btree.create(identifier.getPath().c_str(), i)) != 0)

      return ecode;

  }

  /* Write the table definition to system table. */

  if ((ecode = dict.open_system_table(identifier.getPath(), HDBOWRITER)) != 0)

    return ecode;

  if (!dict.write_table_definition(proto)) {

    dict.close_system_table();

    return HA_ERR_CRASHED_ON_USAGE;

  }

  dict.close_system_table();

  return 0;

}

int BlitzEngine::doRenameTable(drizzled::Session &,

                               const drizzled::TableIdentifier &from,

                               const drizzled::TableIdentifier &to) {

  int rv = 0;

  BlitzData blitz_table;

  uint32_t nkeys;

  /* Find out the number of indexes in this table. This information

     is required because BlitzDB creates a file for each indexes.*/

  if (blitz_table.open_data_table(from.getPath().c_str(), HDBOREADER) != 0)

    return HA_ERR_CRASHED_ON_USAGE;

  nkeys = blitz_table.read_meta_keycount();

  if (blitz_table.close_data_table() != 0)

    return HA_ERR_CRASHED_ON_USAGE;

  /* We're now ready to rename the file(s) for this table. Start by

     attempting to rename the data and system files. */

  if (rename_file_ext(from.getPath().c_str(),

                      to.getPath().c_str(), BLITZ_DATA_EXT)) {

    if ((rv = errno) != ENOENT)

      return rv;

  }

  if (rename_file_ext(from.getPath().c_str(),

                      to.getPath().c_str(), BLITZ_SYSTEM_EXT)) {

    if ((rv = errno) != ENOENT)

      return rv;

  }

  /* So far so good. Rename the index file(s) and we're done. */

  BlitzTree btree;

  for (uint32_t i = 0; i < nkeys; i++) {

    if (btree.rename(from.getPath().c_str(), to.getPath().c_str(), i) != 0)

      return HA_ERR_CRASHED_ON_USAGE;

  }

  return rv;

}

int BlitzEngine::doDropTable(drizzled::Session &,

                             const drizzled::TableIdentifier &identifier) {

  BlitzData dict;

  BlitzTree btree;

  char buf[FN_REFLEN];

  uint32_t nkeys;

  int err;

  /* We open the dictionary to extract meta data from it */

  if ((err = dict.open_data_table(identifier.getPath().c_str(),

                                  HDBOREADER)) != 0) {

    return err;

  }

  nkeys = dict.read_meta_keycount();

  /* We no longer need the dictionary to be open */

  dict.close_data_table();

  /* Drop the Data Dictionary */

  snprintf(buf, FN_REFLEN, "%s%s", identifier.getPath().c_str(), BLITZ_DATA_EXT);

  if ((err = unlink(buf)) == -1) {

    return err;

  }

  /* Drop the System Table */

  snprintf(buf, FN_REFLEN, "%s%s", identifier.getPath().c_str(), BLITZ_SYSTEM_EXT);

  if ((err = unlink(buf)) == -1) {

    return err;

  }

  /* Drop Index file(s) */

  for (uint32_t i = 0; i < nkeys; i++) {

    if ((err = btree.drop(identifier.getPath().c_str(), i)) != 0) {

      return err;

    }

  }

  return 0;

}

int BlitzEngine::doGetTableDefinition(drizzled::Session &,

                                      const drizzled::TableIdentifier &identifier,

                                      drizzled::message::Table &proto) {

  struct stat stat_info;

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

  path.append(BLITZ_SYSTEM_EXT);

  if (stat(path.c_str(), &stat_info)) {

    return errno;

  }

  BlitzData db;

  char *proto_string;

  int proto_string_len;

  if (db.open_system_table(identifier.getPath(), HDBOREADER) != 0) {

    return HA_ERR_CRASHED_ON_USAGE;

  }

  proto_string = db.get_system_entry(BLITZ_TABLE_PROTO_KEY.c_str(),

                                     BLITZ_TABLE_PROTO_KEY.length(),

                                     &proto_string_len);

  if (db.close_system_table() != 0) {

    return HA_ERR_CRASHED_ON_USAGE;

  }

  if (proto_string == NULL) {

    return ENOMEM;

  }

  if (!proto.ParseFromArray(proto_string, proto_string_len)) {

    free(proto_string);

    return HA_ERR_CRASHED_ON_USAGE;

  }

  free(proto_string);

  return EEXIST;

}

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

                                        const drizzled::SchemaIdentifier &schema_id,

                                        drizzled::TableIdentifiers &ids) {

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

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

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

    drizzled::CachedDirectory::Entry *entry = *entry_iter;

    const std::string *filename = &entry->filename;

    assert(filename->size());

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

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

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

    } else {

      char uname[NAME_LEN + 1];

      uint32_t file_name_len;

      file_name_len = TableIdentifier::filename_to_tablename(filename->c_str(),

                                                             uname,

                                                             sizeof(uname));

      uname[file_name_len - sizeof(BLITZ_DATA_EXT) + 1]= '\0';

      ids.push_back(TableIdentifier(schema_id, uname));

    }

  }

}

bool BlitzEngine::doDoesTableExist(drizzled::Session &,

                                   const drizzled::TableIdentifier &identifier) {

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

  proto_path.append(BLITZ_DATA_EXT);

  return (access(proto_path.c_str(), F_OK)) ? false : true;

}

bool BlitzEngine::validateCreateTableOption(const std::string &key,

                                            const std::string &state) {

  if (key == "ESTIMATED_ROWS" || key == "estimated_rows") {

    if (str_is_numeric(state))

      return true;

  }

  return false;

}

bool BlitzEngine::doCreateTableCache(void) {

  return ((blitz_table_cache = tcmapnew()) == NULL) ? false : true;

}

BlitzShare *BlitzEngine::getTableShare(const std::string &table_name) {

  int vlen;

  const void *fetched;

  BlitzShare *rv = NULL;

  fetched = tcmapget(blitz_table_cache, table_name.c_str(),

                     table_name.length(), &vlen);

  /* dereference the object */

  if (fetched)

    rv = *(BlitzShare **)fetched;

  return rv;

}

void BlitzEngine::cacheTableShare(const std::string &table_name,

                                  BlitzShare *share) {

  /* Cache the memory address of the share object */

  tcmapput(blitz_table_cache, table_name.c_str(), table_name.length(),

           &share, sizeof(share));

}

void BlitzEngine::deleteTableShare(const std::string &table_name) {

  tcmapout2(blitz_table_cache, table_name.c_str());

}

ha_blitz::ha_blitz(drizzled::plugin::StorageEngine &engine_arg,

                   TableShare &table_arg) : Cursor(engine_arg, table_arg),

                                            btree_cursor(NULL),

                                            table_scan(false),

                                            table_based(false),

                                            thread_locked(false),

                                            held_key(NULL),

                                            held_key_len(0),

                                            current_key(NULL),

                                            current_key_len(0),

                                            key_buffer(NULL),

                                            errkey_id(0) {}

int ha_blitz::open(const char *table_name, int, uint32_t) {

  if ((share = get_share(table_name)) == NULL)

    return HA_ERR_CRASHED_ON_USAGE;

  pthread_mutex_lock(&blitz_utility_mutex);

  btree_cursor = new BlitzCursor[share->nkeys];

  for (uint32_t i = 0; i < share->nkeys; i++) {

    if (!share->btrees[i].create_cursor(&btree_cursor[i])) {

      free_share();

      pthread_mutex_unlock(&blitz_utility_mutex);

      return HA_ERR_OUT_OF_MEM;

    }

  }

  if ((key_buffer = (char *)malloc(BLITZ_MAX_KEY_LEN)) == NULL) {

    free_share();

    pthread_mutex_unlock(&blitz_utility_mutex);

    return HA_ERR_OUT_OF_MEM;

  }

  if ((key_merge_buffer = (char *)malloc(BLITZ_MAX_KEY_LEN)) == NULL) {

    free_share();

    pthread_mutex_unlock(&blitz_utility_mutex);

    return HA_ERR_OUT_OF_MEM;

  }

  if ((held_key_buf = (char *)malloc(BLITZ_MAX_KEY_LEN)) == NULL) {

    free_share();

    free(key_buffer);

    free(key_merge_buffer);

    pthread_mutex_unlock(&blitz_utility_mutex);

    return HA_ERR_OUT_OF_MEM;

  }

  secondary_row_buffer = NULL;

  secondary_row_buffer_size = 0;

  key_merge_buffer_len = BLITZ_MAX_KEY_LEN;

  /* 'ref_length' determines the size of the buffer that the kernel

     will use to uniquely identify a row. The actual allocation is

     done by the kernel so all we do here is specify the size of it.*/

  if (share->primary_key_exists) {

    ref_length = table->key_info[table->s->getPrimaryKey()].key_length;

  } else {

    ref_length = sizeof(held_key_len) + sizeof(uint64_t);

  }

  pthread_mutex_unlock(&blitz_utility_mutex);

  return 0;

}

int ha_blitz::close(void) {

  for (uint32_t i = 0; i < share->nkeys; i++) {

    share->btrees[i].destroy_cursor(&btree_cursor[i]);

  }

  delete [] btree_cursor;

  free(key_buffer);

  free(key_merge_buffer);

  free(held_key_buf);

  free(secondary_row_buffer);

  return free_share();

}

int ha_blitz::info(uint32_t flag) {

  if (flag & HA_STATUS_VARIABLE) {

    stats.records = share->dict.nrecords();

    stats.data_file_length = share->dict.table_size();

  }

  if (flag & HA_STATUS_AUTO)

    stats.auto_increment_value = share->auto_increment_value + 1;

  if (flag & HA_STATUS_ERRKEY)

    errkey = errkey_id;

  return 0;

}

int ha_blitz::doStartTableScan(bool scan) {

  /* Obtain the query type for this scan */

  sql_command_type = session_sql_command(table->getSession());

  table_scan = scan;

  table_based = true;

  /* Obtain the most suitable lock for the given statement type. */

  blitz_optimal_lock();

  /* Get the first record from TCHDB. Let the scanner take

     care of checking return value errors. */

  if (table_scan) {

    current_key = share->dict.next_key_and_row(NULL, 0,

                                               &current_key_len,

                                               &current_row,

                                               &current_row_len);

  }

  return 0;

}

int ha_blitz::rnd_next(unsigned char *drizzle_buf) {

  char *next_key;

  const char *next_row;

  int next_row_len;

  int next_key_len;

  free(held_key);

  held_key = NULL;

  if (current_key == NULL) {

    table->status = STATUS_NOT_FOUND;

    return HA_ERR_END_OF_FILE;

  }

  ha_statistic_increment(&system_status_var::ha_read_rnd_next_count);

  /* Unpack and copy the current row to Drizzle's result buffer. */

  unpack_row(drizzle_buf, current_row, current_row_len);

  /* Retrieve both key and row of the next record with one allocation. */

  next_key = share->dict.next_key_and_row(current_key, current_key_len,

                                          &next_key_len, &next_row,

                                          &next_row_len);

  /* Memory region for "current_row" will be freed as "held key" on

     the next iteration. This is because "current_key" points to the

     region of memory that contains "current_row" and "held_key" points

     to it. If there isn't another iteration then it is freed in doEndTableScan(). */

  current_row = next_row;

  current_row_len = next_row_len;

  /* Remember the current row because delete, update or replace

     function could be called after this function. This pointer is

     also used to free the previous key and row, which resides on

     the same buffer. */

  held_key = current_key;

  held_key_len = current_key_len;

  /* It is now memory-leak-safe to point current_key to next_key. */

  current_key = next_key;

  current_key_len = next_key_len;

  table->status = 0;

  return 0;

}

int ha_blitz::doEndTableScan() {

  if (table_scan && current_key)

    free(current_key);

  if (table_scan && held_key)

    free(held_key);

  current_key = NULL;

  held_key = NULL;

  current_key_len = 0;

  held_key_len = 0;

  table_scan = false;

  table_based = false;

  if (thread_locked)

    blitz_optimal_unlock();

  return 0;

}

int ha_blitz::rnd_pos(unsigned char *copy_to, unsigned char *pos) {

  char *row;

  char *key = NULL;

  int key_len, row_len;

  memcpy(&key_len, pos, sizeof(key_len));

  key = (char *)(pos + sizeof(key_len));

  /* TODO: Find a better error type. */

  if (key == NULL)

    return HA_ERR_KEY_NOT_FOUND;

  row = share->dict.get_row(key, key_len, &row_len);

  if (row == NULL)

    return HA_ERR_KEY_NOT_FOUND;

  unpack_row(copy_to, row, row_len);

  /* Remember the key location on memory if the thread is not doing

     a table scan. This is because either update_row() or delete_row()

     might be called after this function. */

  if (!table_scan) {

    held_key = key;

    held_key_len = key_len;

  }

  free(row);

  return 0;

}

void ha_blitz::position(const unsigned char *) {

  int length = sizeof(held_key_len);

  memcpy(ref, &held_key_len, length);

  memcpy(ref + length, (unsigned char *)held_key, held_key_len);

}

const char *ha_blitz::index_type(uint32_t /*key_num*/) {

  return "BTREE";

}

int ha_blitz::doStartIndexScan(uint32_t key_num, bool) {

  active_index = key_num;

  sql_command_type = session_sql_command(table->getSession());

  /* This is unlikely to happen but just for assurance, re-obtain

     the lock if this thread already has a certain lock. This makes

     sure that this thread will get the most appropriate lock for

     the current statement. */

  if (thread_locked)

    blitz_optimal_unlock();

  blitz_optimal_lock();

  return 0;

}

int ha_blitz::index_first(unsigned char *buf) {

  char *dict_key, *bt_key, *row;

  int dict_klen, bt_klen, prefix_len, rlen;

  bt_key = btree_cursor[active_index].first_key(&bt_klen);

  if (bt_key == NULL)

    return HA_ERR_END_OF_FILE;

  prefix_len = btree_key_length(bt_key, active_index);

  dict_key = skip_btree_key(bt_key, prefix_len, &dict_klen);

  if ((row = share->dict.get_row(dict_key, dict_klen, &rlen)) == NULL) {

    free(bt_key);

    return HA_ERR_KEY_NOT_FOUND;

  }

  unpack_row(buf, row, rlen);

  keep_track_of_key(bt_key, bt_klen);

  free(bt_key);

  free(row);

  return 0;

}

int ha_blitz::index_next(unsigned char *buf) {

  char *dict_key, *bt_key, *row;

  int dict_klen, bt_klen, prefix_len, rlen;

  bt_key = btree_cursor[active_index].next_key(&bt_klen);

  if (bt_key == NULL) {

    table->status = STATUS_NOT_FOUND;

    return HA_ERR_END_OF_FILE;

  }

  prefix_len = btree_key_length(bt_key, active_index);

  dict_key = skip_btree_key(bt_key, prefix_len, &dict_klen);

  if ((row = share->dict.get_row(dict_key, dict_klen, &rlen)) == NULL) {