lib/DBIx/Class/Storage/DBI.pm

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