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