← Back to branch summary

~drizzle-trunk/drizzle/development

« back to all changes in this revision

Viewing changes to drizzled/transaction_services.cc

  • Committer: Mark Atwood
  • Date: 2011-08-11 03:05:03 UTC
  • mfrom: (2385.1.12 refactor4)
  • Revision ID: me@mark.atwood.name-20110811030503-rp9xjihc5x3y0x4q
merge lp:~olafvdspek/drizzle/refactor4

Show diffs side-by-side

added added

removed removed

Lines of Context:
drizzled/transaction_services.cc
47
47
 * plugins can understand.
48
48
 */
49
49
 
50
 
#include "config.h"
51
 
#include "drizzled/my_hash.h"
52
 
#include "drizzled/error.h"
53
 
#include "drizzled/gettext.h"
54
 
#include "drizzled/probes.h"
55
 
#include "drizzled/sql_parse.h"
56
 
#include "drizzled/session.h"
57
 
#include "drizzled/sql_base.h"
58
 
#include "drizzled/replication_services.h"
59
 
#include "drizzled/transaction_services.h"
60
 
#include "drizzled/transaction_context.h"
61
 
#include "drizzled/message/transaction.pb.h"
62
 
#include "drizzled/message/statement_transform.h"
63
 
#include "drizzled/resource_context.h"
64
 
#include "drizzled/lock.h"
65
 
#include "drizzled/item/int.h"
66
 
#include "drizzled/item/empty_string.h"
67
 
#include "drizzled/field/epoch.h"
68
 
#include "drizzled/plugin/client.h"
69
 
#include "drizzled/plugin/monitored_in_transaction.h"
70
 
#include "drizzled/plugin/transactional_storage_engine.h"
71
 
#include "drizzled/plugin/xa_resource_manager.h"
72
 
#include "drizzled/plugin/xa_storage_engine.h"
73
 
#include "drizzled/internal/my_sys.h"
 
50
#include <config.h>
 
51
#include <drizzled/current_session.h>
 
52
#include <drizzled/error.h>
 
53
#include <drizzled/gettext.h>
 
54
#include <drizzled/probes.h>
 
55
#include <drizzled/sql_parse.h>
 
56
#include <drizzled/session.h>
 
57
#include <drizzled/session/times.h>
 
58
#include <drizzled/sql_base.h>
 
59
#include <drizzled/replication_services.h>
 
60
#include <drizzled/transaction_services.h>
 
61
#include <drizzled/transaction_context.h>
 
62
#include <drizzled/message/transaction.pb.h>
 
63
#include <drizzled/message/statement_transform.h>
 
64
#include <drizzled/resource_context.h>
 
65
#include <drizzled/lock.h>
 
66
#include <drizzled/item/int.h>
 
67
#include <drizzled/item/empty_string.h>
 
68
#include <drizzled/field/epoch.h>
 
69
#include <drizzled/plugin/client.h>
 
70
#include <drizzled/plugin/monitored_in_transaction.h>
 
71
#include <drizzled/plugin/transactional_storage_engine.h>
 
72
#include <drizzled/plugin/xa_resource_manager.h>
 
73
#include <drizzled/plugin/xa_storage_engine.h>
 
74
#include <drizzled/internal/my_sys.h>
 
75
#include <drizzled/statistics_variables.h>
 
76
#include <drizzled/system_variables.h>
 
77
#include <drizzled/session/transactions.h>
74
78
 
75
79
#include <vector>
76
80
#include <algorithm>
80
84
using namespace std;
81
85
using namespace google;
82
86
 
83
 
namespace drizzled
84
 
{
 
87
namespace drizzled {
85
88
 
86
89
/**
87
90
 * @defgroup Transactions
300
303
 * transaction after all DDLs, just like the statement transaction
301
304
 * is always committed at the end of all statements.
302
305
 */
303
 
TransactionServices::TransactionServices()
 
306
 
 
307
static plugin::XaStorageEngine& xa_storage_engine()
304
308
{
305
 
  plugin::StorageEngine *engine= plugin::StorageEngine::findByName("InnoDB");
306
 
  if (engine)
307
 
  {
308
 
    xa_storage_engine= (plugin::XaStorageEngine*)engine; 
309
 
  }
310
 
  else 
311
 
  {
312
 
    xa_storage_engine= NULL;
313
 
  }
 
309
  static plugin::XaStorageEngine& engine= static_cast<plugin::XaStorageEngine&>(*plugin::StorageEngine::findByName("InnoDB"));
 
310
  return engine;
314
311
}
315
312
 
316
 
void TransactionServices::registerResourceForStatement(Session::reference session,
 
313
void TransactionServices::registerResourceForStatement(Session& session,
317
314
                                                       plugin::MonitoredInTransaction *monitored,
318
315
                                                       plugin::TransactionalStorageEngine *engine)
319
316
{
329
326
    registerResourceForTransaction(session, monitored, engine);
330
327
  }
331
328
 
332
 
  TransactionContext *trans= &session.transaction.stmt;
333
 
  ResourceContext *resource_context= session.getResourceContext(monitored, 0);
 
329
  TransactionContext& trans= session.transaction.stmt;
 
330
  ResourceContext& resource_context= session.getResourceContext(*monitored, 0);
334
331
 
335
 
  if (resource_context->isStarted())
 
332
  if (resource_context.isStarted())
336
333
    return; /* already registered, return */
337
334
 
338
335
  assert(monitored->participatesInSqlTransaction());
339
336
  assert(not monitored->participatesInXaTransaction());
340
337
 
341
 
  resource_context->setMonitored(monitored);
342
 
  resource_context->setTransactionalStorageEngine(engine);
343
 
  trans->registerResource(resource_context);
344
 
 
345
 
  trans->no_2pc|= true;
 
338
  resource_context.setMonitored(monitored);
 
339
  resource_context.setTransactionalStorageEngine(engine);
 
340
  trans.registerResource(&resource_context);
 
341
  trans.no_2pc= true;
346
342
}
347
343
 
348
 
void TransactionServices::registerResourceForStatement(Session::reference session,
 
344
void TransactionServices::registerResourceForStatement(Session& session,
349
345
                                                       plugin::MonitoredInTransaction *monitored,
350
346
                                                       plugin::TransactionalStorageEngine *engine,
351
347
                                                       plugin::XaResourceManager *resource_manager)
362
358
    registerResourceForTransaction(session, monitored, engine, resource_manager);
363
359
  }
364
360
 
365
 
  TransactionContext *trans= &session.transaction.stmt;
366
 
  ResourceContext *resource_context= session.getResourceContext(monitored, 0);
 
361
  TransactionContext& trans= session.transaction.stmt;
 
362
  ResourceContext& resource_context= session.getResourceContext(*monitored, 0);
367
363
 
368
 
  if (resource_context->isStarted())
 
364
  if (resource_context.isStarted())
369
365
    return; /* already registered, return */
370
366
 
371
367
  assert(monitored->participatesInXaTransaction());
372
368
  assert(monitored->participatesInSqlTransaction());
373
369
 
374
 
  resource_context->setMonitored(monitored);
375
 
  resource_context->setTransactionalStorageEngine(engine);
376
 
  resource_context->setXaResourceManager(resource_manager);
377
 
  trans->registerResource(resource_context);
378
 
 
379
 
  trans->no_2pc|= false;
 
370
  resource_context.setMonitored(monitored);
 
371
  resource_context.setTransactionalStorageEngine(engine);
 
372
  resource_context.setXaResourceManager(resource_manager);
 
373
  trans.registerResource(&resource_context);
380
374
}
381
375
 
382
 
void TransactionServices::registerResourceForTransaction(Session::reference session,
 
376
void TransactionServices::registerResourceForTransaction(Session& session,
383
377
                                                         plugin::MonitoredInTransaction *monitored,
384
378
                                                         plugin::TransactionalStorageEngine *engine)
385
379
{
386
 
  TransactionContext *trans= &session.transaction.all;
387
 
  ResourceContext *resource_context= session.getResourceContext(monitored, 1);
 
380
  TransactionContext& trans= session.transaction.all;
 
381
  ResourceContext& resource_context= session.getResourceContext(*monitored, 1);
388
382
 
389
 
  if (resource_context->isStarted())
 
383
  if (resource_context.isStarted())
390
384
    return; /* already registered, return */
391
385
 
392
386
  session.server_status|= SERVER_STATUS_IN_TRANS;
393
387
 
394
 
  trans->registerResource(resource_context);
 
388
  trans.registerResource(&resource_context);
395
389
 
396
390
  assert(monitored->participatesInSqlTransaction());
397
391
  assert(not monitored->participatesInXaTransaction());
398
392
 
399
 
  resource_context->setMonitored(monitored);
400
 
  resource_context->setTransactionalStorageEngine(engine);
401
 
  trans->no_2pc|= true;
 
393
  resource_context.setMonitored(monitored);
 
394
  resource_context.setTransactionalStorageEngine(engine);
 
395
  trans.no_2pc= true;
402
396
 
403
397
  if (session.transaction.xid_state.xid.is_null())
404
398
    session.transaction.xid_state.xid.set(session.getQueryId());
405
399
 
406
400
  /* Only true if user is executing a BEGIN WORK/START TRANSACTION */
407
 
  if (! session.getResourceContext(monitored, 0)->isStarted())
 
401
  if (not session.getResourceContext(*monitored, 0).isStarted())
408
402
    registerResourceForStatement(session, monitored, engine);
409
403
}
410
404
 
411
 
void TransactionServices::registerResourceForTransaction(Session::reference session,
 
405
void TransactionServices::registerResourceForTransaction(Session& session,
412
406
                                                         plugin::MonitoredInTransaction *monitored,
413
407
                                                         plugin::TransactionalStorageEngine *engine,
414
408
                                                         plugin::XaResourceManager *resource_manager)
415
409
{
416
410
  TransactionContext *trans= &session.transaction.all;
417
 
  ResourceContext *resource_context= session.getResourceContext(monitored, 1);
 
411
  ResourceContext& resource_context= session.getResourceContext(*monitored, 1);
418
412
 
419
 
  if (resource_context->isStarted())
 
413
  if (resource_context.isStarted())
420
414
    return; /* already registered, return */
421
415
 
422
416
  session.server_status|= SERVER_STATUS_IN_TRANS;
423
417
 
424
 
  trans->registerResource(resource_context);
 
418
  trans->registerResource(&resource_context);
425
419
 
426
420
  assert(monitored->participatesInSqlTransaction());
427
421
 
428
 
  resource_context->setMonitored(monitored);
429
 
  resource_context->setXaResourceManager(resource_manager);
430
 
  resource_context->setTransactionalStorageEngine(engine);
431
 
  trans->no_2pc|= true;
 
422
  resource_context.setMonitored(monitored);
 
423
  resource_context.setXaResourceManager(resource_manager);
 
424
  resource_context.setTransactionalStorageEngine(engine);
 
425
  trans->no_2pc= true;
432
426
 
433
427
  if (session.transaction.xid_state.xid.is_null())
434
428
    session.transaction.xid_state.xid.set(session.getQueryId());
436
430
  engine->startTransaction(&session, START_TRANS_NO_OPTIONS);
437
431
 
438
432
  /* Only true if user is executing a BEGIN WORK/START TRANSACTION */
439
 
  if (! session.getResourceContext(monitored, 0)->isStarted())
 
433
  if (! session.getResourceContext(*monitored, 0).isStarted())
440
434
    registerResourceForStatement(session, monitored, engine, resource_manager);
441
435
}
442
436
 
443
437
void TransactionServices::allocateNewTransactionId()
444
438
{
445
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
446
 
  if (! replication_services.isActive())
 
439
  if (! ReplicationServices::isActive())
447
440
  {
448
441
    return;
449
442
  }
450
443
 
451
444
  Session *my_session= current_session;
452
 
  uint64_t xa_id= xa_storage_engine->getNewTransactionId(my_session);
 
445
  uint64_t xa_id= xa_storage_engine().getNewTransactionId(my_session);
453
446
  my_session->setXaId(xa_id);
454
447
}
455
448
 
456
 
uint64_t TransactionServices::getCurrentTransactionId(Session::reference session)
 
449
uint64_t TransactionServices::getCurrentTransactionId(Session& session)
457
450
{
458
451
  if (session.getXaId() == 0)
459
452
  {
460
 
    session.setXaId(xa_storage_engine->getNewTransactionId(&session)); 
 
453
    session.setXaId(xa_storage_engine().getNewTransactionId(&session)); 
461
454
  }
462
455
 
463
456
  return session.getXaId();
464
457
}
465
458
 
466
 
int TransactionServices::commitTransaction(Session::reference session,
 
459
int TransactionServices::commitTransaction(Session& session,
467
460
                                           bool normal_transaction)
468
461
{
469
462
  int error= 0, cookie= 0;
500
493
     */
501
494
    if (shouldConstructMessages())
502
495
    {
503
 
      for (TransactionContext::ResourceContexts::iterator it= resource_contexts.begin();
504
 
           it != resource_contexts.end() && ! error;
505
 
           ++it)
 
496
      BOOST_FOREACH(TransactionContext::ResourceContexts::reference resource_context, resource_contexts)
506
497
      {
507
 
        ResourceContext *resource_context= *it;
508
 
        int err;
 
498
        if (error)
 
499
          break;
509
500
        /*
510
501
          Do not call two-phase commit if this particular
511
502
          transaction is read-only. This allows for simpler
518
509
 
519
510
        if (resource->participatesInXaTransaction())
520
511
        {
521
 
          if ((err= resource_context->getXaResourceManager()->xaPrepare(&session, normal_transaction)))
 
512
          if (int err= resource_context->getXaResourceManager()->xaPrepare(&session, normal_transaction))
522
513
          {
523
514
            my_error(ER_ERROR_DURING_COMMIT, MYF(0), err);
524
515
            error= 1;
556
547
  @note
557
548
  This function does not care about global read lock. A caller should.
558
549
*/
559
 
int TransactionServices::commitPhaseOne(Session::reference session,
 
550
int TransactionServices::commitPhaseOne(Session& session,
560
551
                                        bool normal_transaction)
561
552
{
562
553
  int error=0;
574
565
 
575
566
  if (resource_contexts.empty() == false)
576
567
  {
577
 
    for (TransactionContext::ResourceContexts::iterator it= resource_contexts.begin();
578
 
         it != resource_contexts.end();
579
 
         ++it)
 
568
    BOOST_FOREACH(TransactionContext::ResourceContexts::reference resource_context, resource_contexts)
580
569
    {
581
 
      int err;
582
 
      ResourceContext *resource_context= *it;
583
 
 
584
570
      plugin::MonitoredInTransaction *resource= resource_context->getMonitored();
585
571
 
586
572
      if (resource->participatesInXaTransaction())
587
573
      {
588
 
        if ((err= resource_context->getXaResourceManager()->xaCommit(&session, all)))
 
574
        if (int err= resource_context->getXaResourceManager()->xaCommit(&session, all))
589
575
        {
590
576
          my_error(ER_ERROR_DURING_COMMIT, MYF(0), err);
591
577
          error= 1;
597
583
      }
598
584
      else if (resource->participatesInSqlTransaction())
599
585
      {
600
 
        if ((err= resource_context->getTransactionalStorageEngine()->commit(&session, all)))
 
586
        if (int err= resource_context->getTransactionalStorageEngine()->commit(&session, all))
601
587
        {
602
588
          my_error(ER_ERROR_DURING_COMMIT, MYF(0), err);
603
589
          error= 1;
611
597
    }
612
598
 
613
599
    if (is_real_trans)
614
 
      session.transaction.xid_state.xid.null();
 
600
      session.transaction.xid_state.xid.set_null();
615
601
 
616
602
    if (normal_transaction)
617
603
    {
623
609
  return error;
624
610
}
625
611
 
626
 
int TransactionServices::rollbackTransaction(Session::reference session,
 
612
int TransactionServices::rollbackTransaction(Session& session,
627
613
                                             bool normal_transaction)
628
614
{
629
615
  int error= 0;
637
623
    We must not rollback the normal transaction if a statement
638
624
    transaction is pending.
639
625
  */
640
 
  assert(session.transaction.stmt.getResourceContexts().empty() ||
641
 
              trans == &session.transaction.stmt);
 
626
  assert(session.transaction.stmt.getResourceContexts().empty() || trans == &session.transaction.stmt);
642
627
 
643
628
  if (resource_contexts.empty() == false)
644
629
  {
645
 
    for (TransactionContext::ResourceContexts::iterator it= resource_contexts.begin();
646
 
         it != resource_contexts.end();
647
 
         ++it)
 
630
    BOOST_FOREACH(TransactionContext::ResourceContexts::reference resource_context, resource_contexts)
648
631
    {
649
 
      int err;
650
 
      ResourceContext *resource_context= *it;
651
 
 
652
632
      plugin::MonitoredInTransaction *resource= resource_context->getMonitored();
653
633
 
654
634
      if (resource->participatesInXaTransaction())
655
635
      {
656
 
        if ((err= resource_context->getXaResourceManager()->xaRollback(&session, all)))
 
636
        if (int err= resource_context->getXaResourceManager()->xaRollback(&session, all))
657
637
        {
658
638
          my_error(ER_ERROR_DURING_ROLLBACK, MYF(0), err);
659
639
          error= 1;
665
645
      }
666
646
      else if (resource->participatesInSqlTransaction())
667
647
      {
668
 
        if ((err= resource_context->getTransactionalStorageEngine()->rollback(&session, all)))
 
648
        if (int err= resource_context->getTransactionalStorageEngine()->rollback(&session, all))
669
649
        {
670
650
          my_error(ER_ERROR_DURING_ROLLBACK, MYF(0), err);
671
651
          error= 1;
691
671
      rollbackStatementMessage(session);
692
672
 
693
673
    if (is_real_trans)
694
 
      session.transaction.xid_state.xid.null();
 
674
      session.transaction.xid_state.xid.set_null();
695
675
    if (normal_transaction)
696
676
    {
697
677
      session.variables.tx_isolation=session.session_tx_isolation;
716
696
  return error;
717
697
}
718
698
 
719
 
int TransactionServices::autocommitOrRollback(Session::reference session,
 
699
int TransactionServices::autocommitOrRollback(Session& session,
720
700
                                              int error)
721
701
{
722
702
  /* One GPB Statement message per SQL statement */
728
708
  {
729
709
    TransactionContext *trans = &session.transaction.stmt;
730
710
    TransactionContext::ResourceContexts &resource_contexts= trans->getResourceContexts();
731
 
    for (TransactionContext::ResourceContexts::iterator it= resource_contexts.begin();
732
 
         it != resource_contexts.end();
733
 
         ++it)
 
711
    BOOST_FOREACH(TransactionContext::ResourceContexts::reference resource_context, resource_contexts)
734
712
    {
735
 
      ResourceContext *resource_context= *it;
736
 
 
737
713
      resource_context->getTransactionalStorageEngine()->endStatement(&session);
738
714
    }
739
715
 
768
744
  }
769
745
};
770
746
 
771
 
int TransactionServices::rollbackToSavepoint(Session::reference session,
 
747
int TransactionServices::rollbackToSavepoint(Session& session,
772
748
                                             NamedSavepoint &sv)
773
749
{
774
750
  int error= 0;
781
757
    rolling back to savepoint in all storage engines that were part of the
782
758
    transaction when the savepoint was set
783
759
  */
784
 
  for (TransactionContext::ResourceContexts::iterator it= sv_resource_contexts.begin();
785
 
       it != sv_resource_contexts.end();
786
 
       ++it)
 
760
  BOOST_FOREACH(TransactionContext::ResourceContexts::reference resource_context, sv_resource_contexts)
787
761
  {
788
 
    int err;
789
 
    ResourceContext *resource_context= *it;
790
 
 
791
762
    plugin::MonitoredInTransaction *resource= resource_context->getMonitored();
792
763
 
793
764
    if (resource->participatesInSqlTransaction())
794
765
    {
795
 
      if ((err= resource_context->getTransactionalStorageEngine()->rollbackToSavepoint(&session, sv)))
 
766
      if (int err= resource_context->getTransactionalStorageEngine()->rollbackToSavepoint(&session, sv))
796
767
      {
797
768
        my_error(ER_ERROR_DURING_ROLLBACK, MYF(0), err);
798
769
        error= 1;
838
809
     * savepoint's resource contexts.
839
810
     */
840
811
        
841
 
    for (TransactionContext::ResourceContexts::iterator it= set_difference_contexts.begin();
842
 
         it != set_difference_contexts.end();
843
 
         ++it)
 
812
    BOOST_FOREACH(TransactionContext::ResourceContexts::reference resource_context, set_difference_contexts)
844
813
    {
845
 
      ResourceContext *resource_context= *it;
846
 
      int err;
847
 
 
848
814
      plugin::MonitoredInTransaction *resource= resource_context->getMonitored();
849
815
 
850
816
      if (resource->participatesInSqlTransaction())
851
817
      {
852
 
        if ((err= resource_context->getTransactionalStorageEngine()->rollback(&session, !(0))))
 
818
        if (int err= resource_context->getTransactionalStorageEngine()->rollback(&session, true))
853
819
        {
854
820
          my_error(ER_ERROR_DURING_ROLLBACK, MYF(0), err);
855
821
          error= 1;
898
864
  section "4.33.4 SQL-statements and transaction states",
899
865
  NamedSavepoint is *not* transaction-initiating SQL-statement
900
866
*/
901
 
int TransactionServices::setSavepoint(Session::reference session,
 
867
int TransactionServices::setSavepoint(Session& session,
902
868
                                      NamedSavepoint &sv)
903
869
{
904
870
  int error= 0;
907
873
 
908
874
  if (resource_contexts.empty() == false)
909
875
  {
910
 
    for (TransactionContext::ResourceContexts::iterator it= resource_contexts.begin();
911
 
         it != resource_contexts.end();
912
 
         ++it)
 
876
    BOOST_FOREACH(TransactionContext::ResourceContexts::reference resource_context, resource_contexts)
913
877
    {
914
 
      ResourceContext *resource_context= *it;
915
 
      int err;
916
 
 
917
878
      plugin::MonitoredInTransaction *resource= resource_context->getMonitored();
918
879
 
919
880
      if (resource->participatesInSqlTransaction())
920
881
      {
921
 
        if ((err= resource_context->getTransactionalStorageEngine()->setSavepoint(&session, sv)))
 
882
        if (int err= resource_context->getTransactionalStorageEngine()->setSavepoint(&session, sv))
922
883
        {
923
884
          my_error(ER_GET_ERRNO, MYF(0), err);
924
885
          error= 1;
950
911
  return error;
951
912
}
952
913
 
953
 
int TransactionServices::releaseSavepoint(Session::reference session,
 
914
int TransactionServices::releaseSavepoint(Session& session,
954
915
                                          NamedSavepoint &sv)
955
916
{
956
917
  int error= 0;
957
918
 
958
919
  TransactionContext::ResourceContexts &resource_contexts= sv.getResourceContexts();
959
920
 
960
 
  for (TransactionContext::ResourceContexts::iterator it= resource_contexts.begin();
961
 
       it != resource_contexts.end();
962
 
       ++it)
 
921
  BOOST_FOREACH(TransactionContext::ResourceContexts::reference resource_context, resource_contexts)
963
922
  {
964
 
    int err;
965
 
    ResourceContext *resource_context= *it;
966
 
 
967
923
    plugin::MonitoredInTransaction *resource= resource_context->getMonitored();
968
924
 
969
925
    if (resource->participatesInSqlTransaction())
970
926
    {
971
 
      if ((err= resource_context->getTransactionalStorageEngine()->releaseSavepoint(&session, sv)))
 
927
      if (int err= resource_context->getTransactionalStorageEngine()->releaseSavepoint(&session, sv))
972
928
      {
973
929
        my_error(ER_GET_ERRNO, MYF(0), err);
974
930
        error= 1;
981
937
 
982
938
bool TransactionServices::shouldConstructMessages()
983
939
{
984
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
985
 
  return replication_services.isActive();
 
940
  return ReplicationServices::isActive();
986
941
}
987
942
 
988
 
message::Transaction *TransactionServices::getActiveTransactionMessage(Session::reference session,
 
943
message::Transaction *TransactionServices::getActiveTransactionMessage(Session& session,
989
944
                                                                       bool should_inc_trx_id)
990
945
{
991
946
  message::Transaction *transaction= session.getTransactionMessage();
997
952
     * for this Session object.  Session is responsible for
998
953
     * deleting transaction message when done with it.
999
954
     */
1000
 
    transaction= new (nothrow) message::Transaction();
 
955
    transaction= new message::Transaction();
1001
956
    initTransactionMessage(*transaction, session, should_inc_trx_id);
1002
957
    session.setTransactionMessage(transaction);
1003
 
    return transaction;
1004
958
  }
1005
 
  else
1006
 
    return transaction;
 
959
  return transaction;
1007
960
}
1008
961
 
1009
962
void TransactionServices::initTransactionMessage(message::Transaction &transaction,
1010
 
                                                 Session::reference session,
 
963
                                                 Session& session,
1011
964
                                                 bool should_inc_trx_id)
1012
965
{
1013
966
  message::TransactionContext *trx= transaction.mutable_transaction_context();
1024
977
    trx->set_transaction_id(0);
1025
978
  }
1026
979
 
1027
 
  trx->set_start_timestamp(session.getCurrentTimestamp());
 
980
  trx->set_start_timestamp(session.times.getCurrentTimestamp());
1028
981
  
1029
982
  /* segment info may get set elsewhere as needed */
1030
983
  transaction.set_segment_id(1);
1032
985
}
1033
986
 
1034
987
void TransactionServices::finalizeTransactionMessage(message::Transaction &transaction,
1035
 
                                                     Session::const_reference session)
 
988
                                                     const Session& session)
1036
989
{
1037
990
  message::TransactionContext *trx= transaction.mutable_transaction_context();
1038
 
  trx->set_end_timestamp(session.getCurrentTimestamp());
 
991
  trx->set_end_timestamp(session.times.getCurrentTimestamp());
1039
992
}
1040
993
 
1041
994
void TransactionServices::cleanupTransactionMessage(message::Transaction *transaction,
1042
 
                                                    Session::reference session)
 
995
                                                    Session& session)
1043
996
{
1044
997
  delete transaction;
1045
998
  session.setStatementMessage(NULL);
1047
1000
  session.setXaId(0);
1048
1001
}
1049
1002
 
1050
 
int TransactionServices::commitTransactionMessage(Session::reference session)
 
1003
int TransactionServices::commitTransactionMessage(Session& session)
1051
1004
{
1052
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
1053
 
  if (! replication_services.isActive())
 
1005
  if (! ReplicationServices::isActive())
1054
1006
    return 0;
1055
1007
 
1056
1008
  /*
1084
1036
  
1085
1037
  finalizeTransactionMessage(*transaction, session);
1086
1038
  
1087
 
  plugin::ReplicationReturnCode result= replication_services.pushTransactionMessage(session, *transaction);
 
1039
  plugin::ReplicationReturnCode result= ReplicationServices::pushTransactionMessage(session, *transaction);
1088
1040
 
1089
1041
  cleanupTransactionMessage(transaction, session);
1090
1042
 
1093
1045
 
1094
1046
void TransactionServices::initStatementMessage(message::Statement &statement,
1095
1047
                                               message::Statement::Type type,
1096
 
                                               Session::const_reference session)
 
1048
                                               const Session& session)
1097
1049
{
1098
1050
  statement.set_type(type);
1099
 
  statement.set_start_timestamp(session.getCurrentTimestamp());
 
1051
  statement.set_start_timestamp(session.times.getCurrentTimestamp());
1100
1052
 
1101
1053
  if (session.variables.replicate_query)
1102
1054
    statement.set_sql(session.getQueryString()->c_str());
1103
1055
}
1104
1056
 
1105
1057
void TransactionServices::finalizeStatementMessage(message::Statement &statement,
1106
 
                                                   Session::reference session)
 
1058
                                                   Session& session)
1107
1059
{
1108
 
  statement.set_end_timestamp(session.getCurrentTimestamp());
 
1060
  statement.set_end_timestamp(session.times.getCurrentTimestamp());
1109
1061
  session.setStatementMessage(NULL);
1110
1062
}
1111
1063
 
1112
 
void TransactionServices::rollbackTransactionMessage(Session::reference session)
 
1064
void TransactionServices::rollbackTransactionMessage(Session& session)
1113
1065
{
1114
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
1115
 
  if (! replication_services.isActive())
 
1066
  if (! ReplicationServices::isActive())
1116
1067
    return;
1117
1068
  
1118
1069
  message::Transaction *transaction= getActiveTransactionMessage(session);
1159
1110
 
1160
1111
    finalizeTransactionMessage(*transaction, session);
1161
1112
    
1162
 
    (void) replication_services.pushTransactionMessage(session, *transaction);
 
1113
    (void) ReplicationServices::pushTransactionMessage(session, *transaction);
1163
1114
  }
1164
1115
 
1165
1116
  cleanupTransactionMessage(transaction, session);
1166
1117
}
1167
1118
 
1168
 
void TransactionServices::rollbackStatementMessage(Session::reference session)
 
1119
void TransactionServices::rollbackStatementMessage(Session& session)
1169
1120
{
1170
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
1171
 
  if (! replication_services.isActive())
 
1121
  if (! ReplicationServices::isActive())
1172
1122
    return;
1173
1123
 
1174
1124
  message::Statement *current_statement= session.getStatementMessage();
1231
1181
  }
1232
1182
}
1233
1183
 
1234
 
message::Transaction *TransactionServices::segmentTransactionMessage(Session::reference session,
 
1184
message::Transaction *TransactionServices::segmentTransactionMessage(Session& session,
1235
1185
                                                                     message::Transaction *transaction)
1236
1186
{
1237
1187
  uint64_t trx_id= transaction->transaction_context().transaction_id();
1249
1199
  return transaction;
1250
1200
}
1251
1201
 
1252
 
message::Statement &TransactionServices::getInsertStatement(Session::reference session,
 
1202
message::Statement &TransactionServices::getInsertStatement(Session& session,
1253
1203
                                                            Table &table,
1254
1204
                                                            uint32_t *next_segment_id)
1255
1205
{
1340
1290
}
1341
1291
 
1342
1292
void TransactionServices::setInsertHeader(message::Statement &statement,
1343
 
                                          Session::const_reference session,
 
1293
                                          const Session& session,
1344
1294
                                          Table &table)
1345
1295
{
1346
1296
  initStatementMessage(statement, message::Statement::INSERT, session);
1353
1303
  message::InsertHeader *header= statement.mutable_insert_header();
1354
1304
  message::TableMetadata *table_metadata= header->mutable_table_metadata();
1355
1305
 
1356
 
  string schema_name;
1357
 
  (void) table.getShare()->getSchemaName(schema_name);
1358
 
  string table_name;
1359
 
  (void) table.getShare()->getTableName(table_name);
1360
 
 
1361
 
  table_metadata->set_schema_name(schema_name.c_str(), schema_name.length());
1362
 
  table_metadata->set_table_name(table_name.c_str(), table_name.length());
1363
 
 
1364
 
  Field *current_field;
 
1306
  table_metadata->set_schema_name(table.getShare()->getSchemaName());
 
1307
  table_metadata->set_table_name(table.getShare()->getTableName());
 
1308
 
1365
1309
  Field **table_fields= table.getFields();
1366
1310
 
1367
1311
  message::FieldMetadata *field_metadata;
1369
1313
  /* We will read all the table's fields... */
1370
1314
  table.setReadSet();
1371
1315
 
1372
 
  while ((current_field= *table_fields++) != NULL) 
 
1316
  while (Field* current_field= *table_fields++) 
1373
1317
  {
1374
1318
    field_metadata= header->add_field_metadata();
1375
1319
    field_metadata->set_name(current_field->field_name);
1377
1321
  }
1378
1322
}
1379
1323
 
1380
 
bool TransactionServices::insertRecord(Session::reference session,
 
1324
bool TransactionServices::insertRecord(Session& session,
1381
1325
                                       Table &table)
1382
1326
{
1383
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
1384
 
  if (! replication_services.isActive())
 
1327
  if (! ReplicationServices::isActive())
 
1328
    return false;
 
1329
 
 
1330
  if (not table.getShare()->is_replicated())
1385
1331
    return false;
1386
1332
 
1387
1333
  /**
1433
1379
  return false;
1434
1380
}
1435
1381
 
1436
 
message::Statement &TransactionServices::getUpdateStatement(Session::reference session,
 
1382
message::Statement &TransactionServices::getUpdateStatement(Session& session,
1437
1383
                                                            Table &table,
1438
1384
                                                            const unsigned char *old_record, 
1439
1385
                                                            const unsigned char *new_record,
1526
1472
}
1527
1473
 
1528
1474
void TransactionServices::setUpdateHeader(message::Statement &statement,
1529
 
                                          Session::const_reference session,
 
1475
                                          const Session& session,
1530
1476
                                          Table &table,
1531
1477
                                          const unsigned char *old_record, 
1532
1478
                                          const unsigned char *new_record)
1541
1487
  message::UpdateHeader *header= statement.mutable_update_header();
1542
1488
  message::TableMetadata *table_metadata= header->mutable_table_metadata();
1543
1489
 
1544
 
  string schema_name;
1545
 
  (void) table.getShare()->getSchemaName(schema_name);
1546
 
  string table_name;
1547
 
  (void) table.getShare()->getTableName(table_name);
1548
 
 
1549
 
  table_metadata->set_schema_name(schema_name.c_str(), schema_name.length());
1550
 
  table_metadata->set_table_name(table_name.c_str(), table_name.length());
 
1490
  table_metadata->set_schema_name(table.getShare()->getSchemaName());
 
1491
  table_metadata->set_table_name(table.getShare()->getTableName());
1551
1492
 
1552
1493
  Field *current_field;
1553
1494
  Field **table_fields= table.getFields();
1579
1520
    }
1580
1521
  }
1581
1522
}
1582
 
void TransactionServices::updateRecord(Session::reference session,
 
1523
 
 
1524
void TransactionServices::updateRecord(Session& session,
1583
1525
                                       Table &table, 
1584
1526
                                       const unsigned char *old_record, 
1585
1527
                                       const unsigned char *new_record)
1586
1528
{
1587
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
1588
 
  if (! replication_services.isActive())
 
1529
  if (! ReplicationServices::isActive())
 
1530
    return;
 
1531
 
 
1532
  if (not table.getShare()->is_replicated())
1589
1533
    return;
1590
1534
 
1591
1535
  uint32_t next_segment_id= 1;
1707
1651
  return isUpdated;
1708
1652
}  
1709
1653
 
1710
 
message::Statement &TransactionServices::getDeleteStatement(Session::reference session,
 
1654
message::Statement &TransactionServices::getDeleteStatement(Session& session,
1711
1655
                                                            Table &table,
1712
1656
                                                            uint32_t *next_segment_id)
1713
1657
{
1798
1742
}
1799
1743
 
1800
1744
void TransactionServices::setDeleteHeader(message::Statement &statement,
1801
 
                                          Session::const_reference session,
 
1745
                                          const Session& session,
1802
1746
                                          Table &table)
1803
1747
{
1804
1748
  initStatementMessage(statement, message::Statement::DELETE, session);
1810
1754
  message::DeleteHeader *header= statement.mutable_delete_header();
1811
1755
  message::TableMetadata *table_metadata= header->mutable_table_metadata();
1812
1756
 
1813
 
  string schema_name;
1814
 
  (void) table.getShare()->getSchemaName(schema_name);
1815
 
  string table_name;
1816
 
  (void) table.getShare()->getTableName(table_name);
1817
 
 
1818
 
  table_metadata->set_schema_name(schema_name.c_str(), schema_name.length());
1819
 
  table_metadata->set_table_name(table_name.c_str(), table_name.length());
 
1757
  table_metadata->set_schema_name(table.getShare()->getSchemaName());
 
1758
  table_metadata->set_table_name(table.getShare()->getTableName());
1820
1759
 
1821
1760
  Field *current_field;
1822
1761
  Field **table_fields= table.getFields();
1839
1778
  }
1840
1779
}
1841
1780
 
1842
 
void TransactionServices::deleteRecord(Session::reference session,
 
1781
void TransactionServices::deleteRecord(Session& session,
1843
1782
                                       Table &table,
1844
1783
                                       bool use_update_record)
1845
1784
{
1846
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
1847
 
  if (! replication_services.isActive())
 
1785
  if (! ReplicationServices::isActive())
 
1786
    return;
 
1787
 
 
1788
  if (not table.getShare()->is_replicated())
1848
1789
    return;
1849
1790
 
1850
1791
  uint32_t next_segment_id= 1;
1896
1837
  }
1897
1838
}
1898
1839
 
1899
 
void TransactionServices::createTable(Session::reference session,
 
1840
void TransactionServices::createTable(Session& session,
1900
1841
                                      const message::Table &table)
1901
1842
{
1902
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
1903
 
  if (! replication_services.isActive())
1904
 
    return;
1905
 
  
 
1843
  if (not ReplicationServices::isActive())
 
1844
    return;
 
1845
 
 
1846
  if (not message::is_replicated(table))
 
1847
    return;
 
1848
 
1906
1849
  message::Transaction *transaction= getActiveTransactionMessage(session);
1907
1850
  message::Statement *statement= transaction->add_statement();
1908
1851
 
1920
1863
 
1921
1864
  finalizeTransactionMessage(*transaction, session);
1922
1865
  
1923
 
  (void) replication_services.pushTransactionMessage(session, *transaction);
 
1866
  (void) ReplicationServices::pushTransactionMessage(session, *transaction);
1924
1867
 
1925
1868
  cleanupTransactionMessage(transaction, session);
1926
1869
 
1927
1870
}
1928
1871
 
1929
 
void TransactionServices::createSchema(Session::reference session,
 
1872
void TransactionServices::createSchema(Session& session,
1930
1873
                                       const message::Schema &schema)
1931
1874
{
1932
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
1933
 
  if (! replication_services.isActive())
1934
 
    return;
1935
 
  
 
1875
  if (! ReplicationServices::isActive())
 
1876
    return;
 
1877
 
 
1878
  if (not message::is_replicated(schema))
 
1879
    return;
 
1880
 
1936
1881
  message::Transaction *transaction= getActiveTransactionMessage(session);
1937
1882
  message::Statement *statement= transaction->add_statement();
1938
1883
 
1950
1895
 
1951
1896
  finalizeTransactionMessage(*transaction, session);
1952
1897
  
1953
 
  (void) replication_services.pushTransactionMessage(session, *transaction);
 
1898
  (void) ReplicationServices::pushTransactionMessage(session, *transaction);
1954
1899
 
1955
1900
  cleanupTransactionMessage(transaction, session);
1956
1901
 
1957
1902
}
1958
1903
 
1959
 
void TransactionServices::dropSchema(Session::reference session,
1960
 
                                     identifier::Schema::const_reference identifier)
 
1904
void TransactionServices::dropSchema(Session& session,
 
1905
                                     const identifier::Schema& identifier,
 
1906
                                     message::schema::const_reference schema)
1961
1907
{
1962
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
1963
 
  if (! replication_services.isActive())
1964
 
    return;
1965
 
  
 
1908
  if (not ReplicationServices::isActive())
 
1909
    return;
 
1910
 
 
1911
  if (not message::is_replicated(schema))
 
1912
    return;
 
1913
 
1966
1914
  message::Transaction *transaction= getActiveTransactionMessage(session);
1967
1915
  message::Statement *statement= transaction->add_statement();
1968
1916
 
1980
1928
 
1981
1929
  finalizeTransactionMessage(*transaction, session);
1982
1930
  
1983
 
  (void) replication_services.pushTransactionMessage(session, *transaction);
 
1931
  (void) ReplicationServices::pushTransactionMessage(session, *transaction);
1984
1932
 
1985
1933
  cleanupTransactionMessage(transaction, session);
1986
1934
}
1987
1935
 
1988
 
void TransactionServices::alterSchema(Session::reference session,
1989
 
                                      const message::schema::shared_ptr &old_schema,
 
1936
void TransactionServices::alterSchema(Session& session,
 
1937
                                      const message::Schema &old_schema,
1990
1938
                                      const message::Schema &new_schema)
1991
1939
{
1992
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
1993
 
  if (! replication_services.isActive())
1994
 
    return;
1995
 
  
 
1940
  if (! ReplicationServices::isActive())
 
1941
    return;
 
1942
 
 
1943
  if (not message::is_replicated(old_schema))
 
1944
    return;
 
1945
 
1996
1946
  message::Transaction *transaction= getActiveTransactionMessage(session);
1997
1947
  message::Statement *statement= transaction->add_statement();
1998
1948
 
2007
1957
  message::Schema *before= alter_schema_statement->mutable_before();
2008
1958
  message::Schema *after= alter_schema_statement->mutable_after();
2009
1959
 
2010
 
  *before= *old_schema;
 
1960
  *before= old_schema;
2011
1961
  *after= new_schema;
2012
1962
 
2013
1963
  finalizeStatementMessage(*statement, session);
2014
1964
 
2015
1965
  finalizeTransactionMessage(*transaction, session);
2016
1966
  
2017
 
  (void) replication_services.pushTransactionMessage(session, *transaction);
 
1967
  (void) ReplicationServices::pushTransactionMessage(session, *transaction);
2018
1968
 
2019
1969
  cleanupTransactionMessage(transaction, session);
2020
1970
}
2021
1971
 
2022
 
void TransactionServices::dropTable(Session::reference session,
2023
 
                                    const identifier::Table &table,
 
1972
void TransactionServices::dropTable(Session& session,
 
1973
                                    const identifier::Table& identifier,
 
1974
                                    message::table::const_reference table,
2024
1975
                                    bool if_exists)
2025
1976
{
2026
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
2027
 
  if (! replication_services.isActive())
2028
 
    return;
2029
 
  
 
1977
  if (! ReplicationServices::isActive())
 
1978
    return;
 
1979
 
 
1980
  if (not message::is_replicated(table))
 
1981
    return;
 
1982
 
2030
1983
  message::Transaction *transaction= getActiveTransactionMessage(session);
2031
1984
  message::Statement *statement= transaction->add_statement();
2032
1985
 
2042
1995
 
2043
1996
  message::TableMetadata *table_metadata= drop_table_statement->mutable_table_metadata();
2044
1997
 
2045
 
  table_metadata->set_schema_name(table.getSchemaName());
2046
 
  table_metadata->set_table_name(table.getTableName());
 
1998
  table_metadata->set_schema_name(identifier.getSchemaName());
 
1999
  table_metadata->set_table_name(identifier.getTableName());
2047
2000
 
2048
2001
  finalizeStatementMessage(*statement, session);
2049
2002
 
2050
2003
  finalizeTransactionMessage(*transaction, session);
2051
2004
  
2052
 
  (void) replication_services.pushTransactionMessage(session, *transaction);
 
2005
  (void) ReplicationServices::pushTransactionMessage(session, *transaction);
2053
2006
 
2054
2007
  cleanupTransactionMessage(transaction, session);
2055
2008
}
2056
2009
 
2057
 
void TransactionServices::truncateTable(Session::reference session,
2058
 
                                        Table &table)
 
2010
void TransactionServices::truncateTable(Session& session, Table &table)
2059
2011
{
2060
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
2061
 
  if (! replication_services.isActive())
2062
 
    return;
2063
 
  
 
2012
  if (! ReplicationServices::isActive())
 
2013
    return;
 
2014
 
 
2015
  if (not table.getShare()->is_replicated())
 
2016
    return;
 
2017
 
2064
2018
  message::Transaction *transaction= getActiveTransactionMessage(session);
2065
2019
  message::Statement *statement= transaction->add_statement();
2066
2020
 
2073
2027
  message::TruncateTableStatement *truncate_statement= statement->mutable_truncate_table_statement();
2074
2028
  message::TableMetadata *table_metadata= truncate_statement->mutable_table_metadata();
2075
2029
 
2076
 
  string schema_name;
2077
 
  (void) table.getShare()->getSchemaName(schema_name);
2078
 
  string table_name;
2079
 
  (void) table.getShare()->getTableName(table_name);
2080
 
 
2081
 
  table_metadata->set_schema_name(schema_name.c_str(), schema_name.length());
2082
 
  table_metadata->set_table_name(table_name.c_str(), table_name.length());
 
2030
  table_metadata->set_schema_name(table.getShare()->getSchemaName());
 
2031
  table_metadata->set_table_name(table.getShare()->getTableName());
2083
2032
 
2084
2033
  finalizeStatementMessage(*statement, session);
2085
2034
 
2086
2035
  finalizeTransactionMessage(*transaction, session);
2087
2036
  
2088
 
  (void) replication_services.pushTransactionMessage(session, *transaction);
 
2037
  (void) ReplicationServices::pushTransactionMessage(session, *transaction);
2089
2038
 
2090
2039
  cleanupTransactionMessage(transaction, session);
2091
2040
}
2092
2041
 
2093
 
void TransactionServices::rawStatement(Session::reference session,
2094
 
                                       const string &query)
 
2042
void TransactionServices::rawStatement(Session& session,
 
2043
                                       const string &query,
 
2044
                                       const string &schema)
2095
2045
{
2096
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
2097
 
  if (! replication_services.isActive())
 
2046
  if (! ReplicationServices::isActive())
2098
2047
    return;
2099
2048
 
2100
2049
  message::Transaction *transaction= getActiveTransactionMessage(session);
2102
2051
 
2103
2052
  initStatementMessage(*statement, message::Statement::RAW_SQL, session);
2104
2053
  statement->set_sql(query);
 
2054
  if (not schema.empty())
 
2055
    statement->set_raw_sql_schema(schema);
2105
2056
  finalizeStatementMessage(*statement, session);
2106
2057
 
2107
2058
  finalizeTransactionMessage(*transaction, session);
2108
2059
  
2109
 
  (void) replication_services.pushTransactionMessage(session, *transaction);
 
2060
  (void) ReplicationServices::pushTransactionMessage(session, *transaction);
2110
2061
 
2111
2062
  cleanupTransactionMessage(transaction, session);
2112
2063
}
2113
2064
 
2114
 
int TransactionServices::sendEvent(Session::reference session,
2115
 
                                   const message::Event &event)
 
2065
int TransactionServices::sendEvent(Session& session, const message::Event &event)
2116
2066
{
2117
 
  ReplicationServices &replication_services= ReplicationServices::singleton();
2118
 
  if (! replication_services.isActive())
 
2067
  if (not ReplicationServices::isActive())
2119
2068
    return 0;
2120
 
 
2121
 
  message::Transaction *transaction= new (nothrow) message::Transaction();
 
2069
  message::Transaction transaction;
2122
2070
 
2123
2071
  // set server id, start timestamp
2124
 
  initTransactionMessage(*transaction, session, true);
 
2072
  initTransactionMessage(transaction, session, true);
2125
2073
 
2126
2074
  // set end timestamp
2127
 
  finalizeTransactionMessage(*transaction, session);
2128
 
 
2129
 
  message::Event *trx_event= transaction->mutable_event();
2130
 
 
 
2075
  finalizeTransactionMessage(transaction, session);
 
2076
 
 
2077
  message::Event *trx_event= transaction.mutable_event();
2131
2078
  trx_event->CopyFrom(event);
2132
 
 
2133
 
  plugin::ReplicationReturnCode result= replication_services.pushTransactionMessage(session, *transaction);
2134
 
 
2135
 
  delete transaction;
2136
 
 
2137
 
  return static_cast<int>(result);
 
2079
  plugin::ReplicationReturnCode result= ReplicationServices::pushTransactionMessage(session, transaction);
 
2080
  return result;
2138
2081
}
2139
2082
 
2140
 
bool TransactionServices::sendStartupEvent(Session::reference session)
 
2083
bool TransactionServices::sendStartupEvent(Session& session)
2141
2084
{
2142
2085
  message::Event event;
2143
2086
  event.set_type(message::Event::STARTUP);
2144
 
  if (sendEvent(session, event) != 0)
2145
 
    return false;
2146
 
  return true;
 
2087
  return not sendEvent(session, event);
2147
2088
}
2148
2089
 
2149
 
bool TransactionServices::sendShutdownEvent(Session::reference session)
 
2090
bool TransactionServices::sendShutdownEvent(Session& session)
2150
2091
{
2151
2092
  message::Event event;
2152
2093
  event.set_type(message::Event::SHUTDOWN);
2153
 
  if (sendEvent(session, event) != 0)
2154
 
    return false;
2155
 
  return true;
 
2094
  return not sendEvent(session, event);
2156
2095
}
2157
2096
 
2158
2097
} /* namespace drizzled */