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