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