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