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