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] || '';
1002           ($driver) = $dsn =~ /dbi:([^:]+):/i;
1003         }
1004       }
1005
1006       if ($driver) {
1007         my $storage_class = "DBIx::Class::Storage::DBI::${driver}";
1008         if ($self->load_optional_class($storage_class)) {
1009           mro::set_mro($storage_class, 'c3');
1010           bless $self, $storage_class;
1011           $self->_rebless();
1012         }
1013       }
1014     }
1015
1016     $self->_driver_determined(1);
1017
1018     $self->_init; # run driver-specific initializations
1019
1020     $self->_run_connection_actions
1021         if !$started_connected && defined $self->_dbh;
1022   }
1023 }
1024
1025 sub _do_connection_actions {
1026   my $self          = shift;
1027   my $method_prefix = shift;
1028   my $call          = shift;
1029
1030   if (not ref($call)) {
1031     my $method = $method_prefix . $call;
1032     $self->$method(@_);
1033   } elsif (ref($call) eq 'CODE') {
1034     $self->$call(@_);
1035   } elsif (ref($call) eq 'ARRAY') {
1036     if (ref($call->[0]) ne 'ARRAY') {
1037       $self->_do_connection_actions($method_prefix, $_) for @$call;
1038     } else {
1039       $self->_do_connection_actions($method_prefix, @$_) for @$call;
1040     }
1041   } else {
1042     $self->throw_exception (sprintf ("Don't know how to process conection actions of type '%s'", ref($call)) );
1043   }
1044
1045   return $self;
1046 }
1047
1048 sub connect_call_do_sql {
1049   my $self = shift;
1050   $self->_do_query(@_);
1051 }
1052
1053 sub disconnect_call_do_sql {
1054   my $self = shift;
1055   $self->_do_query(@_);
1056 }
1057
1058 # override in db-specific backend when necessary
1059 sub connect_call_datetime_setup { 1 }
1060
1061 sub _do_query {
1062   my ($self, $action) = @_;
1063
1064   if (ref $action eq 'CODE') {
1065     $action = $action->($self);
1066     $self->_do_query($_) foreach @$action;
1067   }
1068   else {
1069     # Most debuggers expect ($sql, @bind), so we need to exclude
1070     # the attribute hash which is the second argument to $dbh->do
1071     # furthermore the bind values are usually to be presented
1072     # as named arrayref pairs, so wrap those here too
1073     my @do_args = (ref $action eq 'ARRAY') ? (@$action) : ($action);
1074     my $sql = shift @do_args;
1075     my $attrs = shift @do_args;
1076     my @bind = map { [ undef, $_ ] } @do_args;
1077
1078     $self->_query_start($sql, @bind);
1079     $self->_get_dbh->do($sql, $attrs, @do_args);
1080     $self->_query_end($sql, @bind);
1081   }
1082
1083   return $self;
1084 }
1085
1086 sub _connect {
1087   my ($self, @info) = @_;
1088
1089   $self->throw_exception("You failed to provide any connection info")
1090     if !@info;
1091
1092   my ($old_connect_via, $dbh);
1093
1094   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
1095     $old_connect_via = $DBI::connect_via;
1096     $DBI::connect_via = 'connect';
1097   }
1098
1099   eval {
1100     if(ref $info[0] eq 'CODE') {
1101        $dbh = $info[0]->();
1102     }
1103     else {
1104        $dbh = DBI->connect(@info);
1105     }
1106
1107     if($dbh && !$self->unsafe) {
1108       my $weak_self = $self;
1109       Scalar::Util::weaken($weak_self);
1110       $dbh->{HandleError} = sub {
1111           if ($weak_self) {
1112             $weak_self->throw_exception("DBI Exception: $_[0]");
1113           }
1114           else {
1115             # the handler may be invoked by something totally out of
1116             # the scope of DBIC
1117             croak ("DBI Exception: $_[0]");
1118           }
1119       };
1120       $dbh->{ShowErrorStatement} = 1;
1121       $dbh->{RaiseError} = 1;
1122       $dbh->{PrintError} = 0;
1123     }
1124   };
1125
1126   $DBI::connect_via = $old_connect_via if $old_connect_via;
1127
1128   $self->throw_exception("DBI Connection failed: " . ($@||$DBI::errstr))
1129     if !$dbh || $@;
1130
1131   $self->_dbh_autocommit($dbh->{AutoCommit});
1132
1133   $dbh;
1134 }
1135
1136 sub svp_begin {
1137   my ($self, $name) = @_;
1138
1139   $name = $self->_svp_generate_name
1140     unless defined $name;
1141
1142   $self->throw_exception ("You can't use savepoints outside a transaction")
1143     if $self->{transaction_depth} == 0;
1144
1145   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1146     unless $self->can('_svp_begin');
1147
1148   push @{ $self->{savepoints} }, $name;
1149
1150   $self->debugobj->svp_begin($name) if $self->debug;
1151
1152   return $self->_svp_begin($name);
1153 }
1154
1155 sub svp_release {
1156   my ($self, $name) = @_;
1157
1158   $self->throw_exception ("You can't use savepoints outside a transaction")
1159     if $self->{transaction_depth} == 0;
1160
1161   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1162     unless $self->can('_svp_release');
1163
1164   if (defined $name) {
1165     $self->throw_exception ("Savepoint '$name' does not exist")
1166       unless grep { $_ eq $name } @{ $self->{savepoints} };
1167
1168     # Dig through the stack until we find the one we are releasing.  This keeps
1169     # the stack up to date.
1170     my $svp;
1171
1172     do { $svp = pop @{ $self->{savepoints} } } while $svp ne $name;
1173   } else {
1174     $name = pop @{ $self->{savepoints} };
1175   }
1176
1177   $self->debugobj->svp_release($name) if $self->debug;
1178
1179   return $self->_svp_release($name);
1180 }
1181
1182 sub svp_rollback {
1183   my ($self, $name) = @_;
1184
1185   $self->throw_exception ("You can't use savepoints outside a transaction")
1186     if $self->{transaction_depth} == 0;
1187
1188   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1189     unless $self->can('_svp_rollback');
1190
1191   if (defined $name) {
1192       # If they passed us a name, verify that it exists in the stack
1193       unless(grep({ $_ eq $name } @{ $self->{savepoints} })) {
1194           $self->throw_exception("Savepoint '$name' does not exist!");
1195       }
1196
1197       # Dig through the stack until we find the one we are releasing.  This keeps
1198       # the stack up to date.
1199       while(my $s = pop(@{ $self->{savepoints} })) {
1200           last if($s eq $name);
1201       }
1202       # Add the savepoint back to the stack, as a rollback doesn't remove the
1203       # named savepoint, only everything after it.
1204       push(@{ $self->{savepoints} }, $name);
1205   } else {
1206       # We'll assume they want to rollback to the last savepoint
1207       $name = $self->{savepoints}->[-1];
1208   }
1209
1210   $self->debugobj->svp_rollback($name) if $self->debug;
1211
1212   return $self->_svp_rollback($name);
1213 }
1214
1215 sub _svp_generate_name {
1216     my ($self) = @_;
1217
1218     return 'savepoint_'.scalar(@{ $self->{'savepoints'} });
1219 }
1220
1221 sub txn_begin {
1222   my $self = shift;
1223
1224   # this means we have not yet connected and do not know the AC status
1225   # (e.g. coderef $dbh)
1226   $self->ensure_connected if (! defined $self->_dbh_autocommit);
1227
1228   if($self->{transaction_depth} == 0) {
1229     $self->debugobj->txn_begin()
1230       if $self->debug;
1231     $self->_dbh_begin_work;
1232   }
1233   elsif ($self->auto_savepoint) {
1234     $self->svp_begin;
1235   }
1236   $self->{transaction_depth}++;
1237 }
1238
1239 sub _dbh_begin_work {
1240   my $self = shift;
1241
1242   # if the user is utilizing txn_do - good for him, otherwise we need to
1243   # ensure that the $dbh is healthy on BEGIN.
1244   # We do this via ->dbh_do instead of ->dbh, so that the ->dbh "ping"
1245   # will be replaced by a failure of begin_work itself (which will be
1246   # then retried on reconnect)
1247   if ($self->{_in_dbh_do}) {
1248     $self->_dbh->begin_work;
1249   } else {
1250     $self->dbh_do(sub { $_[1]->begin_work });
1251   }
1252 }
1253
1254 sub txn_commit {
1255   my $self = shift;
1256   if ($self->{transaction_depth} == 1) {
1257     $self->debugobj->txn_commit()
1258       if ($self->debug);
1259     $self->_dbh_commit;
1260     $self->{transaction_depth} = 0
1261       if $self->_dbh_autocommit;
1262   }
1263   elsif($self->{transaction_depth} > 1) {
1264     $self->{transaction_depth}--;
1265     $self->svp_release
1266       if $self->auto_savepoint;
1267   }
1268 }
1269
1270 sub _dbh_commit {
1271   my $self = shift;
1272   my $dbh  = $self->_dbh
1273     or $self->throw_exception('cannot COMMIT on a disconnected handle');
1274   $dbh->commit;
1275 }
1276
1277 sub txn_rollback {
1278   my $self = shift;
1279   my $dbh = $self->_dbh;
1280   eval {
1281     if ($self->{transaction_depth} == 1) {
1282       $self->debugobj->txn_rollback()
1283         if ($self->debug);
1284       $self->{transaction_depth} = 0
1285         if $self->_dbh_autocommit;
1286       $self->_dbh_rollback;
1287     }
1288     elsif($self->{transaction_depth} > 1) {
1289       $self->{transaction_depth}--;
1290       if ($self->auto_savepoint) {
1291         $self->svp_rollback;
1292         $self->svp_release;
1293       }
1294     }
1295     else {
1296       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
1297     }
1298   };
1299   if ($@) {
1300     my $error = $@;
1301     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
1302     $error =~ /$exception_class/ and $self->throw_exception($error);
1303     # ensure that a failed rollback resets the transaction depth
1304     $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
1305     $self->throw_exception($error);
1306   }
1307 }
1308
1309 sub _dbh_rollback {
1310   my $self = shift;
1311   my $dbh  = $self->_dbh
1312     or $self->throw_exception('cannot ROLLBACK on a disconnected handle');
1313   $dbh->rollback;
1314 }
1315
1316 # This used to be the top-half of _execute.  It was split out to make it
1317 #  easier to override in NoBindVars without duping the rest.  It takes up
1318 #  all of _execute's args, and emits $sql, @bind.
1319 sub _prep_for_execute {
1320   my ($self, $op, $extra_bind, $ident, $args) = @_;
1321
1322   if( Scalar::Util::blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1323     $ident = $ident->from();
1324   }
1325
1326   my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
1327
1328   unshift(@bind,
1329     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
1330       if $extra_bind;
1331   return ($sql, \@bind);
1332 }
1333
1334
1335 sub _fix_bind_params {
1336     my ($self, @bind) = @_;
1337
1338     ### Turn @bind from something like this:
1339     ###   ( [ "artist", 1 ], [ "cdid", 1, 3 ] )
1340     ### to this:
1341     ###   ( "'1'", "'1'", "'3'" )
1342     return
1343         map {
1344             if ( defined( $_ && $_->[1] ) ) {
1345                 map { qq{'$_'}; } @{$_}[ 1 .. $#$_ ];
1346             }
1347             else { q{'NULL'}; }
1348         } @bind;
1349 }
1350
1351 sub _query_start {
1352     my ( $self, $sql, @bind ) = @_;
1353
1354     if ( $self->debug ) {
1355         @bind = $self->_fix_bind_params(@bind);
1356
1357         $self->debugobj->query_start( $sql, @bind );
1358     }
1359 }
1360
1361 sub _query_end {
1362     my ( $self, $sql, @bind ) = @_;
1363
1364     if ( $self->debug ) {
1365         @bind = $self->_fix_bind_params(@bind);
1366         $self->debugobj->query_end( $sql, @bind );
1367     }
1368 }
1369
1370 sub _dbh_execute {
1371   my ($self, $dbh, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
1372
1373   my ($sql, $bind) = $self->_prep_for_execute($op, $extra_bind, $ident, \@args);
1374
1375   $self->_query_start( $sql, @$bind );
1376
1377   my $sth = $self->sth($sql,$op);
1378
1379   my $placeholder_index = 1;
1380
1381   foreach my $bound (@$bind) {
1382     my $attributes = {};
1383     my($column_name, @data) = @$bound;
1384
1385     if ($bind_attributes) {
1386       $attributes = $bind_attributes->{$column_name}
1387       if defined $bind_attributes->{$column_name};
1388     }
1389
1390     foreach my $data (@data) {
1391       my $ref = ref $data;
1392       $data = $ref && $ref ne 'ARRAY' ? ''.$data : $data; # stringify args (except arrayrefs)
1393
1394       $sth->bind_param($placeholder_index, $data, $attributes);
1395       $placeholder_index++;
1396     }
1397   }
1398
1399   # Can this fail without throwing an exception anyways???
1400   my $rv = $sth->execute();
1401   $self->throw_exception($sth->errstr) if !$rv;
1402
1403   $self->_query_end( $sql, @$bind );
1404
1405   return (wantarray ? ($rv, $sth, @$bind) : $rv);
1406 }
1407
1408 sub _execute {
1409     my $self = shift;
1410     $self->dbh_do('_dbh_execute', @_);  # retry over disconnects
1411 }
1412
1413 sub _prefetch_insert_auto_nextvals {
1414   my ($self, $source, $to_insert) = @_;
1415
1416   my $upd = {};
1417
1418   foreach my $col ( $source->columns ) {
1419     if ( !defined $to_insert->{$col} ) {
1420       my $col_info = $source->column_info($col);
1421
1422       if ( $col_info->{auto_nextval} ) {
1423         $upd->{$col} = $to_insert->{$col} = $self->_sequence_fetch(
1424           'nextval',
1425           $col_info->{sequence} ||=
1426             $self->_dbh_get_autoinc_seq($self->_get_dbh, $source, $col)
1427         );
1428       }
1429     }
1430   }
1431
1432   return $upd;
1433 }
1434
1435 sub insert {
1436   my $self = shift;
1437   my ($source, $to_insert, $opts) = @_;
1438
1439   my $updated_cols = $self->_prefetch_insert_auto_nextvals (@_);
1440
1441   my $bind_attributes = $self->source_bind_attributes($source);
1442
1443   my ($rv, $sth) = $self->_execute('insert' => [], $source, $bind_attributes, $to_insert, $opts);
1444
1445   if ($opts->{returning}) {
1446     my @ret_cols = @{$opts->{returning}};
1447
1448     my @ret_vals = eval {
1449       local $SIG{__WARN__} = sub {};
1450       my @r = $sth->fetchrow_array;
1451       $sth->finish;
1452       @r;
1453     };
1454
1455     my %ret;
1456     @ret{@ret_cols} = @ret_vals if (@ret_vals);
1457
1458     $updated_cols = {
1459       %$updated_cols,
1460       %ret,
1461     };
1462   }
1463
1464   return $updated_cols;
1465 }
1466
1467 ## Currently it is assumed that all values passed will be "normal", i.e. not
1468 ## scalar refs, or at least, all the same type as the first set, the statement is
1469 ## only prepped once.
1470 sub insert_bulk {
1471   my ($self, $source, $cols, $data) = @_;
1472
1473   my %colvalues;
1474   @colvalues{@$cols} = (0..$#$cols);
1475
1476   for my $i (0..$#$cols) {
1477     my $first_val = $data->[0][$i];
1478     next unless ref $first_val eq 'SCALAR';
1479
1480     $colvalues{ $cols->[$i] } = $first_val;
1481   }
1482
1483   # check for bad data and stringify stringifiable objects
1484   my $bad_slice = sub {
1485     my ($msg, $col_idx, $slice_idx) = @_;
1486     $self->throw_exception(sprintf "%s for column '%s' in populate slice:\n%s",
1487       $msg,
1488       $cols->[$col_idx],
1489       do {
1490         local $Data::Dumper::Maxdepth = 1; # don't dump objects, if any
1491         Data::Dumper::Concise::Dumper({
1492           map { $cols->[$_] => $data->[$slice_idx][$_] } (0 .. $#$cols)
1493         }),
1494       }
1495     );
1496   };
1497
1498   for my $datum_idx (0..$#$data) {
1499     my $datum = $data->[$datum_idx];
1500
1501     for my $col_idx (0..$#$cols) {
1502       my $val            = $datum->[$col_idx];
1503       my $sqla_bind      = $colvalues{ $cols->[$col_idx] };
1504       my $is_literal_sql = (ref $sqla_bind) eq 'SCALAR';
1505
1506       if ($is_literal_sql) {
1507         if (not ref $val) {
1508           $bad_slice->('bind found where literal SQL expected', $col_idx, $datum_idx);
1509         }
1510         elsif ((my $reftype = ref $val) ne 'SCALAR') {
1511           $bad_slice->("$reftype reference found where literal SQL expected",
1512             $col_idx, $datum_idx);
1513         }
1514         elsif ($$val ne $$sqla_bind){
1515           $bad_slice->("inconsistent literal SQL value, expecting: '$$sqla_bind'",
1516             $col_idx, $datum_idx);
1517         }
1518       }
1519       elsif (my $reftype = ref $val) {
1520         require overload;
1521         if (overload::Method($val, '""')) {
1522           $datum->[$col_idx] = "".$val;
1523         }
1524         else {
1525           $bad_slice->("$reftype reference found where bind expected",
1526             $col_idx, $datum_idx);
1527         }
1528       }
1529     }
1530   }
1531
1532   my ($sql, $bind) = $self->_prep_for_execute (
1533     'insert', undef, $source, [\%colvalues]
1534   );
1535   my @bind = @$bind;
1536
1537   my $empty_bind = 1 if (not @bind) &&
1538     (grep { ref $_ eq 'SCALAR' } values %colvalues) == @$cols;
1539
1540   if ((not @bind) && (not $empty_bind)) {
1541     $self->throw_exception(
1542       'Cannot insert_bulk without support for placeholders'
1543     );
1544   }
1545
1546   # neither _execute_array, nor _execute_inserts_with_no_binds are
1547   # atomic (even if _execute _array is a single call). Thus a safety
1548   # scope guard
1549   my $guard = $self->txn_scope_guard;
1550
1551   $self->_query_start( $sql, ['__BULK__'] );
1552   my $sth = $self->sth($sql);
1553   my $rv = do {
1554     if ($empty_bind) {
1555       # bind_param_array doesn't work if there are no binds
1556       $self->_dbh_execute_inserts_with_no_binds( $sth, scalar @$data );
1557     }
1558     else {
1559 #      @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
1560       $self->_execute_array( $source, $sth, \@bind, $cols, $data );
1561     }
1562   };
1563
1564   $self->_query_end( $sql, ['__BULK__'] );
1565
1566   $guard->commit;
1567
1568   return (wantarray ? ($rv, $sth, @bind) : $rv);
1569 }
1570
1571 sub _execute_array {
1572   my ($self, $source, $sth, $bind, $cols, $data, @extra) = @_;
1573
1574   ## This must be an arrayref, else nothing works!
1575   my $tuple_status = [];
1576
1577   ## Get the bind_attributes, if any exist
1578   my $bind_attributes = $self->source_bind_attributes($source);
1579
1580   ## Bind the values and execute
1581   my $placeholder_index = 1;
1582
1583   foreach my $bound (@$bind) {
1584
1585     my $attributes = {};
1586     my ($column_name, $data_index) = @$bound;
1587
1588     if( $bind_attributes ) {
1589       $attributes = $bind_attributes->{$column_name}
1590       if defined $bind_attributes->{$column_name};
1591     }
1592
1593     my @data = map { $_->[$data_index] } @$data;
1594
1595     $sth->bind_param_array(
1596       $placeholder_index,
1597       [@data],
1598       (%$attributes ?  $attributes : ()),
1599     );
1600     $placeholder_index++;
1601   }
1602
1603   my $rv = eval {
1604     $self->_dbh_execute_array($sth, $tuple_status, @extra);
1605   };
1606   my $err = $@ || $sth->errstr;
1607
1608 # Statement must finish even if there was an exception.
1609   eval { $sth->finish };
1610   $err = $@ unless $err;
1611
1612   if ($err) {
1613     my $i = 0;
1614     ++$i while $i <= $#$tuple_status && !ref $tuple_status->[$i];
1615
1616     $self->throw_exception("Unexpected populate error: $err")
1617       if ($i > $#$tuple_status);
1618
1619     $self->throw_exception(sprintf "%s for populate slice:\n%s",
1620       ($tuple_status->[$i][1] || $err),
1621       Data::Dumper::Concise::Dumper({
1622         map { $cols->[$_] => $data->[$i][$_] } (0 .. $#$cols)
1623       }),
1624     );
1625   }
1626   return $rv;
1627 }
1628
1629 sub _dbh_execute_array {
1630     my ($self, $sth, $tuple_status, @extra) = @_;
1631
1632     return $sth->execute_array({ArrayTupleStatus => $tuple_status});
1633 }
1634
1635 sub _dbh_execute_inserts_with_no_binds {
1636   my ($self, $sth, $count) = @_;
1637
1638   eval {
1639     my $dbh = $self->_get_dbh;
1640     local $dbh->{RaiseError} = 1;
1641     local $dbh->{PrintError} = 0;
1642
1643     $sth->execute foreach 1..$count;
1644   };
1645   my $exception = $@;
1646
1647 # Make sure statement is finished even if there was an exception.
1648   eval { $sth->finish };
1649   $exception = $@ unless $exception;
1650
1651   $self->throw_exception($exception) if $exception;
1652
1653   return $count;
1654 }
1655
1656 sub update {
1657   my ($self, $source, @args) = @_;
1658
1659   my $bind_attrs = $self->source_bind_attributes($source);
1660
1661   return $self->_execute('update' => [], $source, $bind_attrs, @args);
1662 }
1663
1664
1665 sub delete {
1666   my ($self, $source, @args) = @_;
1667
1668   my $bind_attrs = $self->source_bind_attributes($source);
1669
1670   return $self->_execute('delete' => [], $source, $bind_attrs, @args);
1671 }
1672
1673 # We were sent here because the $rs contains a complex search
1674 # which will require a subquery to select the correct rows
1675 # (i.e. joined or limited resultsets, or non-introspectable conditions)
1676 #
1677 # Generating a single PK column subquery is trivial and supported
1678 # by all RDBMS. However if we have a multicolumn PK, things get ugly.
1679 # Look at _multipk_update_delete()
1680 sub _subq_update_delete {
1681   my $self = shift;
1682   my ($rs, $op, $values) = @_;
1683
1684   my $rsrc = $rs->result_source;
1685
1686   # quick check if we got a sane rs on our hands
1687   my @pcols = $rsrc->_pri_cols;
1688
1689   my $sel = $rs->_resolved_attrs->{select};
1690   $sel = [ $sel ] unless ref $sel eq 'ARRAY';
1691
1692   if (
1693       join ("\x00", map { join '.', $rs->{attrs}{alias}, $_ } sort @pcols)
1694         ne
1695       join ("\x00", sort @$sel )
1696   ) {
1697     $self->throw_exception (
1698       '_subq_update_delete can not be called on resultsets selecting columns other than the primary keys'
1699     );
1700   }
1701
1702   if (@pcols == 1) {
1703     return $self->$op (
1704       $rsrc,
1705       $op eq 'update' ? $values : (),
1706       { $pcols[0] => { -in => $rs->as_query } },
1707     );
1708   }
1709
1710   else {
1711     return $self->_multipk_update_delete (@_);
1712   }
1713 }
1714
1715 # ANSI SQL does not provide a reliable way to perform a multicol-PK
1716 # resultset update/delete involving subqueries. So by default resort
1717 # to simple (and inefficient) delete_all style per-row opearations,
1718 # while allowing specific storages to override this with a faster
1719 # implementation.
1720 #
1721 sub _multipk_update_delete {
1722   return shift->_per_row_update_delete (@_);
1723 }
1724
1725 # This is the default loop used to delete/update rows for multi PK
1726 # resultsets, and used by mysql exclusively (because it can't do anything
1727 # else).
1728 #
1729 # We do not use $row->$op style queries, because resultset update/delete
1730 # is not expected to cascade (this is what delete_all/update_all is for).
1731 #
1732 # There should be no race conditions as the entire operation is rolled
1733 # in a transaction.
1734 #
1735 sub _per_row_update_delete {
1736   my $self = shift;
1737   my ($rs, $op, $values) = @_;
1738
1739   my $rsrc = $rs->result_source;
1740   my @pcols = $rsrc->_pri_cols;
1741
1742   my $guard = $self->txn_scope_guard;
1743
1744   # emulate the return value of $sth->execute for non-selects
1745   my $row_cnt = '0E0';
1746
1747   my $subrs_cur = $rs->cursor;
1748   my @all_pk = $subrs_cur->all;
1749   for my $pks ( @all_pk) {
1750
1751     my $cond;
1752     for my $i (0.. $#pcols) {
1753       $cond->{$pcols[$i]} = $pks->[$i];
1754     }
1755
1756     $self->$op (
1757       $rsrc,
1758       $op eq 'update' ? $values : (),
1759       $cond,
1760     );
1761
1762     $row_cnt++;
1763   }
1764
1765   $guard->commit;
1766
1767   return $row_cnt;
1768 }
1769
1770 sub _select {
1771   my $self = shift;
1772
1773   # localization is neccessary as
1774   # 1) there is no infrastructure to pass this around before SQLA2
1775   # 2) _select_args sets it and _prep_for_execute consumes it
1776   my $sql_maker = $self->sql_maker;
1777   local $sql_maker->{_dbic_rs_attrs};
1778
1779   return $self->_execute($self->_select_args(@_));
1780 }
1781
1782 sub _select_args_to_query {
1783   my $self = shift;
1784
1785   # localization is neccessary as
1786   # 1) there is no infrastructure to pass this around before SQLA2
1787   # 2) _select_args sets it and _prep_for_execute consumes it
1788   my $sql_maker = $self->sql_maker;
1789   local $sql_maker->{_dbic_rs_attrs};
1790
1791   # my ($op, $bind, $ident, $bind_attrs, $select, $cond, $order, $rows, $offset)
1792   #  = $self->_select_args($ident, $select, $cond, $attrs);
1793   my ($op, $bind, $ident, $bind_attrs, @args) =
1794     $self->_select_args(@_);
1795
1796   # my ($sql, $prepared_bind) = $self->_prep_for_execute($op, $bind, $ident, [ $select, $cond, $order, $rows, $offset ]);
1797   my ($sql, $prepared_bind) = $self->_prep_for_execute($op, $bind, $ident, \@args);
1798   $prepared_bind ||= [];
1799
1800   return wantarray
1801     ? ($sql, $prepared_bind, $bind_attrs)
1802     : \[ "($sql)", @$prepared_bind ]
1803   ;
1804 }
1805
1806 sub _select_args {
1807   my ($self, $ident, $select, $where, $attrs) = @_;
1808
1809   my ($alias2source, $rs_alias) = $self->_resolve_ident_sources ($ident);
1810
1811   my $sql_maker = $self->sql_maker;
1812   $sql_maker->{_dbic_rs_attrs} = {
1813     %$attrs,
1814     select => $select,
1815     from => $ident,
1816     where => $where,
1817     $rs_alias && $alias2source->{$rs_alias}
1818       ? ( _source_handle => $alias2source->{$rs_alias}->handle )
1819       : ()
1820     ,
1821   };
1822
1823   # calculate bind_attrs before possible $ident mangling
1824   my $bind_attrs = {};
1825   for my $alias (keys %$alias2source) {
1826     my $bindtypes = $self->source_bind_attributes ($alias2source->{$alias}) || {};
1827     for my $col (keys %$bindtypes) {
1828
1829       my $fqcn = join ('.', $alias, $col);
1830       $bind_attrs->{$fqcn} = $bindtypes->{$col} if $bindtypes->{$col};
1831
1832       # Unqialified column names are nice, but at the same time can be
1833       # rather ambiguous. What we do here is basically go along with
1834       # the loop, adding an unqualified column slot to $bind_attrs,
1835       # alongside the fully qualified name. As soon as we encounter
1836       # another column by that name (which would imply another table)
1837       # we unset the unqualified slot and never add any info to it
1838       # to avoid erroneous type binding. If this happens the users
1839       # only choice will be to fully qualify his column name
1840
1841       if (exists $bind_attrs->{$col}) {
1842         $bind_attrs->{$col} = {};
1843       }
1844       else {
1845         $bind_attrs->{$col} = $bind_attrs->{$fqcn};
1846       }
1847     }
1848   }
1849
1850   # adjust limits
1851   if (
1852     $attrs->{software_limit}
1853       ||
1854     $sql_maker->_default_limit_syntax eq "GenericSubQ"
1855   ) {
1856     $attrs->{software_limit} = 1;
1857   }
1858   else {
1859     $self->throw_exception("rows attribute must be positive if present")
1860       if (defined($attrs->{rows}) && !($attrs->{rows} > 0));
1861
1862     # MySQL actually recommends this approach.  I cringe.
1863     $attrs->{rows} = 2**48 if not defined $attrs->{rows} and defined $attrs->{offset};
1864   }
1865
1866   my @limit;
1867
1868   # see if we need to tear the prefetch apart otherwise delegate the limiting to the
1869   # storage, unless software limit was requested
1870   if (
1871     #limited has_many
1872     ( $attrs->{rows} && keys %{$attrs->{collapse}} )
1873        ||
1874     # limited prefetch with RNO subqueries
1875     (
1876       $attrs->{rows}
1877         &&
1878       $sql_maker->limit_dialect eq 'RowNumberOver'
1879         &&
1880       $attrs->{_prefetch_select}
1881         &&
1882       @{$attrs->{_prefetch_select}}
1883     )
1884       ||
1885     # grouped prefetch
1886     ( $attrs->{group_by}
1887         &&
1888       @{$attrs->{group_by}}
1889         &&
1890       $attrs->{_prefetch_select}
1891         &&
1892       @{$attrs->{_prefetch_select}}
1893     )
1894   ) {
1895     ($ident, $select, $where, $attrs)
1896       = $self->_adjust_select_args_for_complex_prefetch ($ident, $select, $where, $attrs);
1897   }
1898
1899   elsif (
1900     ($attrs->{rows} || $attrs->{offset})
1901       &&
1902     $sql_maker->limit_dialect eq 'RowNumberOver'
1903       &&
1904     (ref $ident eq 'ARRAY' && @$ident > 1)  # indicates a join
1905       &&
1906     scalar $self->_parse_order_by ($attrs->{order_by})
1907   ) {
1908     # the RNO limit dialect above mangles the SQL such that the join gets lost
1909     # wrap a subquery here
1910
1911     push @limit, delete @{$attrs}{qw/rows offset/};
1912
1913     my $subq = $self->_select_args_to_query (
1914       $ident,
1915       $select,
1916       $where,
1917       $attrs,
1918     );
1919
1920     $ident = {
1921       -alias => $attrs->{alias},
1922       -source_handle => $ident->[0]{-source_handle},
1923       $attrs->{alias} => $subq,
1924     };
1925
1926     # all part of the subquery now
1927     delete @{$attrs}{qw/order_by group_by having/};
1928     $where = undef;
1929   }
1930
1931   elsif (! $attrs->{software_limit} ) {
1932     push @limit, $attrs->{rows}, $attrs->{offset};
1933   }
1934
1935   # try to simplify the joinmap further (prune unreferenced type-single joins)
1936   $ident = $self->_prune_unused_joins ($ident, $select, $where, $attrs);
1937
1938 ###
1939   # This would be the point to deflate anything found in $where
1940   # (and leave $attrs->{bind} intact). Problem is - inflators historically
1941   # expect a row object. And all we have is a resultsource (it is trivial
1942   # to extract deflator coderefs via $alias2source above).
1943   #
1944   # I don't see a way forward other than changing the way deflators are
1945   # invoked, and that's just bad...
1946 ###
1947
1948   my $order = { map
1949     { $attrs->{$_} ? ( $_ => $attrs->{$_} ) : ()  }
1950     (qw/order_by group_by having/ )
1951   };
1952
1953   return ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $where, $order, @limit);
1954 }
1955
1956 # Returns a counting SELECT for a simple count
1957 # query. Abstracted so that a storage could override
1958 # this to { count => 'firstcol' } or whatever makes
1959 # sense as a performance optimization
1960 sub _count_select {
1961   #my ($self, $source, $rs_attrs) = @_;
1962   return { count => '*' };
1963 }
1964
1965 # Returns a SELECT which will end up in the subselect
1966 # There may or may not be a group_by, as the subquery
1967 # might have been called to accomodate a limit
1968 #
1969 # Most databases would be happy with whatever ends up
1970 # here, but some choke in various ways.
1971 #
1972 sub _subq_count_select {
1973   my ($self, $source, $rs_attrs) = @_;
1974
1975   if (my $groupby = $rs_attrs->{group_by}) {
1976
1977     my $avail_columns = $self->_resolve_column_info ($rs_attrs->{from});
1978
1979     my $sel_index;
1980     for my $sel (@{$rs_attrs->{select}}) {
1981       if (ref $sel eq 'HASH' and $sel->{-as}) {
1982         $sel_index->{$sel->{-as}} = $sel;
1983       }
1984     }
1985
1986     my @selection;
1987     for my $g_part (@$groupby) {
1988       if (ref $g_part or $avail_columns->{$g_part}) {
1989         push @selection, $g_part;
1990       }
1991       elsif ($sel_index->{$g_part}) {
1992         push @selection, $sel_index->{$g_part};
1993       }
1994       else {
1995         $self->throw_exception ("group_by criteria '$g_part' not contained within current resultset source(s)");
1996       }
1997     }
1998
1999     return \@selection;
2000   }
2001
2002   my @pcols = map { join '.', $rs_attrs->{alias}, $_ } ($source->primary_columns);
2003   return @pcols ? \@pcols : [ 1 ];
2004 }
2005
2006 sub source_bind_attributes {
2007   my ($self, $source) = @_;
2008
2009   my $bind_attributes;
2010   foreach my $column ($source->columns) {
2011
2012     my $data_type = $source->column_info($column)->{data_type} || '';
2013     $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
2014      if $data_type;
2015   }
2016
2017   return $bind_attributes;
2018 }
2019
2020 =head2 select
2021
2022 =over 4
2023
2024 =item Arguments: $ident, $select, $condition, $attrs
2025
2026 =back
2027
2028 Handle a SQL select statement.
2029
2030 =cut
2031
2032 sub select {
2033   my $self = shift;
2034   my ($ident, $select, $condition, $attrs) = @_;
2035   return $self->cursor_class->new($self, \@_, $attrs);
2036 }
2037
2038 sub select_single {
2039   my $self = shift;
2040   my ($rv, $sth, @bind) = $self->_select(@_);
2041   my @row = $sth->fetchrow_array;
2042   my @nextrow = $sth->fetchrow_array if @row;
2043   if(@row && @nextrow) {
2044     carp "Query returned more than one row.  SQL that returns multiple rows is DEPRECATED for ->find and ->single";
2045   }
2046   # Need to call finish() to work round broken DBDs
2047   $sth->finish();
2048   return @row;
2049 }
2050
2051 =head2 sth
2052
2053 =over 4
2054
2055 =item Arguments: $sql
2056
2057 =back
2058
2059 Returns a L<DBI> sth (statement handle) for the supplied SQL.
2060
2061 =cut
2062
2063 sub _dbh_sth {
2064   my ($self, $dbh, $sql) = @_;
2065
2066   # 3 is the if_active parameter which avoids active sth re-use
2067   my $sth = $self->disable_sth_caching
2068     ? $dbh->prepare($sql)
2069     : $dbh->prepare_cached($sql, {}, 3);
2070
2071   # XXX You would think RaiseError would make this impossible,
2072   #  but apparently that's not true :(
2073   $self->throw_exception($dbh->errstr) if !$sth;
2074
2075   $sth;
2076 }
2077
2078 sub sth {
2079   my ($self, $sql) = @_;
2080   $self->dbh_do('_dbh_sth', $sql);  # retry over disconnects
2081 }
2082
2083 sub _dbh_columns_info_for {
2084   my ($self, $dbh, $table) = @_;
2085
2086   if ($dbh->can('column_info')) {
2087     my %result;
2088     eval {
2089       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
2090       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
2091       $sth->execute();
2092       while ( my $info = $sth->fetchrow_hashref() ){
2093         my %column_info;
2094         $column_info{data_type}   = $info->{TYPE_NAME};
2095         $column_info{size}      = $info->{COLUMN_SIZE};
2096         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
2097         $column_info{default_value} = $info->{COLUMN_DEF};
2098         my $col_name = $info->{COLUMN_NAME};
2099         $col_name =~ s/^\"(.*)\"$/$1/;
2100
2101         $result{$col_name} = \%column_info;
2102       }
2103     };
2104     return \%result if !$@ && scalar keys %result;
2105   }
2106
2107   my %result;
2108   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
2109   $sth->execute;
2110   my @columns = @{$sth->{NAME_lc}};
2111   for my $i ( 0 .. $#columns ){
2112     my %column_info;
2113     $column_info{data_type} = $sth->{TYPE}->[$i];
2114     $column_info{size} = $sth->{PRECISION}->[$i];
2115     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
2116
2117     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
2118       $column_info{data_type} = $1;
2119       $column_info{size}    = $2;
2120     }
2121
2122     $result{$columns[$i]} = \%column_info;
2123   }
2124   $sth->finish;
2125
2126   foreach my $col (keys %result) {
2127     my $colinfo = $result{$col};
2128     my $type_num = $colinfo->{data_type};
2129     my $type_name;
2130     if(defined $type_num && $dbh->can('type_info')) {
2131       my $type_info = $dbh->type_info($type_num);
2132       $type_name = $type_info->{TYPE_NAME} if $type_info;
2133       $colinfo->{data_type} = $type_name if $type_name;
2134     }
2135   }
2136
2137   return \%result;
2138 }
2139
2140 sub columns_info_for {
2141   my ($self, $table) = @_;
2142   $self->_dbh_columns_info_for ($self->_get_dbh, $table);
2143 }
2144
2145 =head2 last_insert_id
2146
2147 Return the row id of the last insert.
2148
2149 =cut
2150
2151 sub _dbh_last_insert_id {
2152     my ($self, $dbh, $source, $col) = @_;
2153
2154     my $id = eval { $dbh->last_insert_id (undef, undef, $source->name, $col) };
2155
2156     return $id if defined $id;
2157
2158     my $class = ref $self;
2159     $self->throw_exception ("No storage specific _dbh_last_insert_id() method implemented in $class, and the generic DBI::last_insert_id() failed");
2160 }
2161
2162 sub last_insert_id {
2163   my $self = shift;
2164   $self->_dbh_last_insert_id ($self->_dbh, @_);
2165 }
2166
2167 =head2 _native_data_type
2168
2169 =over 4
2170
2171 =item Arguments: $type_name
2172
2173 =back
2174
2175 This API is B<EXPERIMENTAL>, will almost definitely change in the future, and
2176 currently only used by L<::AutoCast|DBIx::Class::Storage::DBI::AutoCast> and
2177 L<::Sybase::ASE|DBIx::Class::Storage::DBI::Sybase::ASE>.
2178
2179 The default implementation returns C<undef>, implement in your Storage driver if
2180 you need this functionality.
2181
2182 Should map types from other databases to the native RDBMS type, for example
2183 C<VARCHAR2> to C<VARCHAR>.
2184
2185 Types with modifiers should map to the underlying data type. For example,
2186 C<INTEGER AUTO_INCREMENT> should become C<INTEGER>.
2187
2188 Composite types should map to the container type, for example
2189 C<ENUM(foo,bar,baz)> becomes C<ENUM>.
2190
2191 =cut
2192
2193 sub _native_data_type {
2194   #my ($self, $data_type) = @_;
2195   return undef
2196 }
2197
2198 # Check if placeholders are supported at all
2199 sub _placeholders_supported {
2200   my $self = shift;
2201   my $dbh  = $self->_get_dbh;
2202
2203   # some drivers provide a $dbh attribute (e.g. Sybase and $dbh->{syb_dynamic_supported})
2204   # but it is inaccurate more often than not
2205   eval {
2206     local $dbh->{PrintError} = 0;
2207     local $dbh->{RaiseError} = 1;
2208     $dbh->do('select ?', {}, 1);
2209   };
2210   return $@ ? 0 : 1;
2211 }
2212
2213 # Check if placeholders bound to non-string types throw exceptions
2214 #
2215 sub _typeless_placeholders_supported {
2216   my $self = shift;
2217   my $dbh  = $self->_get_dbh;
2218
2219   eval {
2220     local $dbh->{PrintError} = 0;
2221     local $dbh->{RaiseError} = 1;
2222     # this specifically tests a bind that is NOT a string
2223     $dbh->do('select 1 where 1 = ?', {}, 1);
2224   };
2225   return $@ ? 0 : 1;
2226 }
2227
2228 =head2 sqlt_type
2229
2230 Returns the database driver name.
2231
2232 =cut
2233
2234 sub sqlt_type {
2235   shift->_get_dbh->{Driver}->{Name};
2236 }
2237
2238 =head2 bind_attribute_by_data_type
2239
2240 Given a datatype from column info, returns a database specific bind
2241 attribute for C<< $dbh->bind_param($val,$attribute) >> or nothing if we will
2242 let the database planner just handle it.
2243
2244 Generally only needed for special case column types, like bytea in postgres.
2245
2246 =cut
2247
2248 sub bind_attribute_by_data_type {
2249     return;
2250 }
2251
2252 =head2 is_datatype_numeric
2253
2254 Given a datatype from column_info, returns a boolean value indicating if
2255 the current RDBMS considers it a numeric value. This controls how
2256 L<DBIx::Class::Row/set_column> decides whether to mark the column as
2257 dirty - when the datatype is deemed numeric a C<< != >> comparison will
2258 be performed instead of the usual C<eq>.
2259
2260 =cut
2261
2262 sub is_datatype_numeric {
2263   my ($self, $dt) = @_;
2264
2265   return 0 unless $dt;
2266
2267   return $dt =~ /^ (?:
2268     numeric | int(?:eger)? | (?:tiny|small|medium|big)int | dec(?:imal)? | real | float | double (?: \s+ precision)? | (?:big)?serial
2269   ) $/ix;
2270 }
2271
2272
2273 =head2 create_ddl_dir
2274
2275 =over 4
2276
2277 =item Arguments: $schema \@databases, $version, $directory, $preversion, \%sqlt_args
2278
2279 =back
2280
2281 Creates a SQL file based on the Schema, for each of the specified
2282 database engines in C<\@databases> in the given directory.
2283 (note: specify L<SQL::Translator> names, not L<DBI> driver names).
2284
2285 Given a previous version number, this will also create a file containing
2286 the ALTER TABLE statements to transform the previous schema into the
2287 current one. Note that these statements may contain C<DROP TABLE> or
2288 C<DROP COLUMN> statements that can potentially destroy data.
2289
2290 The file names are created using the C<ddl_filename> method below, please
2291 override this method in your schema if you would like a different file
2292 name format. For the ALTER file, the same format is used, replacing
2293 $version in the name with "$preversion-$version".
2294
2295 See L<SQL::Translator/METHODS> for a list of values for C<\%sqlt_args>.
2296 The most common value for this would be C<< { add_drop_table => 1 } >>
2297 to have the SQL produced include a C<DROP TABLE> statement for each table
2298 created. For quoting purposes supply C<quote_table_names> and
2299 C<quote_field_names>.
2300
2301 If no arguments are passed, then the following default values are assumed:
2302
2303 =over 4
2304
2305 =item databases  - ['MySQL', 'SQLite', 'PostgreSQL']
2306
2307 =item version    - $schema->schema_version
2308
2309 =item directory  - './'
2310
2311 =item preversion - <none>
2312
2313 =back
2314
2315 By default, C<\%sqlt_args> will have
2316
2317  { add_drop_table => 1, ignore_constraint_names => 1, ignore_index_names => 1 }
2318
2319 merged with the hash passed in. To disable any of those features, pass in a
2320 hashref like the following
2321
2322  { ignore_constraint_names => 0, # ... other options }
2323
2324
2325 WARNING: You are strongly advised to check all SQL files created, before applying
2326 them.
2327
2328 =cut
2329
2330 sub create_ddl_dir {
2331   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
2332
2333   unless ($dir) {
2334     carp "No directory given, using ./\n";
2335     $dir = './';
2336   }
2337
2338   $self->throw_exception ("Directory '$dir' does not exist\n") unless(-d $dir);
2339
2340   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
2341   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
2342
2343   my $schema_version = $schema->schema_version || '1.x';
2344   $version ||= $schema_version;
2345
2346   $sqltargs = {
2347     add_drop_table => 1,
2348     ignore_constraint_names => 1,
2349     ignore_index_names => 1,
2350     %{$sqltargs || {}}
2351   };
2352
2353   unless (DBIx::Class::Optional::Dependencies->req_ok_for ('deploy')) {
2354     $self->throw_exception("Can't create a ddl file without " . DBIx::Class::Optional::Dependencies->req_missing_for ('deploy') );
2355   }
2356
2357   my $sqlt = SQL::Translator->new( $sqltargs );
2358
2359   $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
2360   my $sqlt_schema = $sqlt->translate({ data => $schema })
2361     or $self->throw_exception ($sqlt->error);
2362
2363   foreach my $db (@$databases) {
2364     $sqlt->reset();
2365     $sqlt->{schema} = $sqlt_schema;
2366     $sqlt->producer($db);
2367
2368     my $file;
2369     my $filename = $schema->ddl_filename($db, $version, $dir);
2370     if (-e $filename && ($version eq $schema_version )) {
2371       # if we are dumping the current version, overwrite the DDL
2372       carp "Overwriting existing DDL file - $filename";
2373       unlink($filename);
2374     }
2375
2376     my $output = $sqlt->translate;
2377     if(!$output) {
2378       carp("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
2379       next;
2380     }
2381     if(!open($file, ">$filename")) {
2382       $self->throw_exception("Can't open $filename for writing ($!)");
2383       next;
2384     }
2385     print $file $output;
2386     close($file);
2387
2388     next unless ($preversion);
2389
2390     require SQL::Translator::Diff;
2391
2392     my $prefilename = $schema->ddl_filename($db, $preversion, $dir);
2393     if(!-e $prefilename) {
2394       carp("No previous schema file found ($prefilename)");
2395       next;
2396     }
2397
2398     my $difffile = $schema->ddl_filename($db, $version, $dir, $preversion);
2399     if(-e $difffile) {
2400       carp("Overwriting existing diff file - $difffile");
2401       unlink($difffile);
2402     }
2403
2404     my $source_schema;
2405     {
2406       my $t = SQL::Translator->new($sqltargs);
2407       $t->debug( 0 );
2408       $t->trace( 0 );
2409
2410       $t->parser( $db )
2411         or $self->throw_exception ($t->error);
2412
2413       my $out = $t->translate( $prefilename )
2414         or $self->throw_exception ($t->error);
2415
2416       $source_schema = $t->schema;
2417
2418       $source_schema->name( $prefilename )
2419         unless ( $source_schema->name );
2420     }
2421
2422     # The "new" style of producers have sane normalization and can support
2423     # diffing a SQL file against a DBIC->SQLT schema. Old style ones don't
2424     # And we have to diff parsed SQL against parsed SQL.
2425     my $dest_schema = $sqlt_schema;
2426
2427     unless ( "SQL::Translator::Producer::$db"->can('preprocess_schema') ) {
2428       my $t = SQL::Translator->new($sqltargs);
2429       $t->debug( 0 );
2430       $t->trace( 0 );
2431
2432       $t->parser( $db )
2433         or $self->throw_exception ($t->error);
2434
2435       my $out = $t->translate( $filename )
2436         or $self->throw_exception ($t->error);
2437
2438       $dest_schema = $t->schema;
2439
2440       $dest_schema->name( $filename )
2441         unless $dest_schema->name;
2442     }
2443
2444     my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
2445                                                   $dest_schema,   $db,
2446                                                   $sqltargs
2447                                                  );
2448     if(!open $file, ">$difffile") {
2449       $self->throw_exception("Can't write to $difffile ($!)");
2450       next;
2451     }
2452     print $file $diff;
2453     close($file);
2454   }
2455 }
2456
2457 =head2 deployment_statements
2458
2459 =over 4
2460
2461 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
2462
2463 =back
2464
2465 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
2466
2467 The L<SQL::Translator> (not L<DBI>) database driver name can be explicitly
2468 provided in C<$type>, otherwise the result of L</sqlt_type> is used as default.
2469
2470 C<$directory> is used to return statements from files in a previously created
2471 L</create_ddl_dir> directory and is optional. The filenames are constructed
2472 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
2473
2474 If no C<$directory> is specified then the statements are constructed on the
2475 fly using L<SQL::Translator> and C<$version> is ignored.
2476
2477 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
2478
2479 =cut
2480
2481 sub deployment_statements {
2482   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
2483   $type ||= $self->sqlt_type;
2484   $version ||= $schema->schema_version || '1.x';
2485   $dir ||= './';
2486   my $filename = $schema->ddl_filename($type, $version, $dir);
2487   if(-f $filename)
2488   {
2489       my $file;
2490       open($file, "<$filename")
2491         or $self->throw_exception("Can't open $filename ($!)");
2492       my @rows = <$file>;
2493       close($file);
2494       return join('', @rows);
2495   }
2496
2497   unless (DBIx::Class::Optional::Dependencies->req_ok_for ('deploy') ) {
2498     $self->throw_exception("Can't deploy without a ddl_dir or " . DBIx::Class::Optional::Dependencies->req_missing_for ('deploy') );
2499   }
2500
2501   # sources needs to be a parser arg, but for simplicty allow at top level
2502   # coming in
2503   $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
2504       if exists $sqltargs->{sources};
2505
2506   my $tr = SQL::Translator->new(
2507     producer => "SQL::Translator::Producer::${type}",
2508     %$sqltargs,
2509     parser => 'SQL::Translator::Parser::DBIx::Class',
2510     data => $schema,
2511   );
2512
2513   my @ret;
2514   my $wa = wantarray;
2515   if ($wa) {
2516     @ret = $tr->translate;
2517   }
2518   else {
2519     $ret[0] = $tr->translate;
2520   }
2521
2522   $self->throw_exception( 'Unable to produce deployment statements: ' . $tr->error)
2523     unless (@ret && defined $ret[0]);
2524
2525   return $wa ? @ret : $ret[0];
2526 }
2527
2528 sub deploy {
2529   my ($self, $schema, $type, $sqltargs, $dir) = @_;
2530   my $deploy = sub {
2531     my $line = shift;
2532     return if($line =~ /^--/);
2533     return if(!$line);
2534     # next if($line =~ /^DROP/m);
2535     return if($line =~ /^BEGIN TRANSACTION/m);
2536     return if($line =~ /^COMMIT/m);
2537     return if $line =~ /^\s+$/; # skip whitespace only
2538     $self->_query_start($line);
2539     eval {
2540       # do a dbh_do cycle here, as we need some error checking in
2541       # place (even though we will ignore errors)
2542       $self->dbh_do (sub { $_[1]->do($line) });
2543     };
2544     if ($@) {
2545       carp qq{$@ (running "${line}")};
2546     }
2547     $self->_query_end($line);
2548   };
2549   my @statements = $schema->deployment_statements($type, undef, $dir, { %{ $sqltargs || {} }, no_comments => 1 } );
2550   if (@statements > 1) {
2551     foreach my $statement (@statements) {
2552       $deploy->( $statement );
2553     }
2554   }
2555   elsif (@statements == 1) {
2556     foreach my $line ( split(";\n", $statements[0])) {
2557       $deploy->( $line );
2558     }
2559   }
2560 }
2561
2562 =head2 datetime_parser
2563
2564 Returns the datetime parser class
2565
2566 =cut
2567
2568 sub datetime_parser {
2569   my $self = shift;
2570   return $self->{datetime_parser} ||= do {
2571     $self->build_datetime_parser(@_);
2572   };
2573 }
2574
2575 =head2 datetime_parser_type
2576
2577 Defines (returns) the datetime parser class - currently hardwired to
2578 L<DateTime::Format::MySQL>
2579
2580 =cut
2581
2582 sub datetime_parser_type { "DateTime::Format::MySQL"; }
2583
2584 =head2 build_datetime_parser
2585
2586 See L</datetime_parser>
2587
2588 =cut
2589
2590 sub build_datetime_parser {
2591   my $self = shift;
2592   my $type = $self->datetime_parser_type(@_);
2593   $self->ensure_class_loaded ($type);
2594   return $type;
2595 }
2596
2597
2598 =head2 is_replicating
2599
2600 A boolean that reports if a particular L<DBIx::Class::Storage::DBI> is set to
2601 replicate from a master database.  Default is undef, which is the result
2602 returned by databases that don't support replication.
2603
2604 =cut
2605
2606 sub is_replicating {
2607     return;
2608
2609 }
2610
2611 =head2 lag_behind_master
2612
2613 Returns a number that represents a certain amount of lag behind a master db
2614 when a given storage is replicating.  The number is database dependent, but
2615 starts at zero and increases with the amount of lag. Default in undef
2616
2617 =cut
2618
2619 sub lag_behind_master {
2620     return;
2621 }
2622
2623 =head2 relname_to_table_alias
2624
2625 =over 4
2626
2627 =item Arguments: $relname, $join_count
2628
2629 =back
2630
2631 L<DBIx::Class> uses L<DBIx::Class::Relationship> names as table aliases in
2632 queries.
2633
2634 This hook is to allow specific L<DBIx::Class::Storage> drivers to change the
2635 way these aliases are named.
2636
2637 The default behavior is C<< "$relname_$join_count" if $join_count > 1 >>,
2638 otherwise C<"$relname">.
2639
2640 =cut
2641
2642 sub relname_to_table_alias {
2643   my ($self, $relname, $join_count) = @_;
2644
2645   my $alias = ($join_count && $join_count > 1 ?
2646     join('_', $relname, $join_count) : $relname);
2647
2648   return $alias;
2649 }
2650
2651 sub DESTROY {
2652   my $self = shift;
2653
2654   $self->_verify_pid if $self->_dbh;
2655
2656   # some databases need this to stop spewing warnings
2657   if (my $dbh = $self->_dbh) {
2658     local $@;
2659     eval {
2660       %{ $dbh->{CachedKids} } = ();
2661       $dbh->disconnect;
2662     };
2663   }
2664
2665   $self->_dbh(undef);
2666 }
2667
2668 1;
2669
2670 =head1 USAGE NOTES
2671
2672 =head2 DBIx::Class and AutoCommit
2673
2674 DBIx::Class can do some wonderful magic with handling exceptions,
2675 disconnections, and transactions when you use C<< AutoCommit => 1 >>
2676 (the default) combined with C<txn_do> for transaction support.
2677
2678 If you set C<< AutoCommit => 0 >> in your connect info, then you are always
2679 in an assumed transaction between commits, and you're telling us you'd
2680 like to manage that manually.  A lot of the magic protections offered by
2681 this module will go away.  We can't protect you from exceptions due to database
2682 disconnects because we don't know anything about how to restart your
2683 transactions.  You're on your own for handling all sorts of exceptional
2684 cases if you choose the C<< AutoCommit => 0 >> path, just as you would
2685 be with raw DBI.
2686
2687
2688 =head1 AUTHORS
2689
2690 Matt S. Trout <mst@shadowcatsystems.co.uk>
2691
2692 Andy Grundman <andy@hybridized.org>
2693
2694 =head1 LICENSE
2695
2696 You may distribute this code under the same terms as Perl itself.
2697
2698 =cut