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