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 _on_connect_do
  18        _on_disconnect_do savepoints/
  19 );
  20
  21 # the values for these accessors are picked out (and deleted) from
  22 # the attribute hashref passed to connect_info
  23 my @storage_options = qw/
  24   on_connect_call on_disconnect_call 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 sub on_connect_do {
 456   my $self = shift;
 457   $self->_setup_connect_do(on_connect_do => @_);
 458 }
 459
 460 =head2 on_disconnect_do
 461
 462 This method is deprecated in favour of setting via L</connect_info>.
 463
 464 =cut
 465
 466 sub on_disconnect_do {
 467   my $self = shift;
 468   $self->_setup_connect_do(on_disconnect_do => @_);
 469 }
 470
 471 sub _setup_connect_do {
 472   my ($self, $opt) = (shift, shift);
 473
 474   my $accessor = "_$opt";
 475
 476   return $self->$accessor if not @_;
 477
 478   my $val = shift;
 479
 480   $self->throw_exception("The value of $opt cannot be 'undef'")
 481     unless defined $val;
 482
 483   my @store;
 484
 485   if (not ref($val)) {
 486     push @store, [ 'do_sql', $val ];
 487   } elsif (ref($val) eq 'CODE') {
 488     push @store, $val;
 489   } elsif (ref($val) eq 'ARRAY') {
 490     push @store, map [ 'do_sql', $_ ], @$val;
 491   } else {
 492     $self->throw_exception("Invalid type for $opt ".ref($val));
 493   }
 494
 495   $self->$accessor(\@store);
 496 }
 497
 498 =head2 dbh_do
 499
 500 Arguments: ($subref | $method_name), @extra_coderef_args?
 501
 502 Execute the given $subref or $method_name using the new exception-based
 503 connection management.
 504
 505 The first two arguments will be the storage object that C<dbh_do> was called
 506 on and a database handle to use.  Any additional arguments will be passed
 507 verbatim to the called subref as arguments 2 and onwards.
 508
 509 Using this (instead of $self->_dbh or $self->dbh) ensures correct
 510 exception handling and reconnection (or failover in future subclasses).
 511
 512 Your subref should have no side-effects outside of the database, as
 513 there is the potential for your subref to be partially double-executed
 514 if the database connection was stale/dysfunctional.
 515
 516 Example:
 517
 518   my @stuff = $schema->storage->dbh_do(
 519     sub {
 520       my ($storage, $dbh, @cols) = @_;
 521       my $cols = join(q{, }, @cols);
 522       $dbh->selectrow_array("SELECT $cols FROM foo");
 523     },
 524     @column_list
 525   );
 526
 527 =cut
 528
 529 sub dbh_do {
 530   my $self = shift;
 531   my $code = shift;
 532
 533   my $dbh = $self->_dbh;
 534
 535   return $self->$code($dbh, @_) if $self->{_in_dbh_do}
 536       || $self->{transaction_depth};
 537
 538   local $self->{_in_dbh_do} = 1;
 539
 540   my @result;
 541   my $want_array = wantarray;
 542
 543   eval {
 544     $self->_verify_pid if $dbh;
 545     if(!$self->_dbh) {
 546         $self->_populate_dbh;
 547         $dbh = $self->_dbh;
 548     }
 549
 550     if($want_array) {
 551         @result = $self->$code($dbh, @_);
 552     }
 553     elsif(defined $want_array) {
 554         $result[0] = $self->$code($dbh, @_);
 555     }
 556     else {
 557         $self->$code($dbh, @_);
 558     }
 559   };
 560
 561   my $exception = $@;
 562   if(!$exception) { return $want_array ? @result : $result[0] }
 563
 564   $self->throw_exception($exception) if $self->connected;
 565
 566   # We were not connected - reconnect and retry, but let any
 567   #  exception fall right through this time
 568   $self->_populate_dbh;
 569   $self->$code($self->_dbh, @_);
 570 }
 571
 572 # This is basically a blend of dbh_do above and DBIx::Class::Storage::txn_do.
 573 # It also informs dbh_do to bypass itself while under the direction of txn_do,
 574 #  via $self->{_in_dbh_do} (this saves some redundant eval and errorcheck, etc)
 575 sub txn_do {
 576   my $self = shift;
 577   my $coderef = shift;
 578
 579   ref $coderef eq 'CODE' or $self->throw_exception
 580     ('$coderef must be a CODE reference');
 581
 582   return $coderef->(@_) if $self->{transaction_depth} && ! $self->auto_savepoint;
 583
 584   local $self->{_in_dbh_do} = 1;
 585
 586   my @result;
 587   my $want_array = wantarray;
 588
 589   my $tried = 0;
 590   while(1) {
 591     eval {
 592       $self->_verify_pid if $self->_dbh;
 593       $self->_populate_dbh if !$self->_dbh;
 594
 595       $self->txn_begin;
 596       if($want_array) {
 597           @result = $coderef->(@_);
 598       }
 599       elsif(defined $want_array) {
 600           $result[0] = $coderef->(@_);
 601       }
 602       else {
 603           $coderef->(@_);
 604       }
 605       $self->txn_commit;
 606     };
 607
 608     my $exception = $@;
 609     if(!$exception) { return $want_array ? @result : $result[0] }
 610
 611     if($tried++ > 0 || $self->connected) {
 612       eval { $self->txn_rollback };
 613       my $rollback_exception = $@;
 614       if($rollback_exception) {
 615         my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
 616         $self->throw_exception($exception)  # propagate nested rollback
 617           if $rollback_exception =~ /$exception_class/;
 618
 619         $self->throw_exception(
 620           "Transaction aborted: ${exception}. "
 621           . "Rollback failed: ${rollback_exception}"
 622         );
 623       }
 624       $self->throw_exception($exception)
 625     }
 626
 627     # We were not connected, and was first try - reconnect and retry
 628     # via the while loop
 629     $self->_populate_dbh;
 630   }
 631 }
 632
 633 =head2 disconnect
 634
 635 Our C<disconnect> method also performs a rollback first if the
 636 database is not in C<AutoCommit> mode.
 637
 638 =cut
 639
 640 sub disconnect {
 641   my ($self) = @_;
 642
 643   if( $self->connected ) {
 644     if (my $connection_call = $self->on_disconnect_call) {
 645       $self->_do_connection_actions(disconnect_call_ => $connection_call)
 646     }
 647     if (my $connection_do   = $self->_on_disconnect_do) {
 648       $self->_do_connection_actions(disconnect_call_ => $connection_do)
 649     }
 650
 651     $self->_dbh->rollback unless $self->_dbh_autocommit;
 652     $self->_dbh->disconnect;
 653     $self->_dbh(undef);
 654     $self->{_dbh_gen}++;
 655   }
 656 }
 657
 658 =head2 with_deferred_fk_checks
 659
 660 =over 4
 661
 662 =item Arguments: C<$coderef>
 663
 664 =item Return Value: The return value of $coderef
 665
 666 =back
 667
 668 Storage specific method to run the code ref with FK checks deferred or
 669 in MySQL's case disabled entirely.
 670
 671 =cut
 672
 673 # Storage subclasses should override this
 674 sub with_deferred_fk_checks {
 675   my ($self, $sub) = @_;
 676
 677   $sub->();
 678 }
 679
 680 sub connected {
 681   my ($self) = @_;
 682
 683   if(my $dbh = $self->_dbh) {
 684       if(defined $self->_conn_tid && $self->_conn_tid != threads->tid) {
 685           $self->_dbh(undef);
 686           $self->{_dbh_gen}++;
 687           return;
 688       }
 689       else {
 690           $self->_verify_pid;
 691           return 0 if !$self->_dbh;
 692       }
 693       return ($dbh->FETCH('Active') && $dbh->ping);
 694   }
 695
 696   return 0;
 697 }
 698
 699 # handle pid changes correctly
 700 #  NOTE: assumes $self->_dbh is a valid $dbh
 701 sub _verify_pid {
 702   my ($self) = @_;
 703
 704   return if defined $self->_conn_pid && $self->_conn_pid == $$;
 705
 706   $self->_dbh->{InactiveDestroy} = 1;
 707   $self->_dbh(undef);
 708   $self->{_dbh_gen}++;
 709
 710   return;
 711 }
 712
 713 sub ensure_connected {
 714   my ($self) = @_;
 715
 716   unless ($self->connected) {
 717     $self->_populate_dbh;
 718   }
 719 }
 720
 721 =head2 dbh
 722
 723 Returns the dbh - a data base handle of class L<DBI>.
 724
 725 =cut
 726
 727 sub dbh {
 728   my ($self) = @_;
 729
 730   $self->ensure_connected;
 731   return $self->_dbh;
 732 }
 733
 734 sub _sql_maker_args {
 735     my ($self) = @_;
 736
 737     return ( bindtype=>'columns', array_datatypes => 1, limit_dialect => $self->dbh, %{$self->_sql_maker_opts} );
 738 }
 739
 740 sub sql_maker {
 741   my ($self) = @_;
 742   unless ($self->_sql_maker) {
 743     my $sql_maker_class = $self->sql_maker_class;
 744     $self->ensure_class_loaded ($sql_maker_class);
 745     $self->_sql_maker($sql_maker_class->new( $self->_sql_maker_args ));
 746   }
 747   return $self->_sql_maker;
 748 }
 749
 750 sub _rebless {}
 751
 752 sub _populate_dbh {
 753   my ($self) = @_;
 754   my @info = @{$self->_dbi_connect_info || []};
 755   $self->_dbh($self->_connect(@info));
 756
 757   $self->_conn_pid($$);
 758   $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
 759
 760   $self->_determine_driver;
 761
 762   # Always set the transaction depth on connect, since
 763   #  there is no transaction in progress by definition
 764   $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
 765
 766   if (my $connection_call = $self->on_connect_call) {
 767     $self->_do_connection_actions(connect_call_ => $connection_call)
 768   }
 769   if (my $connection_do = $self->_on_connect_do) {
 770     $self->_do_connection_actions(connect_call_ => $connection_do)
 771   }
 772 }
 773
 774 sub _determine_driver {
 775   my ($self) = @_;
 776
 777   if (ref $self eq 'DBIx::Class::Storage::DBI') {
 778     my $driver;
 779
 780     if ($self->_dbh) { # we are connected
 781       $driver = $self->_dbh->{Driver}{Name};
 782     } else {
 783       # try to use dsn to not require being connected, the driver may still
 784       # force a connection in _rebless to determine version
 785       ($driver) = $self->_dbi_connect_info->[0] =~ /dbi:([^:]+):/i;
 786     }
 787
 788     if ($self->load_optional_class("DBIx::Class::Storage::DBI::${driver}")) {
 789       bless $self, "DBIx::Class::Storage::DBI::${driver}";
 790       $self->_rebless();
 791     }
 792   }
 793 }
 794
 795 sub _do_connection_actions {
 796   my $self          = shift;
 797   my $method_prefix = shift;
 798   my $call          = shift;
 799
 800   if (not ref($call)) {
 801     my $method = $method_prefix . $call;
 802     $self->$method(@_);
 803   } elsif (ref($call) eq 'CODE') {
 804     $self->$call(@_);
 805   } elsif (ref($call) eq 'ARRAY') {
 806     if (ref($call->[0]) ne 'ARRAY') {
 807       $self->_do_connection_actions($method_prefix, $_) for @$call;
 808     } else {
 809       $self->_do_connection_actions($method_prefix, @$_) for @$call;
 810     }
 811   } else {
 812     $self->throw_exception (sprintf ("Don't know how to process conection actions of type '%s'", ref($call)) );
 813   }
 814
 815   return $self;
 816 }
 817
 818 sub connect_call_do_sql {
 819   my $self = shift;
 820   $self->_do_query(@_);
 821 }
 822
 823 sub disconnect_call_do_sql {
 824   my $self = shift;
 825   $self->_do_query(@_);
 826 }
 827
 828 # override in db-specific backend when necessary
 829 sub connect_call_datetime_setup { 1 }
 830
 831 sub _do_query {
 832   my ($self, $action) = @_;
 833
 834   if (ref $action eq 'CODE') {
 835     $action = $action->($self);
 836     $self->_do_query($_) foreach @$action;
 837   }
 838   else {
 839     # Most debuggers expect ($sql, @bind), so we need to exclude
 840     # the attribute hash which is the second argument to $dbh->do
 841     # furthermore the bind values are usually to be presented
 842     # as named arrayref pairs, so wrap those here too
 843     my @do_args = (ref $action eq 'ARRAY') ? (@$action) : ($action);
 844     my $sql = shift @do_args;
 845     my $attrs = shift @do_args;
 846     my @bind = map { [ undef, $_ ] } @do_args;
 847
 848     $self->_query_start($sql, @bind);
 849     $self->_dbh->do($sql, $attrs, @do_args);
 850     $self->_query_end($sql, @bind);
 851   }
 852
 853   return $self;
 854 }
 855
 856 sub _connect {
 857   my ($self, @info) = @_;
 858
 859   $self->throw_exception("You failed to provide any connection info")
 860     if !@info;
 861
 862   my ($old_connect_via, $dbh);
 863
 864   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
 865     $old_connect_via = $DBI::connect_via;
 866     $DBI::connect_via = 'connect';
 867   }
 868
 869   eval {
 870     if(ref $info[0] eq 'CODE') {
 871        $dbh = &{$info[0]}
 872     }
 873     else {
 874        $dbh = DBI->connect(@info);
 875     }
 876
 877     if($dbh && !$self->unsafe) {
 878       my $weak_self = $self;
 879       Scalar::Util::weaken($weak_self);
 880       $dbh->{HandleError} = sub {
 881           if ($weak_self) {
 882             $weak_self->throw_exception("DBI Exception: $_[0]");
 883           }
 884           else {
 885             croak ("DBI Exception: $_[0]");
 886           }
 887       };
 888       $dbh->{ShowErrorStatement} = 1;
 889       $dbh->{RaiseError} = 1;
 890       $dbh->{PrintError} = 0;
 891     }
 892   };
 893
 894   $DBI::connect_via = $old_connect_via if $old_connect_via;
 895
 896   $self->throw_exception("DBI Connection failed: " . ($@||$DBI::errstr))
 897     if !$dbh || $@;
 898
 899   $self->_dbh_autocommit($dbh->{AutoCommit});
 900
 901   $dbh;
 902 }
 903
 904 sub svp_begin {
 905   my ($self, $name) = @_;
 906
 907   $name = $self->_svp_generate_name
 908     unless defined $name;
 909
 910   $self->throw_exception ("You can't use savepoints outside a transaction")
 911     if $self->{transaction_depth} == 0;
 912
 913   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
 914     unless $self->can('_svp_begin');
 915
 916   push @{ $self->{savepoints} }, $name;
 917
 918   $self->debugobj->svp_begin($name) if $self->debug;
 919
 920   return $self->_svp_begin($name);
 921 }
 922
 923 sub svp_release {
 924   my ($self, $name) = @_;
 925
 926   $self->throw_exception ("You can't use savepoints outside a transaction")
 927     if $self->{transaction_depth} == 0;
 928
 929   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
 930     unless $self->can('_svp_release');
 931
 932   if (defined $name) {
 933     $self->throw_exception ("Savepoint '$name' does not exist")
 934       unless grep { $_ eq $name } @{ $self->{savepoints} };
 935
 936     # Dig through the stack until we find the one we are releasing.  This keeps
 937     # the stack up to date.
 938     my $svp;
 939
 940     do { $svp = pop @{ $self->{savepoints} } } while $svp ne $name;
 941   } else {
 942     $name = pop @{ $self->{savepoints} };
 943   }
 944
 945   $self->debugobj->svp_release($name) if $self->debug;
 946
 947   return $self->_svp_release($name);
 948 }
 949
 950 sub svp_rollback {
 951   my ($self, $name) = @_;
 952
 953   $self->throw_exception ("You can't use savepoints outside a transaction")
 954     if $self->{transaction_depth} == 0;
 955
 956   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
 957     unless $self->can('_svp_rollback');
 958
 959   if (defined $name) {
 960       # If they passed us a name, verify that it exists in the stack
 961       unless(grep({ $_ eq $name } @{ $self->{savepoints} })) {
 962           $self->throw_exception("Savepoint '$name' does not exist!");
 963       }
 964
 965       # Dig through the stack until we find the one we are releasing.  This keeps
 966       # the stack up to date.
 967       while(my $s = pop(@{ $self->{savepoints} })) {
 968           last if($s eq $name);
 969       }
 970       # Add the savepoint back to the stack, as a rollback doesn't remove the
 971       # named savepoint, only everything after it.
 972       push(@{ $self->{savepoints} }, $name);
 973   } else {
 974       # We'll assume they want to rollback to the last savepoint
 975       $name = $self->{savepoints}->[-1];
 976   }
 977
 978   $self->debugobj->svp_rollback($name) if $self->debug;
 979
 980   return $self->_svp_rollback($name);
 981 }
 982
 983 sub _svp_generate_name {
 984     my ($self) = @_;
 985
 986     return 'savepoint_'.scalar(@{ $self->{'savepoints'} });
 987 }
 988
 989 sub txn_begin {
 990   my $self = shift;
 991   $self->ensure_connected();
 992   if($self->{transaction_depth} == 0) {
 993     $self->debugobj->txn_begin()
 994       if $self->debug;
 995     # this isn't ->_dbh-> because
 996     #  we should reconnect on begin_work
 997     #  for AutoCommit users
 998     $self->dbh->begin_work;
 999   } elsif ($self->auto_savepoint) {
1000     $self->svp_begin;
1001   }
1002   $self->{transaction_depth}++;
1003 }
1004
1005 sub txn_commit {
1006   my $self = shift;
1007   if ($self->{transaction_depth} == 1) {
1008     my $dbh = $self->_dbh;
1009     $self->debugobj->txn_commit()
1010       if ($self->debug);
1011     $dbh->commit;
1012     $self->{transaction_depth} = 0
1013       if $self->_dbh_autocommit;
1014   }
1015   elsif($self->{transaction_depth} > 1) {
1016     $self->{transaction_depth}--;
1017     $self->svp_release
1018       if $self->auto_savepoint;
1019   }
1020 }
1021
1022 sub txn_rollback {
1023   my $self = shift;
1024   my $dbh = $self->_dbh;
1025   eval {
1026     if ($self->{transaction_depth} == 1) {
1027       $self->debugobj->txn_rollback()
1028         if ($self->debug);
1029       $self->{transaction_depth} = 0
1030         if $self->_dbh_autocommit;
1031       $dbh->rollback;
1032     }
1033     elsif($self->{transaction_depth} > 1) {
1034       $self->{transaction_depth}--;
1035       if ($self->auto_savepoint) {
1036         $self->svp_rollback;
1037         $self->svp_release;
1038       }
1039     }
1040     else {
1041       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
1042     }
1043   };
1044   if ($@) {
1045     my $error = $@;
1046     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
1047     $error =~ /$exception_class/ and $self->throw_exception($error);
1048     # ensure that a failed rollback resets the transaction depth
1049     $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
1050     $self->throw_exception($error);
1051   }
1052 }
1053
1054 # This used to be the top-half of _execute.  It was split out to make it
1055 #  easier to override in NoBindVars without duping the rest.  It takes up
1056 #  all of _execute's args, and emits $sql, @bind.
1057 sub _prep_for_execute {
1058   my ($self, $op, $extra_bind, $ident, $args) = @_;
1059
1060   if( Scalar::Util::blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1061     $ident = $ident->from();
1062   }
1063
1064   my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
1065
1066   unshift(@bind,
1067     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
1068       if $extra_bind;
1069   return ($sql, \@bind);
1070 }
1071
1072
1073 sub _fix_bind_params {
1074     my ($self, @bind) = @_;
1075
1076     ### Turn @bind from something like this:
1077     ###   ( [ "artist", 1 ], [ "cdid", 1, 3 ] )
1078     ### to this:
1079     ###   ( "'1'", "'1'", "'3'" )
1080     return
1081         map {
1082             if ( defined( $_ && $_->[1] ) ) {
1083                 map { qq{'$_'}; } @{$_}[ 1 .. $#$_ ];
1084             }
1085             else { q{'NULL'}; }
1086         } @bind;
1087 }
1088
1089 sub _query_start {
1090     my ( $self, $sql, @bind ) = @_;
1091
1092     if ( $self->debug ) {
1093         @bind = $self->_fix_bind_params(@bind);
1094
1095         $self->debugobj->query_start( $sql, @bind );
1096     }
1097 }
1098
1099 sub _query_end {
1100     my ( $self, $sql, @bind ) = @_;
1101
1102     if ( $self->debug ) {
1103         @bind = $self->_fix_bind_params(@bind);
1104         $self->debugobj->query_end( $sql, @bind );
1105     }
1106 }
1107
1108 sub _dbh_execute {
1109   my ($self, $dbh, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
1110
1111   my ($sql, $bind) = $self->_prep_for_execute($op, $extra_bind, $ident, \@args);
1112
1113   $self->_query_start( $sql, @$bind );
1114
1115   my $sth = $self->sth($sql,$op);
1116
1117   my $placeholder_index = 1;
1118
1119   foreach my $bound (@$bind) {
1120     my $attributes = {};
1121     my($column_name, @data) = @$bound;
1122
1123     if ($bind_attributes) {
1124       $attributes = $bind_attributes->{$column_name}
1125       if defined $bind_attributes->{$column_name};
1126     }
1127
1128     foreach my $data (@data) {
1129       my $ref = ref $data;
1130       $data = $ref && $ref ne 'ARRAY' ? ''.$data : $data; # stringify args (except arrayrefs)
1131
1132       $sth->bind_param($placeholder_index, $data, $attributes);
1133       $placeholder_index++;
1134     }
1135   }
1136
1137   # Can this fail without throwing an exception anyways???
1138   my $rv = $sth->execute();
1139   $self->throw_exception($sth->errstr) if !$rv;
1140
1141   $self->_query_end( $sql, @$bind );
1142
1143   return (wantarray ? ($rv, $sth, @$bind) : $rv);
1144 }
1145
1146 sub _execute {
1147     my $self = shift;
1148     $self->dbh_do('_dbh_execute', @_)
1149 }
1150
1151 sub insert {
1152   my ($self, $source, $to_insert) = @_;
1153
1154   my $ident = $source->from;
1155   my $bind_attributes = $self->source_bind_attributes($source);
1156
1157   my $updated_cols = {};
1158
1159   $self->ensure_connected;
1160   foreach my $col ( $source->columns ) {
1161     if ( !defined $to_insert->{$col} ) {
1162       my $col_info = $source->column_info($col);
1163
1164       if ( $col_info->{auto_nextval} ) {
1165         $updated_cols->{$col} = $to_insert->{$col} = $self->_sequence_fetch( 'nextval', $col_info->{sequence} || $self->_dbh_get_autoinc_seq($self->dbh, $source) );
1166       }
1167     }
1168   }
1169
1170   $self->_execute('insert' => [], $source, $bind_attributes, $to_insert);
1171
1172   return $updated_cols;
1173 }
1174
1175 ## Still not quite perfect, and EXPERIMENTAL
1176 ## Currently it is assumed that all values passed will be "normal", i.e. not
1177 ## scalar refs, or at least, all the same type as the first set, the statement is
1178 ## only prepped once.
1179 sub insert_bulk {
1180   my ($self, $source, $cols, $data) = @_;
1181   my %colvalues;
1182   my $table = $source->from;
1183   @colvalues{@$cols} = (0..$#$cols);
1184   my ($sql, @bind) = $self->sql_maker->insert($table, \%colvalues);
1185
1186   $self->_query_start( $sql, @bind );
1187   my $sth = $self->sth($sql);
1188
1189 #  @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
1190
1191   ## This must be an arrayref, else nothing works!
1192   my $tuple_status = [];
1193
1194   ## Get the bind_attributes, if any exist
1195   my $bind_attributes = $self->source_bind_attributes($source);
1196
1197   ## Bind the values and execute
1198   my $placeholder_index = 1;
1199
1200   foreach my $bound (@bind) {
1201
1202     my $attributes = {};
1203     my ($column_name, $data_index) = @$bound;
1204
1205     if( $bind_attributes ) {
1206       $attributes = $bind_attributes->{$column_name}
1207       if defined $bind_attributes->{$column_name};
1208     }
1209
1210     my @data = map { $_->[$data_index] } @$data;
1211
1212     $sth->bind_param_array( $placeholder_index, [@data], $attributes );
1213     $placeholder_index++;
1214   }
1215   my $rv = eval { $sth->execute_array({ArrayTupleStatus => $tuple_status}) };
1216   if (my $err = $@) {
1217     my $i = 0;
1218     ++$i while $i <= $#$tuple_status && !ref $tuple_status->[$i];
1219
1220     $self->throw_exception($sth->errstr || "Unexpected populate error: $err")
1221       if ($i > $#$tuple_status);
1222
1223     require Data::Dumper;
1224     local $Data::Dumper::Terse = 1;
1225     local $Data::Dumper::Indent = 1;
1226     local $Data::Dumper::Useqq = 1;
1227     local $Data::Dumper::Quotekeys = 0;
1228
1229     $self->throw_exception(sprintf "%s for populate slice:\n%s",
1230       $tuple_status->[$i][1],
1231       Data::Dumper::Dumper(
1232         { map { $cols->[$_] => $data->[$i][$_] } (0 .. $#$cols) }
1233       ),
1234     );
1235   }
1236   $self->throw_exception($sth->errstr) if !$rv;
1237
1238   $self->_query_end( $sql, @bind );
1239   return (wantarray ? ($rv, $sth, @bind) : $rv);
1240 }
1241
1242 sub update {
1243   my $self = shift @_;
1244   my $source = shift @_;
1245   my $bind_attributes = $self->source_bind_attributes($source);
1246
1247   return $self->_execute('update' => [], $source, $bind_attributes, @_);
1248 }
1249
1250
1251 sub delete {
1252   my $self = shift @_;
1253   my $source = shift @_;
1254
1255   my $bind_attrs = $self->source_bind_attributes($source);
1256
1257   return $self->_execute('delete' => [], $source, $bind_attrs, @_);
1258 }
1259
1260 # We were sent here because the $rs contains a complex search
1261 # which will require a subquery to select the correct rows
1262 # (i.e. joined or limited resultsets)
1263 #
1264 # Genarating a single PK column subquery is trivial and supported
1265 # by all RDBMS. However if we have a multicolumn PK, things get ugly.
1266 # Look at _multipk_update_delete()
1267 sub _subq_update_delete {
1268   my $self = shift;
1269   my ($rs, $op, $values) = @_;
1270
1271   my $rsrc = $rs->result_source;
1272
1273   # we already check this, but double check naively just in case. Should be removed soon
1274   my $sel = $rs->_resolved_attrs->{select};
1275   $sel = [ $sel ] unless ref $sel eq 'ARRAY';
1276   my @pcols = $rsrc->primary_columns;
1277   if (@$sel != @pcols) {
1278     $self->throw_exception (
1279       'Subquery update/delete can not be called on resultsets selecting a'
1280      .' number of columns different than the number of primary keys'
1281     );
1282   }
1283
1284   if (@pcols == 1) {
1285     return $self->$op (
1286       $rsrc,
1287       $op eq 'update' ? $values : (),
1288       { $pcols[0] => { -in => $rs->as_query } },
1289     );
1290   }
1291
1292   else {
1293     return $self->_multipk_update_delete (@_);
1294   }
1295 }
1296
1297 # ANSI SQL does not provide a reliable way to perform a multicol-PK
1298 # resultset update/delete involving subqueries. So by default resort
1299 # to simple (and inefficient) delete_all style per-row opearations,
1300 # while allowing specific storages to override this with a faster
1301 # implementation.
1302 #
1303 sub _multipk_update_delete {
1304   return shift->_per_row_update_delete (@_);
1305 }
1306
1307 # This is the default loop used to delete/update rows for multi PK
1308 # resultsets, and used by mysql exclusively (because it can't do anything
1309 # else).
1310 #
1311 # We do not use $row->$op style queries, because resultset update/delete
1312 # is not expected to cascade (this is what delete_all/update_all is for).
1313 #
1314 # There should be no race conditions as the entire operation is rolled
1315 # in a transaction.
1316 #
1317 sub _per_row_update_delete {
1318   my $self = shift;
1319   my ($rs, $op, $values) = @_;
1320
1321   my $rsrc = $rs->result_source;
1322   my @pcols = $rsrc->primary_columns;
1323
1324   my $guard = $self->txn_scope_guard;
1325
1326   # emulate the return value of $sth->execute for non-selects
1327   my $row_cnt = '0E0';
1328
1329   my $subrs_cur = $rs->cursor;
1330   while (my @pks = $subrs_cur->next) {
1331
1332     my $cond;
1333     for my $i (0.. $#pcols) {
1334       $cond->{$pcols[$i]} = $pks[$i];
1335     }
1336
1337     $self->$op (
1338       $rsrc,
1339       $op eq 'update' ? $values : (),
1340       $cond,
1341     );
1342
1343     $row_cnt++;
1344   }
1345
1346   $guard->commit;
1347
1348   return $row_cnt;
1349 }
1350
1351 sub _select {
1352   my $self = shift;
1353   my $sql_maker = $self->sql_maker;
1354   local $sql_maker->{for};
1355   return $self->_execute($self->_select_args(@_));
1356 }
1357
1358 sub _select_args_to_query {
1359   my $self = shift;
1360
1361   my $sql_maker = $self->sql_maker;
1362   local $sql_maker->{for};
1363
1364   # my ($op, $bind, $ident, $bind_attrs, $select, $cond, $order, $rows, $offset)
1365   #  = $self->_select_args($ident, $select, $cond, $attrs);
1366   my ($op, $bind, $ident, $bind_attrs, @args) =
1367     $self->_select_args(@_);
1368
1369   # my ($sql, $bind) = $self->_prep_for_execute($op, $bind, $ident, [ $select, $cond, $order, $rows, $offset ]);
1370   my ($sql, $prepared_bind) = $self->_prep_for_execute($op, $bind, $ident, \@args);
1371
1372   return \[ "($sql)", @{ $prepared_bind || [] }];
1373 }
1374
1375 sub _select_args {
1376   my ($self, $ident, $select, $condition, $attrs) = @_;
1377
1378   my $for = delete $attrs->{for};
1379   my $sql_maker = $self->sql_maker;
1380   $sql_maker->{for} = $for;
1381
1382   my $order = { map
1383     { $attrs->{$_} ? ( $_ => $attrs->{$_} ) : ()  }
1384     (qw/order_by group_by having _virtual_order_by/ )
1385   };
1386
1387
1388   my $bind_attrs = {};
1389
1390   my $alias2source = $self->_resolve_ident_sources ($ident);
1391
1392   for my $alias (keys %$alias2source) {
1393     my $bindtypes = $self->source_bind_attributes ($alias2source->{$alias}) || {};
1394     for my $col (keys %$bindtypes) {
1395
1396       my $fqcn = join ('.', $alias, $col);
1397       $bind_attrs->{$fqcn} = $bindtypes->{$col} if $bindtypes->{$col};
1398
1399       # so that unqualified searches can be bound too
1400       $bind_attrs->{$col} = $bind_attrs->{$fqcn} if $alias eq 'me';
1401     }
1402   }
1403
1404   # This would be the point to deflate anything found in $condition
1405   # (and leave $attrs->{bind} intact). Problem is - inflators historically
1406   # expect a row object. And all we have is a resultsource (it is trivial
1407   # to extract deflator coderefs via $alias2source above).
1408   #
1409   # I don't see a way forward other than changing the way deflators are
1410   # invoked, and that's just bad...
1411
1412   my @args = ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $condition, $order);
1413   if ($attrs->{software_limit} ||
1414       $sql_maker->_default_limit_syntax eq "GenericSubQ") {
1415         $attrs->{software_limit} = 1;
1416   } else {
1417     $self->throw_exception("rows attribute must be positive if present")
1418       if (defined($attrs->{rows}) && !($attrs->{rows} > 0));
1419
1420     # MySQL actually recommends this approach.  I cringe.
1421     $attrs->{rows} = 2**48 if not defined $attrs->{rows} and defined $attrs->{offset};
1422     push @args, $attrs->{rows}, $attrs->{offset};
1423   }
1424   return @args;
1425 }
1426
1427 sub _resolve_ident_sources {
1428   my ($self, $ident) = @_;
1429
1430   my $alias2source = {};
1431
1432   # the reason this is so contrived is that $ident may be a {from}
1433   # structure, specifying multiple tables to join
1434   if ( Scalar::Util::blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1435     # this is compat mode for insert/update/delete which do not deal with aliases
1436     $alias2source->{me} = $ident;
1437   }
1438   elsif (ref $ident eq 'ARRAY') {
1439
1440     for (@$ident) {
1441       my $tabinfo;
1442       if (ref $_ eq 'HASH') {
1443         $tabinfo = $_;
1444       }
1445       if (ref $_ eq 'ARRAY' and ref $_->[0] eq 'HASH') {
1446         $tabinfo = $_->[0];
1447       }
1448
1449       $alias2source->{$tabinfo->{-alias}} = $tabinfo->{-result_source}
1450         if ($tabinfo->{-result_source});
1451     }
1452   }
1453
1454   return $alias2source;
1455 }
1456
1457 sub count {
1458   my ($self, $source, $attrs) = @_;
1459
1460   my $tmp_attrs = { %$attrs };
1461
1462   # take off any pagers, record_filter is cdbi, and no point of ordering a count
1463   delete $tmp_attrs->{$_} for (qw/select as rows offset page order_by record_filter/);
1464
1465   # overwrite the selector
1466   $tmp_attrs->{select} = { count => '*' };
1467
1468   my $tmp_rs = $source->resultset_class->new($source, $tmp_attrs);
1469   my ($count) = $tmp_rs->cursor->next;
1470
1471   # if the offset/rows attributes are still present, we did not use
1472   # a subquery, so we need to make the calculations in software
1473   $count -= $attrs->{offset} if $attrs->{offset};
1474   $count = $attrs->{rows} if $attrs->{rows} and $attrs->{rows} < $count;
1475   $count = 0 if ($count < 0);
1476
1477   return $count;
1478 }
1479
1480 sub count_grouped {
1481   my ($self, $source, $attrs) = @_;
1482
1483   # copy for the subquery, we need to do some adjustments to it too
1484   my $sub_attrs = { %$attrs };
1485
1486   # these can not go in the subquery, and there is no point of ordering it
1487   delete $sub_attrs->{$_} for qw/prefetch collapse select as order_by/;
1488
1489   # if we prefetch, we group_by primary keys only as this is what we would get out of the rs via ->next/->all
1490   # simply deleting group_by suffices, as the code below will re-fill it
1491   # Note: we check $attrs, as $sub_attrs has collapse deleted
1492   if (ref $attrs->{collapse} and keys %{$attrs->{collapse}} ) {
1493     delete $sub_attrs->{group_by};
1494   }
1495
1496   $sub_attrs->{group_by} ||= [ map { "$attrs->{alias}.$_" } ($source->primary_columns) ];
1497   $sub_attrs->{select} = $self->_grouped_count_select ($source, $sub_attrs);
1498
1499   $attrs->{from} = [{
1500     count_subq => $source->resultset_class->new ($source, $sub_attrs )->as_query
1501   }];
1502
1503   # the subquery replaces this
1504   delete $attrs->{$_} for qw/where bind prefetch collapse group_by having having_bind rows offset page pager/;
1505
1506   return $self->count ($source, $attrs);
1507 }
1508
1509 #
1510 # Returns a SELECT to go with a supplied GROUP BY
1511 # (caled by count_grouped so a group_by is present)
1512 # Most databases expect them to match, but some
1513 # choke in various ways.
1514 #
1515 sub _grouped_count_select {
1516   my ($self, $source, $rs_args) = @_;
1517   return $rs_args->{group_by};
1518 }
1519
1520 sub source_bind_attributes {
1521   my ($self, $source) = @_;
1522
1523   my $bind_attributes;
1524   foreach my $column ($source->columns) {
1525
1526     my $data_type = $source->column_info($column)->{data_type} || '';
1527     $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
1528      if $data_type;
1529   }
1530
1531   return $bind_attributes;
1532 }
1533
1534 =head2 select
1535
1536 =over 4
1537
1538 =item Arguments: $ident, $select, $condition, $attrs
1539
1540 =back
1541
1542 Handle a SQL select statement.
1543
1544 =cut
1545
1546 sub select {
1547   my $self = shift;
1548   my ($ident, $select, $condition, $attrs) = @_;
1549   return $self->cursor_class->new($self, \@_, $attrs);
1550 }
1551
1552 sub select_single {
1553   my $self = shift;
1554   my ($rv, $sth, @bind) = $self->_select(@_);
1555   my @row = $sth->fetchrow_array;
1556   my @nextrow = $sth->fetchrow_array if @row;
1557   if(@row && @nextrow) {
1558     carp "Query returned more than one row.  SQL that returns multiple rows is DEPRECATED for ->find and ->single";
1559   }
1560   # Need to call finish() to work round broken DBDs
1561   $sth->finish();
1562   return @row;
1563 }
1564
1565 =head2 sth
1566
1567 =over 4
1568
1569 =item Arguments: $sql
1570
1571 =back
1572
1573 Returns a L<DBI> sth (statement handle) for the supplied SQL.
1574
1575 =cut
1576
1577 sub _dbh_sth {
1578   my ($self, $dbh, $sql) = @_;
1579
1580   # 3 is the if_active parameter which avoids active sth re-use
1581   my $sth = $self->disable_sth_caching
1582     ? $dbh->prepare($sql)
1583     : $dbh->prepare_cached($sql, {}, 3);
1584
1585   # XXX You would think RaiseError would make this impossible,
1586   #  but apparently that's not true :(
1587   $self->throw_exception($dbh->errstr) if !$sth;
1588
1589   $sth;
1590 }
1591
1592 sub sth {
1593   my ($self, $sql) = @_;
1594   $self->dbh_do('_dbh_sth', $sql);
1595 }
1596
1597 sub _dbh_columns_info_for {
1598   my ($self, $dbh, $table) = @_;
1599
1600   if ($dbh->can('column_info')) {
1601     my %result;
1602     eval {
1603       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
1604       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
1605       $sth->execute();
1606       while ( my $info = $sth->fetchrow_hashref() ){
1607         my %column_info;
1608         $column_info{data_type}   = $info->{TYPE_NAME};
1609         $column_info{size}      = $info->{COLUMN_SIZE};
1610         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
1611         $column_info{default_value} = $info->{COLUMN_DEF};
1612         my $col_name = $info->{COLUMN_NAME};
1613         $col_name =~ s/^\"(.*)\"$/$1/;
1614
1615         $result{$col_name} = \%column_info;
1616       }
1617     };
1618     return \%result if !$@ && scalar keys %result;
1619   }
1620
1621   my %result;
1622   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
1623   $sth->execute;
1624   my @columns = @{$sth->{NAME_lc}};
1625   for my $i ( 0 .. $#columns ){
1626     my %column_info;
1627     $column_info{data_type} = $sth->{TYPE}->[$i];
1628     $column_info{size} = $sth->{PRECISION}->[$i];
1629     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
1630
1631     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
1632       $column_info{data_type} = $1;
1633       $column_info{size}    = $2;
1634     }
1635
1636     $result{$columns[$i]} = \%column_info;
1637   }
1638   $sth->finish;
1639
1640   foreach my $col (keys %result) {
1641     my $colinfo = $result{$col};
1642     my $type_num = $colinfo->{data_type};
1643     my $type_name;
1644     if(defined $type_num && $dbh->can('type_info')) {
1645       my $type_info = $dbh->type_info($type_num);
1646       $type_name = $type_info->{TYPE_NAME} if $type_info;
1647       $colinfo->{data_type} = $type_name if $type_name;
1648     }
1649   }
1650
1651   return \%result;
1652 }
1653
1654 sub columns_info_for {
1655   my ($self, $table) = @_;
1656   $self->dbh_do('_dbh_columns_info_for', $table);
1657 }
1658
1659 =head2 last_insert_id
1660
1661 Return the row id of the last insert.
1662
1663 =cut
1664
1665 sub _dbh_last_insert_id {
1666     # All Storage's need to register their own _dbh_last_insert_id
1667     # the old SQLite-based method was highly inappropriate
1668
1669     my $self = shift;
1670     my $class = ref $self;
1671     $self->throw_exception (<<EOE);
1672
1673 No _dbh_last_insert_id() method found in $class.
1674 Since the method of obtaining the autoincrement id of the last insert
1675 operation varies greatly between different databases, this method must be
1676 individually implemented for every storage class.
1677 EOE
1678 }
1679
1680 sub last_insert_id {
1681   my $self = shift;
1682   $self->dbh_do('_dbh_last_insert_id', @_);
1683 }
1684
1685 =head2 sqlt_type
1686
1687 Returns the database driver name.
1688
1689 =cut
1690
1691 sub sqlt_type { shift->dbh->{Driver}->{Name} }
1692
1693 =head2 bind_attribute_by_data_type
1694
1695 Given a datatype from column info, returns a database specific bind
1696 attribute for C<< $dbh->bind_param($val,$attribute) >> or nothing if we will
1697 let the database planner just handle it.
1698
1699 Generally only needed for special case column types, like bytea in postgres.
1700
1701 =cut
1702
1703 sub bind_attribute_by_data_type {
1704     return;
1705 }
1706
1707 =head2 is_datatype_numeric
1708
1709 Given a datatype from column_info, returns a boolean value indicating if
1710 the current RDBMS considers it a numeric value. This controls how
1711 L<DBIx::Class::Row/set_column> decides whether to mark the column as
1712 dirty - when the datatype is deemed numeric a C<< != >> comparison will
1713 be performed instead of the usual C<eq>.
1714
1715 =cut
1716
1717 sub is_datatype_numeric {
1718   my ($self, $dt) = @_;
1719
1720   return 0 unless $dt;
1721
1722   return $dt =~ /^ (?:
1723     numeric | int(?:eger)? | (?:tiny|small|medium|big)int | dec(?:imal)? | real | float | double (?: \s+ precision)? | (?:big)?serial
1724   ) $/ix;
1725 }
1726
1727
1728 =head2 create_ddl_dir (EXPERIMENTAL)
1729
1730 =over 4
1731
1732 =item Arguments: $schema \@databases, $version, $directory, $preversion, \%sqlt_args
1733
1734 =back
1735
1736 Creates a SQL file based on the Schema, for each of the specified
1737 database engines in C<\@databases> in the given directory.
1738 (note: specify L<SQL::Translator> names, not L<DBI> driver names).
1739
1740 Given a previous version number, this will also create a file containing
1741 the ALTER TABLE statements to transform the previous schema into the
1742 current one. Note that these statements may contain C<DROP TABLE> or
1743 C<DROP COLUMN> statements that can potentially destroy data.
1744
1745 The file names are created using the C<ddl_filename> method below, please
1746 override this method in your schema if you would like a different file
1747 name format. For the ALTER file, the same format is used, replacing
1748 $version in the name with "$preversion-$version".
1749
1750 See L<SQL::Translator/METHODS> for a list of values for C<\%sqlt_args>.
1751 The most common value for this would be C<< { add_drop_table => 1 } >>
1752 to have the SQL produced include a C<DROP TABLE> statement for each table
1753 created. For quoting purposes supply C<quote_table_names> and
1754 C<quote_field_names>.
1755
1756 If no arguments are passed, then the following default values are assumed:
1757
1758 =over 4
1759
1760 =item databases  - ['MySQL', 'SQLite', 'PostgreSQL']
1761
1762 =item version    - $schema->schema_version
1763
1764 =item directory  - './'
1765
1766 =item preversion - <none>
1767
1768 =back
1769
1770 By default, C<\%sqlt_args> will have
1771
1772  { add_drop_table => 1, ignore_constraint_names => 1, ignore_index_names => 1 }
1773
1774 merged with the hash passed in. To disable any of those features, pass in a
1775 hashref like the following
1776
1777  { ignore_constraint_names => 0, # ... other options }
1778
1779
1780 Note that this feature is currently EXPERIMENTAL and may not work correctly
1781 across all databases, or fully handle complex relationships.
1782
1783 WARNING: Please check all SQL files created, before applying them.
1784
1785 =cut
1786
1787 sub create_ddl_dir {
1788   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
1789
1790   if(!$dir || !-d $dir) {
1791     carp "No directory given, using ./\n";
1792     $dir = "./";
1793   }
1794   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
1795   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
1796
1797   my $schema_version = $schema->schema_version || '1.x';
1798   $version ||= $schema_version;
1799
1800   $sqltargs = {
1801     add_drop_table => 1,
1802     ignore_constraint_names => 1,
1803     ignore_index_names => 1,
1804     %{$sqltargs || {}}
1805   };
1806
1807   $self->throw_exception(q{Can't create a ddl file without SQL::Translator 0.09003: '}
1808       . $self->_check_sqlt_message . q{'})
1809           if !$self->_check_sqlt_version;
1810
1811   my $sqlt = SQL::Translator->new( $sqltargs );
1812
1813   $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
1814   my $sqlt_schema = $sqlt->translate({ data => $schema })
1815     or $self->throw_exception ($sqlt->error);
1816
1817   foreach my $db (@$databases) {
1818     $sqlt->reset();
1819     $sqlt->{schema} = $sqlt_schema;
1820     $sqlt->producer($db);
1821
1822     my $file;
1823     my $filename = $schema->ddl_filename($db, $version, $dir);
1824     if (-e $filename && ($version eq $schema_version )) {
1825       # if we are dumping the current version, overwrite the DDL
1826       carp "Overwriting existing DDL file - $filename";
1827       unlink($filename);
1828     }
1829
1830     my $output = $sqlt->translate;
1831     if(!$output) {
1832       carp("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
1833       next;
1834     }
1835     if(!open($file, ">$filename")) {
1836       $self->throw_exception("Can't open $filename for writing ($!)");
1837       next;
1838     }
1839     print $file $output;
1840     close($file);
1841
1842     next unless ($preversion);
1843
1844     require SQL::Translator::Diff;
1845
1846     my $prefilename = $schema->ddl_filename($db, $preversion, $dir);
1847     if(!-e $prefilename) {
1848       carp("No previous schema file found ($prefilename)");
1849       next;
1850     }
1851
1852     my $difffile = $schema->ddl_filename($db, $version, $dir, $preversion);
1853     if(-e $difffile) {
1854       carp("Overwriting existing diff file - $difffile");
1855       unlink($difffile);
1856     }
1857
1858     my $source_schema;
1859     {
1860       my $t = SQL::Translator->new($sqltargs);
1861       $t->debug( 0 );
1862       $t->trace( 0 );
1863
1864       $t->parser( $db )
1865         or $self->throw_exception ($t->error);
1866
1867       my $out = $t->translate( $prefilename )
1868         or $self->throw_exception ($t->error);
1869
1870       $source_schema = $t->schema;
1871
1872       $source_schema->name( $prefilename )
1873         unless ( $source_schema->name );
1874     }
1875
1876     # The "new" style of producers have sane normalization and can support
1877     # diffing a SQL file against a DBIC->SQLT schema. Old style ones don't
1878     # And we have to diff parsed SQL against parsed SQL.
1879     my $dest_schema = $sqlt_schema;
1880
1881     unless ( "SQL::Translator::Producer::$db"->can('preprocess_schema') ) {
1882       my $t = SQL::Translator->new($sqltargs);
1883       $t->debug( 0 );
1884       $t->trace( 0 );
1885
1886       $t->parser( $db )
1887         or $self->throw_exception ($t->error);
1888
1889       my $out = $t->translate( $filename )
1890         or $self->throw_exception ($t->error);
1891
1892       $dest_schema = $t->schema;
1893
1894       $dest_schema->name( $filename )
1895         unless $dest_schema->name;
1896     }
1897
1898     my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
1899                                                   $dest_schema,   $db,
1900                                                   $sqltargs
1901                                                  );
1902     if(!open $file, ">$difffile") {
1903       $self->throw_exception("Can't write to $difffile ($!)");
1904       next;
1905     }
1906     print $file $diff;
1907     close($file);
1908   }
1909 }
1910
1911 =head2 deployment_statements
1912
1913 =over 4
1914
1915 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
1916
1917 =back
1918
1919 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
1920
1921 The L<SQL::Translator> (not L<DBI>) database driver name can be explicitly
1922 provided in C<$type>, otherwise the result of L</sqlt_type> is used as default.
1923
1924 C<$directory> is used to return statements from files in a previously created
1925 L</create_ddl_dir> directory and is optional. The filenames are constructed
1926 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
1927
1928 If no C<$directory> is specified then the statements are constructed on the
1929 fly using L<SQL::Translator> and C<$version> is ignored.
1930
1931 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
1932
1933 =cut
1934
1935 sub deployment_statements {
1936   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
1937   # Need to be connected to get the correct sqlt_type
1938   $self->ensure_connected() unless $type;
1939   $type ||= $self->sqlt_type;
1940   $version ||= $schema->schema_version || '1.x';
1941   $dir ||= './';
1942   my $filename = $schema->ddl_filename($type, $version, $dir);
1943   if(-f $filename)
1944   {
1945       my $file;
1946       open($file, "<$filename")
1947         or $self->throw_exception("Can't open $filename ($!)");
1948       my @rows = <$file>;
1949       close($file);
1950       return join('', @rows);
1951   }
1952
1953   $self->throw_exception(q{Can't deploy without SQL::Translator 0.09003: '}
1954       . $self->_check_sqlt_message . q{'})
1955           if !$self->_check_sqlt_version;
1956
1957   require SQL::Translator::Parser::DBIx::Class;
1958   eval qq{use SQL::Translator::Producer::${type}};
1959   $self->throw_exception($@) if $@;
1960
1961   # sources needs to be a parser arg, but for simplicty allow at top level
1962   # coming in
1963   $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
1964       if exists $sqltargs->{sources};
1965
1966   my $tr = SQL::Translator->new(%$sqltargs);
1967   SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
1968   return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
1969 }
1970
1971 sub deploy {
1972   my ($self, $schema, $type, $sqltargs, $dir) = @_;
1973   my $deploy = sub {
1974     my $line = shift;
1975     return if($line =~ /^--/);
1976     return if(!$line);
1977     # next if($line =~ /^DROP/m);
1978     return if($line =~ /^BEGIN TRANSACTION/m);
1979     return if($line =~ /^COMMIT/m);
1980     return if $line =~ /^\s+$/; # skip whitespace only
1981     $self->_query_start($line);
1982     eval {
1983       $self->dbh->do($line); # shouldn't be using ->dbh ?
1984     };
1985     if ($@) {
1986       carp qq{$@ (running "${line}")};
1987     }
1988     $self->_query_end($line);
1989   };
1990   my @statements = $self->deployment_statements($schema, $type, undef, $dir, { %{ $sqltargs || {} }, no_comments => 1 } );
1991   if (@statements > 1) {
1992     foreach my $statement (@statements) {
1993       $deploy->( $statement );
1994     }
1995   }
1996   elsif (@statements == 1) {
1997     foreach my $line ( split(";\n", $statements[0])) {
1998       $deploy->( $line );
1999     }
2000   }
2001 }
2002
2003 =head2 datetime_parser
2004
2005 Returns the datetime parser class
2006
2007 =cut
2008
2009 sub datetime_parser {
2010   my $self = shift;
2011   return $self->{datetime_parser} ||= do {
2012     $self->ensure_connected;
2013     $self->build_datetime_parser(@_);
2014   };
2015 }
2016
2017 =head2 datetime_parser_type
2018
2019 Defines (returns) the datetime parser class - currently hardwired to
2020 L<DateTime::Format::MySQL>
2021
2022 =cut
2023
2024 sub datetime_parser_type { "DateTime::Format::MySQL"; }
2025
2026 =head2 build_datetime_parser
2027
2028 See L</datetime_parser>
2029
2030 =cut
2031
2032 sub build_datetime_parser {
2033   my $self = shift;
2034   my $type = $self->datetime_parser_type(@_);
2035   eval "use ${type}";
2036   $self->throw_exception("Couldn't load ${type}: $@") if $@;
2037   return $type;
2038 }
2039
2040 {
2041     my $_check_sqlt_version; # private
2042     my $_check_sqlt_message; # private
2043     sub _check_sqlt_version {
2044         return $_check_sqlt_version if defined $_check_sqlt_version;
2045         eval 'use SQL::Translator "0.09003"';
2046         $_check_sqlt_message = $@ || '';
2047         $_check_sqlt_version = !$@;
2048     }
2049
2050     sub _check_sqlt_message {
2051         _check_sqlt_version if !defined $_check_sqlt_message;
2052         $_check_sqlt_message;
2053     }
2054 }
2055
2056 =head2 is_replicating
2057
2058 A boolean that reports if a particular L<DBIx::Class::Storage::DBI> is set to
2059 replicate from a master database.  Default is undef, which is the result
2060 returned by databases that don't support replication.
2061
2062 =cut
2063
2064 sub is_replicating {
2065     return;
2066
2067 }
2068
2069 =head2 lag_behind_master
2070
2071 Returns a number that represents a certain amount of lag behind a master db
2072 when a given storage is replicating.  The number is database dependent, but
2073 starts at zero and increases with the amount of lag. Default in undef
2074
2075 =cut
2076
2077 sub lag_behind_master {
2078     return;
2079 }
2080
2081 sub DESTROY {
2082   my $self = shift;
2083   return if !$self->_dbh;
2084   $self->_verify_pid;
2085   $self->_dbh(undef);
2086 }
2087
2088 1;
2089
2090 =head1 USAGE NOTES
2091
2092 =head2 DBIx::Class and AutoCommit
2093
2094 DBIx::Class can do some wonderful magic with handling exceptions,
2095 disconnections, and transactions when you use C<< AutoCommit => 1 >>
2096 combined with C<txn_do> for transaction support.
2097
2098 If you set C<< AutoCommit => 0 >> in your connect info, then you are always
2099 in an assumed transaction between commits, and you're telling us you'd
2100 like to manage that manually.  A lot of the magic protections offered by
2101 this module will go away.  We can't protect you from exceptions due to database
2102 disconnects because we don't know anything about how to restart your
2103 transactions.  You're on your own for handling all sorts of exceptional
2104 cases if you choose the C<< AutoCommit => 0 >> path, just as you would
2105 be with raw DBI.
2106
2107
2108
2109 =head1 AUTHORS
2110
2111 Matt S. Trout <mst@shadowcatsystems.co.uk>
2112
2113 Andy Grundman <andy@hybridized.org>
2114
2115 =head1 LICENSE
2116
2117 You may distribute this code under the same terms as Perl itself.
2118
2119 =cut