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