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