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