lib/DBIx/Class/Storage/DBI.pm

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