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