lib/DBIx/Class/Storage/DBI.pm

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