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