~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 Implementation of the Session class and API

 */

#include <config.h>

#include <boost/checked_delete.hpp>

#include <boost/filesystem.hpp>

#include <boost/ptr_container/ptr_container.hpp>

#include <drizzled/copy_field.h>

#include <drizzled/data_home.h>

#include <drizzled/diagnostics_area.h>

#include <drizzled/display.h>

#include <drizzled/drizzled.h>

#include <drizzled/error.h>

#include <drizzled/gettext.h>

#include <drizzled/ha_data.h>

#include <drizzled/identifier.h>

#include <drizzled/internal/iocache.h>

#include <drizzled/internal/thread_var.h>

#include <drizzled/item/cache.h>

#include <drizzled/item/empty_string.h>

#include <drizzled/item/float.h>

#include <drizzled/item/return_int.h>

#include <drizzled/item/subselect.h>

#include <drizzled/lock.h>

#include <drizzled/open_tables_state.h>

#include <drizzled/plugin/authentication.h>

#include <drizzled/plugin/authorization.h>

#include <drizzled/plugin/client.h>

#include <drizzled/plugin/event_observer.h>

#include <drizzled/plugin/logging.h>

#include <drizzled/plugin/query_rewrite.h>

#include <drizzled/plugin/scheduler.h>

#include <drizzled/plugin/transactional_storage_engine.h>

#include <drizzled/probes.h>

#include <drizzled/pthread_globals.h>

#include <drizzled/query_id.h>

#include <drizzled/schema.h>

#include <drizzled/select_dump.h>

#include <drizzled/select_exists_subselect.h>

#include <drizzled/select_export.h>

#include <drizzled/select_max_min_finder_subselect.h>

#include <drizzled/select_singlerow_subselect.h>

#include <drizzled/select_subselect.h>

#include <drizzled/select_to_file.h>

#include <drizzled/session.h>

#include <drizzled/session/cache.h>

#include <drizzled/session/state.h>

#include <drizzled/session/table_messages.h>

#include <drizzled/session/times.h>

#include <drizzled/session/transactions.h>

#include <drizzled/show.h>

#include <drizzled/sql_base.h>

#include <drizzled/sql_lex.h>

#include <drizzled/system_variables.h>

#include <drizzled/statement.h>

#include <drizzled/statistics_variables.h>

#include <drizzled/table/singular.h>

#include <drizzled/table_proto.h>

#include <drizzled/tmp_table_param.h>

#include <drizzled/transaction_services.h>

#include <drizzled/user_var_entry.h>

#include <drizzled/util/backtrace.h>

#include <drizzled/util/find_ptr.h>

#include <drizzled/util/functors.h>

#include <drizzled/util/storable.h>

#include <plugin/myisam/myisam.h>

#include <algorithm>

#include <climits>

#include <fcntl.h>

#include <sys/stat.h>

using namespace std;

namespace fs= boost::filesystem;

namespace drizzled {

const char* const Session::DEFAULT_WHERE= "field list";

uint64_t g_refresh_version = 1;

bool Key_part_spec::operator==(const Key_part_spec& other) const

{

  return length == other.length &&

         field_name.length == other.field_name.length &&

    !my_strcasecmp(system_charset_info, field_name.str, other.field_name.str);

}

Open_tables_state::Open_tables_state(Session& session, uint64_t version_arg) :

  version(version_arg),

  session_(session)

{

  open_tables_= temporary_tables= derived_tables= NULL;

  extra_lock= lock= NULL;

}

/*

  The following functions form part of the C plugin API

*/

int tmpfile(const char *prefix)

{

  char filename[FN_REFLEN];

  int fd = internal::create_temp_file(filename, drizzle_tmpdir.c_str(), prefix, MYF(MY_WME));

  if (fd >= 0)

    unlink(filename);

  return fd;

}

void **Session::getEngineData(const plugin::MonitoredInTransaction *monitored)

{

  return static_cast<void **>(&ha_data[monitored->getId()].ha_ptr);

}

ResourceContext& Session::getResourceContext(const plugin::MonitoredInTransaction& monitored, size_t index)

{

  return ha_data[monitored.getId()].resource_context[index];

}

int64_t session_test_options(const Session *session, int64_t test_options)

{

  return session->options & test_options;

}

class Session::impl_c

{

public:

  typedef boost::unordered_map<std::string, util::Storable*, util::insensitive_hash, util::insensitive_equal_to> properties_t;

  typedef std::map<std::string, plugin::EventObserverList*> schema_event_observers_t;

  impl_c(Session& session) :

    open_tables(session, g_refresh_version),

    schema(boost::make_shared<std::string>())

  {

  }

  ~impl_c()

  {

    BOOST_FOREACH(properties_t::reference it, properties)

      delete it.second;

  }

  Diagnostics_area diagnostics;

  memory::Root mem_root;

  /**

    The lex to hold the parsed tree of conventional (non-prepared) queries.

    Whereas for prepared and stored procedure statements we use an own lex

    instance for each new query, for conventional statements we reuse

    the same lex. (@see mysql_parse for details).

  */

  LEX lex;

  Open_tables_state open_tables;

  properties_t properties;

  schema_event_observers_t schema_event_observers;

  system_status_var status_var;

  session::TableMessages table_message_cache;

  util::string::mptr schema;

  boost::shared_ptr<session::State> state;

  boost::ptr_vector<table::Singular> temporary_shares;

	session::Times times;

	session::Transactions transaction;

  drizzle_system_variables variables;

};

Session::Session(plugin::Client *client_arg, catalog::Instance::shared_ptr catalog_arg) :

  impl_(new impl_c(*this)),

  mem(impl_->mem_root),

  mem_root(&impl_->mem_root),

  query(new std::string),

  scheduler(NULL),

  variables(impl_->variables),

  status_var(impl_->status_var),

  lock_id(&main_lock_id),

  thread_stack(NULL),

  _where(Session::DEFAULT_WHERE),

  mysys_var(0),

  command(COM_CONNECT),

  ha_data(plugin::num_trx_monitored_objects),

  query_id(0),

  warn_query_id(0),

	transaction(impl_->transaction),

  open_tables(impl_->open_tables),

	times(impl_->times),

  first_successful_insert_id_in_prev_stmt(0),

  first_successful_insert_id_in_cur_stmt(0),

  limit_found_rows(0),

  options(session_startup_options),

  row_count_func(-1),

  sent_row_count(0),

  examined_row_count(0),

  used_tables(0),

  total_warn_count(0),

  row_count(0),

  thread_id(0),

  tmp_table(0),

  _global_read_lock(NONE),

  count_cuted_fields(CHECK_FIELD_ERROR_FOR_NULL),

  _killed(NOT_KILLED),

  no_errors(false),

  is_fatal_error(false),

  transaction_rollback_request(false),

  is_fatal_sub_stmt_error(0),

  derived_tables_processing(false),

  m_lip(NULL),

  arg_of_last_insert_id_function(false),

  _catalog(catalog_arg),

  transaction_message(NULL),

  statement_message(NULL),

  session_event_observers(NULL),

  xa_id(0),

  concurrent_execute_allowed(true),

  tablespace_op(false),

  use_usage(false),

  security_ctx(identifier::User::make_shared()),

  originating_server_uuid_set(false),

  client(client_arg)

{

  client->setSession(this);

  /*

    Pass nominal parameters to init only to ensure that

    the destructor works OK in case of an error. The main_mem_root

    will be re-initialized in init_for_queries().

  */

  mem.init(memory::ROOT_MIN_BLOCK_SIZE);

  cuted_fields= sent_row_count= row_count= 0L;

  // Must be reset to handle error with Session's created for init of mysqld

  lex().current_select= 0;

  memset(&variables, 0, sizeof(variables));

  scoreboard_index= -1;

  originating_server_uuid= "";

  originating_commit_id= 0;

  cleanup_done= abort_on_warning= no_warnings_for_error= false;

  /* query_cache init */

  query_cache_key= "";

  resultset= NULL;

  /* Variables with default values */

  proc_info="login";

  plugin_sessionvar_init(this);

  /*

    variables= global_system_variables above has reset

    variables.pseudo_thread_id to 0. We need to correct it here to

    avoid temporary tables replication failure.

  */

  variables.pseudo_thread_id= thread_id;

  server_status= SERVER_STATUS_AUTOCOMMIT;

  if (variables.max_join_size == HA_POS_ERROR)

    options |= OPTION_BIG_SELECTS;

  else

    options &= ~OPTION_BIG_SELECTS;

  open_options=ha_open_options;

  update_lock_default= TL_WRITE;

  session_tx_isolation= (enum_tx_isolation) variables.tx_isolation;

  memset(warn_count, 0, sizeof(warn_count));

  memset(&status_var, 0, sizeof(status_var));

  /* Initialize sub structures */

  warn_root.init(WARN_ALLOC_BLOCK_SIZE);

  substitute_null_with_insert_id = false;

  lock_info.init(); /* safety: will be reset after start */

  thr_lock_owner_init(&main_lock_id, &lock_info);

  plugin::EventObserver::registerSessionEvents(*this);

}

Diagnostics_area& Session::main_da()

{

  return impl_->diagnostics;

}

const LEX& Session::lex() const

{

  return impl_->lex;

}

LEX& Session::lex()

{

  return impl_->lex;

}

enum_sql_command Session::getSqlCommand() const

{

  return lex().sql_command;

}

session::TableMessages& Session::getMessageCache()

{

  return impl_->table_message_cache;

}

void statement::Statement::set_command(enum_sql_command v)

{

	session().lex().sql_command= v;

}

LEX& statement::Statement::lex()

{

	return session().lex();

}

session::Transactions& statement::Statement::transaction()

{

	return session().transaction;

}

void Session::add_item_to_list(Item *item)

{

  lex().current_select->add_item_to_list(this, item);

}

void Session::add_value_to_list(Item *value)

{

	lex().value_list.push_back(value);

}

void Session::add_order_to_list(Item *item, bool asc)

{

  lex().current_select->add_order_to_list(this, item, asc);

}

void Session::add_group_to_list(Item *item, bool asc)

{

  lex().current_select->add_group_to_list(this, item, asc);

}

void Session::free_items()

{

  /* This works because items are allocated with memory::sql_alloc() */

  for (Item* next; free_list; free_list= next)

  {

    next= free_list->next;

    free_list->delete_self();

  }

}

void Session::setAbort(bool arg)

{

  mysys_var->abort= arg;

}

void Session::lockOnSys()

{

  if (not mysys_var)

    return;

  setAbort(true);

  boost::mutex::scoped_lock scopedLock(mysys_var->mutex);

  if (mysys_var->current_cond)

  {

    mysys_var->current_mutex->lock();

    mysys_var->current_cond->notify_all();

    mysys_var->current_mutex->unlock();

  }

}

void Session::get_xid(DrizzleXid *xid)

{

  *xid = *(DrizzleXid *) &transaction.xid_state.xid;

}

/* Do operations that may take a long time */

void Session::cleanup()

{

  assert(not cleanup_done);

  setKilled(KILL_CONNECTION);

#ifdef ENABLE_WHEN_BINLOG_WILL_BE_ABLE_TO_PREPARE

  if (transaction.xid_state.xa_state == XA_PREPARED)

  {

#error xid_state in the cache should be replaced by the allocated value

  }

#endif

  {

    TransactionServices::rollbackTransaction(*this, true);

  }

  BOOST_FOREACH(UserVars::reference iter, user_vars)

    boost::checked_delete(iter.second);

  user_vars.clear();

  open_tables.close_temporary_tables();

  if (global_read_lock)

    unlockGlobalReadLock();

  cleanup_done= true;

}

Session::~Session()

{

  if (client and client->isConnected())

  {

    assert(security_ctx);

    if (global_system_variables.log_warnings)

    {

      errmsg_printf(error::WARN, ER(ER_FORCING_CLOSE),

                    internal::my_progname,

                    thread_id,

                    security_ctx->username().c_str());

    }

    disconnect();

  }

  /* Close connection */

  if (client)

  {

    client->close();

    boost::checked_delete(client);

    client= NULL;

  }

  if (not cleanup_done)

    cleanup();

  plugin::StorageEngine::closeConnection(*this);

  plugin_sessionvar_cleanup(this);

  warn_root.free_root(MYF(0));

  mysys_var=0;					// Safety (shouldn't be needed)

  impl_->mem_root.free_root(MYF(0));

  setCurrentMemRoot(NULL);

  setCurrentSession(NULL);

  plugin::Logging::postEndDo(this);

  plugin::EventObserver::deregisterSessionEvents(session_event_observers); 

	BOOST_FOREACH(impl_c::schema_event_observers_t::reference it, impl_->schema_event_observers)

    plugin::EventObserver::deregisterSchemaEvents(it.second);

}

void Session::setClient(plugin::Client *client_arg)

{

  client= client_arg;

  client->setSession(this);

}

void Session::awake(Session::killed_state_t state_to_set)

{

  if (state_to_set == Session::KILL_QUERY && command == COM_SLEEP)

    return;

  setKilled(state_to_set);

  scheduler->killSession(this);

  if (state_to_set != Session::KILL_QUERY)

  {

    DRIZZLE_CONNECTION_DONE(thread_id);

  }

  if (mysys_var)

  {

    boost::mutex::scoped_lock scopedLock(mysys_var->mutex);

    /*

      "

      This broadcast could be up in the air if the victim thread

      exits the cond in the time between read and broadcast, but that is

      ok since all we want to do is to make the victim thread get out

      of waiting on current_cond.

      If we see a non-zero current_cond: it cannot be an old value (because

      then exit_cond() should have run and it can't because we have mutex); so

      it is the true value but maybe current_mutex is not yet non-zero (we're

      in the middle of enter_cond() and there is a "memory order

      inversion"). So we test the mutex too to not lock 0.

      Note that there is a small chance we fail to kill. If victim has locked

      current_mutex, but hasn't yet entered enter_cond() (which means that

      current_cond and current_mutex are 0), then the victim will not get

      a signal and it may wait "forever" on the cond (until

      we issue a second KILL or the status it's waiting for happens).

      It's true that we have set its session->killed but it may not

      see it immediately and so may have time to reach the cond_wait().

    */

    if (mysys_var->current_cond && mysys_var->current_mutex)

    {

      mysys_var->current_mutex->lock();

      mysys_var->current_cond->notify_all();

      mysys_var->current_mutex->unlock();

    }

  }

}

/*

  Remember the location of thread info, the structure needed for

  memory::sql_alloc() and the structure for the net buffer

*/

void Session::storeGlobals()

{

  /*

    Assert that thread_stack is initialized: it's necessary to be able

    to track stack overrun.

  */

  assert(thread_stack);

  setCurrentSession(this);

  setCurrentMemRoot(&mem);

  mysys_var= my_thread_var;

  /*

    Let mysqld define the thread id (not mysys)

    This allows us to move Session to different threads if needed.

  */

  mysys_var->id= thread_id;

  /*

    We have to call thr_lock_info_init() again here as Session may have been

    created in another thread

  */

  lock_info.init();

}

/*

  Init Session for query processing.

  This has to be called once before we call mysql_parse.

  See also comments in session.h.

*/

void Session::prepareForQueries()

{

  if (variables.max_join_size == HA_POS_ERROR)

    options |= OPTION_BIG_SELECTS;

  open_tables.version= g_refresh_version;

  set_proc_info(NULL);

  command= COM_SLEEP;

  times.set_time();

  mem.reset_defaults(variables.query_alloc_block_size, variables.query_prealloc_size);

  transaction.xid_state.xid.set_null();

  transaction.xid_state.in_session=1;

  if (use_usage)

    resetUsage();

}

void Session::run()

{

  storeGlobals();

  if (authenticate())

  {

    disconnect();

    return;

  }

  prepareForQueries();

  while (not client->haveError() && getKilled() != KILL_CONNECTION)

  {

    if (not executeStatement())

      break;

  }

  disconnect();

}

bool Session::schedule(const shared_ptr& arg)

{

  arg->scheduler= plugin::Scheduler::getScheduler();

  assert(arg->scheduler);

  ++connection_count;

  long current_connections= connection_count;

  if (current_connections > 0 and static_cast<uint64_t>(current_connections) > current_global_counters.max_used_connections)

  {

    current_global_counters.max_used_connections= static_cast<uint64_t>(connection_count);

  }

  current_global_counters.connections++;

  arg->thread_id= arg->variables.pseudo_thread_id= global_thread_id++;

  session::Cache::insert(arg);

  if (unlikely(plugin::EventObserver::connectSession(*arg)))

  {

    // We should do something about an error...

  }

  if (plugin::Scheduler::getScheduler()->addSession(arg))

  {

    DRIZZLE_CONNECTION_START(arg->getSessionId());

    char error_message_buff[DRIZZLE_ERRMSG_SIZE];

    arg->setKilled(Session::KILL_CONNECTION);

    arg->status_var.aborted_connects++;

    /* Can't use my_error() since store_globals has not been called. */

    /* TODO replace will better error message */

    snprintf(error_message_buff, sizeof(error_message_buff), ER(ER_CANT_CREATE_THREAD), 1);

    arg->client->sendError(ER_CANT_CREATE_THREAD, error_message_buff);

    return true;

  }

  return false;

}

/*

  Is this session viewable by the current user?

*/

bool Session::isViewable(const identifier::User& user_arg) const

{

  return plugin::Authorization::isAuthorized(user_arg, *this, false);

}

const char* Session::enter_cond(boost::condition_variable_any &cond, boost::mutex &mutex, const char* msg)

{

  const char* old_msg = get_proc_info();

  safe_mutex_assert_owner(mutex);

  mysys_var->current_mutex = &mutex;

  mysys_var->current_cond = &cond;

  this->set_proc_info(msg);

  return old_msg;

}

void Session::exit_cond(const char* old_msg)