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