~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

 */

#include "config.h"

#include <fcntl.h>

#include <unistd.h>

#include <string>

#include <vector>

#include <set>

#include <fstream>

#include <algorithm>

#include <functional>

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

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

#include "drizzled/cached_directory.h"

#include <drizzled/definitions.h>

#include <drizzled/base.h>

#include <drizzled/cursor.h>

#include <drizzled/plugin/storage_engine.h>

#include <drizzled/session.h>

#include <drizzled/error.h>

#include <drizzled/gettext.h>

#include <drizzled/unireg.h>

#include <drizzled/data_home.h>

#include "drizzled/errmsg_print.h"

#include "drizzled/xid.h"

#include "drizzled/sql_table.h"

#include "drizzled/global_charset_info.h"

#include "drizzled/charset.h"

#include "drizzled/internal/my_sys.h"

#include "drizzled/db.h"

#include <drizzled/table_proto.h>

#include <drizzled/plugin/event_observer.h>

#include <drizzled/table/shell.h>

#include "drizzled/message/cache.h"

#include <boost/algorithm/string/compare.hpp>

static bool shutdown_has_begun= false; // Once we put in the container for the vector/etc for engines this will go away.

namespace drizzled

{

namespace plugin

{

static EngineVector vector_of_engines;

static EngineVector vector_of_schema_engines;

const std::string DEFAULT_STRING("default");

const std::string UNKNOWN_STRING("UNKNOWN");

const std::string DEFAULT_DEFINITION_FILE_EXT(".dfe");

static std::set<std::string> set_of_table_definition_ext;

EngineVector &StorageEngine::getSchemaEngines()

{

  return vector_of_schema_engines;

}

StorageEngine::StorageEngine(const std::string name_arg,

                             const std::bitset<HTON_BIT_SIZE> &flags_arg) :

  Plugin(name_arg, "StorageEngine"),

  MonitoredInTransaction(), /* This gives the storage engine a "slot" or ID */

  flags(flags_arg)

{

}

StorageEngine::~StorageEngine()

{

}

void StorageEngine::setTransactionReadWrite(Session& session)

{

  TransactionContext &statement_ctx= session.transaction.stmt;

  statement_ctx.markModifiedNonTransData();

}

int StorageEngine::renameTable(Session &session, const identifier::Table &from, const identifier::Table &to)

{

  int error;

  setTransactionReadWrite(session);

  if (unlikely(plugin::EventObserver::beforeRenameTable(session, from, to)))

  {

    error= ER_EVENT_OBSERVER_PLUGIN;

  }

  else

  {

    error =  doRenameTable(session, from, to);

    if (unlikely(plugin::EventObserver::afterRenameTable(session, from, to, error)))

    {

      error= ER_EVENT_OBSERVER_PLUGIN;

    }

  }

  return error;

}

/**

  Delete all files with extension from bas_ext().

  @param name		Base name of table

  @note

    We assume that the Cursor may return more extensions than

    was actually used for the file.

  @retval

    0   If we successfully deleted at least one file from base_ext and

    didn't get any other errors than ENOENT

  @retval

    !0  Error

*/

int StorageEngine::doDropTable(Session&, const identifier::Table &identifier)

{

  int error= 0;

  int enoent_or_zero= ENOENT;                   // Error if no file was deleted

  char buff[FN_REFLEN];

  for (const char **ext= bas_ext(); *ext ; ext++)

  {

    internal::fn_format(buff, identifier.getPath().c_str(), "", *ext,

                        MY_UNPACK_FILENAME|MY_APPEND_EXT);

    if (internal::my_delete_with_symlink(buff, MYF(0)))

    {

      if ((error= errno) != ENOENT)

        break;

    }

    else

    {

      enoent_or_zero= 0;                        // No error for ENOENT

    }

    error= enoent_or_zero;

  }

  return error;

}

bool StorageEngine::addPlugin(StorageEngine *engine)

{

  vector_of_engines.push_back(engine);

  if (engine->getTableDefinitionFileExtension().length())

  {

    assert(engine->getTableDefinitionFileExtension().length() == DEFAULT_DEFINITION_FILE_EXT.length());

    set_of_table_definition_ext.insert(engine->getTableDefinitionFileExtension());

  }

  if (engine->check_flag(HTON_BIT_SCHEMA_DICTIONARY))

    vector_of_schema_engines.push_back(engine);

  return false;

}

void StorageEngine::removePlugin(StorageEngine *)

{

  if (shutdown_has_begun == false)

  {

    vector_of_engines.clear();

    vector_of_schema_engines.clear();

    shutdown_has_begun= true;

  }

}

class FindEngineByName

  : public std::unary_function<StorageEngine *, bool>

{

  const std::string &predicate;

public:

  explicit FindEngineByName(const std::string &target_arg) :

    predicate(target_arg)

  {

  }

  result_type operator() (argument_type engine)

  {

    return boost::iequals(engine->getName(), predicate);

  }

};

StorageEngine *StorageEngine::findByName(const std::string &predicate)

{

  EngineVector::iterator iter= std::find_if(vector_of_engines.begin(),

                                            vector_of_engines.end(),

                                            FindEngineByName(predicate));

  if (iter != vector_of_engines.end())

  {

    StorageEngine *engine= *iter;

    if (engine->is_user_selectable())

      return engine;

  }

  return NULL;

}

StorageEngine *StorageEngine::findByName(Session& session, const std::string &predicate)

{

  if (boost::iequals(predicate, DEFAULT_STRING))

    return session.getDefaultStorageEngine();

  EngineVector::iterator iter= std::find_if(vector_of_engines.begin(),

                                            vector_of_engines.end(),

                                            FindEngineByName(predicate));

  if (iter != vector_of_engines.end())

  {

    StorageEngine *engine= *iter;

    if (engine->is_user_selectable())

      return engine;

  }

  return NULL;

}

class StorageEngineCloseConnection : public std::unary_function<StorageEngine *, void>

{

  Session *session;

public:

  StorageEngineCloseConnection(Session *session_arg) : session(session_arg) {}

  /*

    there's no need to rollback here as all transactions must

    be rolled back already

  */

  inline result_type operator() (argument_type engine)

  {

    if (*session->getEngineData(engine))

      engine->close_connection(session);

  }

};

/**

  @note

    don't bother to rollback here, it's done already

*/

void StorageEngine::closeConnection(Session* session)

{

  std::for_each(vector_of_engines.begin(), vector_of_engines.end(),

                StorageEngineCloseConnection(session));

}

bool StorageEngine::flushLogs(StorageEngine *engine)

{

  if (engine == NULL)

  {

    if (std::find_if(vector_of_engines.begin(), vector_of_engines.end(),

                     std::mem_fun(&StorageEngine::flush_logs))

        != vector_of_engines.begin())

      return true;

  }

  else

  {

    if (engine->flush_logs())

      return true;

  }

  return false;

}

class StorageEngineGetTableDefinition: public std::unary_function<StorageEngine *,bool>

{

  Session& session;

  const identifier::Table &identifier;

  message::Table &table_message;

  drizzled::error_t &err;

public:

  StorageEngineGetTableDefinition(Session& session_arg,

                                  const identifier::Table &identifier_arg,

                                  message::Table &table_message_arg,

                                  drizzled::error_t &err_arg) :

    session(session_arg), 

    identifier(identifier_arg),

    table_message(table_message_arg), 

    err(err_arg) {}

  result_type operator() (argument_type engine)

  {

    int ret= engine->doGetTableDefinition(session, identifier, table_message);

    if (ret != ENOENT)

      err= static_cast<drizzled::error_t>(ret);

    return err == static_cast<drizzled::error_t>(EEXIST) or err != static_cast<drizzled::error_t>(ENOENT);

  }

};

class StorageEngineDoesTableExist: public std::unary_function<StorageEngine *, bool>

{

  Session& session;

  const identifier::Table &identifier;

public:

  StorageEngineDoesTableExist(Session& session_arg, const identifier::Table &identifier_arg) :

    session(session_arg), 

    identifier(identifier_arg) 

  { }

  result_type operator() (argument_type engine)

  {

    return engine->doDoesTableExist(session, identifier);

  }

};

/**

  Utility method which hides some of the details of getTableDefinition()

*/

bool plugin::StorageEngine::doesTableExist(Session &session,

                                           const identifier::Table &identifier,

                                           bool include_temporary_tables)

{

  if (include_temporary_tables)

  {

    if (session.doDoesTableExist(identifier))

      return true;

  }

  EngineVector::iterator iter=

    std::find_if(vector_of_engines.begin(), vector_of_engines.end(),

                 StorageEngineDoesTableExist(session, identifier));

  if (iter == vector_of_engines.end())

  {

    return false;

  }

  return true;

}

bool plugin::StorageEngine::doDoesTableExist(Session&, const drizzled::identifier::Table&)

{

  std::cerr << " Engine was called for doDoesTableExist() and does not implement it: " << this->getName() << "\n";

  assert(0);

  return false;

}

/**

  Call this function in order to give the Cursor the possiblity

  to ask engine if there are any new tables that should be written to disk

  or any dropped tables that need to be removed from disk

*/

int StorageEngine::getTableDefinition(Session& session,

                                      const identifier::Table &identifier,

                                      message::table::shared_ptr &table_message,

                                      bool include_temporary_tables)

{

  drizzled::error_t err= static_cast<drizzled::error_t>(ENOENT);

  if (include_temporary_tables)

  {

    Table *table= session.find_temporary_table(identifier);

    if (table)

    {

      table_message.reset(new message::Table(*table->getShare()->getTableMessage()));

      return EEXIST;

    }

  }

  drizzled::message::table::shared_ptr table_ptr;

  if ((table_ptr= drizzled::message::Cache::singleton().find(identifier)))

  {

    table_message= table_ptr;

  }

  message::Table message;

  EngineVector::iterator iter=

    std::find_if(vector_of_engines.begin(), vector_of_engines.end(),

                 StorageEngineGetTableDefinition(session, identifier, message, err));

  if (iter == vector_of_engines.end())

  {

    return ENOENT;

  }

  table_message.reset(new message::Table(message));

 drizzled::message::Cache::singleton().insert(identifier, table_message);

  return err;

}

message::table::shared_ptr StorageEngine::getTableMessage(Session& session,

                                                          identifier::Table::const_reference identifier,

                                                          drizzled::error_t &error,

                                                          bool include_temporary_tables)

{

  error= static_cast<drizzled::error_t>(ENOENT);

  if (include_temporary_tables)

  {

    Table *table= session.find_temporary_table(identifier);

    if (table)

    {

      error= EE_OK;

      return message::table::shared_ptr(new message::Table(*table->getShare()->getTableMessage()));

    }

  }

  drizzled::message::table::shared_ptr table_ptr;

  if ((table_ptr= drizzled::message::Cache::singleton().find(identifier)))

  {

    (void)table_ptr;

  }

  message::Table message;

  EngineVector::iterator iter=

    std::find_if(vector_of_engines.begin(), vector_of_engines.end(),

                 StorageEngineGetTableDefinition(session, identifier, message, error));

  if (iter == vector_of_engines.end())

  {

    error= static_cast<drizzled::error_t>(ENOENT);

    return message::table::shared_ptr();

  }

  message::table::shared_ptr table_message(new message::Table(message));

  drizzled::message::Cache::singleton().insert(identifier, table_message);

  return table_message;

}

/**

  An interceptor to hijack the text of the error message without

  setting an error in the thread. We need the text to present it

  in the form of a warning to the user.

*/

class Ha_delete_table_error_handler: public Internal_error_handler

{

public:

  Ha_delete_table_error_handler() : Internal_error_handler() {}

  virtual bool handle_error(drizzled::error_t sql_errno,

                            const char *message,

                            DRIZZLE_ERROR::enum_warning_level level,

                            Session *session);

  char buff[DRIZZLE_ERRMSG_SIZE];

};

bool

Ha_delete_table_error_handler::

handle_error(drizzled::error_t ,

             const char *message,

             DRIZZLE_ERROR::enum_warning_level ,

             Session *)

{

  /* Grab the error message */

  strncpy(buff, message, sizeof(buff)-1);

  return true;

}

class DropTableByIdentifier: public std::unary_function<EngineVector::value_type, bool>

{

  Session::reference session;

  identifier::Table::const_reference identifier;

  drizzled::error_t &error;

public:

  DropTableByIdentifier(Session::reference session_arg,

                        identifier::Table::const_reference identifier_arg,

                        drizzled::error_t &error_arg) :

    session(session_arg),

    identifier(identifier_arg),

    error(error_arg)

  { }

  result_type operator() (argument_type engine)

  {

    if (not engine->doDoesTableExist(session, identifier))

      return false;

    int local_error= engine->doDropTable(session, identifier);

    if (not local_error)

      return true;

    switch (local_error)

    {

    case HA_ERR_NO_SUCH_TABLE:

    case ENOENT:

      error= static_cast<drizzled::error_t>(HA_ERR_NO_SUCH_TABLE);

      return false;

    default:

      error= static_cast<drizzled::error_t>(local_error);

      return true;

    }

  } 

};

bool StorageEngine::dropTable(Session::reference session,

                              identifier::Table::const_reference identifier,

                              drizzled::error_t &error)

{

  error= EE_OK;

  EngineVector::const_iterator iter= std::find_if(vector_of_engines.begin(), vector_of_engines.end(),

                                                  DropTableByIdentifier(session, identifier, error));

  if (error)

  {

    return false;

  }

  else if (iter == vector_of_engines.end())

  {

    error= ER_BAD_TABLE_ERROR;

    return false;

  }

  drizzled::message::Cache::singleton().erase(identifier);

  return true;

}

bool StorageEngine::dropTable(Session& session,

                              const identifier::Table &identifier)

{

  drizzled::error_t error;

  if (not dropTable(session, identifier, error))

  {

    return false;

  }

  return true;

}

bool StorageEngine::dropTable(Session::reference session,

                              StorageEngine &engine,

                              identifier::Table::const_reference identifier,

                              drizzled::error_t &error)

{

  error= EE_OK;

  engine.setTransactionReadWrite(session);

  assert(identifier.isTmp());

  if (unlikely(plugin::EventObserver::beforeDropTable(session, identifier)))

  {

    error= ER_EVENT_OBSERVER_PLUGIN;

  }

  else

  {

    error= static_cast<drizzled::error_t>(engine.doDropTable(session, identifier));

    if (unlikely(plugin::EventObserver::afterDropTable(session, identifier, error)))

    {

      error= ER_EVENT_OBSERVER_PLUGIN;

    }

  }

  drizzled::message::Cache::singleton().erase(identifier);

  if (error)

  {

    return false;

  }

  return true;

}

/**

  Initiates table-file and calls appropriate database-creator.

  @retval

   0  ok

  @retval

   1  error

*/

bool StorageEngine::createTable(Session &session,

                                const identifier::Table &identifier,

                                message::Table& table_message)

{

  drizzled::error_t error= EE_OK;

  TableShare share(identifier);

  table::Shell table(share);

  message::Table tmp_proto;

  if (share.parse_table_proto(session, table_message) || share.open_table_from_share(&session, identifier, "", 0, 0, table))

  { 

    // @note Error occured, we should probably do a little more here.

    // ER_CORRUPT_TABLE_DEFINITION,ER_CORRUPT_TABLE_DEFINITION_ENUM 

    my_error(ER_CORRUPT_TABLE_DEFINITION_UNKNOWN, identifier);

    return false;

  }

  else

  {

    /* Check for legal operations against the Engine using the proto (if used) */

    if (table_message.type() == message::Table::TEMPORARY &&

        share.storage_engine->check_flag(HTON_BIT_TEMPORARY_NOT_SUPPORTED) == true)

    {

      error= HA_ERR_UNSUPPORTED;

    }

    else if (table_message.type() != message::Table::TEMPORARY &&

             share.storage_engine->check_flag(HTON_BIT_TEMPORARY_ONLY) == true)

    {

      error= HA_ERR_UNSUPPORTED;

    }

    else

    {

      share.storage_engine->setTransactionReadWrite(session);

      error= static_cast<drizzled::error_t>(share.storage_engine->doCreateTable(session,

                                                                                table,

                                                                                identifier,

                                                                                table_message));

    }

    if (error == ER_TABLE_PERMISSION_DENIED)

    {

      my_error(ER_TABLE_PERMISSION_DENIED, identifier);

    }

    else if (error)

    {

      std::string path;

      identifier.getSQLPath(path);

      my_error(ER_CANT_CREATE_TABLE, MYF(ME_BELL+ME_WAITTANG), path.c_str(), error);

    }

    table.delete_table();

  }

  return(error == EE_OK);

}

Cursor *StorageEngine::getCursor(Table &arg)

{

  return create(arg);

}

class AddTableIdentifier : 

  public std::unary_function<StorageEngine *, void>

{

  CachedDirectory &directory;