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