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