RequestManager.java

/**
 * C-JDBC: Clustered JDBC.
 * Copyright (C) 2002-2005 French National Institute For Research In Computer
 * Science And Control (INRIA).
 * Contact: c-jdbc@objectweb.org
 * 
 * This library is free software; you can redistribute it and/or modify it
 * under the terms of the GNU Lesser General Public License as published by the
 * Free Software Foundation; either version 2.1 of the License, or any later
 * version.
 * 
 * This library 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 Lesser General Public License
 * for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library; if not, write to the Free Software Foundation,
 * Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
 *
 * Initial developer(s): Emmanuel Cecchet.
 * Contributor(s): Julie Marguerite, Greg Ward, Nicolas Modrzyk, Vadim Kassin.
 */

package org.objectweb.cjdbc.controller.requestmanager;

import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Hashtable;

import javax.management.NotCompliantMBeanException;

import org.objectweb.cjdbc.common.exceptions.BackupException;
import org.objectweb.cjdbc.common.exceptions.ExceptionTypes;
import org.objectweb.cjdbc.common.exceptions.NoMoreBackendException;
import org.objectweb.cjdbc.common.exceptions.OctopusException;
import org.objectweb.cjdbc.common.exceptions.RollbackException;
import org.objectweb.cjdbc.common.i18n.Translate;
import org.objectweb.cjdbc.common.jmx.JmxConstants;
import org.objectweb.cjdbc.common.jmx.mbeans.RequestManagerMBean;
import org.objectweb.cjdbc.common.log.Trace;
import org.objectweb.cjdbc.common.shared.BackendState;
import org.objectweb.cjdbc.common.shared.BackupListener;
import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
import org.objectweb.cjdbc.common.sql.AlterRequest;
import org.objectweb.cjdbc.common.sql.CreateRequest;
import org.objectweb.cjdbc.common.sql.ParsingGranularities;
import org.objectweb.cjdbc.common.sql.RequestType;
import org.objectweb.cjdbc.common.sql.SelectRequest;
import org.objectweb.cjdbc.common.sql.StoredProcedure;
import org.objectweb.cjdbc.common.sql.UpdateRequest;
import org.objectweb.cjdbc.common.sql.schema.DatabaseSchema;
import org.objectweb.cjdbc.common.sql.schema.DatabaseTable;
import org.objectweb.cjdbc.common.xml.DatabasesXmlTags;
import org.objectweb.cjdbc.common.xml.XmlComponent;
import org.objectweb.cjdbc.controller.backend.BackendStateListener;
import org.objectweb.cjdbc.controller.backend.DatabaseBackend;
import org.objectweb.cjdbc.controller.backup.BackupManager;
import org.objectweb.cjdbc.controller.cache.metadata.MetadataCache;
import org.objectweb.cjdbc.controller.cache.parsing.ParsingCache;
import org.objectweb.cjdbc.controller.cache.result.AbstractResultCache;
import org.objectweb.cjdbc.controller.cache.result.entries.CacheEntry;
import org.objectweb.cjdbc.controller.jmx.AbstractStandardMBean;
import org.objectweb.cjdbc.controller.jmx.MBeanServerManager;
import org.objectweb.cjdbc.controller.loadbalancer.AbstractLoadBalancer;
import org.objectweb.cjdbc.controller.loadbalancer.AllBackendsFailedException;
import org.objectweb.cjdbc.controller.recoverylog.AbstractRecoveryLog;
import org.objectweb.cjdbc.controller.recoverylog.BackendRecoveryInfo;
import org.objectweb.cjdbc.controller.recoverylog.JDBCRecoverThread;
import org.objectweb.cjdbc.controller.scheduler.AbstractScheduler;
import org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet;
import org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase;

/**
 * This class defines the Request Manager.
 * <p>
 * The RM is composed of a Request Scheduler, an optional Query Cache, and a
 * Load Balancer and an optional Recovery Log.
 * 
 * @author <a href="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
 * @author <a href="mailto:Julie.Marguerite@inria.fr">Julie Marguerite </a>
 * @author <a href="mailto:Nicolas.Modrzyk@inrialpes.fr">Nicolas Modrzyk </a>
 * @author <a href="mailto:vadim@kase.kz">Vadim Kassin </a>
 * @version 1.0
 */
public class RequestManager extends AbstractStandardMBean
    implements
      XmlComponent,
      RequestManagerMBean
{

  //
  // How the code is organized ?
  //
  // 1. Member variables
  // 2. Constructor(s)
  // 3. Request handling
  // 4. Transaction handling
  // 5. Database backend management
  // 6. Getter/Setter (possibly in alphabetical order)
  //

  /** begin timeout in ms */
  protected long                 beginTimeout;

  /** commit timeout in ms */
  protected long                 commitTimeout;

  /** rollback timeout in ms */
  protected long                 rollbackTimeout;

  /** The virtual database owning this Request Manager */
  protected VirtualDatabase      vdb;

  /** The request scheduler to order and schedule requests */
  protected AbstractScheduler    scheduler;

  /** An optional request cache to cache responses to SQL requests */
  protected AbstractResultCache  resultCache;

  /** The request load balancer to use to send requests to the databases */
  protected AbstractLoadBalancer loadBalancer;

  /** An optional recovery log */
  protected AbstractRecoveryLog  recoveryLog;

  /** The backup manager responsible for backup and restore of backends */
  protected BackupManager        backupManager;

  // Database schema pointing to the virtual dabase schema
  protected DatabaseSchema       dbs;

  private boolean                schemaIsStatic                = false;

  private boolean                isCaseSensitiveParsing        = false;

  protected ParsingCache         parsingCache                  = null;

  private MetadataCache          metadataCache                 = null;

  // SQL queries parsing granularity according to Scheduler, ResultCache and
  // LoadBalancer required granularity
  protected int                  schedulerParsingranularity    = ParsingGranularities.NO_PARSING;

  private int                    cacheParsingranularity        = ParsingGranularities.NO_PARSING;

  private int                    loadBalancerParsingranularity = ParsingGranularities.NO_PARSING;

  protected int                  requiredGranularity           = ParsingGranularities.NO_PARSING;

  /** Transaction id/Login mapping */
  protected Hashtable            tidLoginTable;

  protected static Trace         logger                        = null;

  private BackendStateListener   backendStateListener;

  //
  // Constructors
  //

  /**
   * Creates a new <code>RequestManager</code> instance.
   * 
   * @param vdb the virtual database this request manager belongs to
   * @param scheduler the Request Scheduler to use
   * @param cache a Query Cache implementation
   * @param loadBalancer the Request Load Balancer to use
   * @param recoveryLog the Log Recovery to use
   * @param beginTimeout timeout in seconds for begin
   * @param commitTimeout timeout in seconds for commit
   * @param rollbackTimeout timeout in seconds for rollback
   * @throws SQLException if an error occurs
   * @throws NotCompliantMBeanException if the MBean is not JMX compliant
   */
  public RequestManager(VirtualDatabase vdb, AbstractScheduler scheduler,
      AbstractResultCache cache, AbstractLoadBalancer loadBalancer,
      AbstractRecoveryLog recoveryLog, long beginTimeout, long commitTimeout,
      long rollbackTimeout) throws SQLException, NotCompliantMBeanException
  {
    super(RequestManagerMBean.class);
    this.vdb = vdb;
    assignAndCheckSchedulerLoadBalancerValidity(scheduler, loadBalancer);
    // requiredGranularity is the maximum of each component granularity
    this.resultCache = cache;
    if (resultCache != null)
    {
      cacheParsingranularity = cache.getParsingGranularity();
      if (cacheParsingranularity > requiredGranularity)
        requiredGranularity = cacheParsingranularity;
    }
    setRecoveryLog(recoveryLog);
    initRequestManagerVariables(vdb, beginTimeout, commitTimeout,
        rollbackTimeout);
    setBackendsLastKnownCheckpointFromRecoveryLog();
    logger.info(Translate.get("requestmanager.parsing.granularity",
        ParsingGranularities.getInformation(requiredGranularity)));

    if (MBeanServerManager.isJmxEnabled())
    {
      try
      {
        MBeanServerManager.registerMBean(this, JmxConstants
            .getRequestManagerObjectName(vdb.getVirtualDatabaseName()));

      }
      catch (Exception e)
      {
        logger.error(Translate.get("jmx.failed.register.mbean.requestmanager"));
      }
    }
  }

  /**
   * Retrieve the last known checkpoint from the recovery log and set it for
   * each backend.
   */
  private void setBackendsLastKnownCheckpointFromRecoveryLog()
  {

    if (recoveryLog == null)
      return;
    String databaseName = vdb.getVirtualDatabaseName();
    ArrayList backends = vdb.getBackends();
    int size = backends.size();
    DatabaseBackend backend;
    BackendRecoveryInfo info;
    for (int i = 0; i < size; i++)
    {
      backend = (DatabaseBackend) backends.get(i);
      try
      {
        info = recoveryLog.getBackendRecoveryInfo(databaseName, backend
            .getName());
        backend.setLastKnownCheckpoint(info.getCheckpoint());
      }
      catch (SQLException e)
      {
        logger.error(Translate.get("requestmanager.checkpoint.not.found",
            backend.getName()), e);
      }
    }

  }

  /**
   * Check that Scheduler and Load Balancer are not null and have compatible
   * RAIDb levels.
   * 
   * @param scheduler
   * @param loadBalancer
   * @throws SQLException if an error occurs
   */
  private void assignAndCheckSchedulerLoadBalancerValidity(
      AbstractScheduler scheduler, AbstractLoadBalancer loadBalancer)
      throws SQLException
  {
    if (scheduler == null)
      throw new SQLException(Translate.get("requestmanager.null.scheduler"));

    if (loadBalancer == null)
      throw new SQLException(Translate.get("requestmanager.null.loadbalancer"));

    if (scheduler.getRAIDbLevel() != loadBalancer.getRAIDbLevel())
      throw new SQLException(Translate.get(
          "requestmanager.incompatible.raidb.levels",
          new String[]{"" + scheduler.getRAIDbLevel(),
              "" + loadBalancer.getRAIDbLevel()}));

    // requiredGranularity is the maximum of each component granularity
    setScheduler(scheduler);
    schedulerParsingranularity = scheduler.getParsingGranularity();
    requiredGranularity = schedulerParsingranularity;
    setLoadBalancer(loadBalancer);
    loadBalancerParsingranularity = loadBalancer.getParsingGranularity();
    if (loadBalancerParsingranularity > requiredGranularity)
      requiredGranularity = loadBalancerParsingranularity;
  }

  /**
   * Method initRequestManagerVariables.
   * 
   * @param vdb
   * @param beginTimeout
   * @param commitTimeout
   * @param rollbackTimeout
   */
  private void initRequestManagerVariables(VirtualDatabase vdb,
      long beginTimeout, long commitTimeout, long rollbackTimeout)
  {
    this.tidLoginTable = new Hashtable();
    this.beginTimeout = beginTimeout;
    this.commitTimeout = commitTimeout;
    this.rollbackTimeout = rollbackTimeout;
    this.vdb = vdb;
    logger = Trace.getLogger("org.objectweb.cjdbc.controller.RequestManager."
        + vdb.getDatabaseName());
  }

  //
  // Request Handling
  //

  /**
   * Perform a read request and return the reply. Call first the scheduler, then
   * the cache (if defined) and finally the load balancer.
   * 
   * @param request the request to execute
   * @return a <code>ControllerResultSet</code> value
   * @exception SQLException if an error occurs
   */
  public ControllerResultSet execReadRequest(SelectRequest request)
      throws SQLException
  {
    // Sanity check
    if (!request.isAutoCommit())
    { // Check that the transaction has been
      // started
      long tid = request.getTransactionId();
      if (!tidLoginTable.containsKey(new Long(tid)))
        throw new SQLException(Translate.get("transaction.not.started", tid));
    }

    // If we need to parse the request, try to get the parsing from the
    // cache.
    // Note that if we have a cache miss but backgroundParsing has been
    // turned
    // on, then this call will start a ParsedThread in background.
    if ((requiredGranularity != ParsingGranularities.NO_PARSING)
        && (!request.isParsed()))
    {
      if (parsingCache == null)
        request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
      else
        parsingCache.getParsingFromCache(request);
    }

    //
    // SCHEDULER
    //

    // Get the parsing now if the request is not yet parsed. The parsing is
    // handled by the ParsingCache that may already have parsed the request
    // in background (if backgroundParsing is set).
    if ((schedulerParsingranularity != ParsingGranularities.NO_PARSING)
        && !request.isParsed())
    {
      if (parsingCache == null)
        request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
      else
        parsingCache.getParsingFromCacheAndParseIfMissing(request);
    }

    if (logger.isDebugEnabled())
      logger.debug(Translate.get("requestmanager.read.request.schedule",
          new String[]{String.valueOf(request.getId()),
              request.getSQLShortForm(vdb.getSQLShortFormLength())}));

    // Wait for the scheduler to give us the authorization to execute
    scheduler.scheduleReadRequest(request);

    //
    // CACHE
    //

    ControllerResultSet result = null;
    try
    { // Check cache if any
      if (resultCache != null)
      {
        if (logger.isDebugEnabled())
          logger.debug(Translate.get("requestmanager.read.request.cache.get",
              new String[]{String.valueOf(request.getId()),
                  request.getSQLShortForm(vdb.getSQLShortFormLength())}));

        CacheEntry qce = resultCache.getFromCache(request, true);
        if (qce != null)
        {
          result = qce.getResult();
          if (result != null)
          { // Cache hit !
            if (vdb.getSQLMonitor() != null)
              vdb.getSQLMonitor().logCacheHit(request);

            scheduler.readCompleted(request);
            return result;
          }
        }
      }

      //
      // LOAD BALANCER
      //

      if (logger.isDebugEnabled())
        logger.debug(Translate.get("requestmanager.read.request.balance",
            new String[]{String.valueOf(request.getId()),
                request.getSQLShortForm(vdb.getSQLShortFormLength())}));

      // At this point, we have a result cache miss.
      // If we had a parsing cache miss too, wait for the parsing to be
      // done if
      // needed.
      if ((loadBalancerParsingranularity != ParsingGranularities.NO_PARSING)
          && !request.isParsed())
      {
        if (parsingCache == null)
          request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
        else
          parsingCache.getParsingFromCacheAndParseIfMissing(request);
      }

      // Send the request to the load balancer
      result = loadBalancer.execReadRequest(request, metadataCache);

      //
      // UPDATES & NOTIFICATIONS
      //

      // Update cache
      if ((resultCache != null)
          && (request.getCacheAbility() != RequestType.UNCACHEABLE))
      {
        if (logger.isDebugEnabled())
          logger.debug(Translate.get(
              "requestmanager.read.request.cache.update", new String[]{
                  String.valueOf(request.getId()),
                  request.getSQLShortForm(vdb.getSQLShortFormLength())}));

        if (!request.isParsed()
            && (cacheParsingranularity != ParsingGranularities.NO_PARSING))
        { // The cache was the only one to need parsing and the request was not
          // previously in the cache
          {
            if (parsingCache == null)
              request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
            else
              parsingCache.getParsingFromCacheAndParseIfMissing(request);
          }
        }
        resultCache.addToCache(request, result);
      }
    }
    catch (Exception failed)
    {
      if (resultCache != null)
        resultCache.removeFromPendingQueries(request);
      scheduler.readCompleted(request);
      if (failed instanceof NoMoreBackendException)
        throw (NoMoreBackendException) failed;
      String msg = Translate.get("requestmanager.request.failed", new String[]{
          request.getSQLShortForm(vdb.getSQLShortFormLength()),
          failed.getMessage()});
      if (failed instanceof RuntimeException)
        logger.warn(msg, failed);
      else
        logger.warn(msg);
      if (failed instanceof SQLException)
        throw (SQLException) failed;

      throw new SQLException(msg);
    }

    // Notify scheduler of completion
    scheduler.readCompleted(request);

    return result;
  }

  /**
   * Perform a write request and return the number of rows affected Call first
   * the scheduler (if defined), then notify the cache (if defined) and finally
   * call the load balancer.
   * 
   * @param request the request to execute
   * @return number of rows affected
   * @exception SQLException if an error occurs
   */
  public int execWriteRequest(AbstractWriteRequest request) throws SQLException
  {
    scheduleExecWriteRequest(request);
    int execWriteRequestResult = 0;
    try
    {
      execWriteRequestResult = loadBalanceExecWriteRequest(request);
    }
    catch (AllBackendsFailedException e)
    {
      String msg = Translate
          .get("requestmanager.write.request.failed.unexpected");
      logger.fatal(msg, e);
      throw new RuntimeException(msg, e);
    }
    updateAndNotifyExecWriteRequest(request, true);
    return execWriteRequestResult;
  }

  /**
   * Perform a write request and return the auto generated keys. Call first the
   * scheduler (if defined), then notify the cache (if defined) and finally call
   * the load balancer.
   * 
   * @param request the request to execute
   * @return auto generated keys.
   * @exception SQLException if an error occurs
   */
  public ControllerResultSet execWriteRequestWithKeys(
      AbstractWriteRequest request) throws SQLException
  {
    scheduleExecWriteRequest(request);
    ControllerResultSet execWriteRequestWithKeysResult = null;
    try
    {
      execWriteRequestWithKeysResult = loadBalanceExecWriteRequestWithKeys(request);
    }
    catch (AllBackendsFailedException e)
    {
      String msg = Translate
          .get("requestmanager.write.request.keys.failed.unexpected");
      logger.fatal(msg, e);
      throw new RuntimeException(msg, e);
    }
    updateAndNotifyExecWriteRequest(request, true);
    return execWriteRequestWithKeysResult;
  }

  /**
   * Schedule a request for execution.
   * 
   * @param request the request to execute
   * @throws SQLException if an error occurs
   */
  public void scheduleExecWriteRequest(AbstractWriteRequest request)
      throws SQLException
  {
    // Sanity check
    if (!request.isAutoCommit())
    { // Check that the transaction has been
      // started
      long tid = request.getTransactionId();
      if (!tidLoginTable.containsKey(new Long(tid)))
        throw new SQLException(Translate.get("transaction.not.started", tid));
    }

    // If we need to parse the request, try to get the parsing from the
    // cache.
    // Note that if we have a cache miss but backgroundParsing has been
    // turned
    // on, then this call will start a ParsedThread in background.
    if ((requiredGranularity != ParsingGranularities.NO_PARSING)
        && (!request.isParsed()))
    {
      if (parsingCache == null)
        request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
      else
        parsingCache.getParsingFromCache(request);
    }

    //
    // SCHEDULER
    //

    // Get the parsing now if the request is not yet parsed. The parsing is
    // handled by the ParsingCache that may already have parsed the request
    // in background (if backgroundParsing is set).
    if ((schedulerParsingranularity != ParsingGranularities.NO_PARSING)
        && !request.isParsed())
    {
      if (parsingCache == null)
        request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
      else
        parsingCache.getParsingFromCacheAndParseIfMissing(request);
    }

    if (logger.isDebugEnabled())
      logger.debug(Translate.get("requestmanager.write.request.schedule",
          new String[]{String.valueOf(request.getId()),
              request.getSQLShortForm(vdb.getSQLShortFormLength())}));

    // Wait for the scheduler to give us the authorization to execute
    try
    {
      scheduler.scheduleWriteRequest(request);
    }
    catch (RollbackException e)
    { // Something bad happened and we need to
      // rollback this transaction
      rollback(request.getTransactionId());
      throw new SQLException(e.getMessage());
    }

    // If we have a parsing cache miss, wait for the parsing to be done if
    // needed. Note that even if the cache was the only one to require
    // parsing,
    // we wait for the parsing result here, because if it fails, we must not
    // execute the query.
    try
    {
      if ((requiredGranularity != ParsingGranularities.NO_PARSING)
          && !request.isParsed())
      {
        if (parsingCache == null)
          request.parse(dbs, requiredGranularity, isCaseSensitiveParsing);
        else
          parsingCache.getParsingFromCacheAndParseIfMissing(request);
      }
    }
    catch (SQLException e)
    {
      // If the parsing fail, we must release the lock acquired ...
      scheduler.writeCompleted(request);
      throw e;
    }
  }

  /**
   * Send the given query to the load balancer. If the request fails, the
   * scheduler is properly notified.
   * 
   * @param request the request to execute
   * @throws AllBackendsFailedException if all backends failed to execute the
   *           query
   * @throws SQLException if an error occurs
   * @return auto-generated keys
   */
  public ControllerResultSet loadBalanceExecWriteRequestWithKeys(
      AbstractWriteRequest request) throws AllBackendsFailedException,
      SQLException
  {
    if (logger.isDebugEnabled())
      logger.debug(Translate.get("requestmanager.write.request.balance",
          new String[]{String.valueOf(request.getId()),
              request.getSQLShortForm(vdb.getSQLShortFormLength())}));

    try
    { // Send the request to the load balancer
      return loadBalancer.execWriteRequestWithKeys(request, metadataCache);
    }
    catch (Exception failed)
    {
      scheduler.writeCompleted(request);
      String msg = Translate.get("requestmanager.request.failed", new String[]{
          request.getSQLShortForm(vdb.getSQLShortFormLength()),
          failed.getMessage()});
      if (failed instanceof RuntimeException)
        logger.warn(msg, failed);
      else
        logger.warn(msg);
      if (failed instanceof AllBackendsFailedException)
        throw (AllBackendsFailedException) failed;
      else
        throw new SQLException(msg);
    }
  }

  /**
   * Send the given query to the load balancer. If the request fails, the
   * scheduler is properly notified.
   * 
   * @param request the request to execute
   * @throws AllBackendsFailedException if all backends failed to execute the
   *           query
   * @throws SQLException if an error occurs
   * @return number of modified lines
   */
  public int loadBalanceExecWriteRequest(AbstractWriteRequest request)
      throws AllBackendsFailedException, SQLException
  {
    if (logger.isDebugEnabled())
      logger.debug(Translate.get("requestmanager.write.request.balance",
          new String[]{String.valueOf(request.getId()),
              request.getSQLShortForm(vdb.getSQLShortFormLength())}));

    try
    { // Send the request to the load balancer
      if (request.isUpdate() && (resultCache != null))
      { // Try the optimization if we try to update values that are already
        // up-to-date.
        if (!resultCache.isUpdateNecessary((UpdateRequest) request))
          return 0;
      }
      return loadBalancer.execWriteRequest(request);
    }
    catch (Exception failed)
    {
      scheduler.writeCompleted(request);
      String msg = Translate.get("requestmanager.request.failed", new String[]{
          request.getSQLShortForm(vdb.getSQLShortFormLength()),
          failed.getMessage()});
      if (failed instanceof RuntimeException)
        logger.warn(msg, failed);
      else
        logger.warn(msg);
      if (failed instanceof AllBackendsFailedException)
        throw (AllBackendsFailedException) failed;
      else if (failed instanceof SQLException)
        throw (SQLException) failed;
      else
        throw new SQLException(msg);
    }
  }

  /**
   * Update the cache, notify the recovery log and finally the scheduler. It is
   * possible to disable scheduler notifications (unless an error occurs in
   * which case the scheduler is always notified). This is especially useful for
   * distributed schedulers when all backends failed at one controller but we
   * have to wait for the confirmation that all other controllers failed. This
   * piece of code is then generic and reusable.
   * 
   * @param request the request to execute
   * @param notifyRecoveryLogAndScheduler true if the recovery log and the
   *          scheduler must be notified
   * @throws SQLException if an error occurs
   */
  public void updateAndNotifyExecWriteRequest(AbstractWriteRequest request,
      boolean notifyRecoveryLogAndScheduler) throws SQLException
  {

    try
    { // Notify cache if any
      if (resultCache != null)
      { // Update cache
        if (logger.isDebugEnabled())
          logger.debug(Translate.get(
              "requestmanager.write.request.cache.update", new String[]{
                  String.valueOf(request.getId()),
                  request.getSQLShortForm(vdb.getSQLShortFormLength())}));

        resultCache.writeNotify(request);
      }

      // Log the request
      if (notifyRecoveryLogAndScheduler && (recoveryLog != null))
      {
        if (logger.isDebugEnabled())
          logger.debug(Translate.get("requestmanager.write.request.log",
              new String[]{String.valueOf(request.getId()),
                  request.getSQLShortForm(vdb.getSQLShortFormLength())}));

        recoveryLog.logRequest(request);
      }

      // Update the schema if needed
      if (requiredGranularity != ParsingGranularities.NO_PARSING)
      {
        if (request.isCreate())
        { // Add the table to the schema
          dbs.addTable(((CreateRequest) request).getDatabaseTable());
          if (logger.isDebugEnabled())
            logger.debug(Translate.get("requestmanager.schema.add.table",
                request.getTableName()));
        }
        else if (request.isDrop())
        { // Delete the table from the
          // schema
          dbs.removeTable(dbs.getTable(request.getTableName()));
          if (logger.isDebugEnabled())
            logger.debug(Translate.get("requestmanager.schema.remove.table",
                request.getTableName()));
        }
        else if (request.isAlter()
            && (requiredGranularity > ParsingGranularities.TABLE))
        {
          // Add or drop the column from the table
          AlterRequest req = (AlterRequest) request;
          if (req.isDrop())
            dbs.getTable(req.getTableName()).remove(req.getColumn().getName());
          else if (req.isAdd())
          {
            dbs.getTable(req.getTableName()).addColumn(req.getColumn());
          }
        }
      }

      // Notify scheduler
      if (notifyRecoveryLogAndScheduler)
        scheduler.writeCompleted(request);

    }
    catch (Exception failed)
    {
      scheduler.writeCompleted(request);
      String msg = Translate.get("requestmanager.request.failed", new String[]{
          request.getSQLShortForm(vdb.getSQLShortFormLength()),
          failed.getMessage()});
      if (failed instanceof RuntimeException)
        logger.warn(msg, failed);
      else
        logger.warn(msg);
      throw new SQLException(msg);
    }
  }

  /**
   * Call a stored procedure that returns a ResultSet.
   * 
   * @param proc the stored procedure call
   * @return a <code>ControllerResultSet</code> value
   * @exception SQLException if an error occurs
   */
  public ControllerResultSet execReadStoredProcedure(StoredProcedure proc)
      throws SQLException
  {
    // Sanity check
    if (!proc.isAutoCommit())
    { // Check that the transaction has been
      // started
      long tid = proc.getTransactionId();
      if (!tidLoginTable.containsKey(new Long(tid)))
        throw new SQLException(Translate.get("transaction.not.started", tid));
    }

    //
    // SCHEDULER
    //

    // Note that no parsing is required for stored procedures.
    // We build a fake request that span over all tables to be sure
    // that the scheduler will lock everything.
    SelectRequest spanOnAllTables = null;
    if (requiredGranularity != ParsingGranularities.NO_PARSING)
    {
      String sql = "SELECT * FROM";
      ArrayList tables = dbs.getTables();
      int size = tables.size();
      // The following code does not generate a syntaxically correct SQL
      // request (an extra coma at the beginning) but that is not an
      // issue.
      for (int i = 0; i < size; i++)
        sql += " ," + ((DatabaseTable) tables.get(i)).getName();

      spanOnAllTables = new SelectRequest(sql, false, 0, proc
          .getLineSeparator());
    }
    else
      spanOnAllTables = new SelectRequest("select * from x", false, 0, proc
          .getLineSeparator());

    // Wait for the scheduler to give us the authorization to execute
    scheduler.scheduleReadRequest(spanOnAllTables);

    if (logger.isDebugEnabled())
      logger.debug(Translate.get("requestmanager.read.store.procedure",
          new String[]{String.valueOf(proc.getId()),
              proc.getSQLShortForm(vdb.getSQLShortFormLength())}));

    //
    // CACHE
    //

    // Cache is always flushed unless the user has explicitely set the
    // connection to read-only mode in which case we assume that the
    // users deliberately forces the cache not to be flushed when calling
    // this stored procedure.
    if ((resultCache != null) && (!proc.isReadOnly()))
      resultCache.flushCache();

    ControllerResultSet result = null;
    try
    {
      //
      // LOAD BALANCER
      //

      // Send the request to the load balancer
      if (proc.isReadOnly())
        result = loadBalancer.execReadOnlyReadStoredProcedure(proc,
            metadataCache);
      else
        result = loadBalancer.execReadStoredProcedure(proc, metadataCache);

      //
      // RECOVERY LOG
      //

      if (recoveryLog != null)
        recoveryLog.logRequest(proc, true);

    }
    catch (Exception failed)
    {
      scheduler.readCompleted(spanOnAllTables);
      String msg = Translate.get("requestmanager.store.procedure.failed",
          new String[]{proc.getSQLShortForm(vdb.getSQLShortFormLength()),
              failed.getMessage()});
      logger.warn(msg);
      if (failed instanceof SQLException)
      {
        throw (SQLException) failed;
      }
      throw new SQLException(msg);
    }

    // Notify scheduler of completion
    scheduler.readCompleted(spanOnAllTables);

    return result;
  }

  /**
   * Call a stored procedure that performs an update.
   * 
   * @param proc the stored procedure call
   * @return number of rows affected
   * @exception SQLException if an error occurs
   */
  public int execWriteStoredProcedure(StoredProcedure proc) throws SQLException
  {
    // Sanity check
    if (!proc.isAutoCommit())
    { // Check that the transaction has been
      // started
      long tid = proc.getTransactionId();
      if (!tidLoginTable.containsKey(new Long(tid)))
        throw new SQLException(Translate.get("transaction.not.started", String
            .valueOf(tid)));
    }

    //
    // SCHEDULER
    //

    // Note that no parsing is required for stored procedures.
    // We build a fake request that span over all tables to be sure
    // that the scheduler will lock everything.
    SelectRequest spanOnAllTables = null;
    if (requiredGranularity != ParsingGranularities.NO_PARSING)
    {
      String sql = "SELECT * FROM";
      ArrayList tables = dbs.getTables();
      int size = tables.size();
      // The following code does not generate a syntaxically correct SQL
      // request (an extra coma at the beginning) but that is not an
      // issue.
      for (int i = 0; i < size; i++)
        sql += " ," + ((DatabaseTable) tables.get(i)).getName();

      spanOnAllTables = new SelectRequest(sql, false, 0, proc
          .getLineSeparator());
    }
    else
      spanOnAllTables = new SelectRequest("select * from x", false, 0, proc
          .getLineSeparator());

    // Wait for the scheduler to give us the authorization to execute
    // We schedule a read even if it is a write because there is no
    // way to lock all tables for write in a single query. FIXME!
    scheduler.scheduleReadRequest(spanOnAllTables);

    if (logger.isDebugEnabled())
      logger.debug(Translate.get("requestmanager.write.store.procedure",
          new String[]{String.valueOf(proc.getId()),
              proc.getSQLShortForm(vdb.getSQLShortFormLength())}));

    //
    // CACHE
    //

    // Flush cache if any
    if (resultCache != null)
      resultCache.flushCache();

    int result;
    try
    {
      //
      // LOAD BALANCER
      //

      // Send the request to the load balancer
      result = loadBalancer.execWriteStoredProcedure(proc);

      //
      // RECOVERY LOG
      //

      if (recoveryLog != null)
        recoveryLog.logRequest(proc, false);

    }
    catch (Exception failed)
    {
      scheduler.readCompleted(spanOnAllTables);
      String msg = Translate.get("requestmanager.store.procedure.failed",
          new String[]{proc.getSQLShortForm(vdb.getSQLShortFormLength()),
              failed.getMessage()});
      logger.warn(msg);
      throw new SQLException(msg);
    }

    // Notify scheduler of completion
    scheduler.readCompleted(spanOnAllTables);

    return result;
  }

  //
  // Transaction management
  //

  /**
   * Begin a new transaction and return the corresponding transaction
   * identifier. This method is called from the driver when setAutoCommit(false)
   * is called.
   * 
   * @param login the login used by the connection
   * @return int a unique transaction identifier
   * @throws SQLException if an error occurs
   */
  public long begin(String login) throws SQLException
  {
    try
    {
      TransactionMarkerMetaData tm = new TransactionMarkerMetaData(0,
          beginTimeout, login);

      // Wait for the scheduler to give us the authorization to execute
      long tid = scheduler.begin(tm);
      tm.setTransactionId(tid);

      if (logger.isDebugEnabled())
        logger.debug(Translate.get("transaction.begin", String.valueOf(tid)));

      try
      {
        // Send to load balancer
        loadBalancer.begin(tm);

        // Log the begin
        if (recoveryLog != null)
        {
          recoveryLog.begin(tm);
        }
      }
      catch (SQLException e)
      {
        throw e;
      }
      finally
      {
        // Notify scheduler for completion in any case
        scheduler.beginCompleted(tid);
      }

      tidLoginTable.put(new Long(tid), tm);
      return tid;
    }
    catch (RuntimeException e)
    {
      logger.fatal(Translate.get(
          "fatal.runtime.exception.requestmanager.begin", e));
      throw new SQLException(e.getMessage());
    }
  }

  /**
   * Abort a transaction that has been started but in which no query was
   * executed. As we use lazy transaction begin, there is no need to rollback
   * such transaction but just to cleanup the metadata associated with this not
   * effectively started transaction.
   * 
   * @param transactionId id of the transaction to abort
   * @throws SQLException if an error occurs
   */
  public void abort(long transactionId) throws SQLException
  {
    try
    {
      Long tid = new Long(transactionId);
      TransactionMarkerMetaData tm = getTransactionMarker(tid);

      if (logger.isDebugEnabled())
        logger.debug(Translate.get("transaction.abort", String
            .valueOf(transactionId)));

      try
      {
        // Notify the scheduler to abort which is the same as a rollback
        // from a
        // scheduler point of view.
        scheduler.rollback(tm);

        // Notify the recovery log manager
        if (recoveryLog != null)
        {
          recoveryLog.abort(tm);
        }
      }
      catch (SQLException e)
      {
        throw e;
      }
      finally
      {
        // Notify scheduler for completion
        scheduler.rollbackCompleted(transactionId);

        completeTransaction(tid);
      }
    }
    catch (RuntimeException e)
    {
      logger.fatal(Translate
          .get("fatal.runtime.exception.requestmanager.rollback"), e);
      throw new SQLException(e.getMessage());
    }
  }

  /**
   * Get the TransactionMarkerMetaData for the given transaction id.
   * 
   * @param tid transaction id
   * @return the TransactionMarkerMetaData
   * @throws SQLException if no marker has been found for this transaction
   */
  public TransactionMarkerMetaData getTransactionMarker(Long tid)
      throws SQLException
  {
    TransactionMarkerMetaData tm = (TransactionMarkerMetaData) tidLoginTable
        .get(tid);

    if (tm == null)
      throw new SQLException(Translate.get("transaction.marker.not.found", ""
          + tid));

    tm.setTimeout(commitTimeout);
    return tm;
  }

  /**
   * Complete the transaction by removing it from the tidLoginTable.
   * 
   * @param tid transaction id
   */
  public void completeTransaction(Long tid)
  {
    tidLoginTable.remove(tid);
  }

  /**
   * Commit a transaction given its id.
   * 
   * @param transactionId the transaction id
   * @throws SQLException if an error occurs
   */
  public void commit(long transactionId) throws SQLException
  {
    try
    {
      Long tid = new Long(transactionId);
      TransactionMarkerMetaData tm = getTransactionMarker(tid);

      // Wait for the scheduler to give us the authorization to execute
      scheduler.commit(tm);

      if (logger.isDebugEnabled())
        logger.debug(Translate.get("transaction.commit", String.valueOf(tid)));

      try
      {
        // Send to load balancer
        loadBalancer.commit(tm);

        // Notify the cache
        if (resultCache != null)
          resultCache.commit(tm.getTransactionId());

        // Notify the recovery log manager
        if (recoveryLog != null)
          recoveryLog.commit(tm);
      }
      catch (SQLException e)
      {
        throw e;
      }
      catch (AllBackendsFailedException e)
      {
        String msg = "All backends failed to commit transaction "
            + transactionId + " (" + e + ")";
        logger.error(msg);
        throw new SQLException(msg);
      }
      finally
      {
        // Notify scheduler for completion
        scheduler.commitCompleted(transactionId);

        completeTransaction(tid);
      }
    }
    catch (RuntimeException e)
    {
      logger.fatal(Translate.get(
          "fatal.runtime.exception.requestmanager.commit", e));
      throw new SQLException(e.getMessage());
    }
  }

  /**
   * Rollback a transaction given its id.
   * 
   * @param transactionId the transaction id
   * @throws SQLException if an error occurs
   */
  public void rollback(long transactionId) throws SQLException
  {
    try
    {
      Long tid = new Long(transactionId);
      TransactionMarkerMetaData tm = getTransactionMarker(tid);

      // Wait for the scheduler to give us the authorization to execute
      scheduler.rollback(tm);

      if (logger.isDebugEnabled())
        logger.debug(Translate.get("transaction.rollback", String
            .valueOf(transactionId)));

      try
      {
        // Send to load balancer
        loadBalancer.rollback(tm);

        // Send to cache
        if (this.resultCache != null)
          resultCache.rollback(transactionId);

        // Notify the recovery log manager
        if (recoveryLog != null)
        {
          recoveryLog.rollback(tm);
        }
      }
      catch (SQLException e)
      {
        throw e;
      }
      catch (AllBackendsFailedException e)
      {
        String msg = Translate.get("requestmanager.rollback.failed.all",
            new String[]{String.valueOf(transactionId), e.getMessage()});
        logger.error(msg);
        throw new SQLException(msg);
      }
      finally
      {
        // Notify scheduler for completion
        scheduler.rollbackCompleted(transactionId);

        completeTransaction(tid);
      }
    }
    catch (RuntimeException e)
    {
      logger.fatal(Translate
          .get("fatal.runtime.exception.requestmanager.rollback"), e);
      throw new SQLException(e.getMessage());
    }
  }

  //
  // Database Backends management
  //

  /**
   * Enable a backend that has been previously added to this virtual database
   * and that is in the disabled state.
   * <p>
   * The backend is enabled without further check.
   * <p>
   * The enableBackend method of the load balancer is called.
   * 
   * @param db The database backend to enable
   * @throws SQLException if an error occurs
   */
  public void enableBackend(DatabaseBackend db) throws SQLException
  {
    loadBalancer.enableBackend(db, true);
    logger.info(Translate.get("backend.state.enabled", db.getName()));
  }

  /**
   * The backend must have been previously added to this virtual database and be
   * in the disabled state.
   * <p>
   * All the queries since the given checkpoint are played and the backend state
   * is set to enabled when it is completely synchronized.
   * <p>
   * Note that the job is performed in background by a
   * <code>JDBCRecoverThread</code>. You can synchronize on thread
   * termination if you need to wait for completion of this task and listen to
   * JMX notifications for the success status.
   * 
   * @param db The database backend to enable
   * @param checkpointName The checkpoint name to restart from
   * @return the JDBC reocver thread synchronizing the backend
   * @throws SQLException if an error occurs
   */
  public JDBCRecoverThread enableBackendFromCheckpoint(DatabaseBackend db,
      String checkpointName) throws SQLException
  {
    // Sanity check
    if (recoveryLog == null)
    {
      String msg = Translate.get(
          "recovery.restore.checkpoint.failed.cause.null", checkpointName);
      logger.error(msg);
      throw new SQLException(msg);
    }

    if (db.getStateValue() == BackendState.REPLAYING
        || db.getStateValue() == BackendState.RECOVERING)
      throw new SQLException(Translate.get(
          "recovery.restore.backend.state.invalid", db.getName()));

    JDBCRecoverThread recoverThread = new JDBCRecoverThread(scheduler,
        recoveryLog, db, loadBalancer, checkpointName);

    // fire the thread and forget
    // exception will be reported in a jmx notification on the
    // backend.
    recoverThread.start();
    return recoverThread;
  }

  /**
   * Disable a backend that is currently enabled on this virtual database.
   * <p>
   * The backend is disabled without further check.
   * <p>
   * The load balancer disabled method is called on the specified backend.
   * 
   * @param db The database backend to disable
   * @throws SQLException if an error occurs
   */
  public void disableBackend(DatabaseBackend db) throws SQLException
  {
    if (db.isReadEnabled() || db.isWriteEnabled())
    {
      loadBalancer.disableBackend(db);
      logger.info(Translate.get("backend.state.disabled", db.getName()));
    }
    else
    {
      throw new SQLException(Translate.get("backend.already.disabled", db
          .getName()));
    }
  }

  /**
   * The backend must belong to this virtual database and be in the enabled
   * state.
   * <p>
   * The backend is disabled once all the pending write queries are executed. A
   * checkpoint is inserted in the recovery log.
   * 
   * @param db The database backend to enable
   * @param checkpointName The checkpoint name to restart from
   * @throws SQLException if an error occurs
   */
  public void disableBackendForCheckpoint(DatabaseBackend db,
      String checkpointName) throws SQLException
  {
    // Sanity checks
    if (recoveryLog == null)
    {
      String msg = Translate.get("recovery.store.checkpoint.failed.cause.null",
          checkpointName);
      logger.error(msg);
      throw new SQLException(msg);
    }

    // Wait for all pending writes to finish
    logger.info(Translate.get("requestmanager.wait.pending.writes"));
    scheduler.suspendWrites();

    // Store checkpoint
    recoveryLog.storeCheckpoint(checkpointName);
    logger.info(Translate.get("recovery.checkpoint.stored", checkpointName));

    // Signal the backend should not begin any new transaction
    db.setState(BackendState.DISABLING);
    logger.info(Translate.get("backend.state.disabling", db.getName()));

    // Resume writes
    logger.info(Translate.get("requestmanager.resume.pending.writes"));
    scheduler.resumeWrites();

    // Wait for all current transactions on the backend to finish
    db.waitForAllTransactionsToComplete();

    // Now we can safely disable the backend
    db.setLastKnownCheckpoint(checkpointName);
    loadBalancer.disableBackend(db);
    logger.info(Translate.get("backend.state.disabled", db.getName()));
  }

  /**
   * Disable a list of backends. Only to store only one checkpoint, and to
   * disable all the backends at the same time so the the system is in a
   * coherent state. Consider only the backends that were enabled. The others
   * are left in the state they were before.
   * 
   * @param backendsArrayList backends to disable
   * @param checkpointName to store
   * @throws SQLException if an error occurs
   */
  public void disableBackendsForCheckpoint(ArrayList backendsArrayList,
      String checkpointName) throws SQLException
  {
    // Sanity checks
    if (recoveryLog == null)
    {
      String msg = Translate.get("recovery.store.checkpoint.failed.cause.null",
          checkpointName);
      logger.error(msg);
      throw new SQLException(msg);
    }

    // Wait for all pending writes to finish
    logger.info(Translate.get("requestmanager.wait.pending.writes"));
    scheduler.suspendWrites();

    // Store checkpoint
    recoveryLog.storeCheckpoint(checkpointName);
    logger.info(Translate.get("recovery.checkpoint.stored", checkpointName));

    // Copy the list and consider only the backends that are enabled
    DatabaseBackend db;
    ArrayList backendList = (ArrayList) backendsArrayList.clone();
    for (int i = 0; i < backendList.size(); i++)
    {
      db = (DatabaseBackend) backendList.get(i);
      if (!db.isWriteEnabled())
        backendList.remove(i);
    }

    // Signal all backends that they should not begin any new transaction
    int size = backendList.size();
    for (int i = 0; i < size; i++)
    {
      db = (DatabaseBackend) backendList.get(i);
      db.setState(BackendState.DISABLING);
      logger.info(Translate.get("backend.state.disabling", db.getName()));
    }

    // Resume writes
    logger.info(Translate.get("requestmanager.resume.pending.writes"));
    scheduler.resumeWrites();

    // Wait for all current transactions on backends to finish
    for (int i = 0; i < size; i++)
    {
      db = (DatabaseBackend) backendList.get(i);
      db.waitForAllTransactionsToComplete();
    }

    // Now we can safely disable all backends
    for (int i = 0; i < size; i++)
    {
      db = (DatabaseBackend) backendList.get(i);
      db.setLastKnownCheckpoint(checkpointName);
      loadBalancer.disableBackend(db);
      logger.info(Translate.get("backend.state.disabled", db.getName()));
    }
  }

  /**
   * Call the backup manager on the given backend. Start a fire and forget
   * thread for backup or recovery. The only way to get the results is to wait
   * for jmx notifications
   * 
   * @param backup true if this is a backup, false if this is a restore process
   * @param db backend object
   * @param checkpoint the name of the checkpoint associated with the process
   * @param tables the tables to consider while performing the copy
   * @param enableAfter should be enable the backend automatically after
   * @param listener backup listener for callback
   * @throws Exception if fails
   */
  public void callBackupManager(boolean backup, DatabaseBackend db,
      String checkpoint, ArrayList tables, boolean enableAfter,
      BackupListener listener) throws Exception
  {
    if (backup)
      backupBackendWithCheckpoint(db, checkpoint, tables, enableAfter, false,
          listener);
    else
      restoreBackendFromBackupCheckpoint(db, checkpoint, false, listener);
  }

  /**
   * Creates a new backup with the corresponding checkpoint Note that this will
   * disable the backend for the time of the backup
   * 
   * @param backend the backend to backup
   * @param checkpointName the checkpoint to insert in the log
   * @param tables to copy or null for copy all
   * @param enableAfter if the backend should be enable after backup
   * @param wait4Result if the method should block until the backup is over
   * @param listener object for callback
   * @throws SQLException if fails
   */
  public void backupBackendWithCheckpoint(DatabaseBackend backend,
      String checkpointName, ArrayList tables, boolean enableAfter,
      boolean wait4Result, BackupListener listener) throws SQLException
  {
    if (backend.isReadEnabled())
    { // Disable backend and store checkpoint
      disableBackendForCheckpoint(backend, checkpointName);
      logger.info(Translate.get("backend.state.disabled", backend.getName()));
    }
    // else backend is already disabled, no checkpoint is stored here, it should
    // have been done at disable time.

    try
    {
      logger.info(Translate.get("controller.backup.octopus.start", backend
          .getName()));
      backupManager.backup(backend, checkpointName, tables, listener);
      if (wait4Result)
        backupManager.getResult(backend, 0);
    }
    catch (OctopusException e)
    {
      logger.error(Translate.get("controller.backup.octopus.failed"), e);
      throw new SQLException(e.getMessage());
    }
    catch (BackupException be)
    {
      logger.error(Translate.get("controller.backup.failed"), be);
      throw new SQLException(be.getMessage());
    }
    logger.info(Translate.get("controller.backup.complete", backend.getName()));

    if (enableAfter)
    {
      JDBCRecoverThread thread = enableBackendFromCheckpoint(backend,
          checkpointName);
      if (wait4Result)
        try
        {
          thread.join();
        }
        catch (InterruptedException e)
        {
          logger.error("Recovery thread has been interrupted", e);
        }
    }

  }

  /**
   * Recopy all the data of a previous dump recorded by octopus into the named
   * backend. This disables the backend and leave it disable after recovery
   * process. The user has to call the <code>enableBackendFromCheckpoint</code>
   * after this.
   * 
   * @param db the backend to restore
   * @param checkpointName the name of the checkpoint that has a dump
   * @param wait4Result should we blocked this method until the restore is
   *          finished
   * @param listener object for callback
   * @throws OctopusException if backup failed while in octopus mode
   * @throws BackupException if backup failed for other reasons
   */
  public void restoreBackendFromBackupCheckpoint(DatabaseBackend db,
      String checkpointName, boolean wait4Result, BackupListener listener)
      throws OctopusException, BackupException
  {
    try
    {
      // no check for disable as we are going to overwrite
      // all the database data
      if (db.isReadEnabled())
        loadBalancer.disableBackend(db);

      backupManager.restore(db, checkpointName, null, listener);
      if (wait4Result)
        backupManager.getResult(db, 0);
    }
    catch (SQLException e1)
    {
      // This comes from the loadbalancer
      throw new BackupException(ExceptionTypes.BACKEND_CANNOT_BE_DISABLED);
    }
    catch (OctopusException e)
    {
      logger.error(Translate.get("controller.octopus.recovery.failed"), e);
      throw e;
    }
    catch (BackupException be)
    {
      logger.error(Translate.get("controller.backup.recovery.failed"), be);
      throw be;
    }
    finally
    {
      logger.info(Translate
          .get("controller.backup.recovery.done", db.getName()));
    }
  }

  /**
   * Store all the backends checkpoint in the recoverylog
   * 
   * @param databaseName the virtual database name
   * @param backends the <code>Arraylist</code> of backends
   */
  public void storeBackendsInfo(String databaseName, ArrayList backends)
  {
    if (recoveryLog == null)
      return;
    int size = backends.size();
    DatabaseBackend backend;
    for (int i = 0; i < size; i++)
    {
      backend = (DatabaseBackend) backends.get(i);
      try
      {
        recoveryLog.storeBackendRecoveryInfo(databaseName,
            new BackendRecoveryInfo(backend.getName(), backend
                .getLastKnownCheckpoint(), backend.getStateValue(),
                databaseName));
      }
      catch (SQLException e)
      {
        logger.error(Translate.get("recovery.store.checkpoint.failed",
            new String[]{backend.getName(), e.getMessage()}), e);
      }
    }
  }

  /**
   * Remove a checkpoint and corresponding entries from the log table
   * 
   * @param checkpointName to remove
   * @throws SQLException if fails
   */
  public void removeCheckpoint(String checkpointName) throws SQLException
  {
    recoveryLog.removeCheckpoint(checkpointName);
  }

  //
  // Getter/Setter methods
  //

  /**
   * Returns the vdb value.
   * 
   * @return Returns the vdb.
   */
  public VirtualDatabase getVirtualDatabase()
  {
    return vdb;
  }

  /**
   * Sets the <code>DatabaseSchema</code> to be able to parse the requests and
   * find dependencies.
   * 
   * @param schema a <code>DatabaseSchema</code> value
   * @param isStatic true if the given schema is static
   */
  public void setDatabaseSchema(DatabaseSchema schema, boolean isStatic)
  {
    if (schemaIsStatic)
    {
      if (isStatic)
      {
        logger.warn(Translate
            .get("requestmanager.schema.replace.static.with.new"));
        this.dbs = schema;
      }
      else
        logger.info(Translate.get("requestmanager.schema.ignore.new.dynamic"));
    }
    else
    {
      schemaIsStatic = isStatic;
      this.dbs = schema;
      logger.info(Translate
          .get("requestmanager.schema.set.new.virtualdatabase"));
    }

    if (schedulerParsingranularity != ParsingGranularities.NO_PARSING)
      scheduler.setDatabaseSchema(dbs);

    if (cacheParsingranularity != ParsingGranularities.NO_PARSING)
      resultCache.setDatabaseSchema(dbs);

    // Load balancers do not have a specific database schema to update
  }

  /**
   * Merge the given schema with the existing database schema.
   * 
   * @param backendSchema The virtual database schema to merge.
   */
  public void mergeDatabaseSchema(DatabaseSchema backendSchema)
  {
    try
    {
      if (dbs == null)
        setDatabaseSchema(new DatabaseSchema(backendSchema), false);
      else
      {
        dbs.mergeSchema(backendSchema);
        logger.info(Translate
            .get("requestmanager.schema.virtualdatabase.merged.new"));

        if (schedulerParsingranularity != ParsingGranularities.NO_PARSING)
          scheduler.mergeDatabaseSchema(dbs);

        if (cacheParsingranularity != ParsingGranularities.NO_PARSING)
          resultCache.mergeDatabaseSchema(dbs);
      }
    }
    catch (SQLException e)
    {
      logger.error(Translate.get("requestmanager.schema.merge.failed", e
          .getMessage()), e);
    }
  }

  /**
   * Sets the backup manager for this recovery log
   * 
   * @param currentBackupManager an instance of <code>BackupManager</code>
   */
  public void setBackupManager(BackupManager currentBackupManager)
  {
    this.backupManager = currentBackupManager;
  }

  /**
   * Returns the backupManager value.
   * 
   * @return Returns the backupManager.
   */
  public BackupManager getBackupManager()
  {
    return backupManager;
  }

  /**
   * Get the <code>DatabaseSchema</code> used by this Request Manager.
   * 
   * @return a <code>DatabaseSchema</code> value
   */
  public DatabaseSchema getDatabaseSchema()
  {
    return dbs;
  }

  /**
   * Get the Request Load Balancer used in this Request Controller.
   * 
   * @return an <code>AbstractLoadBalancer</code> value
   */
  public AbstractLoadBalancer getLoadBalancer()
  {
    return loadBalancer;
  }

  /**
   * Set the Request Load Balancer to use in this Request Controller.
   * 
   * @param loadBalancer a Request Load Balancer implementation
   */
  public void setLoadBalancer(AbstractLoadBalancer loadBalancer)
  {
    if (this.loadBalancer != null)
      throw new RuntimeException(
          "It is not possible to dynamically change a load balancer.");
    this.loadBalancer = loadBalancer;
    if (loadBalancer == null)
      return;
    loadBalancerParsingranularity = loadBalancer.getParsingGranularity();
    if (loadBalancerParsingranularity > requiredGranularity)
      requiredGranularity = loadBalancerParsingranularity;

    if (MBeanServerManager.isJmxEnabled())
    {
      try
      {
        MBeanServerManager.registerMBean(loadBalancer, JmxConstants
            .getLoadBalancerObjectName(vdb.getVirtualDatabaseName()));
      }
      catch (Exception e)
      {
        logger.error(Translate.get("jmx.failed.register.mbean.loadbalancer"));
      }
    }

  }

  /**
   * Get the result cache (if any) used in this Request Manager.
   * 
   * @return an <code>AbstractResultCache</code> value or null if no Reqsult
   *         Cache has been defined
   */
  public AbstractResultCache getResultCache()
  {
    return resultCache;
  }

  /**
   * Returns the metadataCache value.
   * 
   * @return Returns the metadataCache.
   */
  public MetadataCache getMetadataCache()
  {
    return metadataCache;
  }

  /**
   * Sets the metadataCache value.
   * 
   * @param metadataCache The metadataCache to set.
   */
  public void setMetadataCache(MetadataCache metadataCache)
  {
    this.metadataCache = metadataCache;
  }

  /**
   * Sets the ParsingCache.
   * 
   * @param parsingCache The parsingCache to set.
   */
  public void setParsingCache(ParsingCache parsingCache)
  {
    parsingCache.setRequestManager(this);
    parsingCache.setGranularity(requiredGranularity);
    parsingCache.setCaseSensitiveParsing(isCaseSensitiveParsing);
    this.parsingCache = parsingCache;
  }

  /**
   * Returns the Recovery Log Manager.
   * 
   * @return AbstractRecoveryLog
   */
  public AbstractRecoveryLog getRecoveryLog()
  {
    return recoveryLog;
  }

  /**
   * Sets the Recovery Log Manager.
   * 
   * @param recoveryLog The log recovery to set
   */
  public void setRecoveryLog(AbstractRecoveryLog recoveryLog)
  {
    if (recoveryLog == null)
      return;
    this.recoveryLog = recoveryLog;
    ArrayList backends = vdb.getBackends();
    int size = backends.size();
    backendStateListener = new BackendStateListener(vdb
        .getVirtualDatabaseName(), recoveryLog);
    for (int i = 0; i < size; i++)
      ((DatabaseBackend) backends.get(i))
          .setStateListener(backendStateListener);

    if (MBeanServerManager.isJmxEnabled())
    {
      try
      {
        MBeanServerManager.registerMBean(recoveryLog, JmxConstants
            .getRecoveryLogObjectName(vdb.getVirtualDatabaseName()));
      }
      catch (Exception e)
      {
        logger.error(Translate.get("jmx.failed.register.mbean.recoverylog"));
      }
    }
  }

  /**
   * Set the Request Cache to use in this Request Controller.
   * 
   * @param cache a Request Cache implementation
   */
  public void setResultCache(AbstractResultCache cache)
  {
    resultCache = cache;
    cacheParsingranularity = cache.getParsingGranularity();
    if (cacheParsingranularity > requiredGranularity)
      requiredGranularity = cacheParsingranularity;
  }

  /**
   * Get the Request Scheduler (if any) used in this Request Controller.
   * 
   * @return an <code>AbstractScheduler</code> value or null if no Request
   *         Scheduler has been defined
   */
  public AbstractScheduler getScheduler()
  {
    return scheduler;
  }

  /**
   * Set the Request Scheduler to use in this Request Controller.
   * 
   * @param scheduler a Request Scheduler implementation
   */
  public void setScheduler(AbstractScheduler scheduler)
  {
    this.scheduler = scheduler;
    schedulerParsingranularity = scheduler.getParsingGranularity();
    if (schedulerParsingranularity > requiredGranularity)
      requiredGranularity = schedulerParsingranularity;
  }

  /**
   * Sets the parsing case sensitivity. If true the request are parsed in a case
   * sensitive way (table/column name must match exactly the case of the names
   * fetched from the database or enforced by a static schema).
   * 
   * @param isCaseSensitiveParsing true if parsing is case sensitive
   */
  public void setCaseSensitiveParsing(boolean isCaseSensitiveParsing)
  {
    this.isCaseSensitiveParsing = isCaseSensitiveParsing;
    if (parsingCache != null)
      parsingCache.setCaseSensitiveParsing(isCaseSensitiveParsing);
  }

  /**
   * @return the parsing granularity required by the current configuration
   */
  public int getRequiredParsingGranularity()
  {
    return requiredGranularity;
  }

  //
  // Debug/Monitoring
  //

  /**
   * Get xml information about this Request Manager
   * 
   * @return <code>String</code> in xml formatted text
   */
  public String getXml()
  {
    StringBuffer info = new StringBuffer();
    info.append("<" + DatabasesXmlTags.ELT_RequestManager + " "
        + DatabasesXmlTags.ATT_caseSensitiveParsing + "=\""
        + isCaseSensitiveParsing + "\" " + DatabasesXmlTags.ATT_beginTimeout
        + "=\"" + beginTimeout / 1000 + "\" "
        + DatabasesXmlTags.ATT_commitTimeout + "=\"" + commitTimeout / 1000
        + "\" " + DatabasesXmlTags.ATT_rollbackTimeout + "=\""
        + rollbackTimeout / 1000 + "\">");
    if (scheduler != null)
      info.append(scheduler.getXml());

    if (metadataCache != null || parsingCache != null || resultCache != null)
    {
      info.append("<" + DatabasesXmlTags.ELT_RequestCache + ">");
      if (metadataCache != null)
        info.append(metadataCache.getXml());
      if (parsingCache != null)
        info.append(parsingCache.getXml());
      if (resultCache != null)
        info.append(resultCache.getXml());
      info.append("</" + DatabasesXmlTags.ELT_RequestCache + ">");
    }

    if (loadBalancer != null)
      info.append(loadBalancer.getXml());
    if (recoveryLog != null)
      info.append(this.recoveryLog.getXml());
    info.append("</" + DatabasesXmlTags.ELT_RequestManager + ">");
    return info.toString();
  }

  /**
   * Returns the backendStateListener value.
   * 
   * @return Returns the backendStateListener.
   */
  public BackendStateListener getBackendStateListener()
  {
    return backendStateListener;
  }

  /**
   * Returns the beginTimeout value.
   * 
   * @return Returns the beginTimeout.
   */
  public long getBeginTimeout()
  {
    return beginTimeout;
  }

  /**
   * Sets the beginTimeout value.
   * 
   * @param beginTimeout The beginTimeout to set.
   */
  public void setBeginTimeout(long beginTimeout)
  {
    this.beginTimeout = beginTimeout;
  }

  /**
   * Returns the cacheParsingranularity value.
   * 
   * @return Returns the cacheParsingranularity.
   */
  public int getCacheParsingranularity()
  {
    return cacheParsingranularity;
  }

  /**
   * Sets the cacheParsingranularity value.
   * 
   * @param cacheParsingranularity The cacheParsingranularity to set.
   */
  public void setCacheParsingranularity(int cacheParsingranularity)
  {
    this.cacheParsingranularity = cacheParsingranularity;
  }

  /**
   * Returns the commitTimeout value.
   * 
   * @return Returns the commitTimeout.
   */
  public long getCommitTimeout()
  {
    return commitTimeout;
  }

  /**
   * Sets the commitTimeout value.
   * 
   * @param commitTimeout The commitTimeout to set.
   */
  public void setCommitTimeout(long commitTimeout)
  {
    this.commitTimeout = commitTimeout;
  }

  /**
   * Returns the loadBalancerParsingranularity value.
   * 
   * @return Returns the loadBalancerParsingranularity.
   */
  public int getLoadBalancerParsingranularity()
  {
    return loadBalancerParsingranularity;
  }

  /**
   * Sets the loadBalancerParsingranularity value.
   * 
   * @param loadBalancerParsingranularity The loadBalancerParsingranularity to
   *          set.
   */
  public void setLoadBalancerParsingranularity(int loadBalancerParsingranularity)
  {
    this.loadBalancerParsingranularity = loadBalancerParsingranularity;
  }

  /**
   * Returns the requiredGranularity value.
   * 
   * @return Returns the requiredGranularity.
   */
  public int getRequiredGranularity()
  {
    return requiredGranularity;
  }

  /**
   * Sets the requiredGranularity value.
   * 
   * @param requiredGranularity The requiredGranularity to set.
   */
  public void setRequiredGranularity(int requiredGranularity)
  {
    this.requiredGranularity = requiredGranularity;
  }

  /**
   * Returns the rollbackTimeout value.
   * 
   * @return Returns the rollbackTimeout.
   */
  public long getRollbackTimeout()
  {
    return rollbackTimeout;
  }

  /**
   * Sets the rollbackTimeout value.
   * 
   * @param rollbackTimeout The rollbackTimeout to set.
   */
  public void setRollbackTimeout(long rollbackTimeout)
  {
    this.rollbackTimeout = rollbackTimeout;
  }

  /**
   * Returns the schedulerParsingranularity value.
   * 
   * @return Returns the schedulerParsingranularity.
   */
  public int getSchedulerParsingranularity()
  {
    return schedulerParsingranularity;
  }

  /**
   * Sets the schedulerParsingranularity value.
   * 
   * @param schedulerParsingranularity The schedulerParsingranularity to set.
   */
  public void setSchedulerParsingranularity(int schedulerParsingranularity)
  {
    this.schedulerParsingranularity = schedulerParsingranularity;
  }

  /**
   * Returns the schemaIsStatic value.
   * 
   * @return Returns the schemaIsStatic.
   */
  public boolean isSchemaIsStatic()
  {
    return schemaIsStatic;
  }

  /**
   * Sets the schemaIsStatic value.
   * 
   * @param schemaIsStatic The schemaIsStatic to set.
   */
  public void setSchemaIsStatic(boolean schemaIsStatic)
  {
    this.schemaIsStatic = schemaIsStatic;
  }

  /**
   * Returns the isCaseSensitiveParsing value.
   * 
   * @return Returns the isCaseSensitiveParsing.
   */
  public boolean isCaseSensitiveParsing()
  {
    return isCaseSensitiveParsing;
  }

  /**
   * @see org.objectweb.cjdbc.controller.jmx.AbstractStandardMBean#getAssociatedString()
   */
  public String getAssociatedString()
  {
    return "requestmanager";
  }
}

---