lib/DBIx/Class/Storage/DBI.pm

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