session.h

/* -*- 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
 */

#pragma once

#include <algorithm>
#include <bitset>
#include <boost/make_shared.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/thread/condition_variable.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/shared_mutex.hpp>
#include <boost/thread/thread.hpp>
#include <map>
#include <netdb.h>
#include <string>
#include <sys/resource.h>
#include <sys/time.h>

#include <drizzled/charset.h>
#include <drizzled/base.h>
#include <drizzled/error.h>
#include <drizzled/lock.h>
#include <drizzled/pthread_globals.h>
#include <drizzled/sql_error.h>
#include <drizzled/sql_locale.h>
#include <drizzled/visibility.h>
#include <drizzled/util/find_ptr.h>
#include <drizzled/util/string.h>
#include <drizzled/type/time.h>

namespace drizzled {

extern uint32_t server_id;
extern std::string server_uuid;

#define TC_HEURISTIC_RECOVER_COMMIT   1
#define TC_HEURISTIC_RECOVER_ROLLBACK 2
extern uint32_t tc_heuristic_recover;

extern DRIZZLED_API struct drizzle_system_variables global_system_variables;

class Open_tables_state;

class DRIZZLED_API Session
{
private:
  class impl_c;

  boost::scoped_ptr<impl_c> impl_;
public:
  typedef boost::shared_ptr<Session> shared_ptr;

  static shared_ptr make_shared(plugin::Client *client, boost::shared_ptr<catalog::Instance> instance_arg)
  {
    assert(instance_arg);
    return boost::make_shared<Session>(client, instance_arg);
  }

  /*
    MARK_COLUMNS_NONE:  Means mark_used_colums is not set and no indicator to
                        handler of fields used is set
    MARK_COLUMNS_READ:  Means a bit in read set is set to inform handler
                  that the field is to be read. If field list contains
                        duplicates, then session->dup_field is set to point
                        to the last found duplicate.
    MARK_COLUMNS_WRITE: Means a bit is set in write set to inform handler
      that it needs to update this field in write_row
                        and update_row.
  */
  enum_mark_columns mark_used_columns;

  void free_items();

  Item *free_list;
  memory::Root& mem;
  memory::Root* mem_root; 
  uint64_t getXaId() const
  {
    return xa_id;
  }

  void setXaId(uint64_t in_xa_id)
  {
    xa_id= in_xa_id;
  }

public:
  Diagnostics_area& main_da();
  const LEX& lex() const;
  LEX& lex();
  enum_sql_command getSqlCommand() const;

  typedef boost::shared_ptr<const std::string> QueryString;

private:
  boost::shared_ptr<std::string> query;

  // Never allow for a modification of this outside of the class. c_str()
  // requires under some setup non const, you must copy the QueryString in
  // order to use it.
public:
  void resetQueryString();
  const boost::shared_ptr<session::State>& state();

  QueryString getQueryString() const
  {
    return query;
  }

  const char* getQueryStringCopy(size_t &length)
  {
    QueryString tmp_string(getQueryString());
    if (not tmp_string)
    {
      length= 0;
      return NULL;
    }
    length= tmp_string->length();
    return mem.strdup(*tmp_string);
  }

  util::string::ptr schema() const;

  static const char* const DEFAULT_WHERE;

  memory::Root warn_root; 
public:
  void setClient(plugin::Client *client_arg);

  plugin::Client *getClient() const
  {
    return client;
  }

  plugin::Scheduler* scheduler; 
  typedef boost::unordered_map<std::string, user_var_entry*, util::insensitive_hash, util::insensitive_equal_to> UserVars;

private:
  typedef std::pair< UserVars::iterator, UserVars::iterator > UserVarsRange;
  UserVars user_vars; 
public:
  const UserVars &getUserVariables() const
  {
    return user_vars;
  }

  drizzle_system_variables& variables; 
  enum_tx_isolation getTxIsolation() const;
  system_status_var& status_var;

  THR_LOCK_INFO lock_info; 
  THR_LOCK_OWNER main_lock_id; 
  THR_LOCK_OWNER *lock_id; 
  char *thread_stack;

  identifier::user::ptr user() const
  {
    return security_ctx;
  }

  void setUser(identifier::user::mptr arg)
  {
    security_ctx= arg;
  }

  int32_t getScoreboardIndex() const
  {
    return scoreboard_index;
  }

  void setScoreboardIndex(int32_t in_scoreboard_index)
  {
    scoreboard_index= in_scoreboard_index;
  }

  bool isOriginatingServerUUIDSet() const
  {
    return originating_server_uuid_set;
  }

  void setOriginatingServerUUID(std::string in_originating_server_uuid)
  {
    originating_server_uuid= in_originating_server_uuid;
    originating_server_uuid_set= true;
  }

  const std::string &getOriginatingServerUUID() const
  {
    return originating_server_uuid;
  }

  void setOriginatingCommitID(uint64_t in_originating_commit_id)
  {
    originating_commit_id= in_originating_commit_id;
  }

  uint64_t getOriginatingCommitID() const
  {
    return originating_commit_id;
  }

  bool isViewable(const identifier::User&) const;

private:
  const char *_where;

public:
  const char *where() const
  {
    return _where;
  }

  void setWhere(const char *arg)
  {
    _where= arg;
  }

  /*
    One thread can hold up to one named user-level lock. This variable
    points to a lock object if the lock is present. See item_func.cc and
    chapter 'Miscellaneous functions', for functions GET_LOCK, RELEASE_LOCK.
  */

private:
  boost::thread::id boost_thread_id;
  thread_ptr _thread;
  boost::this_thread::disable_interruption *interrupt;

  internal::st_my_thread_var *mysys_var;

public:
  thread_ptr &getThread()
  {
    return _thread;
  }

  void pushInterrupt(boost::this_thread::disable_interruption *interrupt_arg)
  {
    interrupt= interrupt_arg;
  }

  boost::this_thread::disable_interruption &getThreadInterupt()
  {
    assert(interrupt);
    return *interrupt;
  }

  internal::st_my_thread_var *getThreadVar()
  {
    return mysys_var;
  }

  enum_server_command command;

  thr_lock_type update_lock_default;

  /*
    Both of the following container points in session will be converted to an API.
  */

private:
  /* container for handler's private per-connection data */
  std::vector<Ha_data> ha_data;
  /*
    Id of current query. Statement can be reused to execute several queries
    query_id is global in context of the whole MySQL server.
    ID is automatically generated from an atomic counter.
    It's used in Cursor code for various purposes: to check which columns
    from table are necessary for this select, to check if it's necessary to
    update auto-updatable fields (like auto_increment and timestamp).
  */
  query_id_t query_id;
  query_id_t warn_query_id;

public:
  void **getEngineData(const plugin::MonitoredInTransaction *monitored);
  ResourceContext& getResourceContext(const plugin::MonitoredInTransaction&, size_t index= 0);

  session::Transactions& transaction;
  Open_tables_state& open_tables;
  session::Times& times;

  Field *dup_field;
  sigset_t signals;

public:
  // As of right now we do not allow a concurrent execute to launch itself
  void setConcurrentExecute(bool arg)
  {
    concurrent_execute_allowed= arg;
  }

  bool isConcurrentExecuteAllowed() const
  {
    return concurrent_execute_allowed;
  }

  /*
    ALL OVER THIS FILE, "insert_id" means "*automatically generated* value for
    insertion into an auto_increment column".
  */
  uint64_t first_successful_insert_id_in_prev_stmt;
  uint64_t first_successful_insert_id_in_cur_stmt;
  uint64_t limit_found_rows;
  uint64_t options; 
  int64_t row_count_func; 
  int64_t rowCount() const
  {
    return row_count_func;
  }

  ha_rows cuted_fields; 
  ha_rows sent_row_count;

  ha_rows examined_row_count;

  table_map used_tables;

  uint32_t warn_count[(uint32_t) DRIZZLE_ERROR::WARN_LEVEL_END];
  uint32_t total_warn_count;

  uint32_t row_count;

  session_id_t thread_id;
  uint32_t tmp_table;
  enum global_read_lock_t
  {
    NONE= 0,
    GOT_GLOBAL_READ_LOCK= 1,
    MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT= 2
  };
private:
  global_read_lock_t _global_read_lock;

public:

  global_read_lock_t isGlobalReadLock() const
  {
    return _global_read_lock;
  }

  void setGlobalReadLock(global_read_lock_t arg)
  {
    _global_read_lock= arg;
  }

  DrizzleLock *lockTables(Table **tables, uint32_t count, uint32_t flags);
  bool lockGlobalReadLock();
  bool lock_table_names(TableList *table_list);
  bool lock_table_names_exclusively(TableList *table_list);
  bool makeGlobalReadLockBlockCommit();
  bool abortLockForThread(Table *table);
  bool wait_if_global_read_lock(bool abort_on_refresh, bool is_not_commit);
  int lock_table_name(TableList *table_list);
  void abortLock(Table *table);
  void removeLock(Table *table);
  void unlockReadTables(DrizzleLock *sql_lock);
  void unlockSomeTables(Table **table, uint32_t count);
  void unlockTables(DrizzleLock *sql_lock);
  void startWaitingGlobalReadLock();
  void unlockGlobalReadLock();

private:
  int unlock_external(Table **table, uint32_t count);
  int lock_external(Table **tables, uint32_t count);
  bool wait_for_locked_table_names(TableList *table_list);
  DrizzleLock *get_lock_data(Table **table_ptr, uint32_t count,
                             bool should_lock, Table **write_lock_used);
public:

  uint32_t server_status;
  uint32_t open_options;
  uint32_t select_number; 
  /* variables.transaction_isolation is reset to this after each commit */
  enum_tx_isolation session_tx_isolation;
  enum_check_fields count_cuted_fields;

  enum killed_state_t
  {
    NOT_KILLED,
    KILL_BAD_DATA,
    KILL_CONNECTION,
    KILL_QUERY,
    KILLED_NO_VALUE /* means none of the above states apply */
  };
private:
  killed_state_t volatile _killed;

public:

  void setKilled(killed_state_t arg)
  {
    _killed= arg;
  }

  killed_state_t getKilled() const
  {
    return _killed;
  }

  volatile killed_state_t *getKilledPtr() // Do not use this method, it is here for historical convience.
  {
    return &_killed;
  }

  bool is_admin_connection;
  bool no_errors;
  bool is_fatal_error;
  bool transaction_rollback_request;
  bool is_fatal_sub_stmt_error;
  bool substitute_null_with_insert_id;
  bool cleanup_done;

public:
  bool got_warning; 
  bool no_warnings_for_error; 
  bool derived_tables_processing;

  bool doing_tablespace_operation() const
  {
    return tablespace_op;
  }

  void setDoingTablespaceOperation(bool doing)
  {
    tablespace_op= doing;
  }


  union
  {
    bool bool_value;
    uint32_t uint32_t_value;
    int32_t int32_t_value;
    uint64_t uint64_t_value;
  } sys_var_tmp;

  Lex_input_stream *m_lip;

  void *session_marker;

  inline void set_proc_info(const char *info)
  {
    proc_info= info;
  }
  inline const char* get_proc_info() const
  {
    return proc_info;
  }

  inline void setQueryId(query_id_t in_query_id)
  {
    query_id= in_query_id;
  }

  query_id_t getQueryId()  const
  {
    return query_id;
  }


  inline void setWarningQueryId(query_id_t in_query_id)
  {
    warn_query_id= in_query_id;
  }

  inline query_id_t getWarningQueryId()  const
  {
    return warn_query_id;
  }

  inline session_id_t getSessionId()  const
  {
    return thread_id;
  }

  inline uint32_t getServerId()  const
  {
    /* We return the global server ID. */
    return server_id;
  }

  inline std::string &getServerUUID() const
  {
    return server_uuid;
  }

  inline void record_first_successful_insert_id_in_cur_stmt(uint64_t id_arg)
  {
    if (first_successful_insert_id_in_cur_stmt == 0)
      first_successful_insert_id_in_cur_stmt= id_arg;
  }
  inline uint64_t read_first_successful_insert_id_in_prev_stmt() const
  {
    return first_successful_insert_id_in_prev_stmt;
  }

  Session(plugin::Client*, boost::shared_ptr<catalog::Instance>);
  ~Session();

  void cleanup();
  void cleanup_after_query();
  void storeGlobals();
  void awake(Session::killed_state_t state_to_set);

  void prepareForQueries();

  bool executeStatement();

  void readAndStoreQuery(const char *in_packet, uint32_t in_packet_length);

  bool endTransaction(enum enum_mysql_completiontype completion);
  bool endActiveTransaction();
  bool startTransaction(start_transaction_option_t opt= START_TRANS_NO_OPTIONS);
  void markTransactionForRollback(bool all);

  bool authenticate();
  void run();
  static bool schedule(const Session::shared_ptr&);
  static void unlink(session_id_t&);
  static void unlink(const Session::shared_ptr&);

  /*
    For enter_cond() / exit_cond() to work the mutex must be got before
    enter_cond(); this mutex is then released by exit_cond().
    Usage must be: lock mutex; enter_cond(); your code; exit_cond().
  */
  const char* enter_cond(boost::condition_variable_any &cond, boost::mutex &mutex, const char* msg);
  void exit_cond(const char* old_msg);

  uint64_t found_rows() const
  {
    return limit_found_rows;
  }

  bool inTransaction() const
  {
    return server_status & SERVER_STATUS_IN_TRANS;
  }

  lex_string_t* make_lex_string(lex_string_t*, str_ref);

  void send_explain_fields(select_result*);

  void clear_error(bool full= false);
  void clearDiagnostics();
  bool is_error() const;

  static const charset_info_st *charset() { return default_charset_info; }

  void end_statement();
  inline int killed_errno() const
  {
    killed_state_t killed_val; /* to cache the volatile 'killed' */
    return (killed_val= _killed) != KILL_BAD_DATA ? killed_val : 0;
  }
  void send_kill_message() const;
  /* return true if we will abort query if we make a warning now */
  inline bool abortOnWarning()
  {
    return abort_on_warning;
  }

  inline void setAbortOnWarning(bool arg)
  {
    abort_on_warning= arg;
  }

  void setAbort(bool arg);
  void lockOnSys();
  void set_status_var_init();
  void set_schema(const std::string&);

  /*
    Copy the current database to the argument. Use the current arena to
    allocate memory for a deep copy: current database may be freed after
    a statement is parsed but before it's executed.
  */
  str_ref copy_db_to() const;

public:

  void reset_for_next_command();

  void disconnect(enum error_t errcode= EE_OK);

  bool checkUser(const std::string &passwd, const std::string &db);

  message::Transaction *getTransactionMessage() const
  {
    return transaction_message;
  }

  message::Statement *getStatementMessage() const
  {
    return statement_message;
  }

  message::Resultset *getResultsetMessage() const
  {
    return resultset;
  }
  void setTransactionMessage(message::Transaction *in_message)
  {
    transaction_message= in_message;
  }

  void setStatementMessage(message::Statement *in_message)
  {
    statement_message= in_message;
  }

  void setResultsetMessage(message::Resultset *in_message)
  {
    resultset= in_message;
  }
  void resetResultsetMessage()
  {
    resultset= NULL;
  }

  plugin::EventObserverList *getSessionObservers()
  {
    return session_event_observers;
  }

  void setSessionObservers(plugin::EventObserverList *observers)
  {
    session_event_observers= observers;
  }

  plugin::EventObserverList* getSchemaObservers(const std::string& schema);
  plugin::EventObserverList* setSchemaObservers(const std::string& schema, plugin::EventObserverList*);

public:
  void my_ok(ha_rows affected_rows= 0, ha_rows found_rows_arg= 0, uint64_t passed_id= 0, const char *message= NULL);
  void my_eof();
  void add_item_to_list(Item *item);
  void add_value_to_list(Item *value);
  void add_order_to_list(Item *item, bool asc);
  void add_group_to_list(Item *item, bool asc);

  void refresh_status();
  user_var_entry* getVariable(str_ref name, bool create_if_not_exists);
  void setVariable(const std::string& name, const std::string& value);

  void close_thread_tables();
  void close_old_data_files(bool morph_locks= false, bool send_refresh= false);
  void close_data_files_and_morph_locks(const identifier::Table&);

  void close_tables_for_reopen(TableList **tables);


  bool openTablesLock(TableList*);
  Table *open_temporary_table(const identifier::Table &identifier, bool link_in_list= true);

  int open_tables_from_list(TableList **start, uint32_t *counter, uint32_t flags= 0);

  Table *openTableLock(TableList *table_list, thr_lock_type lock_type);
  Table *openTable(TableList *table_list, bool *refresh, uint32_t flags= 0);

  void unlink_open_table(Table *find);
  void drop_open_table(Table *table, const identifier::Table &identifier);
  void close_cached_table(Table *table);

  /* Create a lock in the cache */
  table::Placeholder& table_cache_insert_placeholder(const identifier::Table&);
  Table* lock_table_name_if_not_cached(const identifier::Table&);

  session::TableMessages &getMessageCache();

  /* Reopen operations */
  bool reopen_tables();
  bool close_cached_tables(TableList *tables, bool wait_for_refresh, bool wait_for_placeholders);

  void wait_for_condition(boost::mutex &mutex, boost::condition_variable_any &cond);
  int setup_conds(TableList *leaves, COND **conds);
  int lock_tables(TableList *tables, uint32_t count, bool *need_reopen);

  template <class T>
  T* getProperty(const std::string& name)
  {
    return static_cast<T*>(getProperty0(name));
  }

  template <class T>
  T setProperty(const std::string& name, T value)
  {
    setProperty0(name, value);
    return value;
  }

  plugin::StorageEngine *getDefaultStorageEngine();
  void get_xid(DrizzleXid *xid) const; // Innodb only

  table::Singular& getInstanceTable();
  table::Singular& getInstanceTable(std::list<CreateField>&);

  void setUsage(bool arg)
  {
    use_usage= arg;
  }

  const rusage &getUsage()
  {
    return usage;
  }

  const catalog::Instance& catalog() const
  {
    return *_catalog;
  }

  catalog::Instance& catalog()
  {
    return *_catalog;
  }

  bool arg_of_last_insert_id_function; // Tells if LAST_INSERT_ID(#) was called for the current statement
private:
  drizzled::util::Storable* getProperty0(const std::string&);
  void setProperty0(const std::string&, drizzled::util::Storable*);

  bool resetUsage()
  {
    return not getrusage(RUSAGE_THREAD, &usage);
  }

  boost::shared_ptr<catalog::Instance> _catalog;

  message::Transaction *transaction_message;
  message::Statement *statement_message;
  /* Pointer to the current resultset of Select query */
  message::Resultset *resultset;
  plugin::EventObserverList *session_event_observers;

  uint64_t xa_id;
  const char *proc_info;
  bool abort_on_warning;
  bool concurrent_execute_allowed;
  bool tablespace_op; 
  bool use_usage;
  rusage usage;
  identifier::user::mptr security_ctx;
  int32_t scoreboard_index;
  bool originating_server_uuid_set;
  std::string originating_server_uuid;
  uint64_t originating_commit_id;
  plugin::Client *client;
};

#define ESCAPE_CHARS "ntrb0ZN" // keep synchronous with READ_INFO::unescape

/* Bits in sql_command_flags */

enum sql_command_flag_bits
{
  CF_BIT_CHANGES_DATA,
  CF_BIT_HAS_ROW_COUNT,
  CF_BIT_STATUS_COMMAND,
  CF_BIT_SHOW_TABLE_COMMAND,
  CF_BIT_WRITE_LOGS_COMMAND,
  CF_BIT_SIZE
};

static const std::bitset<CF_BIT_SIZE> CF_CHANGES_DATA(1 << CF_BIT_CHANGES_DATA);
static const std::bitset<CF_BIT_SIZE> CF_HAS_ROW_COUNT(1 << CF_BIT_HAS_ROW_COUNT);
static const std::bitset<CF_BIT_SIZE> CF_STATUS_COMMAND(1 << CF_BIT_STATUS_COMMAND);
static const std::bitset<CF_BIT_SIZE> CF_SHOW_TABLE_COMMAND(1 << CF_BIT_SHOW_TABLE_COMMAND);
static const std::bitset<CF_BIT_SIZE> CF_WRITE_LOGS_COMMAND(1 << CF_BIT_WRITE_LOGS_COMMAND);

namespace display  
{
  const std::string &type(Session::global_read_lock_t);
  size_t max_string_length(Session::global_read_lock_t);
} /* namespace display */

} /* namespace drizzled */