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