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