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