~drizzle-trunk/drizzle/development

/* Copyright (C) 2000-2003 MySQL AB

   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>

#define DRIZZLE_LEX 1

#include <drizzled/item/num.h>

#include <drizzled/abort_exception.h>

#include <drizzled/error.h>

#include <drizzled/nested_join.h>

#include <drizzled/query_id.h>

#include <drizzled/transaction_services.h>

#include <drizzled/sql_parse.h>

#include <drizzled/data_home.h>

#include <drizzled/sql_base.h>

#include <drizzled/show.h>

#include <drizzled/function/time/unix_timestamp.h>

#include <drizzled/function/get_system_var.h>

#include <drizzled/item/cmpfunc.h>

#include <drizzled/item/null.h>

#include <drizzled/session.h>

#include <drizzled/session/cache.h>

#include <drizzled/sql_load.h>

#include <drizzled/lock.h>

#include <drizzled/select_send.h>

#include <drizzled/plugin/client.h>

#include <drizzled/statement.h>

#include <drizzled/statement/alter_table.h>

#include <drizzled/probes.h>

#include <drizzled/global_charset_info.h>

#include <drizzled/plugin/logging.h>

#include <drizzled/plugin/query_rewrite.h>

#include <drizzled/plugin/query_cache.h>

#include <drizzled/plugin/authorization.h>

#include <drizzled/optimizer/explain_plan.h>

#include <drizzled/pthread_globals.h>

#include <drizzled/plugin/event_observer.h>

#include <drizzled/display.h>

#include <drizzled/visibility.h>

#include <drizzled/kill.h>

#include <drizzled/schema.h>

#include <drizzled/item/subselect.h>

#include <drizzled/diagnostics_area.h>

#include <drizzled/table_ident.h>

#include <drizzled/statistics_variables.h>

#include <drizzled/system_variables.h>

#include <drizzled/session/transactions.h>

#include <limits.h>

#include <bitset>

#include <algorithm>

#include <boost/date_time.hpp>

#include <drizzled/internal/my_sys.h>

using namespace std;

extern int base_sql_parse(drizzled::Session *session); // from sql_yacc.cc

namespace drizzled {

/* Prototypes */

bool my_yyoverflow(short **a, ParserType **b, ulong *yystacksize);

static bool parse_sql(Session *session, Lex_input_stream *lip);

void parse(Session *session, const char *inBuf, uint32_t length);

/**

  @defgroup Runtime_Environment Runtime Environment

  @{

*/

extern size_t my_thread_stack_size;

extern const CHARSET_INFO *character_set_filesystem;

namespace

{

static const std::string command_name[COM_END+1]={

  "Sleep",

  "Quit",

  "Init DB",

  "Query",

  "Shutdown",

  "Connect",

  "Ping",

  "Kill",

  "Error"  // Last command number

};

}

const char *xa_state_names[]={

  "NON-EXISTING", "ACTIVE", "IDLE", "PREPARED"

};

/**

  Mark all commands that somehow changes a table.

  This is used to check number of updates / hour.

  sql_command is actually set to SQLCOM_END sometimes

  so we need the +1 to include it in the array.

  See COMMAND_FLAG_xxx for different type of commands

     2  - query that returns meaningful ROW_COUNT() -

          a number of modified rows

*/

bitset<CF_BIT_SIZE> sql_command_flags[SQLCOM_END+1];

const std::string &getCommandName(const enum_server_command& command)

{

  return command_name[command];

}

void init_update_queries(void)

{

  uint32_t x;

  for (x= 0; x <= SQLCOM_END; x++)

    sql_command_flags[x].reset();

  sql_command_flags[SQLCOM_CREATE_TABLE]=   CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_CREATE_INDEX]=   CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_ALTER_TABLE]=    CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND;

  sql_command_flags[SQLCOM_TRUNCATE]=       CF_CHANGES_DATA | CF_WRITE_LOGS_COMMAND;

  sql_command_flags[SQLCOM_DROP_TABLE]=     CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_LOAD]=           CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_CREATE_DB]=      CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_DROP_DB]=        CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_RENAME_TABLE]=   CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_DROP_INDEX]=     CF_CHANGES_DATA;

  sql_command_flags[SQLCOM_UPDATE]=	    CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_INSERT]=	    CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_INSERT_SELECT]=  CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_DELETE]=         CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_REPLACE]=        CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_REPLACE_SELECT]= CF_CHANGES_DATA | CF_HAS_ROW_COUNT;

  sql_command_flags[SQLCOM_SHOW_WARNS]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_ERRORS]= CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_CREATE_DB]=  CF_STATUS_COMMAND;

  sql_command_flags[SQLCOM_SHOW_CREATE]=  CF_STATUS_COMMAND;

  /*

    The following admin table operations are allowed

    on log tables.

  */

  sql_command_flags[SQLCOM_ANALYZE]=          CF_WRITE_LOGS_COMMAND;

}

/**

  Perform one connection-level (COM_XXXX) command.

  @param command         type of command to perform

  @param session             connection handle

  @param packet          data for the command, packet is always null-terminated

  @param packet_length   length of packet + 1 (to show that data is

                         null-terminated) except for COM_SLEEP, where it

                         can be zero.

  @todo

    set session->lex().sql_command to SQLCOM_END here.

  @todo

    The following has to be changed to an 8 byte integer

  @retval

    0   ok

  @retval

    1   request of thread shutdown, i. e. if command is

        COM_QUIT/COM_SHUTDOWN

*/

bool dispatch_command(enum_server_command command, Session *session,

                      char* packet, uint32_t packet_length)

{

  bool error= 0;

  Query_id &query_id= Query_id::get_query_id();

  DRIZZLE_COMMAND_START(session->thread_id, command);

  session->command= command;

  session->lex().sql_command= SQLCOM_END; /* to avoid confusing VIEW detectors */

  session->set_time();

  session->setQueryId(query_id.value());

  switch( command ) {

  /* Ignore these statements. */

  case COM_PING:

    break;

  /* Increase id and count all other statements. */

  default:

    session->status_var.questions++;

    query_id.next();

  }

  /* @todo set session->lex().sql_command to SQLCOM_END here */

  plugin::Logging::preDo(session);

  if (unlikely(plugin::EventObserver::beforeStatement(*session)))

  {

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

  }

  session->server_status&=

           ~(SERVER_QUERY_NO_INDEX_USED | SERVER_QUERY_NO_GOOD_INDEX_USED);

  switch (command) {

  case COM_INIT_DB:

  {

    if (packet_length == 0)

    {

      my_message(ER_NO_DB_ERROR, ER(ER_NO_DB_ERROR), MYF(0));

      break;

    }

    string tmp(packet, packet_length);

    identifier::Schema identifier(tmp);

    if (not schema::change(*session, identifier))

    {

      session->my_ok();

    }

    break;

  }

  case COM_QUERY:

  {

    if (not session->readAndStoreQuery(packet, packet_length))

      break;					// fatal error is set

    DRIZZLE_QUERY_START(session->getQueryString()->c_str(),

                        session->thread_id,

                        const_cast<const char *>(session->schema()->c_str()));

    parse(session, session->getQueryString()->c_str(), session->getQueryString()->length());

    break;

  }

  case COM_QUIT:

    /* We don't calculate statistics for this command */

    session->main_da().disable_status();              // Don't send anything back

    error= true;					// End server

    break;

  case COM_KILL:

    {

      if (packet_length != 4)

      {

        my_error(ER_NO_SUCH_THREAD, MYF(0), 0);

        break;

      }

      else

      {

        uint32_t kill_id;

        memcpy(&kill_id, packet, sizeof(uint32_t));

        kill_id= ntohl(kill_id);

        (void)drizzled::kill(*session->user(), kill_id, true);

      }

      session->my_ok();

      break;

    }

  case COM_SHUTDOWN:

  {

    session->status_var.com_other++;

    session->my_eof();

    session->close_thread_tables();			// Free before kill

    kill_drizzle();

    error= true;

    break;

  }

  case COM_PING:

    session->status_var.com_other++;

    session->my_ok();				// Tell client we are alive

    break;

  case COM_SLEEP:

  case COM_CONNECT:				// Impossible here

  case COM_END:

  default:

    my_message(ER_UNKNOWN_COM_ERROR, ER(ER_UNKNOWN_COM_ERROR), MYF(0));

    break;

  }

  /* If commit fails, we should be able to reset the OK status. */

  session->main_da().can_overwrite_status= true;

  TransactionServices &transaction_services= TransactionServices::singleton();

  transaction_services.autocommitOrRollback(*session, session->is_error());

  session->main_da().can_overwrite_status= false;

  session->transaction.stmt.reset();

  /* report error issued during command execution */

  if (session->killed_errno())

  {

    if (! session->main_da().is_set())

      session->send_kill_message();

  }

  if (session->getKilled() == Session::KILL_QUERY || session->getKilled() == Session::KILL_BAD_DATA)

  {

    session->setKilled(Session::NOT_KILLED);

    session->setAbort(false);

  }

  /* Can not be true, but do not take chances in production. */

  assert(! session->main_da().is_sent);

  switch (session->main_da().status())

  {

  case Diagnostics_area::DA_ERROR:

    /* The query failed, send error to log and abort bootstrap. */

    session->getClient()->sendError(session->main_da().sql_errno(),

                               session->main_da().message());

    break;

  case Diagnostics_area::DA_EOF:

    session->getClient()->sendEOF();

    break;

  case Diagnostics_area::DA_OK:

    session->getClient()->sendOK();

    break;

  case Diagnostics_area::DA_DISABLED:

    break;

  case Diagnostics_area::DA_EMPTY:

  default:

    session->getClient()->sendOK();

    break;

  }

  session->main_da().is_sent= true;

  session->set_proc_info("closing tables");

  /* Free tables */

  session->close_thread_tables();

  plugin::Logging::postDo(session);

  if (unlikely(plugin::EventObserver::afterStatement(*session)))

  {

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

  }

  /* Store temp state for processlist */

  session->set_proc_info("cleaning up");

  session->command= COM_SLEEP;

  session->resetQueryString();

  session->set_proc_info(NULL);

  session->mem_root->free_root(MYF(memory::KEEP_PREALLOC));

  if (DRIZZLE_QUERY_DONE_ENABLED() || DRIZZLE_COMMAND_DONE_ENABLED())

  {

    if (command == COM_QUERY)

    {

      DRIZZLE_QUERY_DONE(session->is_error());

    }

    DRIZZLE_COMMAND_DONE(session->is_error());

  }

  return error;

}

/**

  Create a TableList object for an INFORMATION_SCHEMA table.

    This function is used in the parser to convert a SHOW or DESCRIBE

    table_name command to a SELECT from INFORMATION_SCHEMA.

    It prepares a Select_Lex and a TableList object to represent the

    given command as a SELECT parse tree.

  @param session           thread handle

  @param lex               current lex

  @param table_ident       table alias if it's used

  @param schema_table_name the name of the INFORMATION_SCHEMA table to be

                           created

  @note

    Due to the way this function works with memory and LEX it cannot

    be used outside the parser (parse tree transformations outside

    the parser break PS and SP).

  @retval

    0                 success

  @retval

    1                 out of memory or SHOW commands are not allowed

                      in this version of the server.

*/

static bool _schema_select(Session *session, Select_Lex *sel,

                           const string& schema_table_name)

{

  LEX_STRING db, table;

  bitset<NUM_OF_TABLE_OPTIONS> table_options;

  /*

     We have to make non const db_name & table_name

     because of lower_case_table_names

  */

  session->make_lex_string(&db, "data_dictionary", sizeof("data_dictionary"), false);

  session->make_lex_string(&table, schema_table_name, false);

  if (not sel->add_table_to_list(session, new Table_ident(db, table), NULL, table_options, TL_READ))

  {

    return true;

  }

  return false;

}

int prepare_new_schema_table(Session *session, LEX& lex,

                             const string& schema_table_name)

{

  Select_Lex *schema_select_lex= NULL;

  Select_Lex *select_lex= lex.current_select;

  assert(select_lex);

  if (_schema_select(session, select_lex, schema_table_name))

  {

    return(1);

  }

  TableList *table_list= (TableList*) select_lex->table_list.first;

  assert(table_list);

  table_list->schema_select_lex= schema_select_lex;

  return 0;

}

/**

  Execute command saved in session and lex->sql_command.

    Before every operation that can request a write lock for a table

    wait if a global read lock exists. However do not wait if this

    thread has locked tables already. No new locks can be requested

    until the other locks are released. The thread that requests the

    global read lock waits for write locked tables to become unlocked.

    Note that wait_if_global_read_lock() sets a protection against a new

    global read lock when it succeeds. This needs to be released by

    start_waiting_global_read_lock() after the operation.

  @param session                       Thread handle

  @todo

    - Invalidate the table in the query cache if something changed

    after unlocking when changes become visible.

    TODO: this is workaround. right way will be move invalidating in

    the unlock procedure.

    - TODO: use check_change_password()

    - JOIN is not supported yet. TODO

    - SUSPEND and FOR MIGRATE are not supported yet. TODO

  @retval

    false       OK

  @retval

    true        Error

*/

static int execute_command(Session *session)

{

  bool res= false;

  /* first Select_Lex (have special meaning for many of non-SELECTcommands) */

  Select_Lex *select_lex= &session->lex().select_lex;

  /* list of all tables in query */

  TableList *all_tables;

  /*

    In many cases first table of main Select_Lex have special meaning =>

    check that it is first table in global list and relink it first in

    queries_tables list if it is necessary (we need such relinking only

    for queries with subqueries in select list, in this case tables of

    subqueries will go to global list first)

    all_tables will differ from first_table only if most upper Select_Lex

    do not contain tables.

    Because of above in place where should be at least one table in most

    outer Select_Lex we have following check:

    assert(first_table == all_tables);

    assert(first_table == all_tables && first_table != 0);

  */

  session->lex().first_lists_tables_same();

  /* should be assigned after making first tables same */

  all_tables= session->lex().query_tables;

  /* set context for commands which do not use setup_tables */

  select_lex->context.resolve_in_table_list_only((TableList*)select_lex->table_list.first);

  /*

    Reset warning count for each query that uses tables

    A better approach would be to reset this for any commands

    that is not a SHOW command or a select that only access local

    variables, but for now this is probably good enough.

    Don't reset warnings when executing a stored routine.

  */

  if (all_tables || ! session->lex().is_single_level_stmt())

  {

    drizzle_reset_errors(session, 0);

  }

  assert(session->transaction.stmt.hasModifiedNonTransData() == false);

  if (! (session->server_status & SERVER_STATUS_AUTOCOMMIT)

      && ! session->inTransaction()

      && session->lex().statement->isTransactional())

  {

    if (session->startTransaction() == false)

    {

      my_error(drizzled::ER_UNKNOWN_ERROR, MYF(0));

      return true;

    }

  }

  /* now we are ready to execute the statement */

  res= session->lex().statement->execute();

  session->set_proc_info("query end");

  /*

    The return value for ROW_COUNT() is "implementation dependent" if the

    statement is not DELETE, INSERT or UPDATE, but -1 is what JDBC and ODBC

    wants. We also keep the last value in case of SQLCOM_CALL or

    SQLCOM_EXECUTE.

  */

  if (! (sql_command_flags[session->lex().sql_command].test(CF_BIT_HAS_ROW_COUNT)))

  {

    session->row_count_func= -1;

  }

  return (res || session->is_error());

}

bool execute_sqlcom_select(Session *session, TableList *all_tables)

{

  LEX	*lex= &session->lex();

  select_result *result=lex->result;

  bool res= false;

  /* assign global limit variable if limit is not given */

  {

    Select_Lex *param= lex->unit.global_parameters;

    if (!param->explicit_limit)

      param->select_limit=

        new Item_int((uint64_t) session->variables.select_limit);

  }

  if (all_tables

      && ! (session->server_status & SERVER_STATUS_AUTOCOMMIT)

      && ! session->inTransaction()

      && ! lex->statement->isShow())

  {

    if (session->startTransaction() == false)

    {

      my_error(drizzled::ER_UNKNOWN_ERROR, MYF(0));

      return true;

    }

  }

  if (not (res= session->openTablesLock(all_tables)))

  {

    if (lex->describe)

    {

      /*

        We always use select_send for EXPLAIN, even if it's an EXPLAIN

        for SELECT ... INTO OUTFILE: a user application should be able

        to prepend EXPLAIN to any query and receive output for it,

        even if the query itself redirects the output.

      */

      if (!(result= new select_send()))

        return true;

      session->send_explain_fields(result);

      optimizer::ExplainPlan planner;

      res= planner.explainUnion(session, &session->lex().unit, result);

      if (lex->describe & DESCRIBE_EXTENDED)

      {

        char buff[1024];

        String str(buff,(uint32_t) sizeof(buff), system_charset_info);

        str.length(0);

        session->lex().unit.print(&str);

        str.append('\0');

        push_warning(session, DRIZZLE_ERROR::WARN_LEVEL_NOTE,

                     ER_YES, str.ptr());

      }

      if (res)

        result->abort();

      else

        result->send_eof();

      delete result;

    }

    else

    {

      if (!result && !(result= new select_send()))

        return true;

      /* Init the Query Cache plugin */

      plugin::QueryCache::prepareResultset(session);

      res= handle_select(session, lex, result, 0);

      /* Send the Resultset to the cache */

      plugin::QueryCache::setResultset(session);

      if (result != lex->result)

        delete result;

    }

  }

  return res;

}

#define MY_YACC_INIT 1000			// Start with big alloc

#define MY_YACC_MAX  32000			// Because of 'short'

bool my_yyoverflow(short **yyss, ParserType **yyvs, ulong *yystacksize)

{

  LEX	*lex= &current_session->lex();

  ulong old_info=0;

  if ((uint32_t) *yystacksize >= MY_YACC_MAX)

    return 1;

  if (!lex->yacc_yyvs)

    old_info= *yystacksize;

  *yystacksize= set_zone((*yystacksize)*2,MY_YACC_INIT,MY_YACC_MAX);

  unsigned char *tmpptr= NULL;

  if (!(tmpptr= (unsigned char *)realloc(lex->yacc_yyvs,

                                         *yystacksize* sizeof(**yyvs))))

      return 1;

  lex->yacc_yyvs= tmpptr;

  tmpptr= NULL;

  if (!(tmpptr= (unsigned char*)realloc(lex->yacc_yyss,

                                        *yystacksize* sizeof(**yyss))))

      return 1;

  lex->yacc_yyss= tmpptr;

  if (old_info)

  {						// Copy old info from stack

    memcpy(lex->yacc_yyss, *yyss, old_info*sizeof(**yyss));

    memcpy(lex->yacc_yyvs, *yyvs, old_info*sizeof(**yyvs));

  }

  *yyss=(short*) lex->yacc_yyss;

  *yyvs=(ParserType*) lex->yacc_yyvs;

  return 0;

}

void

init_select(LEX *lex)

{

  Select_Lex *select_lex= lex->current_select;

  select_lex->init_select();

  lex->wild= 0;

  if (select_lex == &lex->select_lex)

  {

    assert(lex->result == 0);

    lex->exchange= 0;

  }

}

bool

new_select(LEX *lex, bool move_down)

{

  Select_Lex *select_lex;

  Session *session= lex->session;

  if (!(select_lex= new (session->mem_root) Select_Lex()))

    return true;

  select_lex->select_number= ++session->select_number;

  select_lex->parent_lex= lex; /* Used in init_query. */