Java > Open Source Codes > org > objectweb > cjdbc > controller > requestmanager > RequestManager


1 /**
2  * C-JDBC: Clustered JDBC.
3  * Copyright (C) 2002-2005 French National Institute For Research In Computer
4  * Science And Control (INRIA).
5  * Contact: c-jdbc@objectweb.org
6  *
7  * This library is free software; you can redistribute it and/or modify it
8  * under the terms of the GNU Lesser General Public License as published by the
9  * Free Software Foundation; either version 2.1 of the License, or any later
10  * version.
11  *
12  * This library is distributed in the hope that it will be useful, but WITHOUT
13  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
14  * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
15  * for more details.
16  *
17  * You should have received a copy of the GNU Lesser General Public License
18  * along with this library; if not, write to the Free Software Foundation,
19  * Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA.
20  *
21  * Initial developer(s): Emmanuel Cecchet.
22  * Contributor(s): Julie Marguerite, Greg Ward, Nicolas Modrzyk, Vadim Kassin, Jean-Bernard van Zuylen.
23  */
24
25 package org.objectweb.cjdbc.controller.requestmanager;
26
27 import java.sql.SQLException ;
28 import java.util.ArrayList ;
29 import java.util.Date ;
30 import java.util.Hashtable ;
31 import java.util.LinkedList ;
32 import java.util.Vector ;
33
34 import javax.management.NotCompliantMBeanException ;
35
36 import org.objectweb.cjdbc.common.exceptions.BackupException;
37 import org.objectweb.cjdbc.common.exceptions.NoMoreBackendException;
38 import org.objectweb.cjdbc.common.exceptions.RollbackException;
39 import org.objectweb.cjdbc.common.i18n.Translate;
40 import org.objectweb.cjdbc.common.jmx.JmxConstants;
41 import org.objectweb.cjdbc.common.jmx.mbeans.RequestManagerMBean;
42 import org.objectweb.cjdbc.common.jmx.notifications.CjdbcNotificationList;
43 import org.objectweb.cjdbc.common.log.Trace;
44 import org.objectweb.cjdbc.common.shared.BackendState;
45 import org.objectweb.cjdbc.common.shared.DumpInfo;
46 import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
47 import org.objectweb.cjdbc.common.sql.AlterRequest;
48 import org.objectweb.cjdbc.common.sql.CreateRequest;
49 import org.objectweb.cjdbc.common.sql.ParsingGranularities;
50 import org.objectweb.cjdbc.common.sql.RequestType;
51 import org.objectweb.cjdbc.common.sql.SelectRequest;
52 import org.objectweb.cjdbc.common.sql.StoredProcedure;
53 import org.objectweb.cjdbc.common.sql.UpdateRequest;
54 import org.objectweb.cjdbc.common.sql.schema.DatabaseSchema;
55 import org.objectweb.cjdbc.common.sql.schema.DatabaseTable;
56 import org.objectweb.cjdbc.common.xml.DatabasesXmlTags;
57 import org.objectweb.cjdbc.common.xml.XmlComponent;
58 import org.objectweb.cjdbc.controller.backend.BackendStateListener;
59 import org.objectweb.cjdbc.controller.backend.DatabaseBackend;
60 import org.objectweb.cjdbc.controller.backup.BackupManager;
61 import org.objectweb.cjdbc.controller.backup.Backuper;
62 import org.objectweb.cjdbc.controller.cache.metadata.MetadataCache;
63 import org.objectweb.cjdbc.controller.cache.parsing.ParsingCache;
64 import org.objectweb.cjdbc.controller.cache.result.AbstractResultCache;
65 import org.objectweb.cjdbc.controller.cache.result.entries.AbstractResultCacheEntry;
66 import org.objectweb.cjdbc.controller.jmx.AbstractStandardMBean;
67 import org.objectweb.cjdbc.controller.jmx.MBeanServerManager;
68 import org.objectweb.cjdbc.controller.loadbalancer.AbstractLoadBalancer;
69 import org.objectweb.cjdbc.controller.loadbalancer.AllBackendsFailedException;
70 import org.objectweb.cjdbc.controller.recoverylog.BackendRecoveryInfo;
71 import org.objectweb.cjdbc.controller.recoverylog.RecoverThread;
72 import org.objectweb.cjdbc.controller.recoverylog.RecoveryLog;
73 import org.objectweb.cjdbc.controller.scheduler.AbstractScheduler;
74 import org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet;
75 import org.objectweb.cjdbc.controller.virtualdatabase.VirtualDatabase;
76
77 /**
78  * This class defines the Request Manager.
79  * <p>
80  * The RM is composed of a Request Scheduler, an optional Query Cache, and a
81  * Load Balancer and an optional Recovery Log.
82  *
83  * @author <a HREF="mailto:Emmanuel.Cecchet@inria.fr">Emmanuel Cecchet </a>
84  * @author <a HREF="mailto:Julie.Marguerite@inria.fr">Julie Marguerite </a>
85  * @author <a HREF="mailto:Nicolas.Modrzyk@inrialpes.fr">Nicolas Modrzyk </a>
86  * @author <a HREF="mailto:vadim@kase.kz">Vadim Kassin </a>
87  * @author <a HREF="mailto:jbvanzuylen@transwide.com">Jean-Bernard van Zuylen
88  * </a>
89  * @version 1.0
90  */
91 public class RequestManager extends AbstractStandardMBean
92     implements
93       XmlComponent,
94       RequestManagerMBean
95 {
96
97   //
98 // How the code is organized ?
99 //
100 // 1. Member variables
101 // 2. Constructor(s)
102 // 3. Request handling
103 // 4. Transaction handling
104 // 5. Database backend management
105 // 6. Getter/Setter (possibly in alphabetical order)
106 //
107
108   /** begin timeout in ms */
109   protected long beginTimeout;
110
111   /** commit timeout in ms */
112   protected long commitTimeout;
113
114   /** rollback timeout in ms */
115   protected long rollbackTimeout;
116
117   /** The virtual database owning this Request Manager */
118   protected VirtualDatabase vdb;
119
120   /** The request scheduler to order and schedule requests */
121   protected AbstractScheduler scheduler;
122
123   /** An optional request cache to cache responses to SQL requests */
124   protected AbstractResultCache resultCache;
125
126   /** The request load balancer to use to send requests to the databases */
127   protected AbstractLoadBalancer loadBalancer;
128
129   /** An optional recovery log */
130   protected RecoveryLog recoveryLog;
131
132   /** The backup manager responsible for backup and restore of backends */
133   protected BackupManager backupManager;
134
135   // The virtual dabase schema
136 protected DatabaseSchema dbs;
137
138   /** <code>true</code> if schema is no more up-to-date and needs a refresh */
139   private boolean schemaIsDirty = false;
140
141   private boolean schemaIsStatic = false;
142
143   private boolean isCaseSensitiveParsing = false;
144
145   protected ParsingCache parsingCache = null;
146
147   private MetadataCache metadataCache = null;
148
149   // SQL queries parsing granularity according to Scheduler, ResultCache and
150 // LoadBalancer required granularity
151 protected int schedulerParsingranularity = ParsingGranularities.NO_PARSING;
152
153   private int cacheParsingranularity = ParsingGranularities.NO_PARSING;
154
155   private int loadBalancerParsingranularity = ParsingGranularities.NO_PARSING;
156
157   protected int requiredParsingGranularity = ParsingGranularities.NO_PARSING;
158
159   /** Transaction id/Login mapping */
160   protected Hashtable tidLoginTable;
161
162   /** Transaction id/Savepoints mapping */
163   protected Hashtable tidSavepoints;
164
165   protected static Trace logger = null;
166
167   private BackendStateListener backendStateListener;
168
169   //
170 // Constructors
171 //
172
173   /**
174    * Creates a new <code>RequestManager</code> instance.
175    *
176    * @param vdb the virtual database this request manager belongs to
177    * @param scheduler the Request Scheduler to use
178    * @param cache a Query Cache implementation
179    * @param loadBalancer the Request Load Balancer to use
180    * @param recoveryLog the Log Recovery to use
181    * @param beginTimeout timeout in seconds for begin
182    * @param commitTimeout timeout in seconds for commit
183    * @param rollbackTimeout timeout in seconds for rollback
184    * @throws SQLException if an error occurs
185    * @throws NotCompliantMBeanException if the MBean is not JMX compliant
186    */
187   public RequestManager(VirtualDatabase vdb, AbstractScheduler scheduler,
188       AbstractResultCache cache, AbstractLoadBalancer loadBalancer,
189       RecoveryLog recoveryLog, long beginTimeout, long commitTimeout,
190       long rollbackTimeout) throws SQLException , NotCompliantMBeanException
191   {
192     super(RequestManagerMBean.class);
193     this.vdb = vdb;
194     assignAndCheckSchedulerLoadBalancerValidity(scheduler, loadBalancer);
195     // requiredParsingGranularity is the maximum of each component granularity
196 this.resultCache = cache;
197     if (resultCache != null)
198     {
199       cacheParsingranularity = cache.getParsingGranularity();
200       if (cacheParsingranularity > requiredParsingGranularity)
201         requiredParsingGranularity = cacheParsingranularity;
202     }
203     setRecoveryLog(recoveryLog);
204     initRequestManagerVariables(vdb, beginTimeout, commitTimeout,
205         rollbackTimeout);
206     setBackendsLastKnownCheckpointFromRecoveryLog();
207     logger.info(Translate.get("requestmanager.parsing.granularity",
208         ParsingGranularities.getInformation(requiredParsingGranularity)));
209
210     if (MBeanServerManager.isJmxEnabled())
211     {
212       try
213       {
214         MBeanServerManager.registerMBean(this, JmxConstants
215             .getRequestManagerObjectName(vdb.getVirtualDatabaseName()));
216
217       }
218       catch (Exception e)
219       {
220         logger.error(Translate.get("jmx.failed.register.mbean.requestmanager"));
221       }
222     }
223   }
224
225   /**
226    * Retrieve the last known checkpoint from the recovery log and set it for
227    * each backend.
228    */
229   private void setBackendsLastKnownCheckpointFromRecoveryLog()
230   {
231
232     if (recoveryLog == null)
233       return;
234     String databaseName = vdb.getVirtualDatabaseName();
235     ArrayList backends = vdb.getBackends();
236     int size = backends.size();
237     DatabaseBackend backend;
238     BackendRecoveryInfo info;
239     for (int i = 0; i < size; i++)
240     {
241       backend = (DatabaseBackend) backends.get(i);
242       info = recoveryLog
243           .getBackendRecoveryInfo(databaseName, backend.getName());
244       backend.setLastKnownCheckpoint(info.getCheckpoint());
245     }
246
247   }
248
249   /**
250    * Check that Scheduler and Load Balancer are not null and have compatible
251    * RAIDb levels.
252    *
253    * @param scheduler
254    * @param loadBalancer
255    * @throws SQLException if an error occurs
256    */
257   private void assignAndCheckSchedulerLoadBalancerValidity(
258       AbstractScheduler scheduler, AbstractLoadBalancer loadBalancer)
259       throws SQLException
260   {
261     if (scheduler == null)
262       throw new SQLException (Translate.get("requestmanager.null.scheduler"));
263
264     if (loadBalancer == null)
265       throw new SQLException (Translate.get("requestmanager.null.loadbalancer"));
266
267     if (scheduler.getRAIDbLevel() != loadBalancer.getRAIDbLevel())
268       throw new SQLException (Translate.get(
269           "requestmanager.incompatible.raidb.levels",
270           new String []{"" + scheduler.getRAIDbLevel(),
271               "" + loadBalancer.getRAIDbLevel()}));
272
273     // requiredParsingGranularity is the maximum of each component granularity
274 setScheduler(scheduler);
275     schedulerParsingranularity = scheduler.getParsingGranularity();
276     requiredParsingGranularity = schedulerParsingranularity;
277     setLoadBalancer(loadBalancer);
278     loadBalancerParsingranularity = loadBalancer.getParsingGranularity();
279     if (loadBalancerParsingranularity > requiredParsingGranularity)
280       requiredParsingGranularity = loadBalancerParsingranularity;
281   }
282
283   /**
284    * Method initRequestManagerVariables.
285    *
286    * @param vdb
287    * @param beginTimeout
288    * @param commitTimeout
289    * @param rollbackTimeout
290    */
291   private void initRequestManagerVariables(VirtualDatabase vdb,
292       long beginTimeout, long commitTimeout, long rollbackTimeout)
293   {
294     this.tidLoginTable = new Hashtable ();
295     this.tidSavepoints = new Hashtable ();
296     this.beginTimeout = beginTimeout;
297     this.commitTimeout = commitTimeout;
298     this.rollbackTimeout = rollbackTimeout;
299     this.vdb = vdb;
300     logger = Trace.getLogger("org.objectweb.cjdbc.controller.RequestManager."
301         + vdb.getDatabaseName());
302   }
303
304   //
305 // Request Handling
306 //
307
308   /**
309    * Perform a read request and return the reply. Call first the scheduler, then
310    * the cache (if defined) and finally the load balancer.
311    *
312    * @param request the request to execute
313    * @return a <code>ControllerResultSet</code> value
314    * @exception SQLException if an error occurs
315    */
316   public ControllerResultSet execReadRequest(SelectRequest request)
317       throws SQLException
318   {
319     // Sanity check
320 if (!request.isAutoCommit())
321     { // Check that the transaction has been
322 // started
323 long tid = request.getTransactionId();
324       if (!tidLoginTable.containsKey(new Long (tid)))
325         throw new SQLException (Translate.get("transaction.not.started", tid));
326     }
327
328     // If we need to parse the request, try to get the parsing from the
329 // cache.
330 // Note that if we have a cache miss but backgroundParsing has been
331 // turned
332 // on, then this call will start a ParsedThread in background.
333 if ((requiredParsingGranularity != ParsingGranularities.NO_PARSING)
334         && (!request.isParsed()))
335     {
336       if (parsingCache == null)
337         request.parse(getDatabaseSchema(), requiredParsingGranularity,
338             isCaseSensitiveParsing);
339       else
340         parsingCache.getParsingFromCache(request);
341     }
342
343     //
344 // SCHEDULER
345 //
346
347     // Get the parsing now if the request is not yet parsed. The parsing is
348 // handled by the ParsingCache that may already have parsed the request
349 // in background (if backgroundParsing is set).
350 if ((schedulerParsingranularity != ParsingGranularities.NO_PARSING)
351         && !request.isParsed())
352     {
353       if (parsingCache == null)
354         request.parse(dbs, requiredParsingGranularity, isCaseSensitiveParsing);
355       else
356         parsingCache.getParsingFromCacheAndParseIfMissing(request);
357     }
358
359     if (logger.isDebugEnabled())
360       logger.debug(Translate.get("requestmanager.read.request.schedule",
361           new String []{String.valueOf(request.getId()),
362               request.getSQLShortForm(vdb.getSQLShortFormLength())}));
363
364     // Wait for the scheduler to give us the authorization to execute
365 scheduler.scheduleReadRequest(request);
366
367     //
368 // CACHE
369 //
370
371     ControllerResultSet result = null;
372     try
373     { // Check cache if any
374 if (resultCache != null)
375       {
376         if (logger.isDebugEnabled())
377           logger.debug(Translate.get("requestmanager.read.request.cache.get",
378               new String []{String.valueOf(request.getId()),
379                   request.getSQLShortForm(vdb.getSQLShortFormLength())}));
380
381         AbstractResultCacheEntry qce = resultCache.getFromCache(request, true);
382         if (qce != null)
383         {
384           result = qce.getResult();
385           if (result != null)
386           { // Cache hit !
387 if (vdb.getSQLMonitor() != null)
388               vdb.getSQLMonitor().logCacheHit(request);
389
390             scheduler.readCompleted(request);
391             return result;
392           }
393         }
394       }
395
396       //
397 // LOAD BALANCER
398 //
399
400       if (logger.isDebugEnabled())
401         logger.debug(Translate.get("requestmanager.read.request.balance",
402             new String []{String.valueOf(request.getId()),
403                 request.getSQLShortForm(vdb.getSQLShortFormLength())}));
404
405       // At this point, we have a result cache miss.
406 // If we had a parsing cache miss too, wait for the parsing to be
407 // done if
408 // needed.
409 if ((loadBalancerParsingranularity != ParsingGranularities.NO_PARSING)
410           && !request.isParsed())
411       {
412         if (parsingCache == null)
413           request
414               .parse(dbs, requiredParsingGranularity, isCaseSensitiveParsing);
415         else
416           parsingCache.getParsingFromCacheAndParseIfMissing(request);
417       }
418
419       // Send the request to the load balancer
420 result = loadBalancer.execReadRequest(request, metadataCache);
421
422       //
423 // UPDATES & NOTIFICATIONS
424 //
425
426       // Update cache
427 if ((resultCache != null)
428           && (request.getCacheAbility() != RequestType.UNCACHEABLE))
429       {
430         if (logger.isDebugEnabled())
431           logger.debug(Translate.get(
432               "requestmanager.read.request.cache.update", new String []{
433                   String.valueOf(request.getId()),
434                   request.getSQLShortForm(vdb.getSQLShortFormLength())}));
435
436         if (!request.isParsed()
437             && (cacheParsingranularity != ParsingGranularities.NO_PARSING))
438         { // The cache was the only one to need parsing and the request was not
439 // previously in the cache
440 if (parsingCache == null)
441             request.parse(dbs, requiredParsingGranularity,
442                 isCaseSensitiveParsing);
443           else
444             parsingCache.getParsingFromCacheAndParseIfMissing(request);
445         }
446         resultCache.addToCache(request, result);
447       }
448     }
449     catch (Exception failed)
450     {
451       if (resultCache != null)
452         resultCache.removeFromPendingQueries(request);
453       scheduler.readCompleted(request);
454       if (failed instanceof NoMoreBackendException)
455         throw (NoMoreBackendException) failed;
456       String msg = Translate.get("requestmanager.request.failed", new String []{
457           request.getSQLShortForm(vdb.getSQLShortFormLength()),
458           failed.getMessage()});
459       if (failed instanceof RuntimeException )
460         logger.warn(msg, failed);
461       else
462         logger.warn(msg);
463       if (failed instanceof SQLException )
464         throw (SQLException ) failed;
465
466       throw new SQLException (msg);
467     }
468
469     // Notify scheduler of completion
470 scheduler.readCompleted(request);
471
472     return result;
473   }
474
475   /**
476    * Perform a write request and return the number of rows affected Call first
477    * the scheduler (if defined), then notify the cache (if defined) and finally
478    * call the load balancer.
479    *
480    * @param request the request to execute
481    * @return number of rows affected
482    * @exception SQLException if an error occurs
483    */
484   public int execWriteRequest(AbstractWriteRequest request) throws SQLException
485   {
486     scheduleExecWriteRequest(request);
487     int execWriteRequestResult = 0;
488     try
489     {
490       execWriteRequestResult = loadBalanceExecWriteRequest(request);
491     }
492     catch (AllBackendsFailedException e)
493     {
494       String msg = Translate
495           .get("requestmanager.write.request.failed.unexpected");
496       logger.fatal(msg, e);
497       throw new RuntimeException (msg, e);
498     }
499     updateAndNotifyExecWriteRequest(request);
500     return execWriteRequestResult;
501   }
502
503   /**
504    * Perform a write request and return the auto generated keys. Call first the
505    * scheduler (if defined), then notify the cache (if defined) and finally call
506    * the load balancer.
507    *
508    * @param request the request to execute
509    * @return auto generated keys.
510    * @exception SQLException if an error occurs
511    */
512   public ControllerResultSet execWriteRequestWithKeys(
513       AbstractWriteRequest request) throws SQLException
514   {
515     scheduleExecWriteRequest(request);
516     ControllerResultSet execWriteRequestWithKeysResult = null;
517     try
518     {
519       execWriteRequestWithKeysResult = loadBalanceExecWriteRequestWithKeys(request);
520     }
521     catch (AllBackendsFailedException e)
522     {
523       String msg = Translate
524           .get("requestmanager.write.request.keys.failed.unexpected");
525       logger.fatal(msg, e);
526       throw new RuntimeException (msg, e);
527     }
528     updateAndNotifyExecWriteRequest(request);
529     return execWriteRequestWithKeysResult;
530   }
531
532   /**
533    * Schedule a request for execution.
534    *
535    * @param request the request to execute
536    * @throws SQLException if an error occurs
537    */
538   public void scheduleExecWriteRequest(AbstractWriteRequest request)
539       throws SQLException
540   {
541     // Sanity check
542 if (!request.isAutoCommit())
543     { // Check that the transaction has been
544 // started
545 long tid = request.getTransactionId();
546       if (!tidLoginTable.containsKey(new Long (tid)))
547         throw new SQLException (Translate.get("transaction.not.started", tid));
548     }
549
550     // If we need to parse the request, try to get the parsing from the
551 // cache.
552 // Note that if we have a cache miss but backgroundParsing has been
553 // turned
554 // on, then this call will start a ParsedThread in background.
555 if ((requiredParsingGranularity != ParsingGranularities.NO_PARSING)
556         && (!request.isParsed()))
557     {
558       if (parsingCache == null)
559         request.parse(getDatabaseSchema(), requiredParsingGranularity,
560             isCaseSensitiveParsing);
561       else
562         parsingCache.getParsingFromCache(request);
563     }
564
565     //
566 // SCHEDULER
567 //
568
569     // Get the parsing now if the request is not yet parsed. The parsing is
570 // handled by the ParsingCache that may already have parsed the request
571 // in background (if backgroundParsing is set).
572 if ((schedulerParsingranularity != ParsingGranularities.NO_PARSING)
573         && !request.isParsed())
574     {
575       if (parsingCache == null)
576         request.parse(getDatabaseSchema(), requiredParsingGranularity,
577             isCaseSensitiveParsing);
578       else
579         parsingCache.getParsingFromCacheAndParseIfMissing(request);
580     }
581
582     if (logger.isDebugEnabled())
583       logger.debug(Translate.get("requestmanager.write.request.schedule",
584           new String []{String.valueOf(request.getId()),
585               request.getSQLShortForm(vdb.getSQLShortFormLength())}));
586
587     // Wait for the scheduler to give us the authorization to execute
588 try
589     {
590       scheduler.scheduleWriteRequest(request);
591     }
592     catch (RollbackException e)
593     { // Something bad happened and we need to rollback this transaction
594 rollback(request.getTransactionId(), true);
595       throw new SQLException (e.getMessage());
596     }
597
598     // If we have a parsing cache miss, wait for the parsing to be done if
599 // needed. Note that even if the cache was the only one to require
600 // parsing,
601 // we wait for the parsing result here, because if it fails, we must not
602 // execute the query.
603 try
604     {
605       if ((requiredParsingGranularity != ParsingGranularities.NO_PARSING)
606           && !request.isParsed())
607       {
608         if (parsingCache == null)
609           request.parse(getDatabaseSchema(), requiredParsingGranularity,
610               isCaseSensitiveParsing);
611         else
612           parsingCache.getParsingFromCacheAndParseIfMissing(request);
613       }
614     }
615     catch (SQLException e)
616     {
617       // If the parsing fail, we must release the lock acquired ...
618 scheduler.writeCompleted(request);
619       throw e;
620     }
621   }
622
623   /**
624    * Send the given query to the load balancer. If the request fails, the
625    * scheduler is properly notified.
626    *
627    * @param request the request to execute
628    * @throws AllBackendsFailedException if all backends failed to execute the
629    * query
630    * @throws SQLException if an error occurs
631    * @return auto-generated keys
632    */
633   public ControllerResultSet loadBalanceExecWriteRequestWithKeys(
634       AbstractWriteRequest request) throws AllBackendsFailedException,
635       SQLException
636   {
637     if (logger.isDebugEnabled())
638       logger.debug(Translate.get("requestmanager.write.request.balance",
639           new String []{String.valueOf(request.getId()),
640               request.getSQLShortForm(vdb.getSQLShortFormLength())}));
641
642     try
643     { // Send the request to the load balancer
644 return loadBalancer.execWriteRequestWithKeys(request, metadataCache);
645     }
646     catch (Exception failed)
647     {
648       scheduler.writeCompleted(request);
649       String msg = Translate.get("requestmanager.request.failed", new String []{
650           request.getSQLShortForm(vdb.getSQLShortFormLength()),
651           failed.getMessage()});
652       if (failed instanceof RuntimeException )
653         logger.warn(msg, failed);
654       else
655         logger.warn(msg);
656       if (failed instanceof AllBackendsFailedException)
657         throw (AllBackendsFailedException) failed;
658       else
659         throw new SQLException (msg);
660     }
661   }
662
663   /**
664    * Send the given query to the load balancer. If the request fails, the
665    * scheduler is properly notified.
666    *
667    * @param request the request to execute
668    * @throws AllBackendsFailedException if all backends failed to execute the
669    * query
670    * @throws SQLException if an error occurs
671    * @return number of modified lines
672    */
673   public int loadBalanceExecWriteRequest(AbstractWriteRequest request)
674       throws AllBackendsFailedException, SQLException
675   {
676     if (logger.isDebugEnabled())
677       logger.debug(Translate.get("requestmanager.write.request.balance",
678           new String []{String.valueOf(request.getId()),
679               request.getSQLShortForm(vdb.getSQLShortFormLength())}));
680
681     try
682     { // Send the request to the load balancer
683 if (request.isUpdate() && (resultCache != null))
684       { // Try the optimization if we try to update values that are already
685 // up-to-date.
686 if (!resultCache.isUpdateNecessary((UpdateRequest) request))
687           return 0;
688       }
689       return loadBalancer.execWriteRequest(request);
690     }
691     catch (Exception failed)
692     {
693       scheduler.writeCompleted(request);
694       String msg = Translate.get("requestmanager.request.failed", new String []{
695           request.getSQLShortForm(vdb.getSQLShortFormLength()),
696           failed.getMessage()});
697
698       // Logging
699 if (failed instanceof RuntimeException )
700         logger.warn(msg, failed);
701       else
702         logger.warn(msg);
703
704       // Rethrow exception
705 if (failed instanceof AllBackendsFailedException)
706         throw (AllBackendsFailedException) failed;
707       else if (failed instanceof SQLException )
708         throw (SQLException ) failed;
709       else
710         throw new SQLException (msg);
711     }
712   }
713
714   /**
715    * Update the cache, notify the recovery log, update the database schema if
716    * needed and finally notify the scheduler. Note that if an error occurs, the
717    * scheduler is always notified.
718    *
719    * @param request the request to execute
720    * @throws SQLException if an error occurs
721    */
722   public void updateAndNotifyExecWriteRequest(AbstractWriteRequest request)
723       throws SQLException
724   {
725     try
726     { // Notify cache if any
727 if (resultCache != null)
728       { // Update cache
729 if (logger.isDebugEnabled())
730           logger.debug(Translate.get(
731               "requestmanager.write.request.cache.update", new String []{
732                   String.valueOf(request.getId()),
733                   request.getSQLShortForm(vdb.getSQLShortFormLength())}));
734
735         resultCache.writeNotify(request);
736       }
737
738       if (metadataCache != null)
739       {
740         if (request.isAlter() || request.isDrop())
741           // Note that we also have to flush on Drop in case the same table is
742 // re-created with another schema.
743 metadataCache.flushCache();
744       }
745
746       // Log the request
747 if (recoveryLog != null)
748       {
749         if (logger.isDebugEnabled())
750           logger.debug(Translate.get("requestmanager.write.request.log",
751               new String []{String.valueOf(request.getId()),
752                   request.getSQLShortForm(vdb.getSQLShortFormLength())}));
753
754         recoveryLog.logRequest(request);
755       }
756
757       // Update the schema if needed
758 if ((request.isDDL())
759           && (requiredParsingGranularity != ParsingGranularities.NO_PARSING))
760       {
761         if (request.isCreate())
762         { // Add the table to the schema
763 CreateRequest createRequest = (CreateRequest) request;
764           if (createRequest.altersDatabaseSchema())
765           {
766             if (createRequest.getDatabaseTable() != null)
767             {
768               getDatabaseSchema().addTable(
769                   ((CreateRequest) request).getDatabaseTable());
770               if (logger.isDebugEnabled())
771                 logger.debug(Translate.get("requestmanager.schema.add.table",
772                     request.getTableName()));
773             }
774             else
775               // Some other create statement that modifies the schema, force
776 // refresh
777 schemaIsDirty = true;
778           }
779         }
780         else if (request.isDrop())
781         { // Delete the table from the schema
782 if (getDatabaseSchema().removeTable(
783               getDatabaseSchema().getTable(request.getTableName())))
784             if (logger.isDebugEnabled())
785               logger.debug(Translate.get("requestmanager.schema.remove.table",
786                   request.getTableName()));
787             else
788               // Table not found, force refresh
789 schemaIsDirty = true;
790         }
791         else if (request.isAlter()
792             && (requiredParsingGranularity > ParsingGranularities.TABLE))
793         { // Add or drop the column from the table
794 AlterRequest req = (AlterRequest) request;
795           DatabaseTable alteredTable = getDatabaseSchema().getTable(
796               req.getTableName());
797           if ((alteredTable != null) && (req.getColumn() != null))
798           {
799             if (req.isDrop())
800               alteredTable.remove(req.getColumn().getName());
801             else if (req.isAdd())
802               alteredTable.addColumn(req.getColumn());
803             else
804               // Unsupported, force refresh
805 schemaIsDirty = true;
806           }
807           else
808             // Table not found, force refresh
809 schemaIsDirty = true;
810         }
811         else
812           // Unsupported, force refresh
813 schemaIsDirty = true;
814       }
815     }
816     catch (Exception failed)
817     {
818       scheduler.writeCompleted(request);
819       String msg = Translate.get("requestmanager.request.failed", new String []{
820           request.getSQLShortForm(vdb.getSQLShortFormLength()),
821           failed.getMessage()});
822       if (failed instanceof RuntimeException )
823         logger.warn(msg, failed);
824       else
825         logger.warn(msg);
826       throw new SQLException (msg);
827     }
828     finally
829     {
830       // Notify scheduler
831 scheduler.writeCompleted(request);
832     }
833   }
834
835   /**
836    * Call a stored procedure that returns a ResultSet.
837    *
838    * @param proc the stored procedure call
839    * @return a <code>ControllerResultSet</code> value
840    * @throws AllBackendsFailedException if all backends failed to execute the
841    * stored procedure
842    * @exception SQLException if an error occurs
843    */
844   public ControllerResultSet execReadStoredProcedure(StoredProcedure proc)
845       throws AllBackendsFailedException, SQLException
846   {
847     ControllerResultSet result = null;
848     try
849     {
850       scheduleStoredProcedure(proc);
851
852       if (logger.isDebugEnabled())
853         logger.debug(Translate.get("requestmanager.read.stored.procedure",
854             new String []{String.valueOf(proc.getId()),
855                 proc.getSQLShortForm(vdb.getSQLShortFormLength())}));
856
857       result = loadBalanceReadStoredProcedure(proc);
858
859       flushCacheAndLogStoredProcedure(proc, true);
860
861       // Notify scheduler of completion
862 scheduler.storedProcedureCompleted(proc);
863
864       return result;
865     }
866     catch (AllBackendsFailedException e)
867     {
868       throw e;
869     }
870     catch (Exception failed)
871     {
872       scheduler.storedProcedureCompleted(proc);
873       String msg = Translate.get("requestmanager.stored.procedure.failed",
874           new String []{proc.getSQLShortForm(vdb.getSQLShortFormLength()),
875               failed.getMessage()});
876       logger.warn(msg);
877       if (failed instanceof SQLException )
878       {
879         throw (SQLException ) failed;
880       }
881       throw new SQLException (msg);
882     }
883   }
884
885   /**
886