lib/DBIx/Class/Storage/DBI.pm

   1 package DBIx::Class::Storage::DBI;
   2 # -*- mode: cperl; cperl-indent-level: 2 -*-
   3
   4 use strict;
   5 use warnings;
   6
   7 use base qw/DBIx::Class::Storage::DBIHacks DBIx::Class::Storage/;
   8 use mro 'c3';
   9
  10 use Carp::Clan qw/^DBIx::Class|^Try::Tiny/;
  11 use DBI;
  12 use DBIx::Class::Storage::DBI::Cursor;
  13 use Scalar::Util qw/refaddr weaken reftype blessed/;
  14 use List::Util qw/first/;
  15 use Data::Dumper::Concise 'Dumper';
  16 use Sub::Name 'subname';
  17 use Try::Tiny;
  18 use File::Path 'make_path';
  19 use overload ();
  20 use namespace::clean;
  21
  22
  23 # default cursor class, overridable in connect_info attributes
  24 __PACKAGE__->cursor_class('DBIx::Class::Storage::DBI::Cursor');
  25
  26 __PACKAGE__->mk_group_accessors('inherited' => qw/
  27   sql_maker_class sql_limit_dialect sql_quote_char sql_name_sep
  28 /);
  29
  30 __PACKAGE__->sql_name_sep('.');
  31
  32 __PACKAGE__->sql_maker_class('DBIx::Class::SQLMaker');
  33
  34 __PACKAGE__->mk_group_accessors('simple' => qw/
  35   _connect_info _dbi_connect_info _dbic_connect_attributes _driver_determined
  36   _dbh _dbh_details _conn_pid _sql_maker _sql_maker_opts
  37   transaction_depth _dbh_autocommit  savepoints
  38 /);
  39
  40 # the values for these accessors are picked out (and deleted) from
  41 # the attribute hashref passed to connect_info
  42 my @storage_options = qw/
  43   on_connect_call on_disconnect_call on_connect_do on_disconnect_do
  44   disable_sth_caching unsafe auto_savepoint
  45 /;
  46 __PACKAGE__->mk_group_accessors('simple' => @storage_options);
  47
  48
  49 # capability definitions, using a 2-tiered accessor system
  50 # The rationale is:
  51 #
  52 # A driver/user may define _use_X, which blindly without any checks says:
  53 # "(do not) use this capability", (use_dbms_capability is an "inherited"
  54 # type accessor)
  55 #
  56 # If _use_X is undef, _supports_X is then queried. This is a "simple" style
  57 # accessor, which in turn calls _determine_supports_X, and stores the return
  58 # in a special slot on the storage object, which is wiped every time a $dbh
  59 # reconnection takes place (it is not guaranteed that upon reconnection we
  60 # will get the same rdbms version). _determine_supports_X does not need to
  61 # exist on a driver, as we ->can for it before calling.
  62
  63 my @capabilities = (qw/
  64   insert_returning
  65   insert_returning_bound
  66   placeholders
  67   typeless_placeholders
  68   join_optimizer
  69 /);
  70 __PACKAGE__->mk_group_accessors( dbms_capability => map { "_supports_$_" } @capabilities );
  71 __PACKAGE__->mk_group_accessors( use_dbms_capability => map { "_use_$_" } (@capabilities ) );
  72
  73 # on by default, not strictly a capability (pending rewrite)
  74 __PACKAGE__->_use_join_optimizer (1);
  75 sub _determine_supports_join_optimizer { 1 };
  76
  77 # Each of these methods need _determine_driver called before itself
  78 # in order to function reliably. This is a purely DRY optimization
  79 #
  80 # get_(use)_dbms_capability need to be called on the correct Storage
  81 # class, as _use_X may be hardcoded class-wide, and _supports_X calls
  82 # _determine_supports_X which obv. needs a correct driver as well
  83 my @rdbms_specific_methods = qw/
  84   deployment_statements
  85   sqlt_type
  86   sql_maker
  87   build_datetime_parser
  88   datetime_parser_type
  89
  90   insert
  91   insert_bulk
  92   update
  93   delete
  94   select
  95   select_single
  96
  97   get_use_dbms_capability
  98   get_dbms_capability
  99
 100   _server_info
 101   _get_server_version
 102 /;
 103
 104 for my $meth (@rdbms_specific_methods) {
 105
 106   my $orig = __PACKAGE__->can ($meth)
 107     or die "$meth is not a ::Storage::DBI method!";
 108
 109   no strict qw/refs/;
 110   no warnings qw/redefine/;
 111   *{__PACKAGE__ ."::$meth"} = subname $meth => sub {
 112     if (not $_[0]->_driver_determined and not $_[0]->{_in_determine_driver}) {
 113       $_[0]->_determine_driver;
 114
 115       # This for some reason crashes and burns on perl 5.8.1
 116       # IFF the method ends up throwing an exception
 117       #goto $_[0]->can ($meth);
 118
 119       my $cref = $_[0]->can ($meth);
 120       goto $cref;
 121     }
 122     goto $orig;
 123   };
 124 }
 125
 126
 127 =head1 NAME
 128
 129 DBIx::Class::Storage::DBI - DBI storage handler
 130
 131 =head1 SYNOPSIS
 132
 133   my $schema = MySchema->connect('dbi:SQLite:my.db');
 134
 135   $schema->storage->debug(1);
 136
 137   my @stuff = $schema->storage->dbh_do(
 138     sub {
 139       my ($storage, $dbh, @args) = @_;
 140       $dbh->do("DROP TABLE authors");
 141     },
 142     @column_list
 143   );
 144
 145   $schema->resultset('Book')->search({
 146      written_on => $schema->storage->datetime_parser->format_datetime(DateTime->now)
 147   });
 148
 149 =head1 DESCRIPTION
 150
 151 This class represents the connection to an RDBMS via L<DBI>.  See
 152 L<DBIx::Class::Storage> for general information.  This pod only
 153 documents DBI-specific methods and behaviors.
 154
 155 =head1 METHODS
 156
 157 =cut
 158
 159 sub new {
 160   my $new = shift->next::method(@_);
 161
 162   $new->transaction_depth(0);
 163   $new->_sql_maker_opts({});
 164   $new->_dbh_details({});
 165   $new->{savepoints} = [];
 166   $new->{_in_dbh_do} = 0;
 167   $new->{_dbh_gen} = 0;
 168
 169   # read below to see what this does
 170   $new->_arm_global_destructor;
 171
 172   $new;
 173 }
 174
 175 # This is hack to work around perl shooting stuff in random
 176 # order on exit(). If we do not walk the remaining storage
 177 # objects in an END block, there is a *small but real* chance
 178 # of a fork()ed child to kill the parent's shared DBI handle,
 179 # *before perl reaches the DESTROY in this package*
 180 # Yes, it is ugly and effective.
 181 # Additionally this registry is used by the CLONE method to
 182 # make sure no handles are shared between threads
 183 {
 184   my %seek_and_destroy;
 185
 186   sub _arm_global_destructor {
 187     my $self = shift;
 188     my $key = refaddr ($self);
 189     $seek_and_destroy{$key} = $self;
 190     weaken ($seek_and_destroy{$key});
 191   }
 192
 193   END {
 194     local $?; # just in case the DBI destructor changes it somehow
 195
 196     # destroy just the object if not native to this process/thread
 197     $_->_verify_pid for (grep
 198       { defined $_ }
 199       values %seek_and_destroy
 200     );
 201   }
 202
 203   sub CLONE {
 204     # As per DBI's recommendation, DBIC disconnects all handles as
 205     # soon as possible (DBIC will reconnect only on demand from within
 206     # the thread)
 207     for (values %seek_and_destroy) {
 208       next unless $_;
 209       $_->{_dbh_gen}++;  # so that existing cursors will drop as well
 210       $_->_dbh(undef);
 211     }
 212   }
 213 }
 214
 215 sub DESTROY {
 216   my $self = shift;
 217
 218   # some databases spew warnings on implicit disconnect
 219   local $SIG{__WARN__} = sub {};
 220   $self->_dbh(undef);
 221
 222   # this op is necessary, since the very last perl runtime statement
 223   # triggers a global destruction shootout, and the $SIG localization
 224   # may very well be destroyed before perl actually gets to do the
 225   # $dbh undef
 226   1;
 227 }
 228
 229 # handle pid changes correctly - do not destroy parent's connection
 230 sub _verify_pid {
 231   my $self = shift;
 232
 233   my $pid = $self->_conn_pid;
 234   if( defined $pid and $pid != $$ and my $dbh = $self->_dbh ) {
 235     $dbh->{InactiveDestroy} = 1;
 236     $self->{_dbh_gen}++;
 237     $self->_dbh(undef);
 238   }
 239
 240   return;
 241 }
 242
 243 =head2 connect_info
 244
 245 This method is normally called by L<DBIx::Class::Schema/connection>, which
 246 encapsulates its argument list in an arrayref before passing them here.
 247
 248 The argument list may contain:
 249
 250 =over
 251
 252 =item *
 253
 254 The same 4-element argument set one would normally pass to
 255 L<DBI/connect>, optionally followed by
 256 L<extra attributes|/DBIx::Class specific connection attributes>
 257 recognized by DBIx::Class:
 258
 259   $connect_info_args = [ $dsn, $user, $password, \%dbi_attributes?, \%extra_attributes? ];
 260
 261 =item *
 262
 263 A single code reference which returns a connected
 264 L<DBI database handle|DBI/connect> optionally followed by
 265 L<extra attributes|/DBIx::Class specific connection attributes> recognized
 266 by DBIx::Class:
 267
 268   $connect_info_args = [ sub { DBI->connect (...) }, \%extra_attributes? ];
 269
 270 =item *
 271
 272 A single hashref with all the attributes and the dsn/user/password
 273 mixed together:
 274
 275   $connect_info_args = [{
 276     dsn => $dsn,
 277     user => $user,
 278     password => $pass,
 279     %dbi_attributes,
 280     %extra_attributes,
 281   }];
 282
 283   $connect_info_args = [{
 284     dbh_maker => sub { DBI->connect (...) },
 285     %dbi_attributes,
 286     %extra_attributes,
 287   }];
 288
 289 This is particularly useful for L<Catalyst> based applications, allowing the
 290 following config (L<Config::General> style):
 291
 292   <Model::DB>
 293     schema_class   App::DB
 294     <connect_info>
 295       dsn          dbi:mysql:database=test
 296       user         testuser
 297       password     TestPass
 298       AutoCommit   1
 299     </connect_info>
 300   </Model::DB>
 301
 302 The C<dsn>/C<user>/C<password> combination can be substituted by the
 303 C<dbh_maker> key whose value is a coderef that returns a connected
 304 L<DBI database handle|DBI/connect>
 305
 306 =back
 307
 308 Please note that the L<DBI> docs recommend that you always explicitly
 309 set C<AutoCommit> to either I<0> or I<1>.  L<DBIx::Class> further
 310 recommends that it be set to I<1>, and that you perform transactions
 311 via our L<DBIx::Class::Schema/txn_do> method.  L<DBIx::Class> will set it
 312 to I<1> if you do not do explicitly set it to zero.  This is the default
 313 for most DBDs. See L</DBIx::Class and AutoCommit> for details.
 314
 315 =head3 DBIx::Class specific connection attributes
 316
 317 In addition to the standard L<DBI|DBI/ATTRIBUTES_COMMON_TO_ALL_HANDLES>
 318 L<connection|DBI/Database_Handle_Attributes> attributes, DBIx::Class recognizes
 319 the following connection options. These options can be mixed in with your other
 320 L<DBI> connection attributes, or placed in a separate hashref
 321 (C<\%extra_attributes>) as shown above.
 322
 323 Every time C<connect_info> is invoked, any previous settings for
 324 these options will be cleared before setting the new ones, regardless of
 325 whether any options are specified in the new C<connect_info>.
 326
 327
 328 =over
 329
 330 =item on_connect_do
 331
 332 Specifies things to do immediately after connecting or re-connecting to
 333 the database.  Its value may contain:
 334
 335 =over
 336
 337 =item a scalar
 338
 339 This contains one SQL statement to execute.
 340
 341 =item an array reference
 342
 343 This contains SQL statements to execute in order.  Each element contains
 344 a string or a code reference that returns a string.
 345
 346 =item a code reference
 347
 348 This contains some code to execute.  Unlike code references within an
 349 array reference, its return value is ignored.
 350
 351 =back
 352
 353 =item on_disconnect_do
 354
 355 Takes arguments in the same form as L</on_connect_do> and executes them
 356 immediately before disconnecting from the database.
 357
 358 Note, this only runs if you explicitly call L</disconnect> on the
 359 storage object.
 360
 361 =item on_connect_call
 362
 363 A more generalized form of L</on_connect_do> that calls the specified
 364 C<connect_call_METHOD> methods in your storage driver.
 365
 366   on_connect_do => 'select 1'
 367
 368 is equivalent to:
 369
 370   on_connect_call => [ [ do_sql => 'select 1' ] ]
 371
 372 Its values may contain:
 373
 374 =over
 375
 376 =item a scalar
 377
 378 Will call the C<connect_call_METHOD> method.
 379
 380 =item a code reference
 381
 382 Will execute C<< $code->($storage) >>
 383
 384 =item an array reference
 385
 386 Each value can be a method name or code reference.
 387
 388 =item an array of arrays
 389
 390 For each array, the first item is taken to be the C<connect_call_> method name
 391 or code reference, and the rest are parameters to it.
 392
 393 =back
 394
 395 Some predefined storage methods you may use:
 396
 397 =over
 398
 399 =item do_sql
 400
 401 Executes a SQL string or a code reference that returns a SQL string. This is
 402 what L</on_connect_do> and L</on_disconnect_do> use.
 403
 404 It can take:
 405
 406 =over
 407
 408 =item a scalar
 409
 410 Will execute the scalar as SQL.
 411
 412 =item an arrayref
 413
 414 Taken to be arguments to L<DBI/do>, the SQL string optionally followed by the
 415 attributes hashref and bind values.
 416
 417 =item a code reference
 418
 419 Will execute C<< $code->($storage) >> and execute the return array refs as
 420 above.
 421
 422 =back
 423
 424 =item datetime_setup
 425
 426 Execute any statements necessary to initialize the database session to return
 427 and accept datetime/timestamp values used with
 428 L<DBIx::Class::InflateColumn::DateTime>.
 429
 430 Only necessary for some databases, see your specific storage driver for
 431 implementation details.
 432
 433 =back
 434
 435 =item on_disconnect_call
 436
 437 Takes arguments in the same form as L</on_connect_call> and executes them
 438 immediately before disconnecting from the database.
 439
 440 Calls the C<disconnect_call_METHOD> methods as opposed to the
 441 C<connect_call_METHOD> methods called by L</on_connect_call>.
 442
 443 Note, this only runs if you explicitly call L</disconnect> on the
 444 storage object.
 445
 446 =item disable_sth_caching
 447
 448 If set to a true value, this option will disable the caching of
 449 statement handles via L<DBI/prepare_cached>.
 450
 451 =item limit_dialect
 452
 453 Sets a specific SQL::Abstract::Limit-style limit dialect, overriding the
 454 default L</sql_limit_dialect> setting of the storage (if any). For a list
 455 of available limit dialects see L<DBIx::Class::SQLMaker::LimitDialects>.
 456
 457 =item quote_names
 458
 459 When true automatically sets L</quote_char> and L</name_sep> to the characters
 460 appropriate for your particular RDBMS. This option is preferred over specifying
 461 L</quote_char> directly.
 462
 463 =item quote_char
 464
 465 Specifies what characters to use to quote table and column names.
 466
 467 C<quote_char> expects either a single character, in which case is it
 468 is placed on either side of the table/column name, or an arrayref of length
 469 2 in which case the table/column name is placed between the elements.
 470
 471 For example under MySQL you should use C<< quote_char => '`' >>, and for
 472 SQL Server you should use C<< quote_char => [qw/[ ]/] >>.
 473
 474 =item name_sep
 475
 476 This parameter is only useful in conjunction with C<quote_char>, and is used to
 477 specify the character that separates elements (schemas, tables, columns) from
 478 each other. If unspecified it defaults to the most commonly used C<.>.
 479
 480 =item unsafe
 481
 482 This Storage driver normally installs its own C<HandleError>, sets
 483 C<RaiseError> and C<ShowErrorStatement> on, and sets C<PrintError> off on
 484 all database handles, including those supplied by a coderef.  It does this
 485 so that it can have consistent and useful error behavior.
 486
 487 If you set this option to a true value, Storage will not do its usual
 488 modifications to the database handle's attributes, and instead relies on
 489 the settings in your connect_info DBI options (or the values you set in
 490 your connection coderef, in the case that you are connecting via coderef).
 491
 492 Note that your custom settings can cause Storage to malfunction,
 493 especially if you set a C<HandleError> handler that suppresses exceptions
 494 and/or disable C<RaiseError>.
 495
 496 =item auto_savepoint
 497
 498 If this option is true, L<DBIx::Class> will use savepoints when nesting
 499 transactions, making it possible to recover from failure in the inner
 500 transaction without having to abort all outer transactions.
 501
 502 =item cursor_class
 503
 504 Use this argument to supply a cursor class other than the default
 505 L<DBIx::Class::Storage::DBI::Cursor>.
 506
 507 =back
 508
 509 Some real-life examples of arguments to L</connect_info> and
 510 L<DBIx::Class::Schema/connect>
 511
 512   # Simple SQLite connection
 513   ->connect_info([ 'dbi:SQLite:./foo.db' ]);
 514
 515   # Connect via subref
 516   ->connect_info([ sub { DBI->connect(...) } ]);
 517
 518   # Connect via subref in hashref
 519   ->connect_info([{
 520     dbh_maker => sub { DBI->connect(...) },
 521     on_connect_do => 'alter session ...',
 522   }]);
 523
 524   # A bit more complicated
 525   ->connect_info(
 526     [
 527       'dbi:Pg:dbname=foo',
 528       'postgres',
 529       'my_pg_password',
 530       { AutoCommit => 1 },
 531       { quote_char => q{"} },
 532     ]
 533   );
 534
 535   # Equivalent to the previous example
 536   ->connect_info(
 537     [
 538       'dbi:Pg:dbname=foo',
 539       'postgres',
 540       'my_pg_password',
 541       { AutoCommit => 1, quote_char => q{"}, name_sep => q{.} },
 542     ]
 543   );
 544
 545   # Same, but with hashref as argument
 546   # See parse_connect_info for explanation
 547   ->connect_info(
 548     [{
 549       dsn         => 'dbi:Pg:dbname=foo',
 550       user        => 'postgres',
 551       password    => 'my_pg_password',
 552       AutoCommit  => 1,
 553       quote_char  => q{"},
 554       name_sep    => q{.},
 555     }]
 556   );
 557
 558   # Subref + DBIx::Class-specific connection options
 559   ->connect_info(
 560     [
 561       sub { DBI->connect(...) },
 562       {
 563           quote_char => q{`},
 564           name_sep => q{@},
 565           on_connect_do => ['SET search_path TO myschema,otherschema,public'],
 566           disable_sth_caching => 1,
 567       },
 568     ]
 569   );
 570
 571
 572
 573 =cut
 574
 575 sub connect_info {
 576   my ($self, $info) = @_;
 577
 578   return $self->_connect_info if !$info;
 579
 580   $self->_connect_info($info); # copy for _connect_info
 581
 582   $info = $self->_normalize_connect_info($info)
 583     if ref $info eq 'ARRAY';
 584
 585   for my $storage_opt (keys %{ $info->{storage_options} }) {
 586     my $value = $info->{storage_options}{$storage_opt};
 587
 588     $self->$storage_opt($value);
 589   }
 590
 591   # Kill sql_maker/_sql_maker_opts, so we get a fresh one with only
 592   #  the new set of options
 593   $self->_sql_maker(undef);
 594   $self->_sql_maker_opts({});
 595
 596   for my $sql_maker_opt (keys %{ $info->{sql_maker_options} }) {
 597     my $value = $info->{sql_maker_options}{$sql_maker_opt};
 598
 599     $self->_sql_maker_opts->{$sql_maker_opt} = $value;
 600   }
 601
 602   my %attrs = (
 603     %{ $self->_default_dbi_connect_attributes || {} },
 604     %{ $info->{attributes} || {} },
 605   );
 606
 607   my @args = @{ $info->{arguments} };
 608
 609   if (keys %attrs and ref $args[0] ne 'CODE') {
 610     carp
 611         'You provided explicit AutoCommit => 0 in your connection_info. '
 612       . 'This is almost universally a bad idea (see the footnotes of '
 613       . 'DBIx::Class::Storage::DBI for more info). If you still want to '
 614       . 'do this you can set $ENV{DBIC_UNSAFE_AUTOCOMMIT_OK} to disable '
 615       . 'this warning.'
 616       if ! $attrs{AutoCommit} and ! $ENV{DBIC_UNSAFE_AUTOCOMMIT_OK};
 617
 618     push @args, \%attrs if keys %attrs;
 619   }
 620   $self->_dbi_connect_info(\@args);
 621
 622   # FIXME - dirty:
 623   # save attributes them in a separate accessor so they are always
 624   # introspectable, even in case of a CODE $dbhmaker
 625   $self->_dbic_connect_attributes (\%attrs);
 626
 627   return $self->_connect_info;
 628 }
 629
 630 sub _normalize_connect_info {
 631   my ($self, $info_arg) = @_;
 632   my %info;
 633
 634   my @args = @$info_arg;  # take a shallow copy for further mutilation
 635
 636   # combine/pre-parse arguments depending on invocation style
 637
 638   my %attrs;
 639   if (ref $args[0] eq 'CODE') {     # coderef with optional \%extra_attributes
 640     %attrs = %{ $args[1] || {} };
 641     @args = $args[0];
 642   }
 643   elsif (ref $args[0] eq 'HASH') { # single hashref (i.e. Catalyst config)
 644     %attrs = %{$args[0]};
 645     @args = ();
 646     if (my $code = delete $attrs{dbh_maker}) {
 647       @args = $code;
 648
 649       my @ignored = grep { delete $attrs{$_} } (qw/dsn user password/);
 650       if (@ignored) {
 651         carp sprintf (
 652             'Attribute(s) %s in connect_info were ignored, as they can not be applied '
 653           . "to the result of 'dbh_maker'",
 654
 655           join (', ', map { "'$_'" } (@ignored) ),
 656         );
 657       }
 658     }
 659     else {
 660       @args = delete @attrs{qw/dsn user password/};
 661     }
 662   }
 663   else {                # otherwise assume dsn/user/password + \%attrs + \%extra_attrs
 664     %attrs = (
 665       % { $args[3] || {} },
 666       % { $args[4] || {} },
 667     );
 668     @args = @args[0,1,2];
 669   }
 670
 671   $info{arguments} = \@args;
 672
 673   my @storage_opts = grep exists $attrs{$_},
 674     @storage_options, 'cursor_class';
 675
 676   @{ $info{storage_options} }{@storage_opts} =
 677     delete @attrs{@storage_opts} if @storage_opts;
 678
 679   my @sql_maker_opts = grep exists $attrs{$_},
 680     qw/limit_dialect quote_char name_sep quote_names/;
 681
 682   @{ $info{sql_maker_options} }{@sql_maker_opts} =
 683     delete @attrs{@sql_maker_opts} if @sql_maker_opts;
 684
 685   $info{attributes} = \%attrs if %attrs;
 686
 687   return \%info;
 688 }
 689
 690 sub _default_dbi_connect_attributes () {
 691   +{
 692     AutoCommit => 1,
 693     PrintError => 0,
 694     RaiseError => 1,
 695     ShowErrorStatement => 1,
 696   };
 697 }
 698
 699 =head2 on_connect_do
 700
 701 This method is deprecated in favour of setting via L</connect_info>.
 702
 703 =cut
 704
 705 =head2 on_disconnect_do
 706
 707 This method is deprecated in favour of setting via L</connect_info>.
 708
 709 =cut
 710
 711 sub _parse_connect_do {
 712   my ($self, $type) = @_;
 713
 714   my $val = $self->$type;
 715   return () if not defined $val;
 716
 717   my @res;
 718
 719   if (not ref($val)) {
 720     push @res, [ 'do_sql', $val ];
 721   } elsif (ref($val) eq 'CODE') {
 722     push @res, $val;
 723   } elsif (ref($val) eq 'ARRAY') {
 724     push @res, map { [ 'do_sql', $_ ] } @$val;
 725   } else {
 726     $self->throw_exception("Invalid type for $type: ".ref($val));
 727   }
 728
 729   return \@res;
 730 }
 731
 732 =head2 dbh_do
 733
 734 Arguments: ($subref | $method_name), @extra_coderef_args?
 735
 736 Execute the given $subref or $method_name using the new exception-based
 737 connection management.
 738
 739 The first two arguments will be the storage object that C<dbh_do> was called
 740 on and a database handle to use.  Any additional arguments will be passed
 741 verbatim to the called subref as arguments 2 and onwards.
 742
 743 Using this (instead of $self->_dbh or $self->dbh) ensures correct
 744 exception handling and reconnection (or failover in future subclasses).
 745
 746 Your subref should have no side-effects outside of the database, as
 747 there is the potential for your subref to be partially double-executed
 748 if the database connection was stale/dysfunctional.
 749
 750 Example:
 751
 752   my @stuff = $schema->storage->dbh_do(
 753     sub {
 754       my ($storage, $dbh, @cols) = @_;
 755       my $cols = join(q{, }, @cols);
 756       $dbh->selectrow_array("SELECT $cols FROM foo");
 757     },
 758     @column_list
 759   );
 760
 761 =cut
 762
 763 sub dbh_do {
 764   my $self = shift;
 765   my $code = shift;
 766
 767   my $dbh = $self->_get_dbh;
 768
 769   return $self->$code($dbh, @_)
 770     if ( $self->{_in_dbh_do} || $self->{transaction_depth} );
 771
 772   local $self->{_in_dbh_do} = 1;
 773
 774   # take a ref instead of a copy, to preserve coderef @_ aliasing semantics
 775   my $args = \@_;
 776   return try {
 777     $self->$code ($dbh, @$args);
 778   } catch {
 779     $self->throw_exception($_) if $self->connected;
 780
 781     # We were not connected - reconnect and retry, but let any
 782     #  exception fall right through this time
 783     carp "Retrying $code after catching disconnected exception: $_"
 784       if $ENV{DBIC_DBIRETRY_DEBUG};
 785
 786     $self->_populate_dbh;
 787     $self->$code($self->_dbh, @$args);
 788   };
 789 }
 790
 791 # This is basically a blend of dbh_do above and DBIx::Class::Storage::txn_do.
 792 # It also informs dbh_do to bypass itself while under the direction of txn_do,
 793 #  via $self->{_in_dbh_do} (this saves some redundant eval and errorcheck, etc)
 794 sub txn_do {
 795   my $self = shift;
 796   my $coderef = shift;
 797
 798   ref $coderef eq 'CODE' or $self->throw_exception
 799     ('$coderef must be a CODE reference');
 800
 801   local $self->{_in_dbh_do} = 1;
 802
 803   my @result;
 804   my $want = wantarray;
 805
 806   my $tried = 0;
 807   while(1) {
 808     my $exception;
 809
 810     # take a ref instead of a copy, to preserve coderef @_ aliasing semantics
 811     my $args = \@_;
 812
 813     try {
 814       $self->txn_begin;
 815       my $txn_start_depth = $self->transaction_depth;
 816       if($want) {
 817           @result = $coderef->(@$args);
 818       }
 819       elsif(defined $want) {
 820           $result[0] = $coderef->(@$args);
 821       }
 822       else {
 823           $coderef->(@$args);
 824       }
 825
 826       my $delta_txn = $txn_start_depth - $self->transaction_depth;
 827       if ($delta_txn == 0) {
 828         $self->txn_commit;
 829       }
 830       elsif ($delta_txn != 1) {
 831         # an off-by-one would mean we fired a rollback
 832         carp "Unexpected reduction of transaction depth by $delta_txn after execution of $coderef";
 833       }
 834     } catch {
 835       $exception = $_;
 836     };
 837
 838     if(! defined $exception) { return wantarray ? @result : $result[0] }
 839
 840     if($self->transaction_depth > 1 || $tried++ || $self->connected) {
 841       my $rollback_exception;
 842       try { $self->txn_rollback } catch { $rollback_exception = shift };
 843       if(defined $rollback_exception) {
 844         my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
 845         $self->throw_exception($exception)  # propagate nested rollback
 846           if $rollback_exception =~ /$exception_class/;
 847
 848         $self->throw_exception(
 849           "Transaction aborted: ${exception}. "
 850           . "Rollback failed: ${rollback_exception}"
 851         );
 852       }
 853       $self->throw_exception($exception)
 854     }
 855
 856     # We were not connected, and was first try - reconnect and retry
 857     # via the while loop
 858     carp "Retrying $coderef after catching disconnected exception: $exception"
 859       if $ENV{DBIC_TXNRETRY_DEBUG};
 860     $self->_populate_dbh;
 861   }
 862 }
 863
 864 =head2 disconnect
 865
 866 Our C<disconnect> method also performs a rollback first if the
 867 database is not in C<AutoCommit> mode.
 868
 869 =cut
 870
 871 sub disconnect {
 872   my ($self) = @_;
 873
 874   if( $self->_dbh ) {
 875     my @actions;
 876
 877     push @actions, ( $self->on_disconnect_call || () );
 878     push @actions, $self->_parse_connect_do ('on_disconnect_do');
 879
 880     $self->_do_connection_actions(disconnect_call_ => $_) for @actions;
 881
 882     $self->_dbh_rollback unless $self->_dbh_autocommit;
 883
 884     %{ $self->_dbh->{CachedKids} } = ();
 885     $self->_dbh->disconnect;
 886     $self->_dbh(undef);
 887     $self->{_dbh_gen}++;
 888   }
 889 }
 890
 891 =head2 with_deferred_fk_checks
 892
 893 =over 4
 894
 895 =item Arguments: C<$coderef>
 896
 897 =item Return Value: The return value of $coderef
 898
 899 =back
 900
 901 Storage specific method to run the code ref with FK checks deferred or
 902 in MySQL's case disabled entirely.
 903
 904 =cut
 905
 906 # Storage subclasses should override this
 907 sub with_deferred_fk_checks {
 908   my ($self, $sub) = @_;
 909   $sub->();
 910 }
 911
 912 =head2 connected
 913
 914 =over
 915
 916 =item Arguments: none
 917
 918 =item Return Value: 1|0
 919
 920 =back
 921
 922 Verifies that the current database handle is active and ready to execute
 923 an SQL statement (e.g. the connection did not get stale, server is still
 924 answering, etc.) This method is used internally by L</dbh>.
 925
 926 =cut
 927
 928 sub connected {
 929   my $self = shift;
 930   return 0 unless $self->_seems_connected;
 931
 932   #be on the safe side
 933   local $self->_dbh->{RaiseError} = 1;
 934
 935   return $self->_ping;
 936 }
 937
 938 sub _seems_connected {
 939   my $self = shift;
 940
 941   $self->_verify_pid;
 942
 943   my $dbh = $self->_dbh
 944     or return 0;
 945
 946   return $dbh->FETCH('Active');
 947 }
 948
 949 sub _ping {
 950   my $self = shift;
 951
 952   my $dbh = $self->_dbh or return 0;
 953
 954   return $dbh->ping;
 955 }
 956
 957 sub ensure_connected {
 958   my ($self) = @_;
 959
 960   unless ($self->connected) {
 961     $self->_populate_dbh;
 962   }
 963 }
 964
 965 =head2 dbh
 966
 967 Returns a C<$dbh> - a data base handle of class L<DBI>. The returned handle
 968 is guaranteed to be healthy by implicitly calling L</connected>, and if
 969 necessary performing a reconnection before returning. Keep in mind that this
 970 is very B<expensive> on some database engines. Consider using L</dbh_do>
 971 instead.
 972
 973 =cut
 974
 975 sub dbh {
 976   my ($self) = @_;
 977
 978   if (not $self->_dbh) {
 979     $self->_populate_dbh;
 980   } else {
 981     $self->ensure_connected;
 982   }
 983   return $self->_dbh;
 984 }
 985
 986 # this is the internal "get dbh or connect (don't check)" method
 987 sub _get_dbh {
 988   my $self = shift;
 989   $self->_verify_pid;
 990   $self->_populate_dbh unless $self->_dbh;
 991   return $self->_dbh;
 992 }
 993
 994 sub sql_maker {
 995   my ($self) = @_;
 996   unless ($self->_sql_maker) {
 997     my $sql_maker_class = $self->sql_maker_class;
 998     $self->ensure_class_loaded ($sql_maker_class);
 999
1000     my %opts = %{$self->_sql_maker_opts||{}};
1001     my $dialect =
1002       $opts{limit_dialect}
1003         ||
1004       $self->sql_limit_dialect
1005         ||
1006       do {
1007         my $s_class = (ref $self) || $self;
1008         carp (
1009           "Your storage class ($s_class) does not set sql_limit_dialect and you "
1010         . 'have not supplied an explicit limit_dialect in your connection_info. '
1011         . 'DBIC will attempt to use the GenericSubQ dialect, which works on most '
1012         . 'databases but can be (and often is) painfully slow. '
1013         . "Please file an RT ticket against '$s_class' ."
1014         );
1015
1016         'GenericSubQ';
1017       }
1018     ;
1019
1020     my ($quote_char, $name_sep);
1021
1022     if ($opts{quote_names}) {
1023       $quote_char = (delete $opts{quote_char}) || $self->sql_quote_char || do {
1024         my $s_class = (ref $self) || $self;
1025         carp (
1026           "You requested 'quote_names' but your storage class ($s_class) does "
1027         . 'not explicitly define a default sql_quote_char and you have not '
1028         . 'supplied a quote_char as part of your connection_info. DBIC will '
1029         .q{default to the ANSI SQL standard quote '"', which works most of }
1030         . "the time. Please file an RT ticket against '$s_class'."
1031         );
1032
1033         '"'; # RV
1034       };
1035
1036       $name_sep = (delete $opts{name_sep}) || $self->sql_name_sep;
1037     }
1038
1039     $self->_sql_maker($sql_maker_class->new(
1040       bindtype=>'columns',
1041       array_datatypes => 1,
1042       limit_dialect => $dialect,
1043       ($quote_char ? (quote_char => $quote_char) : ()),
1044       name_sep => ($name_sep || '.'),
1045       %opts,
1046     ));
1047   }
1048   return $self->_sql_maker;
1049 }
1050
1051 # nothing to do by default
1052 sub _rebless {}
1053 sub _init {}
1054
1055 sub _populate_dbh {
1056   my ($self) = @_;
1057
1058   my @info = @{$self->_dbi_connect_info || []};
1059   $self->_dbh(undef); # in case ->connected failed we might get sent here
1060   $self->_dbh_details({}); # reset everything we know
1061
1062   $self->_dbh($self->_connect(@info));
1063
1064   $self->_conn_pid($$) if $^O ne 'MSWin32'; # on win32 these are in fact threads
1065
1066   $self->_determine_driver;
1067
1068   # Always set the transaction depth on connect, since
1069   #  there is no transaction in progress by definition
1070   $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
1071
1072   $self->_run_connection_actions unless $self->{_in_determine_driver};
1073 }
1074
1075 sub _run_connection_actions {
1076   my $self = shift;
1077   my @actions;
1078
1079   push @actions, ( $self->on_connect_call || () );
1080   push @actions, $self->_parse_connect_do ('on_connect_do');
1081
1082   $self->_do_connection_actions(connect_call_ => $_) for @actions;
1083 }
1084
1085
1086
1087 sub set_use_dbms_capability {
1088   $_[0]->set_inherited ($_[1], $_[2]);
1089 }
1090
1091 sub get_use_dbms_capability {
1092   my ($self, $capname) = @_;
1093
1094   my $use = $self->get_inherited ($capname);
1095   return defined $use
1096     ? $use
1097     : do { $capname =~ s/^_use_/_supports_/; $self->get_dbms_capability ($capname) }
1098   ;
1099 }
1100
1101 sub set_dbms_capability {
1102   $_[0]->_dbh_details->{capability}{$_[1]} = $_[2];
1103 }
1104
1105 sub get_dbms_capability {
1106   my ($self, $capname) = @_;
1107
1108   my $cap = $self->_dbh_details->{capability}{$capname};
1109
1110   unless (defined $cap) {
1111     if (my $meth = $self->can ("_determine$capname")) {
1112       $cap = $self->$meth ? 1 : 0;
1113     }
1114     else {
1115       $cap = 0;
1116     }
1117
1118     $self->set_dbms_capability ($capname, $cap);
1119   }
1120
1121   return $cap;
1122 }
1123
1124 sub _server_info {
1125   my $self = shift;
1126
1127   my $info;
1128   unless ($info = $self->_dbh_details->{info}) {
1129
1130     $info = {};
1131
1132     my $server_version = try { $self->_get_server_version };
1133
1134     if (defined $server_version) {
1135       $info->{dbms_version} = $server_version;
1136
1137       my ($numeric_version) = $server_version =~ /^([\d\.]+)/;
1138       my @verparts = split (/\./, $numeric_version);
1139       if (
1140         @verparts
1141           &&
1142         $verparts[0] <= 999
1143       ) {
1144         # consider only up to 3 version parts, iff not more than 3 digits
1145         my @use_parts;
1146         while (@verparts && @use_parts < 3) {
1147           my $p = shift @verparts;
1148           last if $p > 999;
1149           push @use_parts, $p;
1150         }
1151         push @use_parts, 0 while @use_parts < 3;
1152
1153         $info->{normalized_dbms_version} = sprintf "%d.%03d%03d", @use_parts;
1154       }
1155     }
1156
1157     $self->_dbh_details->{info} = $info;
1158   }
1159
1160   return $info;
1161 }
1162
1163 sub _get_server_version {
1164   shift->_get_dbh->get_info(18);
1165 }
1166
1167 sub _determine_driver {
1168   my ($self) = @_;
1169
1170   if ((not $self->_driver_determined) && (not $self->{_in_determine_driver})) {
1171     my $started_connected = 0;
1172     local $self->{_in_determine_driver} = 1;
1173
1174     if (ref($self) eq __PACKAGE__) {
1175       my $driver;
1176       if ($self->_dbh) { # we are connected
1177         $driver = $self->_dbh->{Driver}{Name};
1178         $started_connected = 1;
1179       } else {
1180         # if connect_info is a CODEREF, we have no choice but to connect
1181         if (ref $self->_dbi_connect_info->[0] &&
1182             reftype $self->_dbi_connect_info->[0] eq 'CODE') {
1183           $self->_populate_dbh;
1184           $driver = $self->_dbh->{Driver}{Name};
1185         }
1186         else {
1187           # try to use dsn to not require being connected, the driver may still
1188           # force a connection in _rebless to determine version
1189           # (dsn may not be supplied at all if all we do is make a mock-schema)
1190           my $dsn = $self->_dbi_connect_info->[0] || $ENV{DBI_DSN} || '';
1191           ($driver) = $dsn =~ /dbi:([^:]+):/i;
1192           $driver ||= $ENV{DBI_DRIVER};
1193         }
1194       }
1195
1196       if ($driver) {
1197         my $storage_class = "DBIx::Class::Storage::DBI::${driver}";
1198         if ($self->load_optional_class($storage_class)) {
1199           mro::set_mro($storage_class, 'c3');
1200           bless $self, $storage_class;
1201           $self->_rebless();
1202         }
1203       }
1204     }
1205
1206     $self->_driver_determined(1);
1207
1208     $self->_init; # run driver-specific initializations
1209
1210     $self->_run_connection_actions
1211         if !$started_connected && defined $self->_dbh;
1212   }
1213 }
1214
1215 sub _do_connection_actions {
1216   my $self          = shift;
1217   my $method_prefix = shift;
1218   my $call          = shift;
1219
1220   if (not ref($call)) {
1221     my $method = $method_prefix . $call;
1222     $self->$method(@_);
1223   } elsif (ref($call) eq 'CODE') {
1224     $self->$call(@_);
1225   } elsif (ref($call) eq 'ARRAY') {
1226     if (ref($call->[0]) ne 'ARRAY') {
1227       $self->_do_connection_actions($method_prefix, $_) for @$call;
1228     } else {
1229       $self->_do_connection_actions($method_prefix, @$_) for @$call;
1230     }
1231   } else {
1232     $self->throw_exception (sprintf ("Don't know how to process conection actions of type '%s'", ref($call)) );
1233   }
1234
1235   return $self;
1236 }
1237
1238 sub connect_call_do_sql {
1239   my $self = shift;
1240   $self->_do_query(@_);
1241 }
1242
1243 sub disconnect_call_do_sql {
1244   my $self = shift;
1245   $self->_do_query(@_);
1246 }
1247
1248 # override in db-specific backend when necessary
1249 sub connect_call_datetime_setup { 1 }
1250
1251 sub _do_query {
1252   my ($self, $action) = @_;
1253
1254   if (ref $action eq 'CODE') {
1255     $action = $action->($self);
1256     $self->_do_query($_) foreach @$action;
1257   }
1258   else {
1259     # Most debuggers expect ($sql, @bind), so we need to exclude
1260     # the attribute hash which is the second argument to $dbh->do
1261     # furthermore the bind values are usually to be presented
1262     # as named arrayref pairs, so wrap those here too
1263     my @do_args = (ref $action eq 'ARRAY') ? (@$action) : ($action);
1264     my $sql = shift @do_args;
1265     my $attrs = shift @do_args;
1266     my @bind = map { [ undef, $_ ] } @do_args;
1267
1268     $self->_query_start($sql, @bind);
1269     $self->_get_dbh->do($sql, $attrs, @do_args);
1270     $self->_query_end($sql, @bind);
1271   }
1272
1273   return $self;
1274 }
1275
1276 sub _connect {
1277   my ($self, @info) = @_;
1278
1279   $self->throw_exception("You failed to provide any connection info")
1280     if !@info;
1281
1282   my ($old_connect_via, $dbh);
1283
1284   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
1285     $old_connect_via = $DBI::connect_via;
1286     $DBI::connect_via = 'connect';
1287   }
1288
1289   try {
1290     if(ref $info[0] eq 'CODE') {
1291        $dbh = $info[0]->();
1292     }
1293     else {
1294        $dbh = DBI->connect(@info);
1295     }
1296
1297     if (!$dbh) {
1298       die $DBI::errstr;
1299     }
1300
1301     unless ($self->unsafe) {
1302
1303       $self->throw_exception(
1304         'Refusing clobbering of {HandleError} installed on externally supplied '
1305        ."DBI handle $dbh. Either remove the handler or use the 'unsafe' attribute."
1306       ) if $dbh->{HandleError} and ref $dbh->{HandleError} ne '__DBIC__DBH__ERROR__HANDLER__';
1307
1308       # Default via _default_dbi_connect_attributes is 1, hence it was an explicit
1309       # request, or an external handle. Complain and set anyway
1310       unless ($dbh->{RaiseError}) {
1311         carp( ref $info[0] eq 'CODE'
1312
1313           ? "The 'RaiseError' of the externally supplied DBI handle is set to false. "
1314            ."DBIx::Class will toggle it back to true, unless the 'unsafe' connect "
1315            .'attribute has been supplied'
1316
1317           : 'RaiseError => 0 supplied in your connection_info, without an explicit '
1318            .'unsafe => 1. Toggling RaiseError back to true'
1319         );
1320
1321         $dbh->{RaiseError} = 1;
1322       }
1323
1324       # this odd anonymous coderef dereference is in fact really
1325       # necessary to avoid the unwanted effect described in perl5
1326       # RT#75792
1327       sub {
1328         my $weak_self = $_[0];
1329         weaken $weak_self;
1330
1331         # the coderef is blessed so we can distinguish it from externally
1332         # supplied handles (which must be preserved)
1333         $_[1]->{HandleError} = bless sub {
1334           if ($weak_self) {
1335             $weak_self->throw_exception("DBI Exception: $_[0]");
1336           }
1337           else {
1338             # the handler may be invoked by something totally out of
1339             # the scope of DBIC
1340             croak ("DBI Exception (unhandled by DBIC, ::Schema GCed): $_[0]");
1341           }
1342         }, '__DBIC__DBH__ERROR__HANDLER__';
1343       }->($self, $dbh);
1344     }
1345   }
1346   catch {
1347     $self->throw_exception("DBI Connection failed: $_")
1348   }
1349   finally {
1350     $DBI::connect_via = $old_connect_via if $old_connect_via;
1351   };
1352
1353   $self->_dbh_autocommit($dbh->{AutoCommit});
1354   $dbh;
1355 }
1356
1357 sub svp_begin {
1358   my ($self, $name) = @_;
1359
1360   $name = $self->_svp_generate_name
1361     unless defined $name;
1362
1363   $self->throw_exception ("You can't use savepoints outside a transaction")
1364     if $self->{transaction_depth} == 0;
1365
1366   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1367     unless $self->can('_svp_begin');
1368
1369   push @{ $self->{savepoints} }, $name;
1370
1371   $self->debugobj->svp_begin($name) if $self->debug;
1372
1373   return $self->_svp_begin($name);
1374 }
1375
1376 sub svp_release {
1377   my ($self, $name) = @_;
1378
1379   $self->throw_exception ("You can't use savepoints outside a transaction")
1380     if $self->{transaction_depth} == 0;
1381
1382   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1383     unless $self->can('_svp_release');
1384
1385   if (defined $name) {
1386     $self->throw_exception ("Savepoint '$name' does not exist")
1387       unless grep { $_ eq $name } @{ $self->{savepoints} };
1388
1389     # Dig through the stack until we find the one we are releasing.  This keeps
1390     # the stack up to date.
1391     my $svp;
1392
1393     do { $svp = pop @{ $self->{savepoints} } } while $svp ne $name;
1394   } else {
1395     $name = pop @{ $self->{savepoints} };
1396   }
1397
1398   $self->debugobj->svp_release($name) if $self->debug;
1399
1400   return $self->_svp_release($name);
1401 }
1402
1403 sub svp_rollback {
1404   my ($self, $name) = @_;
1405
1406   $self->throw_exception ("You can't use savepoints outside a transaction")
1407     if $self->{transaction_depth} == 0;
1408
1409   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1410     unless $self->can('_svp_rollback');
1411
1412   if (defined $name) {
1413       # If they passed us a name, verify that it exists in the stack
1414       unless(grep({ $_ eq $name } @{ $self->{savepoints} })) {
1415           $self->throw_exception("Savepoint '$name' does not exist!");
1416       }
1417
1418       # Dig through the stack until we find the one we are releasing.  This keeps
1419       # the stack up to date.
1420       while(my $s = pop(@{ $self->{savepoints} })) {
1421           last if($s eq $name);
1422       }
1423       # Add the savepoint back to the stack, as a rollback doesn't remove the
1424       # named savepoint, only everything after it.
1425       push(@{ $self->{savepoints} }, $name);
1426   } else {
1427       # We'll assume they want to rollback to the last savepoint
1428       $name = $self->{savepoints}->[-1];
1429   }
1430
1431   $self->debugobj->svp_rollback($name) if $self->debug;
1432
1433   return $self->_svp_rollback($name);
1434 }
1435
1436 sub _svp_generate_name {
1437   my ($self) = @_;
1438   return 'savepoint_'.scalar(@{ $self->{'savepoints'} });
1439 }
1440
1441 sub txn_begin {
1442   my $self = shift;
1443
1444   # this means we have not yet connected and do not know the AC status
1445   # (e.g. coderef $dbh)
1446   if (! defined $self->_dbh_autocommit) {
1447     $self->ensure_connected;
1448   }
1449   # otherwise re-connect on pid changes, so
1450   # that the txn_depth is adjusted properly
1451   # the lightweight _get_dbh is good enoug here
1452   # (only superficial handle check, no pings)
1453   else {
1454     $self->_get_dbh;
1455   }
1456
1457   if($self->transaction_depth == 0) {
1458     $self->debugobj->txn_begin()
1459       if $self->debug;
1460     $self->_dbh_begin_work;
1461   }
1462   elsif ($self->auto_savepoint) {
1463     $self->svp_begin;
1464   }
1465   $self->{transaction_depth}++;
1466 }
1467
1468 sub _dbh_begin_work {
1469   my $self = shift;
1470
1471   # if the user is utilizing txn_do - good for him, otherwise we need to
1472   # ensure that the $dbh is healthy on BEGIN.
1473   # We do this via ->dbh_do instead of ->dbh, so that the ->dbh "ping"
1474   # will be replaced by a failure of begin_work itself (which will be
1475   # then retried on reconnect)
1476   if ($self->{_in_dbh_do}) {
1477     $self->_dbh->begin_work;
1478   } else {
1479     $self->dbh_do(sub { $_[1]->begin_work });
1480   }
1481 }
1482
1483 sub txn_commit {
1484   my $self = shift;
1485   if (! $self->_dbh) {
1486     $self->throw_exception('cannot COMMIT on a disconnected handle');
1487   }
1488   elsif ($self->{transaction_depth} == 1) {
1489     $self->debugobj->txn_commit()
1490       if ($self->debug);
1491     $self->_dbh_commit;
1492     $self->{transaction_depth} = 0
1493       if $self->_dbh_autocommit;
1494   }
1495   elsif($self->{transaction_depth} > 1) {
1496     $self->{transaction_depth}--;
1497     $self->svp_release
1498       if $self->auto_savepoint;
1499   }
1500   elsif (! $self->_dbh->FETCH('AutoCommit') ) {
1501
1502     carp "Storage transaction_depth $self->{transaction_depth} does not match "
1503         ."false AutoCommit of $self->{_dbh}, attempting COMMIT anyway";
1504
1505     $self->debugobj->txn_commit()
1506       if ($self->debug);
1507     $self->_dbh_commit;
1508     $self->{transaction_depth} = 0
1509       if $self->_dbh_autocommit;
1510   }
1511   else {
1512     $self->throw_exception( 'Refusing to commit without a started transaction' );
1513   }
1514 }
1515
1516 sub _dbh_commit {
1517   my $self = shift;
1518   my $dbh  = $self->_dbh
1519     or $self->throw_exception('cannot COMMIT on a disconnected handle');
1520   $dbh->commit;
1521 }
1522
1523 sub txn_rollback {
1524   my $self = shift;
1525   my $dbh = $self->_dbh;
1526   try {
1527     if ($self->{transaction_depth} == 1) {
1528       $self->debugobj->txn_rollback()
1529         if ($self->debug);
1530       $self->{transaction_depth} = 0
1531         if $self->_dbh_autocommit;
1532       $self->_dbh_rollback;
1533     }
1534     elsif($self->{transaction_depth} > 1) {
1535       $self->{transaction_depth}--;
1536       if ($self->auto_savepoint) {
1537         $self->svp_rollback;
1538         $self->svp_release;
1539       }
1540     }
1541     else {
1542       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
1543     }
1544   }
1545   catch {
1546     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
1547
1548     if ($_ !~ /$exception_class/) {
1549       # ensure that a failed rollback resets the transaction depth
1550       $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
1551     }
1552
1553     $self->throw_exception($_)
1554   };
1555 }
1556
1557 sub _dbh_rollback {
1558   my $self = shift;
1559   my $dbh  = $self->_dbh
1560     or $self->throw_exception('cannot ROLLBACK on a disconnected handle');
1561   $dbh->rollback;
1562 }
1563
1564 # This used to be the top-half of _execute.  It was split out to make it
1565 #  easier to override in NoBindVars without duping the rest.  It takes up
1566 #  all of _execute's args, and emits $sql, @bind.
1567 sub _prep_for_execute {
1568   my ($self, $op, $extra_bind, $ident, $args) = @_;
1569
1570   if( blessed $ident && $ident->isa("DBIx::Class::ResultSource") ) {
1571     $ident = $ident->from();
1572   }
1573
1574   my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
1575
1576   unshift(@bind,
1577     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
1578       if $extra_bind;
1579   return ($sql, \@bind);
1580 }
1581
1582
1583 sub _fix_bind_params {
1584     my ($self, @bind) = @_;
1585
1586     ### Turn @bind from something like this:
1587     ###   ( [ "artist", 1 ], [ "cdid", 1, 3 ] )
1588     ### to this:
1589     ###   ( "'1'", "'1'", "'3'" )
1590     return
1591         map {
1592             if ( defined( $_ && $_->[1] ) ) {
1593                 map { qq{'$_'}; } @{$_}[ 1 .. $#$_ ];
1594             }
1595             else { q{NULL}; }
1596         } @bind;
1597 }
1598
1599 sub _query_start {
1600     my ( $self, $sql, @bind ) = @_;
1601
1602     if ( $self->debug ) {
1603         @bind = $self->_fix_bind_params(@bind);
1604
1605         $self->debugobj->query_start( $sql, @bind );
1606     }
1607 }
1608
1609 sub _query_end {
1610     my ( $self, $sql, @bind ) = @_;
1611
1612     if ( $self->debug ) {
1613         @bind = $self->_fix_bind_params(@bind);
1614         $self->debugobj->query_end( $sql, @bind );
1615     }
1616 }
1617
1618 sub _dbh_execute {
1619   my ($self, $dbh, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
1620
1621   my ($sql, $bind) = $self->_prep_for_execute($op, $extra_bind, $ident, \@args);
1622
1623   $self->_query_start( $sql, @$bind );
1624
1625   my $sth = $self->sth($sql,$op);
1626
1627   my $placeholder_index = 1;
1628
1629   foreach my $bound (@$bind) {
1630     my $attributes = {};
1631     my($column_name, @data) = @$bound;
1632
1633     if ($bind_attributes) {
1634       $attributes = $bind_attributes->{$column_name}
1635       if defined $bind_attributes->{$column_name};
1636     }
1637
1638     foreach my $data (@data) {
1639       my $ref = ref $data;
1640
1641       if ($ref and overload::Method($data, '""') ) {
1642         $data = "$data";
1643       }
1644       elsif ($ref eq 'SCALAR') {  # any scalarrefs are assumed to be bind_inouts
1645         $sth->bind_param_inout(
1646           $placeholder_index++,
1647           $data,
1648           $self->_max_column_bytesize($ident, $column_name),
1649           $attributes
1650         );
1651         next;
1652       }
1653
1654       $sth->bind_param($placeholder_index++, $data, $attributes);
1655     }
1656   }
1657
1658   # Can this fail without throwing an exception anyways???
1659   my $rv = $sth->execute();
1660   $self->throw_exception(
1661     $sth->errstr || $sth->err || 'Unknown error: execute() returned false, but error flags were not set...'
1662   ) if !$rv;
1663
1664   $self->_query_end( $sql, @$bind );
1665
1666   return (wantarray ? ($rv, $sth, @$bind) : $rv);
1667 }
1668
1669 sub _execute {
1670     my $self = shift;
1671     $self->dbh_do('_dbh_execute', @_);  # retry over disconnects
1672 }
1673
1674 sub _prefetch_autovalues {
1675   my ($self, $source, $to_insert) = @_;
1676
1677   my $colinfo = $source->columns_info;
1678
1679   my %values;
1680   for my $col (keys %$colinfo) {
1681     if (
1682       $colinfo->{$col}{auto_nextval}
1683         and
1684       (
1685         ! exists $to_insert->{$col}
1686           or
1687         ref $to_insert->{$col} eq 'SCALAR'
1688       )
1689     ) {
1690       $values{$col} = $self->_sequence_fetch(
1691         'NEXTVAL',
1692         ( $colinfo->{$col}{sequence} ||=
1693             $self->_dbh_get_autoinc_seq($self->_get_dbh, $source, $col)
1694         ),
1695       );
1696     }
1697   }
1698
1699   \%values;
1700 }
1701
1702 sub insert {
1703   my ($self, $source, $to_insert) = @_;
1704
1705   my $prefetched_values = $self->_prefetch_autovalues($source, $to_insert);
1706
1707   # fuse the values
1708   $to_insert = { %$to_insert, %$prefetched_values };
1709
1710   # list of primary keys we try to fetch from the database
1711   # both not-exsists and scalarrefs are considered
1712   my %fetch_pks;
1713   for ($source->primary_columns) {
1714     $fetch_pks{$_} = scalar keys %fetch_pks  # so we can preserve order for prettyness
1715       if ! exists $to_insert->{$_} or ref $to_insert->{$_} eq 'SCALAR';
1716   }
1717
1718   my ($sqla_opts, @ir_container);
1719   if ($self->_use_insert_returning) {
1720
1721     # retain order as declared in the resultsource
1722     for (sort { $fetch_pks{$a} <=> $fetch_pks{$b} } keys %fetch_pks ) {
1723       push @{$sqla_opts->{returning}}, $_;
1724       $sqla_opts->{returning_container} = \@ir_container
1725         if $self->_use_insert_returning_bound;
1726     }
1727   }
1728
1729   my $bind_attributes = $self->source_bind_attributes($source);
1730
1731   my ($rv, $sth) = $self->_execute('insert' => [], $source, $bind_attributes, $to_insert, $sqla_opts);
1732
1733   my %returned_cols;
1734
1735   if (my $retlist = $sqla_opts->{returning}) {
1736     @ir_container = try {
1737       local $SIG{__WARN__} = sub {};
1738       my @r = $sth->fetchrow_array;
1739       $sth->finish;
1740       @r;
1741     } unless @ir_container;
1742
1743     @returned_cols{@$retlist} = @ir_container if @ir_container;
1744   }
1745
1746   return { %$prefetched_values, %returned_cols };
1747 }
1748
1749
1750 ## Currently it is assumed that all values passed will be "normal", i.e. not
1751 ## scalar refs, or at least, all the same type as the first set, the statement is
1752 ## only prepped once.
1753 sub insert_bulk {
1754   my ($self, $source, $cols, $data) = @_;
1755
1756   my %colvalues;
1757   @colvalues{@$cols} = (0..$#$cols);
1758
1759   for my $i (0..$#$cols) {
1760     my $first_val = $data->[0][$i];
1761     next unless ref $first_val eq 'SCALAR';
1762
1763     $colvalues{ $cols->[$i] } = $first_val;
1764   }
1765
1766   # check for bad data and stringify stringifiable objects
1767   my $bad_slice = sub {
1768     my ($msg, $col_idx, $slice_idx) = @_;
1769     $self->throw_exception(sprintf "%s for column '%s' in populate slice:\n%s",
1770       $msg,
1771       $cols->[$col_idx],
1772       do {
1773         local $Data::Dumper::Maxdepth = 1; # don't dump objects, if any
1774         Dumper {
1775           map { $cols->[$_] => $data->[$slice_idx][$_] } (0 .. $#$cols)
1776         },
1777       }
1778     );
1779   };
1780
1781   for my $datum_idx (0..$#$data) {
1782     my $datum = $data->[$datum_idx];
1783
1784     for my $col_idx (0..$#$cols) {
1785       my $val            = $datum->[$col_idx];
1786       my $sqla_bind      = $colvalues{ $cols->[$col_idx] };
1787       my $is_literal_sql = (ref $sqla_bind) eq 'SCALAR';
1788
1789       if ($is_literal_sql) {
1790         if (not ref $val) {
1791           $bad_slice->('bind found where literal SQL expected', $col_idx, $datum_idx);
1792         }
1793         elsif ((my $reftype = ref $val) ne 'SCALAR') {
1794           $bad_slice->("$reftype reference found where literal SQL expected",
1795             $col_idx, $datum_idx);
1796         }
1797         elsif ($$val ne $$sqla_bind){
1798           $bad_slice->("inconsistent literal SQL value, expecting: '$$sqla_bind'",
1799             $col_idx, $datum_idx);
1800         }
1801       }
1802       elsif (my $reftype = ref $val) {
1803         require overload;
1804         if (overload::Method($val, '""')) {
1805           $datum->[$col_idx] = "".$val;
1806         }
1807         else {
1808           $bad_slice->("$reftype reference found where bind expected",
1809             $col_idx, $datum_idx);
1810         }
1811       }
1812     }
1813   }
1814
1815   my ($sql, $bind) = $self->_prep_for_execute (
1816     'insert', undef, $source, [\%colvalues]
1817   );
1818
1819   if (! @$bind) {
1820     # if the bindlist is empty - make sure all "values" are in fact
1821     # literal scalarrefs. If not the case this means the storage ate
1822     # them away (e.g. the NoBindVars component) and interpolated them
1823     # directly into the SQL. This obviosly can't be good for multi-inserts
1824
1825     $self->throw_exception('Cannot insert_bulk without support for placeholders')
1826       if first { ref $_ ne 'SCALAR' } values %colvalues;
1827   }
1828
1829   # neither _execute_array, nor _execute_inserts_with_no_binds are
1830   # atomic (even if _execute _array is a single call). Thus a safety
1831   # scope guard
1832   my $guard = $self->txn_scope_guard;
1833
1834   $self->_query_start( $sql, @$bind ? [ dummy => '__BULK_INSERT__' ] : () );
1835   my $sth = $self->sth($sql);
1836   my $rv = do {
1837     if (@$bind) {
1838       #@bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
1839       $self->_execute_array( $source, $sth, $bind, $cols, $data );
1840     }
1841     else {
1842       # bind_param_array doesn't work if there are no binds
1843       $self->_dbh_execute_inserts_with_no_binds( $sth, scalar @$data );
1844     }
1845   };
1846
1847   $self->_query_end( $sql, @$bind ? [ dummy => '__BULK_INSERT__' ] : () );
1848
1849   $guard->commit;
1850
1851   return (wantarray ? ($rv, $sth, @$bind) : $rv);
1852 }
1853
1854 sub _execute_array {
1855   my ($self, $source, $sth, $bind, $cols, $data, @extra) = @_;
1856
1857   ## This must be an arrayref, else nothing works!
1858   my $tuple_status = [];
1859
1860   ## Get the bind_attributes, if any exist
1861   my $bind_attributes = $self->source_bind_attributes($source);
1862
1863   ## Bind the values and execute
1864   my $placeholder_index = 1;
1865
1866   foreach my $bound (@$bind) {
1867
1868     my $attributes = {};
1869     my ($column_name, $data_index) = @$bound;
1870
1871     if( $bind_attributes ) {
1872       $attributes = $bind_attributes->{$column_name}
1873       if defined $bind_attributes->{$column_name};
1874     }
1875
1876     my @data = map { $_->[$data_index] } @$data;
1877
1878     $sth->bind_param_array(
1879       $placeholder_index,
1880       [@data],
1881       (%$attributes ?  $attributes : ()),
1882     );
1883     $placeholder_index++;
1884   }
1885
1886   my ($rv, $err);
1887   try {
1888     $rv = $self->_dbh_execute_array($sth, $tuple_status, @extra);
1889   }
1890   catch {
1891     $err = shift;
1892   };
1893
1894   # Not all DBDs are create equal. Some throw on error, some return
1895   # an undef $rv, and some set $sth->err - try whatever we can
1896   $err = ($sth->errstr || 'UNKNOWN ERROR ($sth->errstr is unset)') if (
1897     ! defined $err
1898       and
1899     ( !defined $rv or $sth->err )
1900   );
1901
1902   # Statement must finish even if there was an exception.
1903   try {
1904     $sth->finish
1905   }
1906   catch {
1907     $err = shift unless defined $err
1908   };
1909
1910   if (defined $err) {
1911     my $i = 0;
1912     ++$i while $i <= $#$tuple_status && !ref $tuple_status->[$i];
1913
1914     $self->throw_exception("Unexpected populate error: $err")
1915       if ($i > $#$tuple_status);
1916
1917     $self->throw_exception(sprintf "%s for populate slice:\n%s",
1918       ($tuple_status->[$i][1] || $err),
1919       Dumper { map { $cols->[$_] => $data->[$i][$_] } (0 .. $#$cols) },
1920     );
1921   }
1922
1923   return $rv;
1924 }
1925
1926 sub _dbh_execute_array {
1927     my ($self, $sth, $tuple_status, @extra) = @_;
1928
1929     return $sth->execute_array({ArrayTupleStatus => $tuple_status});
1930 }
1931
1932 sub _dbh_execute_inserts_with_no_binds {
1933   my ($self, $sth, $count) = @_;
1934
1935   my $err;
1936   try {
1937     my $dbh = $self->_get_dbh;
1938     local $dbh->{RaiseError} = 1;
1939     local $dbh->{PrintError} = 0;
1940
1941     $sth->execute foreach 1..$count;
1942   }
1943   catch {
1944     $err = shift;
1945   };
1946
1947   # Make sure statement is finished even if there was an exception.
1948   try {
1949     $sth->finish
1950   }
1951   catch {
1952     $err = shift unless defined $err;
1953   };
1954
1955   $self->throw_exception($err) if defined $err;
1956
1957   return $count;
1958 }
1959
1960 sub update {
1961   my ($self, $source, @args) = @_;
1962
1963   my $bind_attrs = $self->source_bind_attributes($source);
1964
1965   return $self->_execute('update' => [], $source, $bind_attrs, @args);
1966 }
1967
1968
1969 sub delete {
1970   my ($self, $source, @args) = @_;
1971
1972   my $bind_attrs = $self->source_bind_attributes($source);
1973
1974   return $self->_execute('delete' => [], $source, $bind_attrs, @args);
1975 }
1976
1977 # We were sent here because the $rs contains a complex search
1978 # which will require a subquery to select the correct rows
1979 # (i.e. joined or limited resultsets, or non-introspectable conditions)
1980 #
1981 # Generating a single PK column subquery is trivial and supported
1982 # by all RDBMS. However if we have a multicolumn PK, things get ugly.
1983 # Look at _multipk_update_delete()
1984 sub _subq_update_delete {
1985   my $self = shift;
1986   my ($rs, $op, $values) = @_;
1987
1988   my $rsrc = $rs->result_source;
1989
1990   # quick check if we got a sane rs on our hands
1991   my @pcols = $rsrc->_pri_cols;
1992
1993   my $sel = $rs->_resolved_attrs->{select};
1994   $sel = [ $sel ] unless ref $sel eq 'ARRAY';
1995
1996   if (
1997       join ("\x00", map { join '.', $rs->{attrs}{alias}, $_ } sort @pcols)
1998         ne
1999       join ("\x00", sort @$sel )
2000   ) {
2001     $self->throw_exception (
2002       '_subq_update_delete can not be called on resultsets selecting columns other than the primary keys'
2003     );
2004   }
2005
2006   if (@pcols == 1) {
2007     return $self->$op (
2008       $rsrc,
2009       $op eq 'update' ? $values : (),
2010       { $pcols[0] => { -in => $rs->as_query } },
2011     );
2012   }
2013
2014   else {
2015     return $self->_multipk_update_delete (@_);
2016   }
2017 }
2018
2019 # ANSI SQL does not provide a reliable way to perform a multicol-PK
2020 # resultset update/delete involving subqueries. So by default resort
2021 # to simple (and inefficient) delete_all style per-row opearations,
2022 # while allowing specific storages to override this with a faster
2023 # implementation.
2024 #
2025 sub _multipk_update_delete {
2026   return shift->_per_row_update_delete (@_);
2027 }
2028
2029 # This is the default loop used to delete/update rows for multi PK
2030 # resultsets, and used by mysql exclusively (because it can't do anything
2031 # else).
2032 #
2033 # We do not use $row->$op style queries, because resultset update/delete
2034 # is not expected to cascade (this is what delete_all/update_all is for).
2035 #
2036 # There should be no race conditions as the entire operation is rolled
2037 # in a transaction.
2038 #
2039 sub _per_row_update_delete {
2040   my $self = shift;
2041   my ($rs, $op, $values) = @_;
2042
2043   my $rsrc = $rs->result_source;
2044   my @pcols = $rsrc->_pri_cols;
2045
2046   my $guard = $self->txn_scope_guard;
2047
2048   # emulate the return value of $sth->execute for non-selects
2049   my $row_cnt = '0E0';
2050
2051   my $subrs_cur = $rs->cursor;
2052   my @all_pk = $subrs_cur->all;
2053   for my $pks ( @all_pk) {
2054
2055     my $cond;
2056     for my $i (0.. $#pcols) {
2057       $cond->{$pcols[$i]} = $pks->[$i];
2058     }
2059
2060     $self->$op (
2061       $rsrc,
2062       $op eq 'update' ? $values : (),
2063       $cond,
2064     );
2065
2066     $row_cnt++;
2067   }
2068
2069   $guard->commit;
2070
2071   return $row_cnt;
2072 }
2073
2074 sub _select {
2075   my $self = shift;
2076   $self->_execute($self->_select_args(@_));
2077 }
2078
2079 sub _select_args_to_query {
2080   my $self = shift;
2081
2082   # my ($op, $bind, $ident, $bind_attrs, $select, $cond, $rs_attrs, $rows, $offset)
2083   #  = $self->_select_args($ident, $select, $cond, $attrs);
2084   my ($op, $bind, $ident, $bind_attrs, @args) =
2085     $self->_select_args(@_);
2086
2087   # my ($sql, $prepared_bind) = $self->_prep_for_execute($op, $bind, $ident, [ $select, $cond, $rs_attrs, $rows, $offset ]);
2088   my ($sql, $prepared_bind) = $self->_prep_for_execute($op, $bind, $ident, \@args);
2089   $prepared_bind ||= [];
2090
2091   return wantarray
2092     ? ($sql, $prepared_bind, $bind_attrs)
2093     : \[ "($sql)", @$prepared_bind ]
2094   ;
2095 }
2096
2097 sub _select_args {
2098   my ($self, $ident, $select, $where, $attrs) = @_;
2099
2100   my $sql_maker = $self->sql_maker;
2101   my ($alias2source, $rs_alias) = $self->_resolve_ident_sources ($ident);
2102
2103   $attrs = {
2104     %$attrs,
2105     select => $select,
2106     from => $ident,
2107     where => $where,
2108     $rs_alias && $alias2source->{$rs_alias}
2109       ? ( _rsroot_rsrc => $alias2source->{$rs_alias} )
2110       : ()
2111     ,
2112   };
2113
2114   # calculate bind_attrs before possible $ident mangling
2115   my $bind_attrs = {};
2116   for my $alias (keys %$alias2source) {
2117     my $bindtypes = $self->source_bind_attributes ($alias2source->{$alias}) || {};
2118     for my $col (keys %$bindtypes) {
2119
2120       my $fqcn = join ('.', $alias, $col);
2121       $bind_attrs->{$fqcn} = $bindtypes->{$col} if $bindtypes->{$col};
2122
2123       # Unqialified column names are nice, but at the same time can be
2124       # rather ambiguous. What we do here is basically go along with
2125       # the loop, adding an unqualified column slot to $bind_attrs,
2126       # alongside the fully qualified name. As soon as we encounter
2127       # another column by that name (which would imply another table)
2128       # we unset the unqualified slot and never add any info to it
2129       # to avoid erroneous type binding. If this happens the users
2130       # only choice will be to fully qualify his column name
2131
2132       if (exists $bind_attrs->{$col}) {
2133         $bind_attrs->{$col} = {};
2134       }
2135       else {
2136         $bind_attrs->{$col} = $bind_attrs->{$fqcn};
2137       }
2138     }
2139   }
2140
2141   # Sanity check the attributes (SQLMaker does it too, but
2142   # in case of a software_limit we'll never reach there)
2143   if (defined $attrs->{offset}) {
2144     $self->throw_exception('A supplied offset attribute must be a non-negative integer')
2145       if ( $attrs->{offset} =~ /\D/ or $attrs->{offset} < 0 );
2146   }
2147   $attrs->{offset} ||= 0;
2148
2149   if (defined $attrs->{rows}) {
2150     $self->throw_exception("The rows attribute must be a positive integer if present")
2151       if ( $attrs->{rows} =~ /\D/ or $attrs->{rows} <= 0 );
2152   }
2153   elsif ($attrs->{offset}) {
2154     # MySQL actually recommends this approach.  I cringe.
2155     $attrs->{rows} = $sql_maker->__max_int;
2156   }
2157
2158   my @limit;
2159
2160   # see if we need to tear the prefetch apart otherwise delegate the limiting to the
2161   # storage, unless software limit was requested
2162   if (
2163     #limited has_many
2164     ( $attrs->{rows} && keys %{$attrs->{collapse}} )
2165        ||
2166     # grouped prefetch (to satisfy group_by == select)
2167     ( $attrs->{group_by}
2168         &&
2169       @{$attrs->{group_by}}
2170         &&
2171       $attrs->{_prefetch_selector_range}
2172     )
2173   ) {
2174     ($ident, $select, $where, $attrs)
2175       = $self->_adjust_select_args_for_complex_prefetch ($ident, $select, $where, $attrs);
2176   }
2177   elsif (! $attrs->{software_limit} ) {
2178     push @limit, $attrs->{rows}, $attrs->{offset};
2179   }
2180
2181   # try to simplify the joinmap further (prune unreferenced type-single joins)
2182   $ident = $self->_prune_unused_joins ($ident, $select, $where, $attrs);
2183
2184 ###
2185   # This would be the point to deflate anything found in $where
2186   # (and leave $attrs->{bind} intact). Problem is - inflators historically
2187   # expect a row object. And all we have is a resultsource (it is trivial
2188   # to extract deflator coderefs via $alias2source above).
2189   #
2190   # I don't see a way forward other than changing the way deflators are
2191   # invoked, and that's just bad...
2192 ###
2193
2194   return ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $where, $attrs, @limit);
2195 }
2196
2197 # Returns a counting SELECT for a simple count
2198 # query. Abstracted so that a storage could override
2199 # this to { count => 'firstcol' } or whatever makes
2200 # sense as a performance optimization
2201 sub _count_select {
2202   #my ($self, $source, $rs_attrs) = @_;
2203   return { count => '*' };
2204 }
2205
2206
2207 sub source_bind_attributes {
2208   my ($self, $source) = @_;
2209
2210   my $bind_attributes;
2211
2212   my $colinfo = $source->columns_info;
2213
2214   for my $col (keys %$colinfo) {
2215     if (my $dt = $colinfo->{$col}{data_type} ) {
2216       $bind_attributes->{$col} = $self->bind_attribute_by_data_type($dt)
2217     }
2218   }
2219
2220   return $bind_attributes;
2221 }
2222
2223 =head2 select
2224
2225 =over 4
2226
2227 =item Arguments: $ident, $select, $condition, $attrs
2228
2229 =back
2230
2231 Handle a SQL select statement.
2232
2233 =cut
2234
2235 sub select {
2236   my $self = shift;
2237   my ($ident, $select, $condition, $attrs) = @_;
2238   return $self->cursor_class->new($self, \@_, $attrs);
2239 }
2240
2241 sub select_single {
2242   my $self = shift;
2243   my ($rv, $sth, @bind) = $self->_select(@_);
2244   my @row = $sth->fetchrow_array;
2245   my @nextrow = $sth->fetchrow_array if @row;
2246   if(@row && @nextrow) {
2247     carp "Query returned more than one row.  SQL that returns multiple rows is DEPRECATED for ->find and ->single";
2248   }
2249   # Need to call finish() to work round broken DBDs
2250   $sth->finish();
2251   return @row;
2252 }
2253
2254 =head2 sql_limit_dialect
2255
2256 This is an accessor for the default SQL limit dialect used by a particular
2257 storage driver. Can be overriden by supplying an explicit L</limit_dialect>
2258 to L<DBIx::Class::Schema/connect>. For a list of available limit dialects
2259 see L<DBIx::Class::SQLMaker::LimitDialects>.
2260
2261 =head2 sth
2262
2263 =over 4
2264
2265 =item Arguments: $sql
2266
2267 =back
2268
2269 Returns a L<DBI> sth (statement handle) for the supplied SQL.
2270
2271 =cut
2272
2273 sub _dbh_sth {
2274   my ($self, $dbh, $sql) = @_;
2275
2276   # 3 is the if_active parameter which avoids active sth re-use
2277   my $sth = $self->disable_sth_caching
2278     ? $dbh->prepare($sql)
2279     : $dbh->prepare_cached($sql, {}, 3);
2280
2281   # XXX You would think RaiseError would make this impossible,
2282   #  but apparently that's not true :(
2283   $self->throw_exception($dbh->errstr) if !$sth;
2284
2285   $sth;
2286 }
2287
2288 sub sth {
2289   my ($self, $sql) = @_;
2290   $self->dbh_do('_dbh_sth', $sql);  # retry over disconnects
2291 }
2292
2293 sub _dbh_columns_info_for {
2294   my ($self, $dbh, $table) = @_;
2295
2296   if ($dbh->can('column_info')) {
2297     my %result;
2298     my $caught;
2299     try {
2300       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
2301       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
2302       $sth->execute();
2303       while ( my $info = $sth->fetchrow_hashref() ){
2304         my %column_info;
2305         $column_info{data_type}   = $info->{TYPE_NAME};
2306         $column_info{size}      = $info->{COLUMN_SIZE};
2307         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
2308         $column_info{default_value} = $info->{COLUMN_DEF};
2309         my $col_name = $info->{COLUMN_NAME};
2310         $col_name =~ s/^\"(.*)\"$/$1/;
2311
2312         $result{$col_name} = \%column_info;
2313       }
2314     } catch {
2315       $caught = 1;
2316     };
2317     return \%result if !$caught && scalar keys %result;
2318   }
2319
2320   my %result;
2321   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
2322   $sth->execute;
2323   my @columns = @{$sth->{NAME_lc}};
2324   for my $i ( 0 .. $#columns ){
2325     my %column_info;
2326     $column_info{data_type} = $sth->{TYPE}->[$i];
2327     $column_info{size} = $sth->{PRECISION}->[$i];
2328     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
2329
2330     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
2331       $column_info{data_type} = $1;
2332       $column_info{size}    = $2;
2333     }
2334
2335     $result{$columns[$i]} = \%column_info;
2336   }
2337   $sth->finish;
2338
2339   foreach my $col (keys %result) {
2340     my $colinfo = $result{$col};
2341     my $type_num = $colinfo->{data_type};
2342     my $type_name;
2343     if(defined $type_num && $dbh->can('type_info')) {
2344       my $type_info = $dbh->type_info($type_num);
2345       $type_name = $type_info->{TYPE_NAME} if $type_info;
2346       $colinfo->{data_type} = $type_name if $type_name;
2347     }
2348   }
2349
2350   return \%result;
2351 }
2352
2353 sub columns_info_for {
2354   my ($self, $table) = @_;
2355   $self->_dbh_columns_info_for ($self->_get_dbh, $table);
2356 }
2357
2358 =head2 last_insert_id
2359
2360 Return the row id of the last insert.
2361
2362 =cut
2363
2364 sub _dbh_last_insert_id {
2365     my ($self, $dbh, $source, $col) = @_;
2366
2367     my $id = try { $dbh->last_insert_id (undef, undef, $source->name, $col) };
2368
2369     return $id if defined $id;
2370
2371     my $class = ref $self;
2372     $self->throw_exception ("No storage specific _dbh_last_insert_id() method implemented in $class, and the generic DBI::last_insert_id() failed");
2373 }
2374
2375 sub last_insert_id {
2376   my $self = shift;
2377   $self->_dbh_last_insert_id ($self->_dbh, @_);
2378 }
2379
2380 =head2 _native_data_type
2381
2382 =over 4
2383
2384 =item Arguments: $type_name
2385
2386 =back
2387
2388 This API is B<EXPERIMENTAL>, will almost definitely change in the future, and
2389 currently only used by L<::AutoCast|DBIx::Class::Storage::DBI::AutoCast> and
2390 L<::Sybase::ASE|DBIx::Class::Storage::DBI::Sybase::ASE>.
2391
2392 The default implementation returns C<undef>, implement in your Storage driver if
2393 you need this functionality.
2394
2395 Should map types from other databases to the native RDBMS type, for example
2396 C<VARCHAR2> to C<VARCHAR>.
2397
2398 Types with modifiers should map to the underlying data type. For example,
2399 C<INTEGER AUTO_INCREMENT> should become C<INTEGER>.
2400
2401 Composite types should map to the container type, for example
2402 C<ENUM(foo,bar,baz)> becomes C<ENUM>.
2403
2404 =cut
2405
2406 sub _native_data_type {
2407   #my ($self, $data_type) = @_;
2408   return undef
2409 }
2410
2411 # Check if placeholders are supported at all
2412 sub _determine_supports_placeholders {
2413   my $self = shift;
2414   my $dbh  = $self->_get_dbh;
2415
2416   # some drivers provide a $dbh attribute (e.g. Sybase and $dbh->{syb_dynamic_supported})
2417   # but it is inaccurate more often than not
2418   return try {
2419     local $dbh->{PrintError} = 0;
2420     local $dbh->{RaiseError} = 1;
2421     $dbh->do('select ?', {}, 1);
2422     1;
2423   }
2424   catch {
2425     0;
2426   };
2427 }
2428
2429 # Check if placeholders bound to non-string types throw exceptions
2430 #
2431 sub _determine_supports_typeless_placeholders {
2432   my $self = shift;
2433   my $dbh  = $self->_get_dbh;
2434
2435   return try {
2436     local $dbh->{PrintError} = 0;
2437     local $dbh->{RaiseError} = 1;
2438     # this specifically tests a bind that is NOT a string
2439     $dbh->do('select 1 where 1 = ?', {}, 1);
2440     1;
2441   }
2442   catch {
2443     0;
2444   };
2445 }
2446
2447 =head2 sqlt_type
2448
2449 Returns the database driver name.
2450
2451 =cut
2452
2453 sub sqlt_type {
2454   shift->_get_dbh->{Driver}->{Name};
2455 }
2456
2457 =head2 bind_attribute_by_data_type
2458
2459 Given a datatype from column info, returns a database specific bind
2460 attribute for C<< $dbh->bind_param($val,$attribute) >> or nothing if we will
2461 let the database planner just handle it.
2462
2463 Generally only needed for special case column types, like bytea in postgres.
2464
2465 =cut
2466
2467 sub bind_attribute_by_data_type {
2468     return;
2469 }
2470
2471 =head2 is_datatype_numeric
2472
2473 Given a datatype from column_info, returns a boolean value indicating if
2474 the current RDBMS considers it a numeric value. This controls how
2475 L<DBIx::Class::Row/set_column> decides whether to mark the column as
2476 dirty - when the datatype is deemed numeric a C<< != >> comparison will
2477 be performed instead of the usual C<eq>.
2478
2479 =cut
2480
2481 sub is_datatype_numeric {
2482   my ($self, $dt) = @_;
2483
2484   return 0 unless $dt;
2485
2486   return $dt =~ /^ (?:
2487     numeric | int(?:eger)? | (?:tiny|small|medium|big)int | dec(?:imal)? | real | float | double (?: \s+ precision)? | (?:big)?serial
2488   ) $/ix;
2489 }
2490
2491
2492 =head2 create_ddl_dir
2493
2494 =over 4
2495
2496 =item Arguments: $schema \@databases, $version, $directory, $preversion, \%sqlt_args
2497
2498 =back
2499
2500 Creates a SQL file based on the Schema, for each of the specified
2501 database engines in C<\@databases> in the given directory.
2502 (note: specify L<SQL::Translator> names, not L<DBI> driver names).
2503
2504 Given a previous version number, this will also create a file containing
2505 the ALTER TABLE statements to transform the previous schema into the
2506 current one. Note that these statements may contain C<DROP TABLE> or
2507 C<DROP COLUMN> statements that can potentially destroy data.
2508
2509 The file names are created using the C<ddl_filename> method below, please
2510 override this method in your schema if you would like a different file
2511 name format. For the ALTER file, the same format is used, replacing
2512 $version in the name with "$preversion-$version".
2513
2514 See L<SQL::Translator/METHODS> for a list of values for C<\%sqlt_args>.
2515 The most common value for this would be C<< { add_drop_table => 1 } >>
2516 to have the SQL produced include a C<DROP TABLE> statement for each table
2517 created. For quoting purposes supply C<quote_table_names> and
2518 C<quote_field_names>.
2519
2520 If no arguments are passed, then the following default values are assumed:
2521
2522 =over 4
2523
2524 =item databases  - ['MySQL', 'SQLite', 'PostgreSQL']
2525
2526 =item version    - $schema->schema_version
2527
2528 =item directory  - './'
2529
2530 =item preversion - <none>
2531
2532 =back
2533
2534 By default, C<\%sqlt_args> will have
2535
2536  { add_drop_table => 1, ignore_constraint_names => 1, ignore_index_names => 1 }
2537
2538 merged with the hash passed in. To disable any of those features, pass in a
2539 hashref like the following
2540
2541  { ignore_constraint_names => 0, # ... other options }
2542
2543
2544 WARNING: You are strongly advised to check all SQL files created, before applying
2545 them.
2546
2547 =cut
2548
2549 sub create_ddl_dir {
2550   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
2551
2552   unless ($dir) {
2553     carp "No directory given, using ./\n";
2554     $dir = './';
2555   } else {
2556       -d $dir
2557         or
2558       make_path ("$dir")  # make_path does not like objects (i.e. Path::Class::Dir)
2559         or
2560       $self->throw_exception(
2561         "Failed to create '$dir': " . ($! || $@ || 'error unknow')
2562       );
2563   }
2564
2565   $self->throw_exception ("Directory '$dir' does not exist\n") unless(-d $dir);
2566
2567   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
2568   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
2569
2570   my $schema_version = $schema->schema_version || '1.x';
2571   $version ||= $schema_version;
2572
2573   $sqltargs = {
2574     add_drop_table => 1,
2575     ignore_constraint_names => 1,
2576     ignore_index_names => 1,
2577     %{$sqltargs || {}}
2578   };
2579
2580   unless (DBIx::Class::Optional::Dependencies->req_ok_for ('deploy')) {
2581     $self->throw_exception("Can't create a ddl file without " . DBIx::Class::Optional::Dependencies->req_missing_for ('deploy') );
2582   }
2583
2584   my $sqlt = SQL::Translator->new( $sqltargs );
2585
2586   $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
2587   my $sqlt_schema = $sqlt->translate({ data => $schema })
2588     or $self->throw_exception ($sqlt->error);
2589
2590   foreach my $db (@$databases) {
2591     $sqlt->reset();
2592     $sqlt->{schema} = $sqlt_schema;
2593     $sqlt->producer($db);
2594
2595     my $file;
2596     my $filename = $schema->ddl_filename($db, $version, $dir);
2597     if (-e $filename && ($version eq $schema_version )) {
2598       # if we are dumping the current version, overwrite the DDL
2599       carp "Overwriting existing DDL file - $filename";
2600       unlink($filename);
2601     }
2602
2603     my $output = $sqlt->translate;
2604     if(!$output) {
2605       carp("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
2606       next;
2607     }
2608     if(!open($file, ">$filename")) {
2609       $self->throw_exception("Can't open $filename for writing ($!)");
2610       next;
2611     }
2612     print $file $output;
2613     close($file);
2614
2615     next unless ($preversion);
2616
2617     require SQL::Translator::Diff;
2618
2619     my $prefilename = $schema->ddl_filename($db, $preversion, $dir);
2620     if(!-e $prefilename) {
2621       carp("No previous schema file found ($prefilename)");
2622       next;
2623     }
2624
2625     my $difffile = $schema->ddl_filename($db, $version, $dir, $preversion);
2626     if(-e $difffile) {
2627       carp("Overwriting existing diff file - $difffile");
2628       unlink($difffile);
2629     }
2630
2631     my $source_schema;
2632     {
2633       my $t = SQL::Translator->new($sqltargs);
2634       $t->debug( 0 );
2635       $t->trace( 0 );
2636
2637       $t->parser( $db )
2638         or $self->throw_exception ($t->error);
2639
2640       my $out = $t->translate( $prefilename )
2641         or $self->throw_exception ($t->error);
2642
2643       $source_schema = $t->schema;
2644
2645       $source_schema->name( $prefilename )
2646         unless ( $source_schema->name );
2647     }
2648
2649     # The "new" style of producers have sane normalization and can support
2650     # diffing a SQL file against a DBIC->SQLT schema. Old style ones don't
2651     # And we have to diff parsed SQL against parsed SQL.
2652     my $dest_schema = $sqlt_schema;
2653
2654     unless ( "SQL::Translator::Producer::$db"->can('preprocess_schema') ) {
2655       my $t = SQL::Translator->new($sqltargs);
2656       $t->debug( 0 );
2657       $t->trace( 0 );
2658
2659       $t->parser( $db )
2660         or $self->throw_exception ($t->error);
2661
2662       my $out = $t->translate( $filename )
2663         or $self->throw_exception ($t->error);
2664
2665       $dest_schema = $t->schema;
2666
2667       $dest_schema->name( $filename )
2668         unless $dest_schema->name;
2669     }
2670
2671     my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
2672                                                   $dest_schema,   $db,
2673                                                   $sqltargs
2674                                                  );
2675     if(!open $file, ">$difffile") {
2676       $self->throw_exception("Can't write to $difffile ($!)");
2677       next;
2678     }
2679     print $file $diff;
2680     close($file);
2681   }
2682 }
2683
2684 =head2 deployment_statements
2685
2686 =over 4
2687
2688 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
2689
2690 =back
2691
2692 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
2693
2694 The L<SQL::Translator> (not L<DBI>) database driver name can be explicitly
2695 provided in C<$type>, otherwise the result of L</sqlt_type> is used as default.
2696
2697 C<$directory> is used to return statements from files in a previously created
2698 L</create_ddl_dir> directory and is optional. The filenames are constructed
2699 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
2700
2701 If no C<$directory> is specified then the statements are constructed on the
2702 fly using L<SQL::Translator> and C<$version> is ignored.
2703
2704 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
2705
2706 =cut
2707
2708 sub deployment_statements {
2709   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
2710   $type ||= $self->sqlt_type;
2711   $version ||= $schema->schema_version || '1.x';
2712   $dir ||= './';
2713   my $filename = $schema->ddl_filename($type, $version, $dir);
2714   if(-f $filename)
2715   {
2716       my $file;
2717       open($file, "<$filename")
2718         or $self->throw_exception("Can't open $filename ($!)");
2719       my @rows = <$file>;
2720       close($file);
2721       return join('', @rows);
2722   }
2723
2724   unless (DBIx::Class::Optional::Dependencies->req_ok_for ('deploy') ) {
2725     $self->throw_exception("Can't deploy without a ddl_dir or " . DBIx::Class::Optional::Dependencies->req_missing_for ('deploy') );
2726   }
2727
2728   # sources needs to be a parser arg, but for simplicty allow at top level
2729   # coming in
2730   $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
2731       if exists $sqltargs->{sources};
2732
2733   my $tr = SQL::Translator->new(
2734     producer => "SQL::Translator::Producer::${type}",
2735     %$sqltargs,
2736     parser => 'SQL::Translator::Parser::DBIx::Class',
2737     data => $schema,
2738   );
2739
2740   my @ret;
2741   if (wantarray) {
2742     @ret = $tr->translate;
2743   }
2744   else {
2745     $ret[0] = $tr->translate;
2746   }
2747
2748   $self->throw_exception( 'Unable to produce deployment statements: ' . $tr->error)
2749     unless (@ret && defined $ret[0]);
2750
2751   return wantarray ? @ret : $ret[0];
2752 }
2753
2754 sub deploy {
2755   my ($self, $schema, $type, $sqltargs, $dir) = @_;
2756   my $deploy = sub {
2757     my $line = shift;
2758     return if($line =~ /^--/);
2759     return if(!$line);
2760     # next if($line =~ /^DROP/m);
2761     return if($line =~ /^BEGIN TRANSACTION/m);
2762     return if($line =~ /^COMMIT/m);
2763     return if $line =~ /^\s+$/; # skip whitespace only
2764     $self->_query_start($line);
2765     try {
2766       # do a dbh_do cycle here, as we need some error checking in
2767       # place (even though we will ignore errors)
2768       $self->dbh_do (sub { $_[1]->do($line) });
2769     } catch {
2770       carp qq{$_ (running "${line}")};
2771     };
2772     $self->_query_end($line);
2773   };
2774   my @statements = $schema->deployment_statements($type, undef, $dir, { %{ $sqltargs || {} }, no_comments => 1 } );
2775   if (@statements > 1) {
2776     foreach my $statement (@statements) {
2777       $deploy->( $statement );
2778     }
2779   }
2780   elsif (@statements == 1) {
2781     foreach my $line ( split(";\n", $statements[0])) {
2782       $deploy->( $line );
2783     }
2784   }
2785 }
2786
2787 =head2 datetime_parser
2788
2789 Returns the datetime parser class
2790
2791 =cut
2792
2793 sub datetime_parser {
2794   my $self = shift;
2795   return $self->{datetime_parser} ||= do {
2796     $self->build_datetime_parser(@_);
2797   };
2798 }
2799
2800 =head2 datetime_parser_type
2801
2802 Defines (returns) the datetime parser class - currently hardwired to
2803 L<DateTime::Format::MySQL>
2804
2805 =cut
2806
2807 sub datetime_parser_type { "DateTime::Format::MySQL"; }
2808
2809 =head2 build_datetime_parser
2810
2811 See L</datetime_parser>
2812
2813 =cut
2814
2815 sub build_datetime_parser {
2816   my $self = shift;
2817   my $type = $self->datetime_parser_type(@_);
2818   $self->ensure_class_loaded ($type);
2819   return $type;
2820 }
2821
2822
2823 =head2 is_replicating
2824
2825 A boolean that reports if a particular L<DBIx::Class::Storage::DBI> is set to
2826 replicate from a master database.  Default is undef, which is the result
2827 returned by databases that don't support replication.
2828
2829 =cut
2830
2831 sub is_replicating {
2832     return;
2833
2834 }
2835
2836 =head2 lag_behind_master
2837
2838 Returns a number that represents a certain amount of lag behind a master db
2839 when a given storage is replicating.  The number is database dependent, but
2840 starts at zero and increases with the amount of lag. Default in undef
2841
2842 =cut
2843
2844 sub lag_behind_master {
2845     return;
2846 }
2847
2848 =head2 relname_to_table_alias
2849
2850 =over 4
2851
2852 =item Arguments: $relname, $join_count
2853
2854 =back
2855
2856 L<DBIx::Class> uses L<DBIx::Class::Relationship> names as table aliases in
2857 queries.
2858
2859 This hook is to allow specific L<DBIx::Class::Storage> drivers to change the
2860 way these aliases are named.
2861
2862 The default behavior is C<< "$relname_$join_count" if $join_count > 1 >>,
2863 otherwise C<"$relname">.
2864
2865 =cut
2866
2867 sub relname_to_table_alias {
2868   my ($self, $relname, $join_count) = @_;
2869
2870   my $alias = ($join_count && $join_count > 1 ?
2871     join('_', $relname, $join_count) : $relname);
2872
2873   return $alias;
2874 }
2875
2876 # The size in bytes to use for DBI's ->bind_param_inout, this is the generic
2877 # version and it may be necessary to amend or override it for a specific storage
2878 # if such binds are necessary.
2879 sub _max_column_bytesize {
2880   my ($self, $source, $col) = @_;
2881
2882   my $inf = $source->column_info($col);
2883   return $inf->{_max_bytesize} ||= do {
2884
2885     my $max_size;
2886
2887     if (my $data_type = $inf->{data_type}) {
2888       $data_type = lc($data_type);
2889
2890       # String/sized-binary types
2891       if ($data_type =~ /^(?:l?(?:var)?char(?:acter)?(?:\s*varying)?
2892                              |(?:var)?binary(?:\s*varying)?|raw)\b/x
2893       ) {
2894         $max_size = $inf->{size};
2895       }
2896       # Other charset/unicode types, assume scale of 4
2897       elsif ($data_type =~ /^(?:national\s*character(?:\s*varying)?|nchar
2898                               |univarchar
2899                               |nvarchar)\b/x
2900       ) {
2901         $max_size = $inf->{size} * 4 if $inf->{size};
2902       }
2903       # Blob types
2904       elsif ($self->_is_lob_type($data_type)) {
2905         # default to longreadlen
2906       }
2907       else {
2908         $max_size = 100;  # for all other (numeric?) datatypes
2909       }
2910     }
2911
2912     $max_size ||= $self->_get_dbh->{LongReadLen} || 8000;
2913   };
2914 }
2915
2916 # Determine if a data_type is some type of BLOB
2917 sub _is_lob_type {
2918   my ($self, $data_type) = @_;
2919   $data_type && ($data_type =~ /(?:lob|bfile|text|image|bytea|memo)/i
2920     || $data_type =~ /^long(?:\s*(?:raw|bit\s*varying|varbit|binary
2921                                   |varchar|character\s*varying|nvarchar
2922                                   |national\s*character\s*varying))?$/xi);
2923 }
2924
2925 1;
2926
2927 =head1 USAGE NOTES
2928
2929 =head2 DBIx::Class and AutoCommit
2930
2931 DBIx::Class can do some wonderful magic with handling exceptions,
2932 disconnections, and transactions when you use C<< AutoCommit => 1 >>
2933 (the default) combined with C<txn_do> for transaction support.
2934
2935 If you set C<< AutoCommit => 0 >> in your connect info, then you are always
2936 in an assumed transaction between commits, and you're telling us you'd
2937 like to manage that manually.  A lot of the magic protections offered by
2938 this module will go away.  We can't protect you from exceptions due to database
2939 disconnects because we don't know anything about how to restart your
2940 transactions.  You're on your own for handling all sorts of exceptional
2941 cases if you choose the C<< AutoCommit => 0 >> path, just as you would
2942 be with raw DBI.
2943
2944
2945 =head1 AUTHORS
2946
2947 Matt S. Trout <mst@shadowcatsystems.co.uk>
2948
2949 Andy Grundman <andy@hybridized.org>
2950
2951 =head1 LICENSE
2952
2953 You may distribute this code under the same terms as Perl itself.
2954
2955 =cut