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