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