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