~drizzle-trunk/drizzle/development

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

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

 *

 *  Copyright (C) 2011 David Shrewsbury

 *

 *  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; either version 2 of the License, or

 *  (at your option) any later version.

 *

 *  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 <plugin/slave/queue_consumer.h>

#include <drizzled/errmsg_print.h>

#include <drizzled/execute.h>

#include <drizzled/message/transaction.pb.h>

#include <drizzled/message/statement_transform.h>

#include <drizzled/sql/result_set.h>

#include <string>

#include <vector>

#include <boost/thread.hpp>

#include <boost/lexical_cast.hpp>

#include <google/protobuf/text_format.h>

using namespace std;

using namespace drizzled;

namespace slave

{

bool QueueConsumer::init()

{

  setApplierState("", true);

  return true;

}

void QueueConsumer::shutdown()

{

  setApplierState(getErrorMessage(), false);

}

bool QueueConsumer::process()

{

  for (size_t index= 0; index < _master_ids.size(); index++)

  {

    /* We go ahead and get the string version of the master ID

     * so we don't have to keep converting it from int to string.

     */

    const string master_id= boost::lexical_cast<string>(_master_ids[index]);

    if (not processSingleMaster(master_id))

      return false;

  }

  return true;

}

bool QueueConsumer::processSingleMaster(const string &master_id)

{

  TrxIdList completedTransactionIds;

  getListOfCompletedTransactions(master_id, completedTransactionIds);

  for (size_t x= 0; x < completedTransactionIds.size(); x++)

  {

    string commit_id;

    string originating_server_uuid;

    uint64_t originating_commit_id= 0;

    uint64_t trx_id= completedTransactionIds[x];

    vector<string> aggregate_sql;  /* final SQL to execute */

    vector<string> segmented_sql;  /* carryover from segmented statements */

    message::Transaction transaction;

    uint32_t segment_id= 1;

    while (getMessage(transaction, commit_id, master_id, trx_id, originating_server_uuid,

                      originating_commit_id, segment_id++))

    {

      convertToSQL(transaction, aggregate_sql, segmented_sql);

      transaction.Clear();

    }

    /*

     * The last message in a transaction should always have a commit_id

     * value larger than 0, though other messages of the same transaction

     * will have commit_id = 0.

     */

    assert((not commit_id.empty()) && (commit_id != "0"));

    assert(segmented_sql.empty());

    if (not aggregate_sql.empty())

    {

      /*

       * Execution using drizzled::Execute requires some special escaping.

       */

      vector<string>::iterator agg_iter;

      for (agg_iter= aggregate_sql.begin(); agg_iter != aggregate_sql.end(); ++agg_iter)

      {

        string &sql= *agg_iter;

        string::iterator si= sql.begin();

        for (; si != sql.end(); ++si)

        {

          if (*si == '\"')

          {

            si= sql.insert(si, '\\');

            ++si;

          }

          else if (*si == '\\')

          {

            si= sql.insert(si, '\\');

            ++si;

            si= sql.insert(si, '\\');

            ++si;

            si= sql.insert(si, '\\');

            ++si;

          }

          else if (*si == ';')

          {

            si= sql.insert(si, '\\');

            ++si;  /* advance back to the semicolon */

          }

        }

      }

    }

    if (not executeSQLWithCommitId(aggregate_sql, commit_id, 

                                   originating_server_uuid, 

                                   originating_commit_id,

                                   master_id))

    {

      if (_ignore_errors)

      {

        clearErrorState();

        /* Still need to record that we handled this trx */

        vector<string> sql;

        string tmp("UPDATE `sys_replication`.`applier_state`"

                   " SET `last_applied_commit_id` = ");

        tmp.append(commit_id);

        tmp.append(" WHERE `master_id` = ");

        tmp.append(master_id);

        sql.push_back(tmp);

        executeSQL(sql);

      }

      else

      {

        return false;

      }

    }

    if (not deleteFromQueue(master_id, trx_id))

    {

      return false;

    }

  }

  return true;

}

bool QueueConsumer::getMessage(message::Transaction &transaction,

                              string &commit_id,

                              const string &master_id,

                              uint64_t trx_id,

                              string &originating_server_uuid,

                              uint64_t &originating_commit_id,

                              uint32_t segment_id)

{

  string sql("SELECT `msg`, `commit_order`, `originating_server_uuid`, "

             "`originating_commit_id` FROM `sys_replication`.`queue`"

             " WHERE `trx_id` = ");

  sql.append(boost::lexical_cast<string>(trx_id));

  sql.append(" AND `seg_id` = ", 16);

  sql.append(boost::lexical_cast<string>(segment_id));

  sql.append(" AND `master_id` = ", 19),

  sql.append(master_id);

  sql::ResultSet result_set(4);

  Execute execute(*(_session.get()), true);

  execute.run(sql, result_set);

  assert(result_set.getMetaData().getColumnCount() == 4);

  /* Really should only be 1 returned row */

  uint32_t found_rows= 0;

  while (result_set.next())

  {

    string msg= result_set.getString(0);

    string com_id= result_set.getString(1);

    string orig_server_uuid= result_set.getString(2);

    string orig_commit_id= result_set.getString(3);

    if ((msg == "") || (found_rows == 1))

      break;

    /* No columns should be NULL */

    assert(result_set.isNull(0) == false);

    assert(result_set.isNull(1) == false);

    assert(result_set.isNull(2) == false);

    assert(result_set.isNull(3) == false);

    google::protobuf::TextFormat::ParseFromString(msg, &transaction);

    commit_id= com_id;

    originating_server_uuid= orig_server_uuid;

    originating_commit_id= boost::lexical_cast<uint64_t>(orig_commit_id);

    found_rows++;

  }

  if (found_rows == 0)

    return false;

  return true;

}

bool QueueConsumer::getListOfCompletedTransactions(const string &master_id,

                                                   TrxIdList &list)

{

  Execute execute(*(_session.get()), true);

  string sql("SELECT `trx_id` FROM `sys_replication`.`queue`"

             " WHERE `commit_order` IS NOT NULL AND `commit_order` > 0"

             " AND `master_id` = "

             + master_id

             + " ORDER BY `commit_order` ASC");

  /* ResultSet size must match column count */

  sql::ResultSet result_set(1);

  execute.run(sql, result_set);

  assert(result_set.getMetaData().getColumnCount() == 1);

  while (result_set.next())

  {

    assert(result_set.isNull(0) == false);

    string value= result_set.getString(0);

    /* empty string returned when no more results */

    if (value != "")

    {

      list.push_back(boost::lexical_cast<uint64_t>(result_set.getString(0)));

    }

  }

  return true;

}

bool QueueConsumer::convertToSQL(const message::Transaction &transaction,

                                vector<string> &aggregate_sql,

                                vector<string> &segmented_sql)

{

  if (transaction.has_event())

    return true;

  size_t num_statements= transaction.statement_size();

  /*

   * Loop through all Statement messages within this Transaction and

   * convert each to equivalent SQL statements. Complete Statements will

   * be appended to aggregate_sql, while segmented Statements will remain

   * in segmented_sql to be appended to until completed, or rolled back.

   */

  for (size_t idx= 0; idx < num_statements; idx++)

  {

    const message::Statement &statement= transaction.statement(idx);

    /* We won't bother with executing a rolled back transaction */

    if (statement.type() == message::Statement::ROLLBACK)

    {

      assert(idx == (num_statements - 1));  /* should be the final Statement */

      aggregate_sql.clear();

      segmented_sql.clear();

      break;

    }

    switch (statement.type())

    {

      /* DDL cannot be in a transaction, so precede with a COMMIT */

      case message::Statement::TRUNCATE_TABLE:

      case message::Statement::CREATE_SCHEMA:

      case message::Statement::ALTER_SCHEMA:

      case message::Statement::DROP_SCHEMA:

      case message::Statement::CREATE_TABLE:

      case message::Statement::ALTER_TABLE:

      case message::Statement::DROP_TABLE:

      case message::Statement::RAW_SQL:  /* currently ALTER TABLE or RENAME */

      {

        segmented_sql.push_back("COMMIT");

        break;

      }

      /* Cancel any ongoing statement */

      case message::Statement::ROLLBACK_STATEMENT:

      {

        segmented_sql.clear();

        continue;

      }

      default:

      {

        break;

      }

    }

    if (message::transformStatementToSql(statement, segmented_sql,

                                         message::DRIZZLE, true))

    {

      return false;

    }

    if (isEndStatement(statement))

    {

      aggregate_sql.insert(aggregate_sql.end(),

                           segmented_sql.begin(),

                           segmented_sql.end());

      segmented_sql.clear();

    }

  }

  return true;

}

bool QueueConsumer::isEndStatement(const message::Statement &statement)

{

  switch (statement.type())

  {

    case (message::Statement::INSERT):

    {

      const message::InsertData &data= statement.insert_data();

      if (not data.end_segment())

        return false;

      break;

    }

    case (message::Statement::UPDATE):

    {

      const message::UpdateData &data= statement.update_data();

      if (not data.end_segment())

        return false;

      break;

    }

    case (message::Statement::DELETE):

    {

      const message::DeleteData &data= statement.delete_data();

      if (not data.end_segment())

        return false;

      break;

    }

    default:

      return true;

  }

  return true;

}

/*

 * TODO: This currently updates every row in the applier_state table.

 * This use to be a single row. With multi-master support, we now need

 * a row for every master so we can track the last applied commit ID

 * value for each. Eventually, we may want multiple consumer threads,

 * so then we'd need to update each row independently.

 */

void QueueConsumer::setApplierState(const string &err_msg, bool status)

{

  vector<string> statements;

  string sql;

  string msg(err_msg);

  if (not status)

  {

    sql= "UPDATE `sys_replication`.`applier_state` SET `status` = 'STOPPED'";

  }

  else

  {

    sql= "UPDATE `sys_replication`.`applier_state` SET `status` = 'RUNNING'";

  }

  sql.append(", `error_msg` = '", 17);

  /* Escape embedded quotes and statement terminators */

  string::iterator it;

  for (it= msg.begin(); it != msg.end(); ++it)

  {

    if (*it == '\'')

    {

      it= msg.insert(it, '\'');

      ++it;  /* advance back to the quote */

    }

    else if (*it == ';')

    {

      it= msg.insert(it, '\\');

      ++it;  /* advance back to the semicolon */

    }

  }

  sql.append(msg);

  sql.append("'", 1);

  statements.push_back(sql);

  executeSQL(statements);

}

bool QueueConsumer::executeSQLWithCommitId(vector<string> &sql,

                                           const string &commit_id, 

                                           const string &originating_server_uuid, 

                                           uint64_t originating_commit_id,

                                           const string &master_id)

{

  string tmp("UPDATE `sys_replication`.`applier_state`"

             " SET `last_applied_commit_id` = ");

  tmp.append(commit_id);

  tmp.append(", `originating_server_uuid` = '");

  tmp.append(originating_server_uuid);

  tmp.append("' , `originating_commit_id` = ");

  tmp.append(boost::lexical_cast<string>(originating_commit_id));

  tmp.append(" WHERE `master_id` = ");

  tmp.append(master_id);

  sql.push_back(tmp);

  _session->setOriginatingServerUUID(originating_server_uuid);

  _session->setOriginatingCommitID(originating_commit_id);

  return executeSQL(sql);

}

bool QueueConsumer::deleteFromQueue(const string &master_id, uint64_t trx_id)

{

  string sql("DELETE FROM `sys_replication`.`queue` WHERE `trx_id` = ");

  sql.append(boost::lexical_cast<std::string>(trx_id));

  sql.append(" AND `master_id` = ");

  sql.append(master_id);

  vector<string> sql_vect;

  sql_vect.push_back(sql);

  return executeSQL(sql_vect);

}

} /* namespace slave */