lib/DBIx/Class/Storage/DBI.pm

   1 package DBIx::Class::Storage::DBI;
   2 # -*- mode: cperl; cperl-indent-level: 2 -*-
   3
   4 use base 'DBIx::Class::Storage';
   5
   6 use strict;
   7 use warnings;
   8 use Carp::Clan qw/^DBIx::Class/;
   9 use DBI;
  10 use DBIx::Class::Storage::DBI::Cursor;
  11 use DBIx::Class::Storage::Statistics;
  12 use Scalar::Util();
  13 use List::Util();
  14
  15 __PACKAGE__->mk_group_accessors('simple' =>
  16   qw/_connect_info _dbi_connect_info _dbh _sql_maker _sql_maker_opts
  17      _conn_pid _conn_tid transaction_depth _dbh_autocommit savepoints/
  18 );
  19
  20 # the values for these accessors are picked out (and deleted) from
  21 # the attribute hashref passed to connect_info
  22 my @storage_options = qw/
  23   on_connect_call on_disconnect_call on_connect_do on_disconnect_do
  24   disable_sth_caching unsafe auto_savepoint
  25 /;
  26 __PACKAGE__->mk_group_accessors('simple' => @storage_options);
  27
  28
  29 # default cursor class, overridable in connect_info attributes
  30 __PACKAGE__->cursor_class('DBIx::Class::Storage::DBI::Cursor');
  31
  32 __PACKAGE__->mk_group_accessors('inherited' => qw/sql_maker_class/);
  33 __PACKAGE__->sql_maker_class('DBIx::Class::SQLAHacks');
  34
  35
  36 =head1 NAME
  37
  38 DBIx::Class::Storage::DBI - DBI storage handler
  39
  40 =head1 SYNOPSIS
  41
  42   my $schema = MySchema->connect('dbi:SQLite:my.db');
  43
  44   $schema->storage->debug(1);
  45   $schema->dbh_do("DROP TABLE authors");
  46
  47   $schema->resultset('Book')->search({
  48      written_on => $schema->storage->datetime_parser(DateTime->now)
  49   });
  50
  51 =head1 DESCRIPTION
  52
  53 This class represents the connection to an RDBMS via L<DBI>.  See
  54 L<DBIx::Class::Storage> for general information.  This pod only
  55 documents DBI-specific methods and behaviors.
  56
  57 =head1 METHODS
  58
  59 =cut
  60
  61 sub new {
  62   my $new = shift->next::method(@_);
  63
  64   $new->transaction_depth(0);
  65   $new->_sql_maker_opts({});
  66   $new->{savepoints} = [];
  67   $new->{_in_dbh_do} = 0;
  68   $new->{_dbh_gen} = 0;
  69
  70   $new;
  71 }
  72
  73 =head2 connect_info
  74
  75 This method is normally called by L<DBIx::Class::Schema/connection>, which
  76 encapsulates its argument list in an arrayref before passing them here.
  77
  78 The argument list may contain:
  79
  80 =over
  81
  82 =item *
  83
  84 The same 4-element argument set one would normally pass to
  85 L<DBI/connect>, optionally followed by
  86 L<extra attributes|/DBIx::Class specific connection attributes>
  87 recognized by DBIx::Class:
  88
  89   $connect_info_args = [ $dsn, $user, $password, \%dbi_attributes?, \%extra_attributes? ];
  90
  91 =item *
  92
  93 A single code reference which returns a connected
  94 L<DBI database handle|DBI/connect> optionally followed by
  95 L<extra attributes|/DBIx::Class specific connection attributes> recognized
  96 by DBIx::Class:
  97
  98   $connect_info_args = [ sub { DBI->connect (...) }, \%extra_attributes? ];
  99
 100 =item *
 101
 102 A single hashref with all the attributes and the dsn/user/password
 103 mixed together:
 104
 105   $connect_info_args = [{
 106     dsn => $dsn,
 107     user => $user,
 108     password => $pass,
 109     %dbi_attributes,
 110     %extra_attributes,
 111   }];
 112
 113 This is particularly useful for L<Catalyst> based applications, allowing the
 114 following config (L<Config::General> style):
 115
 116   <Model::DB>
 117     schema_class   App::DB
 118     <connect_info>
 119       dsn          dbi:mysql:database=test
 120       user         testuser
 121       password     TestPass
 122       AutoCommit   1
 123     </connect_info>
 124   </Model::DB>
 125
 126 =back
 127
 128 Please note that the L<DBI> docs recommend that you always explicitly
 129 set C<AutoCommit> to either I<0> or I<1>.  L<DBIx::Class> further
 130 recommends that it be set to I<1>, and that you perform transactions
 131 via our L<DBIx::Class::Schema/txn_do> method.  L<DBIx::Class> will set it
 132 to I<1> if you do not do explicitly set it to zero.  This is the default
 133 for most DBDs. See L</DBIx::Class and AutoCommit> for details.
 134
 135 =head3 DBIx::Class specific connection attributes
 136
 137 In addition to the standard L<DBI|DBI/ATTRIBUTES_COMMON_TO_ALL_HANDLES>
 138 L<connection|DBI/Database_Handle_Attributes> attributes, DBIx::Class recognizes
 139 the following connection options. These options can be mixed in with your other
 140 L<DBI> connection attributes, or placed in a seperate hashref
 141 (C<\%extra_attributes>) as shown above.
 142
 143 Every time C<connect_info> is invoked, any previous settings for
 144 these options will be cleared before setting the new ones, regardless of
 145 whether any options are specified in the new C<connect_info>.
 146
 147
 148 =over
 149
 150 =item on_connect_do
 151
 152 Specifies things to do immediately after connecting or re-connecting to
 153 the database.  Its value may contain:
 154
 155 =over
 156
 157 =item a scalar
 158
 159 This contains one SQL statement to execute.
 160
 161 =item an array reference
 162
 163 This contains SQL statements to execute in order.  Each element contains
 164 a string or a code reference that returns a string.
 165
 166 =item a code reference
 167
 168 This contains some code to execute.  Unlike code references within an
 169 array reference, its return value is ignored.
 170
 171 =back
 172
 173 =item on_disconnect_do
 174
 175 Takes arguments in the same form as L</on_connect_do> and executes them
 176 immediately before disconnecting from the database.
 177
 178 Note, this only runs if you explicitly call L</disconnect> on the
 179 storage object.
 180
 181 =item on_connect_call
 182
 183 A more generalized form of L</on_connect_do> that calls the specified
 184 C<connect_call_METHOD> methods in your storage driver.
 185
 186   on_connect_do => 'select 1'
 187
 188 is equivalent to:
 189
 190   on_connect_call => [ [ do_sql => 'select 1' ] ]
 191
 192 Its values may contain:
 193
 194 =over
 195
 196 =item a scalar
 197
 198 Will call the C<connect_call_METHOD> method.
 199
 200 =item a code reference
 201
 202 Will execute C<< $code->($storage) >>
 203
 204 =item an array reference
 205
 206 Each value can be a method name or code reference.
 207
 208 =item an array of arrays
 209
 210 For each array, the first item is taken to be the C<connect_call_> method name
 211 or code reference, and the rest are parameters to it.
 212
 213 =back
 214
 215 Some predefined storage methods you may use:
 216
 217 =over
 218
 219 =item do_sql
 220
 221 Executes a SQL string or a code reference that returns a SQL string. This is
 222 what L</on_connect_do> and L</on_disconnect_do> use.
 223
 224 It can take:
 225
 226 =over
 227
 228 =item a scalar
 229
 230 Will execute the scalar as SQL.
 231
 232 =item an arrayref
 233
 234 Taken to be arguments to L<DBI/do>, the SQL string optionally followed by the
 235 attributes hashref and bind values.
 236
 237 =item a code reference
 238
 239 Will execute C<< $code->($storage) >> and execute the return array refs as
 240 above.
 241
 242 =back
 243
 244 =item datetime_setup
 245
 246 Execute any statements necessary to initialize the database session to return
 247 and accept datetime/timestamp values used with
 248 L<DBIx::Class::InflateColumn::DateTime>.
 249
 250 Only necessary for some databases, see your specific storage driver for
 251 implementation details.
 252
 253 =back
 254
 255 =item on_disconnect_call
 256
 257 Takes arguments in the same form as L</on_connect_call> and executes them
 258 immediately before disconnecting from the database.
 259
 260 Calls the C<disconnect_call_METHOD> methods as opposed to the
 261 C<connect_call_METHOD> methods called by L</on_connect_call>.
 262
 263 Note, this only runs if you explicitly call L</disconnect> on the
 264 storage object.
 265
 266 =item disable_sth_caching
 267
 268 If set to a true value, this option will disable the caching of
 269 statement handles via L<DBI/prepare_cached>.
 270
 271 =item limit_dialect
 272
 273 Sets the limit dialect. This is useful for JDBC-bridge among others
 274 where the remote SQL-dialect cannot be determined by the name of the
 275 driver alone. See also L<SQL::Abstract::Limit>.
 276
 277 =item quote_char
 278
 279 Specifies what characters to use to quote table and column names. If
 280 you use this you will want to specify L</name_sep> as well.
 281
 282 C<quote_char> expects either a single character, in which case is it
 283 is placed on either side of the table/column name, or an arrayref of length
 284 2 in which case the table/column name is placed between the elements.
 285
 286 For example under MySQL you should use C<< quote_char => '`' >>, and for
 287 SQL Server you should use C<< quote_char => [qw/[ ]/] >>.
 288
 289 =item name_sep
 290
 291 This only needs to be used in conjunction with C<quote_char>, and is used to
 292 specify the charecter that seperates elements (schemas, tables, columns) from
 293 each other. In most cases this is simply a C<.>.
 294
 295 The consequences of not supplying this value is that L<SQL::Abstract>
 296 will assume DBIx::Class' uses of aliases to be complete column
 297 names. The output will look like I<"me.name"> when it should actually
 298 be I<"me"."name">.
 299
 300 =item unsafe
 301
 302 This Storage driver normally installs its own C<HandleError>, sets
 303 C<RaiseError> and C<ShowErrorStatement> on, and sets C<PrintError> off on
 304 all database handles, including those supplied by a coderef.  It does this
 305 so that it can have consistent and useful error behavior.
 306
 307 If you set this option to a true value, Storage will not do its usual
 308 modifications to the database handle's attributes, and instead relies on
 309 the settings in your connect_info DBI options (or the values you set in
 310 your connection coderef, in the case that you are connecting via coderef).
 311
 312 Note that your custom settings can cause Storage to malfunction,
 313 especially if you set a C<HandleError> handler that suppresses exceptions
 314 and/or disable C<RaiseError>.
 315
 316 =item auto_savepoint
 317
 318 If this option is true, L<DBIx::Class> will use savepoints when nesting
 319 transactions, making it possible to recover from failure in the inner
 320 transaction without having to abort all outer transactions.
 321
 322 =item cursor_class
 323
 324 Use this argument to supply a cursor class other than the default
 325 L<DBIx::Class::Storage::DBI::Cursor>.
 326
 327 =back
 328
 329 Some real-life examples of arguments to L</connect_info> and
 330 L<DBIx::Class::Schema/connect>
 331
 332   # Simple SQLite connection
 333   ->connect_info([ 'dbi:SQLite:./foo.db' ]);
 334
 335   # Connect via subref
 336   ->connect_info([ sub { DBI->connect(...) } ]);
 337
 338   # A bit more complicated
 339   ->connect_info(
 340     [
 341       'dbi:Pg:dbname=foo',
 342       'postgres',
 343       'my_pg_password',
 344       { AutoCommit => 1 },
 345       { quote_char => q{"}, name_sep => q{.} },
 346     ]
 347   );
 348
 349   # Equivalent to the previous example
 350   ->connect_info(
 351     [
 352       'dbi:Pg:dbname=foo',
 353       'postgres',
 354       'my_pg_password',
 355       { AutoCommit => 1, quote_char => q{"}, name_sep => q{.} },
 356     ]
 357   );
 358
 359   # Same, but with hashref as argument
 360   # See parse_connect_info for explanation
 361   ->connect_info(
 362     [{
 363       dsn         => 'dbi:Pg:dbname=foo',
 364       user        => 'postgres',
 365       password    => 'my_pg_password',
 366       AutoCommit  => 1,
 367       quote_char  => q{"},
 368       name_sep    => q{.},
 369     }]
 370   );
 371
 372   # Subref + DBIx::Class-specific connection options
 373   ->connect_info(
 374     [
 375       sub { DBI->connect(...) },
 376       {
 377           quote_char => q{`},
 378           name_sep => q{@},
 379           on_connect_do => ['SET search_path TO myschema,otherschema,public'],
 380           disable_sth_caching => 1,
 381       },
 382     ]
 383   );
 384
 385
 386
 387 =cut
 388
 389 sub connect_info {
 390   my ($self, $info_arg) = @_;
 391
 392   return $self->_connect_info if !$info_arg;
 393
 394   my @args = @$info_arg;  # take a shallow copy for further mutilation
 395   $self->_connect_info([@args]); # copy for _connect_info
 396
 397
 398   # combine/pre-parse arguments depending on invocation style
 399
 400   my %attrs;
 401   if (ref $args[0] eq 'CODE') {     # coderef with optional \%extra_attributes
 402     %attrs = %{ $args[1] || {} };
 403     @args = $args[0];
 404   }
 405   elsif (ref $args[0] eq 'HASH') { # single hashref (i.e. Catalyst config)
 406     %attrs = %{$args[0]};
 407     @args = ();
 408     for (qw/password user dsn/) {
 409       unshift @args, delete $attrs{$_};
 410     }
 411   }
 412   else {                # otherwise assume dsn/user/password + \%attrs + \%extra_attrs
 413     %attrs = (
 414       % { $args[3] || {} },
 415       % { $args[4] || {} },
 416     );
 417     @args = @args[0,1,2];
 418   }
 419
 420   # Kill sql_maker/_sql_maker_opts, so we get a fresh one with only
 421   #  the new set of options
 422   $self->_sql_maker(undef);
 423   $self->_sql_maker_opts({});
 424
 425   if(keys %attrs) {
 426     for my $storage_opt (@storage_options, 'cursor_class') {    # @storage_options is declared at the top of the module
 427       if(my $value = delete $attrs{$storage_opt}) {
 428         $self->$storage_opt($value);
 429       }
 430     }
 431     for my $sql_maker_opt (qw/limit_dialect quote_char name_sep/) {
 432       if(my $opt_val = delete $attrs{$sql_maker_opt}) {
 433         $self->_sql_maker_opts->{$sql_maker_opt} = $opt_val;
 434       }
 435     }
 436     for my $connect_do_opt (qw/on_connect_do on_disconnect_do/) {
 437       if(my $opt_val = delete $attrs{$connect_do_opt}) {
 438         $self->$connect_do_opt($opt_val);
 439       }
 440     }
 441   }
 442
 443   %attrs = () if (ref $args[0] eq 'CODE');  # _connect() never looks past $args[0] in this case
 444
 445   $self->_dbi_connect_info([@args, keys %attrs ? \%attrs : ()]);
 446   $self->_connect_info;
 447 }
 448
 449 =head2 on_connect_do
 450
 451 This method is deprecated in favour of setting via L</connect_info>.
 452
 453 =cut
 454
 455 =head2 on_disconnect_do
 456
 457 This method is deprecated in favour of setting via L</connect_info>.
 458
 459 =cut
 460
 461 sub _parse_connect_do {
 462   my ($self, $type, $val) = @_;
 463
 464   my @res;
 465
 466   if (not ref($val)) {
 467     push @res, [ 'do_sql', $val ];
 468   } elsif (ref($val) eq 'CODE') {
 469     push @res, $val;
 470   } elsif (ref($val) eq 'ARRAY') {
 471     push @res, map [ 'do_sql', $_ ], @$val;
 472   } else {
 473     $self->throw_exception("Invalid type for $type: ".ref($val));
 474   }
 475
 476   return \@res;
 477 }
 478
 479 =head2 dbh_do
 480
 481 Arguments: ($subref | $method_name), @extra_coderef_args?
 482
 483 Execute the given $subref or $method_name using the new exception-based
 484 connection management.
 485
 486 The first two arguments will be the storage object that C<dbh_do> was called
 487 on and a database handle to use.  Any additional arguments will be passed
 488 verbatim to the called subref as arguments 2 and onwards.
 489
 490 Using this (instead of $self->_dbh or $self->dbh) ensures correct
 491 exception handling and reconnection (or failover in future subclasses).
 492
 493 Your subref should have no side-effects outside of the database, as
 494 there is the potential for your subref to be partially double-executed
 495 if the database connection was stale/dysfunctional.
 496
 497 Example:
 498
 499   my @stuff = $schema->storage->dbh_do(
 500     sub {
 501       my ($storage, $dbh, @cols) = @_;
 502       my $cols = join(q{, }, @cols);
 503       $dbh->selectrow_array("SELECT $cols FROM foo");
 504     },
 505     @column_list
 506   );
 507
 508 =cut
 509
 510 sub dbh_do {
 511   my $self = shift;
 512   my $code = shift;
 513
 514   my $dbh = $self->_dbh;
 515
 516   return $self->$code($dbh, @_) if $self->{_in_dbh_do}
 517       || $self->{transaction_depth};
 518
 519   local $self->{_in_dbh_do} = 1;
 520
 521   my @result;
 522   my $want_array = wantarray;
 523
 524   eval {
 525     $self->_verify_pid if $dbh;
 526     if(!$self->_dbh) {
 527         $self->_populate_dbh;
 528         $dbh = $self->_dbh;
 529     }
 530
 531     if($want_array) {
 532         @result = $self->$code($dbh, @_);
 533     }
 534     elsif(defined $want_array) {
 535         $result[0] = $self->$code($dbh, @_);
 536     }
 537     else {
 538         $self->$code($dbh, @_);
 539     }
 540   };
 541
 542   my $exception = $@;
 543   if(!$exception) { return $want_array ? @result : $result[0] }
 544
 545   $self->throw_exception($exception) if $self->connected;
 546
 547   # We were not connected - reconnect and retry, but let any
 548   #  exception fall right through this time
 549   $self->_populate_dbh;
 550   $self->$code($self->_dbh, @_);
 551 }
 552
 553 # This is basically a blend of dbh_do above and DBIx::Class::Storage::txn_do.
 554 # It also informs dbh_do to bypass itself while under the direction of txn_do,
 555 #  via $self->{_in_dbh_do} (this saves some redundant eval and errorcheck, etc)
 556 sub txn_do {
 557   my $self = shift;
 558   my $coderef = shift;
 559
 560   ref $coderef eq 'CODE' or $self->throw_exception
 561     ('$coderef must be a CODE reference');
 562
 563   return $coderef->(@_) if $self->{transaction_depth} && ! $self->auto_savepoint;
 564
 565   local $self->{_in_dbh_do} = 1;
 566
 567   my @result;
 568   my $want_array = wantarray;
 569
 570   my $tried = 0;
 571   while(1) {
 572     eval {
 573       $self->_verify_pid if $self->_dbh;
 574       $self->_populate_dbh if !$self->_dbh;
 575
 576       $self->txn_begin;
 577       if($want_array) {
 578           @result = $coderef->(@_);
 579       }
 580       elsif(defined $want_array) {
 581           $result[0] = $coderef->(@_);
 582       }
 583       else {
 584           $coderef->(@_);
 585       }
 586       $self->txn_commit;
 587     };
 588
 589     my $exception = $@;
 590     if(!$exception) { return $want_array ? @result : $result[0] }
 591
 592     if($tried++ > 0 || $self->connected) {
 593       eval { $self->txn_rollback };
 594       my $rollback_exception = $@;
 595       if($rollback_exception) {
 596         my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
 597         $self->throw_exception($exception)  # propagate nested rollback
 598           if $rollback_exception =~ /$exception_class/;
 599
 600         $self->throw_exception(
 601           "Transaction aborted: ${exception}. "
 602           . "Rollback failed: ${rollback_exception}"
 603         );
 604       }
 605       $self->throw_exception($exception)
 606     }
 607
 608     # We were not connected, and was first try - reconnect and retry
 609     # via the while loop
 610     $self->_populate_dbh;
 611   }
 612 }
 613
 614 =head2 disconnect
 615
 616 Our C<disconnect> method also performs a rollback first if the
 617 database is not in C<AutoCommit> mode.
 618
 619 =cut
 620
 621 sub disconnect {
 622   my ($self) = @_;
 623
 624   if( $self->connected ) {
 625     if (my $connection_call = $self->on_disconnect_call) {
 626       $self->_do_connection_actions(disconnect_call_ => $connection_call)
 627     }
 628     if (my $connection_do = $self->on_disconnect_do) {
 629       $self->_do_connection_actions(
 630         disconnect_call_ => $self->_parse_connect_do(
 631           on_disconnect_do => $connection_do
 632         )
 633       )
 634     }
 635
 636     $self->_dbh->rollback unless $self->_dbh_autocommit;
 637     $self->_dbh->disconnect;
 638     $self->_dbh(undef);
 639     $self->{_dbh_gen}++;
 640   }
 641 }
 642
 643 =head2 with_deferred_fk_checks
 644
 645 =over 4
 646
 647 =item Arguments: C<$coderef>
 648
 649 =item Return Value: The return value of $coderef
 650
 651 =back
 652
 653 Storage specific method to run the code ref with FK checks deferred or
 654 in MySQL's case disabled entirely.
 655
 656 =cut
 657
 658 # Storage subclasses should override this
 659 sub with_deferred_fk_checks {
 660   my ($self, $sub) = @_;
 661
 662   $sub->();
 663 }
 664
 665 sub connected {
 666   my ($self) = @_;
 667
 668   if(my $dbh = $self->_dbh) {
 669       if(defined $self->_conn_tid && $self->_conn_tid != threads->tid) {
 670           $self->_dbh(undef);
 671           $self->{_dbh_gen}++;
 672           return;
 673       }
 674       else {
 675           $self->_verify_pid;
 676           return 0 if !$self->_dbh;
 677       }
 678       return ($dbh->FETCH('Active') && $dbh->ping);
 679   }
 680
 681   return 0;
 682 }
 683
 684 # handle pid changes correctly
 685 #  NOTE: assumes $self->_dbh is a valid $dbh
 686 sub _verify_pid {
 687   my ($self) = @_;
 688
 689   return if defined $self->_conn_pid && $self->_conn_pid == $$;
 690
 691   $self->_dbh->{InactiveDestroy} = 1;
 692   $self->_dbh(undef);
 693   $self->{_dbh_gen}++;
 694
 695   return;
 696 }
 697
 698 sub ensure_connected {
 699   my ($self) = @_;
 700
 701   unless ($self->connected) {
 702     $self->_populate_dbh;
 703   }
 704 }
 705
 706 =head2 dbh
 707
 708 Returns the dbh - a data base handle of class L<DBI>.
 709
 710 =cut
 711
 712 sub dbh {
 713   my ($self) = @_;
 714
 715   $self->ensure_connected;
 716   return $self->_dbh;
 717 }
 718
 719 sub _sql_maker_args {
 720     my ($self) = @_;
 721
 722     return ( bindtype=>'columns', array_datatypes => 1, limit_dialect => $self->dbh, %{$self->_sql_maker_opts} );
 723 }
 724
 725 sub sql_maker {
 726   my ($self) = @_;
 727   unless ($self->_sql_maker) {
 728     my $sql_maker_class = $self->sql_maker_class;
 729     $self->ensure_class_loaded ($sql_maker_class);
 730     $self->_sql_maker($sql_maker_class->new( $self->_sql_maker_args ));
 731   }
 732   return $self->_sql_maker;
 733 }
 734
 735 sub _rebless {}
 736
 737 sub _populate_dbh {
 738   my ($self) = @_;
 739   my @info = @{$self->_dbi_connect_info || []};
 740   $self->_dbh($self->_connect(@info));
 741
 742   $self->_conn_pid($$);
 743   $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
 744
 745   $self->_determine_driver;
 746
 747   # Always set the transaction depth on connect, since
 748   #  there is no transaction in progress by definition
 749   $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
 750
 751   if (my $connection_call = $self->on_connect_call) {
 752     $self->_do_connection_actions(connect_call_ => $connection_call)
 753   }
 754   if (my $connection_do = $self->on_connect_do) {
 755     $self->_do_connection_actions(
 756      connect_call_ => $self->_parse_connect_do(on_connect_do => $connection_do)
 757     )
 758   }
 759 }
 760
 761 sub _determine_driver {
 762   my ($self) = @_;
 763
 764   if (ref $self eq 'DBIx::Class::Storage::DBI') {
 765     my $driver;
 766
 767     if ($self->_dbh) { # we are connected
 768       $driver = $self->_dbh->{Driver}{Name};
 769     } else {
 770       # try to use dsn to not require being connected, the driver may still
 771       # force a connection in _rebless to determine version
 772       ($driver) = $self->_dbi_connect_info->[0] =~ /dbi:([^:]+):/i;
 773     }
 774
 775     if ($self->load_optional_class("DBIx::Class::Storage::DBI::${driver}")) {
 776       bless $self, "DBIx::Class::Storage::DBI::${driver}";
 777       $self->_rebless();
 778     }
 779   }
 780 }
 781
 782 sub _do_connection_actions {
 783   my $self          = shift;
 784   my $method_prefix = shift;
 785   my $call          = shift;
 786
 787   if (not ref($call)) {
 788     my $method = $method_prefix . $call;
 789     $self->$method(@_);
 790   } elsif (ref($call) eq 'CODE') {
 791     $self->$call(@_);
 792   } elsif (ref($call) eq 'ARRAY') {
 793     if (ref($call->[0]) ne 'ARRAY') {
 794       $self->_do_connection_actions($method_prefix, $_) for @$call;
 795     } else {
 796       $self->_do_connection_actions($method_prefix, @$_) for @$call;
 797     }
 798   } else {
 799     $self->throw_exception (sprintf ("Don't know how to process conection actions of type '%s'", ref($call)) );
 800   }
 801
 802   return $self;
 803 }
 804
 805 sub connect_call_do_sql {
 806   my $self = shift;
 807   $self->_do_query(@_);
 808 }
 809
 810 sub disconnect_call_do_sql {
 811   my $self = shift;
 812   $self->_do_query(@_);
 813 }
 814
 815 # override in db-specific backend when necessary
 816 sub connect_call_datetime_setup { 1 }
 817
 818 sub _do_query {
 819   my ($self, $action) = @_;
 820
 821   if (ref $action eq 'CODE') {
 822     $action = $action->($self);
 823     $self->_do_query($_) foreach @$action;
 824   }
 825   else {
 826     # Most debuggers expect ($sql, @bind), so we need to exclude
 827     # the attribute hash which is the second argument to $dbh->do
 828     # furthermore the bind values are usually to be presented
 829     # as named arrayref pairs, so wrap those here too
 830     my @do_args = (ref $action eq 'ARRAY') ? (@$action) : ($action);
 831     my $sql = shift @do_args;
 832     my $attrs = shift @do_args;
 833     my @bind = map { [ undef, $_ ] } @do_args;
 834
 835     $self->_query_start($sql, @bind);
 836     $self->_dbh->do($sql, $attrs, @do_args);
 837     $self->_query_end($sql, @bind);
 838   }
 839
 840   return $self;
 841 }
 842
 843 sub _connect {
 844   my ($self, @info) = @_;
 845
 846   $self->throw_exception("You failed to provide any connection info")
 847     if !@info;
 848
 849   my ($old_connect_via, $dbh);
 850
 851   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
 852     $old_connect_via = $DBI::connect_via;
 853     $DBI::connect_via = 'connect';
 854   }
 855
 856   eval {
 857     if(ref $info[0] eq 'CODE') {
 858        $dbh = &{$info[0]}
 859     }
 860     else {
 861        $dbh = DBI->connect(@info);
 862     }
 863
 864     if($dbh && !$self->unsafe) {
 865       my $weak_self = $self;
 866       Scalar::Util::weaken($weak_self);
 867       $dbh->{HandleError} = sub {
 868           if ($weak_self) {
 869             $weak_self->throw_exception("DBI Exception: $_[0]");
 870           }
 871           else {
 872             croak ("DBI Exception: $_[0]");
 873           }
 874       };
 875       $dbh->{ShowErrorStatement} = 1;
 876       $dbh->{RaiseError} = 1;
 877       $dbh->{PrintError} = 0;
 878     }
 879   };
 880
 881   $DBI::connect_via = $old_connect_via if $old_connect_via;
 882
 883   $self->throw_exception("DBI Connection failed: " . ($@||$DBI::errstr))
 884     if !$dbh || $@;
 885
 886   $self->_dbh_autocommit($dbh->{AutoCommit});
 887
 888   $dbh;
 889 }
 890
 891 sub svp_begin {
 892   my ($self, $name) = @_;
 893
 894   $name = $self->_svp_generate_name
 895     unless defined $name;
 896
 897   $self->throw_exception ("You can't use savepoints outside a transaction")
 898     if $self->{transaction_depth} == 0;
 899
 900   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
 901     unless $self->can('_svp_begin');
 902
 903   push @{ $self->{savepoints} }, $name;
 904
 905   $self->debugobj->svp_begin($name) if $self->debug;
 906
 907   return $self->_svp_begin($name);
 908 }
 909
 910 sub svp_release {
 911   my ($self, $name) = @_;
 912
 913   $self->throw_exception ("You can't use savepoints outside a transaction")
 914     if $self->{transaction_depth} == 0;
 915
 916   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
 917     unless $self->can('_svp_release');
 918
 919   if (defined $name) {
 920     $self->throw_exception ("Savepoint '$name' does not exist")
 921       unless grep { $_ eq $name } @{ $self->{savepoints} };
 922
 923     # Dig through the stack until we find the one we are releasing.  This keeps
 924     # the stack up to date.
 925     my $svp;
 926
 927     do { $svp = pop @{ $self->{savepoints} } } while $svp ne $name;
 928   } else {
 929     $name = pop @{ $self->{savepoints} };
 930   }
 931
 932   $self->debugobj->svp_release($name) if $self->debug;
 933
 934   return $self->_svp_release($name);
 935 }
 936
 937 sub svp_rollback {
 938   my ($self, $name) = @_;
 939
 940   $self->throw_exception ("You can't use savepoints outside a transaction")
 941     if $self->{transaction_depth} == 0;
 942
 943   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
 944     unless $self->can('_svp_rollback');
 945
 946   if (defined $name) {
 947       # If they passed us a name, verify that it exists in the stack
 948       unless(grep({ $_ eq $name } @{ $self->{savepoints} })) {
 949           $self->throw_exception("Savepoint '$name' does not exist!");
 950       }
 951
 952       # Dig through the stack until we find the one we are releasing.  This keeps
 953       # the stack up to date.
 954       while(my $s = pop(@{ $self->{savepoints} })) {
 955           last if($s eq $name);
 956       }
 957       # Add the savepoint back to the stack, as a rollback doesn't remove the
 958       # named savepoint, only everything after it.
 959       push(@{ $self->{savepoints} }, $name);
 960   } else {
 961       # We'll assume they want to rollback to the last savepoint
 962       $name = $self->{savepoints}->[-1];
 963   }
 964
 965   $self->debugobj->svp_rollback($name) if $self->debug;
 966
 967   return $self->_svp_rollback($name);
 968 }
 969
 970 sub _svp_generate_name {
 971     my ($self) = @_;
 972
 973     return 'savepoint_'.scalar(@{ $self->{'savepoints'} });
 974 }
 975
 976 sub txn_begin {
 977   my $self = shift;
 978   $self->ensure_connected();
 979   if($self->{transaction_depth} == 0) {
 980     $self->debugobj->txn_begin()
 981       if $self->debug;
 982     # this isn't ->_dbh-> because
 983     #  we should reconnect on begin_work
 984     #  for AutoCommit users
 985     $self->dbh->begin_work;
 986   } elsif ($self->auto_savepoint) {
 987     $self->svp_begin;
 988   }
 989   $self->{transaction_depth}++;
 990 }
 991
 992 sub txn_commit {
 993   my $self = shift;
 994   if ($self->{transaction_depth} == 1) {
 995     my $dbh = $self->_dbh;
 996     $self->debugobj->txn_commit()
 997       if ($self->debug);
 998     $dbh->commit;
 999     $self->{transaction_depth} = 0
1000       if $self->_dbh_autocommit;
1001   }
1002   elsif($self->{transaction_depth} > 1) {
1003     $self->{transaction_depth}--;
1004     $self->svp_release
1005       if $self->auto_savepoint;
1006   }
1007 }
1008
1009 sub txn_rollback {
1010   my $self = shift;
1011   my $dbh = $self->_dbh;
1012   eval {
1013     if ($self->{transaction_depth} == 1) {
1014       $self->debugobj->txn_rollback()
1015         if ($self->debug);
1016       $self->{transaction_depth} = 0
1017         if $self->_dbh_autocommit;
1018       $dbh->rollback;
1019     }
1020     elsif($self->{transaction_depth} > 1) {
1021       $self->{transaction_depth}--;
1022       if ($self->auto_savepoint) {
1023         $self->svp_rollback;
1024         $self->svp_release;
1025       }
1026     }
1027     else {
1028       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
1029     }
1030   };
1031   if ($@) {
1032     my $error = $@;
1033     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
1034     $error =~ /$exception_class/ and $self->throw_exception($error);
1035     # ensure that a failed rollback resets the transaction depth
1036     $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
1037     $self->throw_exception($error);
1038   }
1039 }
1040
1041 # This used to be the top-half of _execute.  It was split out to make it
1042 #  easier to override in NoBindVars without duping the rest.  It takes up
1043 #  all of _execute's args, and emits $sql, @bind.
1044 sub _prep_for_execute {
1045   my ($self, $op, $extra_bind, $ident, $args) = @_;
1046
1047   if( Scalar::Util::blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1048     $ident = $ident->from();
1049   }
1050
1051   my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
1052
1053   unshift(@bind,
1054     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
1055       if $extra_bind;
1056   return ($sql, \@bind);
1057 }
1058
1059
1060 sub _fix_bind_params {
1061     my ($self, @bind) = @_;
1062
1063     ### Turn @bind from something like this:
1064     ###   ( [ "artist", 1 ], [ "cdid", 1, 3 ] )
1065     ### to this:
1066     ###   ( "'1'", "'1'", "'3'" )
1067     return
1068         map {
1069             if ( defined( $_ && $_->[1] ) ) {
1070                 map { qq{'$_'}; } @{$_}[ 1 .. $#$_ ];
1071             }
1072             else { q{'NULL'}; }
1073         } @bind;
1074 }
1075
1076 sub _query_start {
1077     my ( $self, $sql, @bind ) = @_;
1078
1079     if ( $self->debug ) {
1080         @bind = $self->_fix_bind_params(@bind);
1081
1082         $self->debugobj->query_start( $sql, @bind );
1083     }
1084 }
1085
1086 sub _query_end {
1087     my ( $self, $sql, @bind ) = @_;
1088
1089     if ( $self->debug ) {
1090         @bind = $self->_fix_bind_params(@bind);
1091         $self->debugobj->query_end( $sql, @bind );
1092     }
1093 }
1094
1095 sub _dbh_execute {
1096   my ($self, $dbh, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
1097
1098   my ($sql, $bind) = $self->_prep_for_execute($op, $extra_bind, $ident, \@args);
1099
1100   $self->_query_start( $sql, @$bind );
1101
1102   my $sth = $self->sth($sql,$op);
1103
1104   my $placeholder_index = 1;
1105
1106   foreach my $bound (@$bind) {
1107     my $attributes = {};
1108     my($column_name, @data) = @$bound;
1109
1110     if ($bind_attributes) {
1111       $attributes = $bind_attributes->{$column_name}
1112       if defined $bind_attributes->{$column_name};
1113     }
1114
1115     foreach my $data (@data) {
1116       my $ref = ref $data;
1117       $data = $ref && $ref ne 'ARRAY' ? ''.$data : $data; # stringify args (except arrayrefs)
1118
1119       $sth->bind_param($placeholder_index, $data, $attributes);
1120       $placeholder_index++;
1121     }
1122   }
1123
1124   # Can this fail without throwing an exception anyways???
1125   my $rv = $sth->execute();
1126   $self->throw_exception($sth->errstr) if !$rv;
1127
1128   $self->_query_end( $sql, @$bind );
1129
1130   return (wantarray ? ($rv, $sth, @$bind) : $rv);
1131 }
1132
1133 sub _execute {
1134     my $self = shift;
1135     $self->dbh_do('_dbh_execute', @_)
1136 }
1137
1138 sub insert {
1139   my ($self, $source, $to_insert) = @_;
1140
1141   my $ident = $source->from;
1142   my $bind_attributes = $self->source_bind_attributes($source);
1143
1144   my $updated_cols = {};
1145
1146   $self->ensure_connected;
1147   foreach my $col ( $source->columns ) {
1148     if ( !defined $to_insert->{$col} ) {
1149       my $col_info = $source->column_info($col);
1150
1151       if ( $col_info->{auto_nextval} ) {
1152         $updated_cols->{$col} = $to_insert->{$col} = $self->_sequence_fetch( 'nextval', $col_info->{sequence} || $self->_dbh_get_autoinc_seq($self->dbh, $source) );
1153       }
1154     }
1155   }
1156
1157   $self->_execute('insert' => [], $source, $bind_attributes, $to_insert);
1158
1159   return $updated_cols;
1160 }
1161
1162 ## Still not quite perfect, and EXPERIMENTAL
1163 ## Currently it is assumed that all values passed will be "normal", i.e. not
1164 ## scalar refs, or at least, all the same type as the first set, the statement is
1165 ## only prepped once.
1166 sub insert_bulk {
1167   my ($self, $source, $cols, $data) = @_;
1168   my %colvalues;
1169   my $table = $source->from;
1170   @colvalues{@$cols} = (0..$#$cols);
1171   my ($sql, @bind) = $self->sql_maker->insert($table, \%colvalues);
1172
1173   $self->_query_start( $sql, @bind );
1174   my $sth = $self->sth($sql);
1175
1176 #  @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
1177
1178   ## This must be an arrayref, else nothing works!
1179   my $tuple_status = [];
1180
1181   ## Get the bind_attributes, if any exist
1182   my $bind_attributes = $self->source_bind_attributes($source);
1183
1184   ## Bind the values and execute
1185   my $placeholder_index = 1;
1186
1187   foreach my $bound (@bind) {
1188
1189     my $attributes = {};
1190     my ($column_name, $data_index) = @$bound;
1191
1192     if( $bind_attributes ) {
1193       $attributes = $bind_attributes->{$column_name}
1194       if defined $bind_attributes->{$column_name};
1195     }
1196
1197     my @data = map { $_->[$data_index] } @$data;
1198
1199     $sth->bind_param_array( $placeholder_index, [@data], $attributes );
1200     $placeholder_index++;
1201   }
1202   my $rv = eval { $sth->execute_array({ArrayTupleStatus => $tuple_status}) };
1203   if (my $err = $@) {
1204     my $i = 0;
1205     ++$i while $i <= $#$tuple_status && !ref $tuple_status->[$i];
1206
1207     $self->throw_exception($sth->errstr || "Unexpected populate error: $err")
1208       if ($i > $#$tuple_status);
1209
1210     require Data::Dumper;
1211     local $Data::Dumper::Terse = 1;
1212     local $Data::Dumper::Indent = 1;
1213     local $Data::Dumper::Useqq = 1;
1214     local $Data::Dumper::Quotekeys = 0;
1215
1216     $self->throw_exception(sprintf "%s for populate slice:\n%s",
1217       $tuple_status->[$i][1],
1218       Data::Dumper::Dumper(
1219         { map { $cols->[$_] => $data->[$i][$_] } (0 .. $#$cols) }
1220       ),
1221     );
1222   }
1223   $self->throw_exception($sth->errstr) if !$rv;
1224
1225   $self->_query_end( $sql, @bind );
1226   return (wantarray ? ($rv, $sth, @bind) : $rv);
1227 }
1228
1229 sub update {
1230   my $self = shift @_;
1231   my $source = shift @_;
1232   my $bind_attributes = $self->source_bind_attributes($source);
1233
1234   return $self->_execute('update' => [], $source, $bind_attributes, @_);
1235 }
1236
1237
1238 sub delete {
1239   my $self = shift @_;
1240   my $source = shift @_;
1241
1242   my $bind_attrs = $self->source_bind_attributes($source);
1243
1244   return $self->_execute('delete' => [], $source, $bind_attrs, @_);
1245 }
1246
1247 # We were sent here because the $rs contains a complex search
1248 # which will require a subquery to select the correct rows
1249 # (i.e. joined or limited resultsets)
1250 #
1251 # Genarating a single PK column subquery is trivial and supported
1252 # by all RDBMS. However if we have a multicolumn PK, things get ugly.
1253 # Look at _multipk_update_delete()
1254 sub _subq_update_delete {
1255   my $self = shift;
1256   my ($rs, $op, $values) = @_;
1257
1258   my $rsrc = $rs->result_source;
1259
1260   # we already check this, but double check naively just in case. Should be removed soon
1261   my $sel = $rs->_resolved_attrs->{select};
1262   $sel = [ $sel ] unless ref $sel eq 'ARRAY';
1263   my @pcols = $rsrc->primary_columns;
1264   if (@$sel != @pcols) {
1265     $self->throw_exception (
1266       'Subquery update/delete can not be called on resultsets selecting a'
1267      .' number of columns different than the number of primary keys'
1268     );
1269   }
1270
1271   if (@pcols == 1) {
1272     return $self->$op (
1273       $rsrc,
1274       $op eq 'update' ? $values : (),
1275       { $pcols[0] => { -in => $rs->as_query } },
1276     );
1277   }
1278
1279   else {
1280     return $self->_multipk_update_delete (@_);
1281   }
1282 }
1283
1284 # ANSI SQL does not provide a reliable way to perform a multicol-PK
1285 # resultset update/delete involving subqueries. So by default resort
1286 # to simple (and inefficient) delete_all style per-row opearations,
1287 # while allowing specific storages to override this with a faster
1288 # implementation.
1289 #
1290 sub _multipk_update_delete {
1291   return shift->_per_row_update_delete (@_);
1292 }
1293
1294 # This is the default loop used to delete/update rows for multi PK
1295 # resultsets, and used by mysql exclusively (because it can't do anything
1296 # else).
1297 #
1298 # We do not use $row->$op style queries, because resultset update/delete
1299 # is not expected to cascade (this is what delete_all/update_all is for).
1300 #
1301 # There should be no race conditions as the entire operation is rolled
1302 # in a transaction.
1303 #
1304 sub _per_row_update_delete {
1305   my $self = shift;
1306   my ($rs, $op, $values) = @_;
1307
1308   my $rsrc = $rs->result_source;
1309   my @pcols = $rsrc->primary_columns;
1310
1311   my $guard = $self->txn_scope_guard;
1312
1313   # emulate the return value of $sth->execute for non-selects
1314   my $row_cnt = '0E0';
1315
1316   my $subrs_cur = $rs->cursor;
1317   while (my @pks = $subrs_cur->next) {
1318
1319     my $cond;
1320     for my $i (0.. $#pcols) {
1321       $cond->{$pcols[$i]} = $pks[$i];
1322     }
1323
1324     $self->$op (
1325       $rsrc,
1326       $op eq 'update' ? $values : (),
1327       $cond,
1328     );
1329
1330     $row_cnt++;
1331   }
1332
1333   $guard->commit;
1334
1335   return $row_cnt;
1336 }
1337
1338 sub _select {
1339   my $self = shift;
1340
1341   # localization is neccessary as
1342   # 1) there is no infrastructure to pass this around (easy to do, but will wait)
1343   # 2) _select_args sets it and _prep_for_execute consumes it
1344   my $sql_maker = $self->sql_maker;
1345   local $sql_maker->{for};
1346
1347   return $self->_execute($self->_select_args(@_));
1348 }
1349
1350 sub _select_args_to_query {
1351   my $self = shift;
1352
1353   # localization is neccessary as
1354   # 1) there is no infrastructure to pass this around (easy to do, but will wait)
1355   # 2) _select_args sets it and _prep_for_execute consumes it
1356   my $sql_maker = $self->sql_maker;
1357   local $sql_maker->{for};
1358
1359   # my ($op, $bind, $ident, $bind_attrs, $select, $cond, $order, $rows, $offset)
1360   #  = $self->_select_args($ident, $select, $cond, $attrs);
1361   my ($op, $bind, $ident, $bind_attrs, @args) =
1362     $self->_select_args(@_);
1363
1364   # my ($sql, $prepared_bind) = $self->_prep_for_execute($op, $bind, $ident, [ $select, $cond, $order, $rows, $offset ]);
1365   my ($sql, $prepared_bind) = $self->_prep_for_execute($op, $bind, $ident, \@args);
1366   $prepared_bind ||= [];
1367
1368   return wantarray
1369     ? ($sql, $prepared_bind, $bind_attrs)
1370     : \[ "($sql)", @$prepared_bind ]
1371   ;
1372 }
1373
1374 sub _select_args {
1375   my ($self, $ident, $select, $where, $attrs) = @_;
1376
1377   my $sql_maker = $self->sql_maker;
1378   my $alias2source = $self->_resolve_ident_sources ($ident);
1379
1380   # calculate bind_attrs before possible $ident mangling
1381   my $bind_attrs = {};
1382   for my $alias (keys %$alias2source) {
1383     my $bindtypes = $self->source_bind_attributes ($alias2source->{$alias}) || {};
1384     for my $col (keys %$bindtypes) {
1385
1386       my $fqcn = join ('.', $alias, $col);
1387       $bind_attrs->{$fqcn} = $bindtypes->{$col} if $bindtypes->{$col};
1388
1389       # so that unqualified searches can be bound too
1390       $bind_attrs->{$col} = $bind_attrs->{$fqcn} if $alias eq 'me';
1391     }
1392   }
1393
1394   my @limit;
1395   if ($attrs->{software_limit} ||
1396       $sql_maker->_default_limit_syntax eq "GenericSubQ") {
1397         $attrs->{software_limit} = 1;
1398   } else {
1399     $self->throw_exception("rows attribute must be positive if present")
1400       if (defined($attrs->{rows}) && !($attrs->{rows} > 0));
1401
1402     # MySQL actually recommends this approach.  I cringe.
1403     $attrs->{rows} = 2**48 if not defined $attrs->{rows} and defined $attrs->{offset};
1404
1405     if ($attrs->{rows} && keys %{$attrs->{collapse}}) {
1406       ($ident, $select, $where, $attrs)
1407         = $self->_adjust_select_args_for_limited_prefetch ($ident, $select, $where, $attrs);
1408     }
1409     else {
1410       push @limit, $attrs->{rows}, $attrs->{offset};
1411     }
1412   }
1413
1414 ###
1415   # This would be the point to deflate anything found in $where
1416   # (and leave $attrs->{bind} intact). Problem is - inflators historically
1417   # expect a row object. And all we have is a resultsource (it is trivial
1418   # to extract deflator coderefs via $alias2source above).
1419   #
1420   # I don't see a way forward other than changing the way deflators are
1421   # invoked, and that's just bad...
1422 ###
1423
1424   my $order = { map
1425     { $attrs->{$_} ? ( $_ => $attrs->{$_} ) : ()  }
1426     (qw/order_by group_by having _virtual_order_by/ )
1427   };
1428
1429
1430   $sql_maker->{for} = delete $attrs->{for};
1431
1432   return ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $where, $order, @limit);
1433 }
1434
1435 sub _adjust_select_args_for_limited_prefetch {
1436   my ($self, $from, $select, $where, $attrs) = @_;
1437
1438   if ($attrs->{group_by} and @{$attrs->{group_by}}) {
1439     $self->throw_exception ('Prefetch with limit (rows/offset) is not supported on resultsets with a group_by attribute');
1440   }
1441
1442   $self->throw_exception ('Prefetch with limit (rows/offset) is not supported on resultsets with a custom from attribute')
1443     if (ref $from ne 'ARRAY');
1444
1445
1446   # separate attributes
1447   my $sub_attrs = { %$attrs };
1448   delete $attrs->{$_} for qw/where bind rows offset/;
1449   delete $sub_attrs->{$_} for qw/for collapse select order_by/;
1450
1451   my $alias = $attrs->{alias};
1452
1453   # create subquery select list
1454   my $sub_select = [ grep { $_ =~ /^$alias\./ } @{$attrs->{select}} ];
1455
1456   # bring over all non-collapse-induced order_by into the inner query (if any)
1457   # the outer one will have to keep them all
1458   if (my $ord_cnt = @{$attrs->{order_by}} - @{$attrs->{_collapse_order_by}} ) {
1459     $sub_attrs->{order_by} = [
1460       @{$attrs->{order_by}}[ 0 .. ($#{$attrs->{order_by}} - $ord_cnt - 1) ]
1461     ];
1462   }
1463
1464   # mangle {from}
1465   $from = [ @$from ];
1466   my $select_root = shift @$from;
1467   my @outer_from = @$from;
1468
1469   my %inner_joins;
1470   my %join_info = map { $_->[0]{-alias} => $_->[0] } (@$from);
1471
1472   # in complex search_related chains $alias may *not* be 'me'
1473   # so always include it in the inner join, and also shift away
1474   # from the outer stack, so that the two datasets actually do
1475   # meet
1476   if ($select_root->{-alias} ne $alias) {
1477     $inner_joins{$alias} = 1;
1478
1479     while (@outer_from && $outer_from[0][0]{-alias} ne $alias) {
1480       shift @outer_from;
1481     }
1482     if (! @outer_from) {
1483       $self->throw_exception ("Unable to find '$alias' in the {from} stack, something is wrong");
1484     }
1485
1486     shift @outer_from; # the new subquery will represent this alias, so get rid of it
1487   }
1488
1489
1490   # decide which parts of the join will remain on the inside
1491   #
1492   # this is not a very viable optimisation, but it was written
1493   # before I realised this, so might as well remain. We can throw
1494   # away _any_ branches of the join tree that are:
1495   # 1) not mentioned in the condition/order
1496   # 2) left-join leaves (or left-join leaf chains)
1497   # Most of the join ocnditions will not satisfy this, but for real
1498   # complex queries some might, and we might make some RDBMS happy.
1499   #
1500   #
1501   # since we do not have introspectable SQLA, we fall back to ugly
1502   # scanning of raw SQL for WHERE, and for pieces of ORDER BY
1503   # in order to determine what goes into %inner_joins
1504   # It may not be very efficient, but it's a reasonable stop-gap
1505   {
1506     # produce stuff unquoted, so it can be scanned
1507     my $sql_maker = $self->sql_maker;
1508     local $sql_maker->{quote_char};
1509
1510     my @order_by = (map
1511       { ref $_ ? $_->[0] : $_ }
1512       $sql_maker->_order_by_chunks ($sub_attrs->{order_by})
1513     );
1514
1515     my $where_sql = $sql_maker->where ($where);
1516
1517     # sort needed joins
1518     for my $alias (keys %join_info) {
1519
1520       # any table alias found on a column name in where or order_by
1521       # gets included in %inner_joins
1522       # Also any parent joins that are needed to reach this particular alias
1523       for my $piece ($where_sql, @order_by ) {
1524         if ($piece =~ /\b$alias\./) {
1525           $inner_joins{$alias} = 1;
1526         }
1527       }
1528     }
1529   }
1530
1531   # scan for non-leaf/non-left joins and mark as needed
1532   # also mark all ancestor joins that are needed to reach this particular alias
1533   # (e.g.  join => { cds => 'tracks' } - tracks will bring cds too )
1534   #
1535   # traverse by the size of the -join_path i.e. reverse depth first
1536   for my $alias (sort { @{$join_info{$b}{-join_path}} <=> @{$join_info{$a}{-join_path}} } (keys %join_info) ) {
1537
1538     my $j = $join_info{$alias};
1539     $inner_joins{$alias} = 1 if (! $j->{-join_type} || ($j->{-join_type} !~ /^left$/i) );
1540
1541     if ($inner_joins{$alias}) {
1542       $inner_joins{$_} = 1 for (@{$j->{-join_path}});
1543     }
1544   }
1545
1546   # construct the inner $from for the subquery
1547   my $inner_from = [ $select_root ];
1548   for my $j (@$from) {
1549     push @$inner_from, $j if $inner_joins{$j->[0]{-alias}};
1550   }
1551
1552   # if a multi-type join was needed in the subquery ("multi" is indicated by
1553   # presence in {collapse}) - add a group_by to simulate the collapse in the subq
1554
1555   for my $alias (keys %inner_joins) {
1556
1557     # the dot comes from some weirdness in collapse
1558     # remove after the rewrite
1559     if ($attrs->{collapse}{".$alias"}) {
1560       $sub_attrs->{group_by} = $sub_select;
1561       last;
1562     }
1563   }
1564
1565   # generate the subquery
1566   my $subq = $self->_select_args_to_query (
1567     $inner_from,
1568     $sub_select,
1569     $where,
1570     $sub_attrs
1571   );
1572
1573   # put it in the new {from}
1574   unshift @outer_from, { $alias => $subq };
1575
1576   # This is totally horrific - the $where ends up in both the inner and outer query
1577   # Unfortunately not much can be done until SQLA2 introspection arrives
1578   #
1579   # OTOH it can be seen as a plus: <ash> (notes that this query would make a DBA cry ;)
1580   return (\@outer_from, $select, $where, $attrs);
1581 }
1582
1583 sub _resolve_ident_sources {
1584   my ($self, $ident) = @_;
1585
1586   my $alias2source = {};
1587
1588   # the reason this is so contrived is that $ident may be a {from}
1589   # structure, specifying multiple tables to join
1590   if ( Scalar::Util::blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1591     # this is compat mode for insert/update/delete which do not deal with aliases
1592     $alias2source->{me} = $ident;
1593   }
1594   elsif (ref $ident eq 'ARRAY') {
1595
1596     for (@$ident) {
1597       my $tabinfo;
1598       if (ref $_ eq 'HASH') {
1599         $tabinfo = $_;
1600       }
1601       if (ref $_ eq 'ARRAY' and ref $_->[0] eq 'HASH') {
1602         $tabinfo = $_->[0];
1603       }
1604
1605       $alias2source->{$tabinfo->{-alias}} = $tabinfo->{-source_handle}->resolve
1606         if ($tabinfo->{-source_handle});
1607     }
1608   }
1609
1610   return $alias2source;
1611 }
1612
1613 # Returns a counting SELECT for a simple count
1614 # query. Abstracted so that a storage could override
1615 # this to { count => 'firstcol' } or whatever makes
1616 # sense as a performance optimization
1617 sub _count_select {
1618   #my ($self, $source, $rs_attrs) = @_;
1619   return { count => '*' };
1620 }
1621
1622 # Returns a SELECT which will end up in the subselect
1623 # There may or may not be a group_by, as the subquery
1624 # might have been called to accomodate a limit
1625 #
1626 # Most databases would be happy with whatever ends up
1627 # here, but some choke in various ways.
1628 #
1629 sub _subq_count_select {
1630   my ($self, $source, $rs_attrs) = @_;
1631   return $rs_attrs->{group_by} if $rs_attrs->{group_by};
1632
1633   my @pcols = map { join '.', $rs_attrs->{alias}, $_ } ($source->primary_columns);
1634   return @pcols ? \@pcols : [ 1 ];
1635 }
1636
1637
1638 sub source_bind_attributes {
1639   my ($self, $source) = @_;
1640
1641   my $bind_attributes;
1642   foreach my $column ($source->columns) {
1643
1644     my $data_type = $source->column_info($column)->{data_type} || '';
1645     $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
1646      if $data_type;
1647   }
1648
1649   return $bind_attributes;
1650 }
1651
1652 =head2 select
1653
1654 =over 4
1655
1656 =item Arguments: $ident, $select, $condition, $attrs
1657
1658 =back
1659
1660 Handle a SQL select statement.
1661
1662 =cut
1663
1664 sub select {
1665   my $self = shift;
1666   my ($ident, $select, $condition, $attrs) = @_;
1667   return $self->cursor_class->new($self, \@_, $attrs);
1668 }
1669
1670 sub select_single {
1671   my $self = shift;
1672   my ($rv, $sth, @bind) = $self->_select(@_);
1673   my @row = $sth->fetchrow_array;
1674   my @nextrow = $sth->fetchrow_array if @row;
1675   if(@row && @nextrow) {
1676     carp "Query returned more than one row.  SQL that returns multiple rows is DEPRECATED for ->find and ->single";
1677   }
1678   # Need to call finish() to work round broken DBDs
1679   $sth->finish();
1680   return @row;
1681 }
1682
1683 =head2 sth
1684
1685 =over 4
1686
1687 =item Arguments: $sql
1688
1689 =back
1690
1691 Returns a L<DBI> sth (statement handle) for the supplied SQL.
1692
1693 =cut
1694
1695 sub _dbh_sth {
1696   my ($self, $dbh, $sql) = @_;
1697
1698   # 3 is the if_active parameter which avoids active sth re-use
1699   my $sth = $self->disable_sth_caching
1700     ? $dbh->prepare($sql)
1701     : $dbh->prepare_cached($sql, {}, 3);
1702
1703   # XXX You would think RaiseError would make this impossible,
1704   #  but apparently that's not true :(
1705   $self->throw_exception($dbh->errstr) if !$sth;
1706
1707   $sth;
1708 }
1709
1710 sub sth {
1711   my ($self, $sql) = @_;
1712   $self->dbh_do('_dbh_sth', $sql);
1713 }
1714
1715 sub _dbh_columns_info_for {
1716   my ($self, $dbh, $table) = @_;
1717
1718   if ($dbh->can('column_info')) {
1719     my %result;
1720     eval {
1721       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
1722       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
1723       $sth->execute();
1724       while ( my $info = $sth->fetchrow_hashref() ){
1725         my %column_info;
1726         $column_info{data_type}   = $info->{TYPE_NAME};
1727         $column_info{size}      = $info->{COLUMN_SIZE};
1728         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
1729         $column_info{default_value} = $info->{COLUMN_DEF};
1730         my $col_name = $info->{COLUMN_NAME};
1731         $col_name =~ s/^\"(.*)\"$/$1/;
1732
1733         $result{$col_name} = \%column_info;
1734       }
1735     };
1736     return \%result if !$@ && scalar keys %result;
1737   }
1738
1739   my %result;
1740   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
1741   $sth->execute;
1742   my @columns = @{$sth->{NAME_lc}};
1743   for my $i ( 0 .. $#columns ){
1744     my %column_info;
1745     $column_info{data_type} = $sth->{TYPE}->[$i];
1746     $column_info{size} = $sth->{PRECISION}->[$i];
1747     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
1748
1749     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
1750       $column_info{data_type} = $1;
1751       $column_info{size}    = $2;
1752     }
1753
1754     $result{$columns[$i]} = \%column_info;
1755   }
1756   $sth->finish;
1757
1758   foreach my $col (keys %result) {
1759     my $colinfo = $result{$col};
1760     my $type_num = $colinfo->{data_type};
1761     my $type_name;
1762     if(defined $type_num && $dbh->can('type_info')) {
1763       my $type_info = $dbh->type_info($type_num);
1764       $type_name = $type_info->{TYPE_NAME} if $type_info;
1765       $colinfo->{data_type} = $type_name if $type_name;
1766     }
1767   }
1768
1769   return \%result;
1770 }
1771
1772 sub columns_info_for {
1773   my ($self, $table) = @_;
1774   $self->dbh_do('_dbh_columns_info_for', $table);
1775 }
1776
1777 =head2 last_insert_id
1778
1779 Return the row id of the last insert.
1780
1781 =cut
1782
1783 sub _dbh_last_insert_id {
1784     # All Storage's need to register their own _dbh_last_insert_id
1785     # the old SQLite-based method was highly inappropriate
1786
1787     my $self = shift;
1788     my $class = ref $self;
1789     $self->throw_exception (<<EOE);
1790
1791 No _dbh_last_insert_id() method found in $class.
1792 Since the method of obtaining the autoincrement id of the last insert
1793 operation varies greatly between different databases, this method must be
1794 individually implemented for every storage class.
1795 EOE
1796 }
1797
1798 sub last_insert_id {
1799   my $self = shift;
1800   $self->dbh_do('_dbh_last_insert_id', @_);
1801 }
1802
1803 =head2 sqlt_type
1804
1805 Returns the database driver name.
1806
1807 =cut
1808
1809 sub sqlt_type { shift->dbh->{Driver}->{Name} }
1810
1811 =head2 bind_attribute_by_data_type
1812
1813 Given a datatype from column info, returns a database specific bind
1814 attribute for C<< $dbh->bind_param($val,$attribute) >> or nothing if we will
1815 let the database planner just handle it.
1816
1817 Generally only needed for special case column types, like bytea in postgres.
1818
1819 =cut
1820
1821 sub bind_attribute_by_data_type {
1822     return;
1823 }
1824
1825 =head2 is_datatype_numeric
1826
1827 Given a datatype from column_info, returns a boolean value indicating if
1828 the current RDBMS considers it a numeric value. This controls how
1829 L<DBIx::Class::Row/set_column> decides whether to mark the column as
1830 dirty - when the datatype is deemed numeric a C<< != >> comparison will
1831 be performed instead of the usual C<eq>.
1832
1833 =cut
1834
1835 sub is_datatype_numeric {
1836   my ($self, $dt) = @_;
1837
1838   return 0 unless $dt;
1839
1840   return $dt =~ /^ (?:
1841     numeric | int(?:eger)? | (?:tiny|small|medium|big)int | dec(?:imal)? | real | float | double (?: \s+ precision)? | (?:big)?serial
1842   ) $/ix;
1843 }
1844
1845
1846 =head2 create_ddl_dir (EXPERIMENTAL)
1847
1848 =over 4
1849
1850 =item Arguments: $schema \@databases, $version, $directory, $preversion, \%sqlt_args
1851
1852 =back
1853
1854 Creates a SQL file based on the Schema, for each of the specified
1855 database engines in C<\@databases> in the given directory.
1856 (note: specify L<SQL::Translator> names, not L<DBI> driver names).
1857
1858 Given a previous version number, this will also create a file containing
1859 the ALTER TABLE statements to transform the previous schema into the
1860 current one. Note that these statements may contain C<DROP TABLE> or
1861 C<DROP COLUMN> statements that can potentially destroy data.
1862
1863 The file names are created using the C<ddl_filename> method below, please
1864 override this method in your schema if you would like a different file
1865 name format. For the ALTER file, the same format is used, replacing
1866 $version in the name with "$preversion-$version".
1867
1868 See L<SQL::Translator/METHODS> for a list of values for C<\%sqlt_args>.
1869 The most common value for this would be C<< { add_drop_table => 1 } >>
1870 to have the SQL produced include a C<DROP TABLE> statement for each table
1871 created. For quoting purposes supply C<quote_table_names> and
1872 C<quote_field_names>.
1873
1874 If no arguments are passed, then the following default values are assumed:
1875
1876 =over 4
1877
1878 =item databases  - ['MySQL', 'SQLite', 'PostgreSQL']
1879
1880 =item version    - $schema->schema_version
1881
1882 =item directory  - './'
1883
1884 =item preversion - <none>
1885
1886 =back
1887
1888 By default, C<\%sqlt_args> will have
1889
1890  { add_drop_table => 1, ignore_constraint_names => 1, ignore_index_names => 1 }
1891
1892 merged with the hash passed in. To disable any of those features, pass in a
1893 hashref like the following
1894
1895  { ignore_constraint_names => 0, # ... other options }
1896
1897
1898 Note that this feature is currently EXPERIMENTAL and may not work correctly
1899 across all databases, or fully handle complex relationships.
1900
1901 WARNING: Please check all SQL files created, before applying them.
1902
1903 =cut
1904
1905 sub create_ddl_dir {
1906   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
1907
1908   if(!$dir || !-d $dir) {
1909     carp "No directory given, using ./\n";
1910     $dir = "./";
1911   }
1912   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
1913   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
1914
1915   my $schema_version = $schema->schema_version || '1.x';
1916   $version ||= $schema_version;
1917
1918   $sqltargs = {
1919     add_drop_table => 1,
1920     ignore_constraint_names => 1,
1921     ignore_index_names => 1,
1922     %{$sqltargs || {}}
1923   };
1924
1925   $self->throw_exception(q{Can't create a ddl file without SQL::Translator 0.09003: '}
1926       . $self->_check_sqlt_message . q{'})
1927           if !$self->_check_sqlt_version;
1928
1929   my $sqlt = SQL::Translator->new( $sqltargs );
1930
1931   $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
1932   my $sqlt_schema = $sqlt->translate({ data => $schema })
1933     or $self->throw_exception ($sqlt->error);
1934
1935   foreach my $db (@$databases) {
1936     $sqlt->reset();
1937     $sqlt->{schema} = $sqlt_schema;
1938     $sqlt->producer($db);
1939
1940     my $file;
1941     my $filename = $schema->ddl_filename($db, $version, $dir);
1942     if (-e $filename && ($version eq $schema_version )) {
1943       # if we are dumping the current version, overwrite the DDL
1944       carp "Overwriting existing DDL file - $filename";
1945       unlink($filename);
1946     }
1947
1948     my $output = $sqlt->translate;
1949     if(!$output) {
1950       carp("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
1951       next;
1952     }
1953     if(!open($file, ">$filename")) {
1954       $self->throw_exception("Can't open $filename for writing ($!)");
1955       next;
1956     }
1957     print $file $output;
1958     close($file);
1959
1960     next unless ($preversion);
1961
1962     require SQL::Translator::Diff;
1963
1964     my $prefilename = $schema->ddl_filename($db, $preversion, $dir);
1965     if(!-e $prefilename) {
1966       carp("No previous schema file found ($prefilename)");
1967       next;
1968     }
1969
1970     my $difffile = $schema->ddl_filename($db, $version, $dir, $preversion);
1971     if(-e $difffile) {
1972       carp("Overwriting existing diff file - $difffile");
1973       unlink($difffile);
1974     }
1975
1976     my $source_schema;
1977     {
1978       my $t = SQL::Translator->new($sqltargs);
1979       $t->debug( 0 );
1980       $t->trace( 0 );
1981
1982       $t->parser( $db )
1983         or $self->throw_exception ($t->error);
1984
1985       my $out = $t->translate( $prefilename )
1986         or $self->throw_exception ($t->error);
1987
1988       $source_schema = $t->schema;
1989
1990       $source_schema->name( $prefilename )
1991         unless ( $source_schema->name );
1992     }
1993
1994     # The "new" style of producers have sane normalization and can support
1995     # diffing a SQL file against a DBIC->SQLT schema. Old style ones don't
1996     # And we have to diff parsed SQL against parsed SQL.
1997     my $dest_schema = $sqlt_schema;
1998
1999     unless ( "SQL::Translator::Producer::$db"->can('preprocess_schema') ) {
2000       my $t = SQL::Translator->new($sqltargs);
2001       $t->debug( 0 );
2002       $t->trace( 0 );
2003
2004       $t->parser( $db )
2005         or $self->throw_exception ($t->error);
2006
2007       my $out = $t->translate( $filename )
2008         or $self->throw_exception ($t->error);
2009
2010       $dest_schema = $t->schema;
2011
2012       $dest_schema->name( $filename )
2013         unless $dest_schema->name;
2014     }
2015
2016     my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
2017                                                   $dest_schema,   $db,
2018                                                   $sqltargs
2019                                                  );
2020     if(!open $file, ">$difffile") {
2021       $self->throw_exception("Can't write to $difffile ($!)");
2022       next;
2023     }
2024     print $file $diff;
2025     close($file);
2026   }
2027 }
2028
2029 =head2 deployment_statements
2030
2031 =over 4
2032
2033 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
2034
2035 =back
2036
2037 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
2038
2039 The L<SQL::Translator> (not L<DBI>) database driver name can be explicitly
2040 provided in C<$type>, otherwise the result of L</sqlt_type> is used as default.
2041
2042 C<$directory> is used to return statements from files in a previously created
2043 L</create_ddl_dir> directory and is optional. The filenames are constructed
2044 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
2045
2046 If no C<$directory> is specified then the statements are constructed on the
2047 fly using L<SQL::Translator> and C<$version> is ignored.
2048
2049 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
2050
2051 =cut
2052
2053 sub deployment_statements {
2054   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
2055   # Need to be connected to get the correct sqlt_type
2056   $self->ensure_connected() unless $type;
2057   $type ||= $self->sqlt_type;
2058   $version ||= $schema->schema_version || '1.x';
2059   $dir ||= './';
2060   my $filename = $schema->ddl_filename($type, $version, $dir);
2061   if(-f $filename)
2062   {
2063       my $file;
2064       open($file, "<$filename")
2065         or $self->throw_exception("Can't open $filename ($!)");
2066       my @rows = <$file>;
2067       close($file);
2068       return join('', @rows);
2069   }
2070
2071   $self->throw_exception(q{Can't deploy without SQL::Translator 0.09003: '}
2072       . $self->_check_sqlt_message . q{'})
2073           if !$self->_check_sqlt_version;
2074
2075   require SQL::Translator::Parser::DBIx::Class;
2076   eval qq{use SQL::Translator::Producer::${type}};
2077   $self->throw_exception($@) if $@;
2078
2079   # sources needs to be a parser arg, but for simplicty allow at top level
2080   # coming in
2081   $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
2082       if exists $sqltargs->{sources};
2083
2084   my $tr = SQL::Translator->new(%$sqltargs);
2085   SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
2086   return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
2087 }
2088
2089 sub deploy {
2090   my ($self, $schema, $type, $sqltargs, $dir) = @_;
2091   my $deploy = sub {
2092     my $line = shift;
2093     return if($line =~ /^--/);
2094     return if(!$line);
2095     # next if($line =~ /^DROP/m);
2096     return if($line =~ /^BEGIN TRANSACTION/m);
2097     return if($line =~ /^COMMIT/m);
2098     return if $line =~ /^\s+$/; # skip whitespace only
2099     $self->_query_start($line);
2100     eval {
2101       $self->dbh->do($line); # shouldn't be using ->dbh ?
2102     };
2103     if ($@) {
2104       carp qq{$@ (running "${line}")};
2105     }
2106     $self->_query_end($line);
2107   };
2108   my @statements = $self->deployment_statements($schema, $type, undef, $dir, { %{ $sqltargs || {} }, no_comments => 1 } );
2109   if (@statements > 1) {
2110     foreach my $statement (@statements) {
2111       $deploy->( $statement );
2112     }
2113   }
2114   elsif (@statements == 1) {
2115     foreach my $line ( split(";\n", $statements[0])) {
2116       $deploy->( $line );
2117     }
2118   }
2119 }
2120
2121 =head2 datetime_parser
2122
2123 Returns the datetime parser class
2124
2125 =cut
2126
2127 sub datetime_parser {
2128   my $self = shift;
2129   return $self->{datetime_parser} ||= do {
2130     $self->ensure_connected;
2131     $self->build_datetime_parser(@_);
2132   };
2133 }
2134
2135 =head2 datetime_parser_type
2136
2137 Defines (returns) the datetime parser class - currently hardwired to
2138 L<DateTime::Format::MySQL>
2139
2140 =cut
2141
2142 sub datetime_parser_type { "DateTime::Format::MySQL"; }
2143
2144 =head2 build_datetime_parser
2145
2146 See L</datetime_parser>
2147
2148 =cut
2149
2150 sub build_datetime_parser {
2151   my $self = shift;
2152   my $type = $self->datetime_parser_type(@_);
2153   eval "use ${type}";
2154   $self->throw_exception("Couldn't load ${type}: $@") if $@;
2155   return $type;
2156 }
2157
2158 {
2159     my $_check_sqlt_version; # private
2160     my $_check_sqlt_message; # private
2161     sub _check_sqlt_version {
2162         return $_check_sqlt_version if defined $_check_sqlt_version;
2163         eval 'use SQL::Translator "0.09003"';
2164         $_check_sqlt_message = $@ || '';
2165         $_check_sqlt_version = !$@;
2166     }
2167
2168     sub _check_sqlt_message {
2169         _check_sqlt_version if !defined $_check_sqlt_message;
2170         $_check_sqlt_message;
2171     }
2172 }
2173
2174 =head2 is_replicating
2175
2176 A boolean that reports if a particular L<DBIx::Class::Storage::DBI> is set to
2177 replicate from a master database.  Default is undef, which is the result
2178 returned by databases that don't support replication.
2179
2180 =cut
2181
2182 sub is_replicating {
2183     return;
2184
2185 }
2186
2187 =head2 lag_behind_master
2188
2189 Returns a number that represents a certain amount of lag behind a master db
2190 when a given storage is replicating.  The number is database dependent, but
2191 starts at zero and increases with the amount of lag. Default in undef
2192
2193 =cut
2194
2195 sub lag_behind_master {
2196     return;
2197 }
2198
2199 sub DESTROY {
2200   my $self = shift;
2201   return if !$self->_dbh;
2202   $self->_verify_pid;
2203   $self->_dbh(undef);
2204 }
2205
2206 1;
2207
2208 =head1 USAGE NOTES
2209
2210 =head2 DBIx::Class and AutoCommit
2211
2212 DBIx::Class can do some wonderful magic with handling exceptions,
2213 disconnections, and transactions when you use C<< AutoCommit => 1 >>
2214 combined with C<txn_do> for transaction support.
2215
2216 If you set C<< AutoCommit => 0 >> in your connect info, then you are always
2217 in an assumed transaction between commits, and you're telling us you'd
2218 like to manage that manually.  A lot of the magic protections offered by
2219 this module will go away.  We can't protect you from exceptions due to database
2220 disconnects because we don't know anything about how to restart your
2221 transactions.  You're on your own for handling all sorts of exceptional
2222 cases if you choose the C<< AutoCommit => 0 >> path, just as you would
2223 be with raw DBI.
2224
2225
2226
2227 =head1 AUTHORS
2228
2229 Matt S. Trout <mst@shadowcatsystems.co.uk>
2230
2231 Andy Grundman <andy@hybridized.org>
2232
2233 =head1 LICENSE
2234
2235 You may distribute this code under the same terms as Perl itself.
2236
2237 =cut