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