lib/DBIx/Class/Storage/DBI.pm

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