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