be1d64d3648f5ad9dca8776af06f64013f947d71
[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 _conn_pid
19      _conn_tid transaction_depth _dbh_autocommit _driver_determined 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   if (ref $args[0] eq 'CODE') {
441     # _connect() never looks past $args[0] in this case
442     %attrs = ()
443   } else {
444     %attrs = (
445       %{ $self->_default_dbi_connect_attributes || {} },
446       %attrs,
447     );
448   }
449
450   $self->_dbi_connect_info([@args, keys %attrs ? \%attrs : ()]);
451   $self->_connect_info;
452 }
453
454 sub _default_dbi_connect_attributes {
455   return {
456     AutoCommit => 1,
457     RaiseError => 1,
458     PrintError => 0,
459   };
460 }
461
462 =head2 on_connect_do
463
464 This method is deprecated in favour of setting via L</connect_info>.
465
466 =cut
467
468 =head2 on_disconnect_do
469
470 This method is deprecated in favour of setting via L</connect_info>.
471
472 =cut
473
474 sub _parse_connect_do {
475   my ($self, $type) = @_;
476
477   my $val = $self->$type;
478   return () if not defined $val;
479
480   my @res;
481
482   if (not ref($val)) {
483     push @res, [ 'do_sql', $val ];
484   } elsif (ref($val) eq 'CODE') {
485     push @res, $val;
486   } elsif (ref($val) eq 'ARRAY') {
487     push @res, map { [ 'do_sql', $_ ] } @$val;
488   } else {
489     $self->throw_exception("Invalid type for $type: ".ref($val));
490   }
491
492   return \@res;
493 }
494
495 =head2 dbh_do
496
497 Arguments: ($subref | $method_name), @extra_coderef_args?
498
499 Execute the given $subref or $method_name using the new exception-based
500 connection management.
501
502 The first two arguments will be the storage object that C<dbh_do> was called
503 on and a database handle to use.  Any additional arguments will be passed
504 verbatim to the called subref as arguments 2 and onwards.
505
506 Using this (instead of $self->_dbh or $self->dbh) ensures correct
507 exception handling and reconnection (or failover in future subclasses).
508
509 Your subref should have no side-effects outside of the database, as
510 there is the potential for your subref to be partially double-executed
511 if the database connection was stale/dysfunctional.
512
513 Example:
514
515   my @stuff = $schema->storage->dbh_do(
516     sub {
517       my ($storage, $dbh, @cols) = @_;
518       my $cols = join(q{, }, @cols);
519       $dbh->selectrow_array("SELECT $cols FROM foo");
520     },
521     @column_list
522   );
523
524 =cut
525
526 sub dbh_do {
527   my $self = shift;
528   my $code = shift;
529
530   my $dbh = $self->_dbh;
531
532   return $self->$code($dbh, @_) if $self->{_in_dbh_do}
533       || $self->{transaction_depth};
534
535   local $self->{_in_dbh_do} = 1;
536
537   my @result;
538   my $want_array = wantarray;
539
540   eval {
541     $self->_verify_pid if $dbh;
542     if(!$self->_dbh) {
543         $self->_populate_dbh;
544         $dbh = $self->_dbh;
545     }
546
547     if($want_array) {
548         @result = $self->$code($dbh, @_);
549     }
550     elsif(defined $want_array) {
551         $result[0] = $self->$code($dbh, @_);
552     }
553     else {
554         $self->$code($dbh, @_);
555     }
556   };
557
558   # ->connected might unset $@ - copy
559   my $exception = $@;
560   if(!$exception) { return $want_array ? @result : $result[0] }
561
562   $self->throw_exception($exception) if $self->connected;
563
564   # We were not connected - reconnect and retry, but let any
565   #  exception fall right through this time
566   carp "Retrying $code after catching disconnected exception: $exception"
567     if $ENV{DBIC_DBIRETRY_DEBUG};
568   $self->_populate_dbh;
569   $self->$code($self->_dbh, @_);
570 }
571
572 # This is basically a blend of dbh_do above and DBIx::Class::Storage::txn_do.
573 # It also informs dbh_do to bypass itself while under the direction of txn_do,
574 #  via $self->{_in_dbh_do} (this saves some redundant eval and errorcheck, etc)
575 sub txn_do {
576   my $self = shift;
577   my $coderef = shift;
578
579   ref $coderef eq 'CODE' or $self->throw_exception
580     ('$coderef must be a CODE reference');
581
582   return $coderef->(@_) if $self->{transaction_depth} && ! $self->auto_savepoint;
583
584   local $self->{_in_dbh_do} = 1;
585
586   my @result;
587   my $want_array = wantarray;
588
589   my $tried = 0;
590   while(1) {
591     eval {
592       $self->_verify_pid if $self->_dbh;
593       $self->_populate_dbh if !$self->_dbh;
594
595       $self->txn_begin;
596       if($want_array) {
597           @result = $coderef->(@_);
598       }
599       elsif(defined $want_array) {
600           $result[0] = $coderef->(@_);
601       }
602       else {
603           $coderef->(@_);
604       }
605       $self->txn_commit;
606     };
607
608     # ->connected might unset $@ - copy
609     my $exception = $@;
610     if(!$exception) { return $want_array ? @result : $result[0] }
611
612     if($tried++ || $self->connected) {
613       eval { $self->txn_rollback };
614       my $rollback_exception = $@;
615       if($rollback_exception) {
616         my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
617         $self->throw_exception($exception)  # propagate nested rollback
618           if $rollback_exception =~ /$exception_class/;
619
620         $self->throw_exception(
621           "Transaction aborted: ${exception}. "
622           . "Rollback failed: ${rollback_exception}"
623         );
624       }
625       $self->throw_exception($exception)
626     }
627
628     # We were not connected, and was first try - reconnect and retry
629     # via the while loop
630     carp "Retrying $coderef after catching disconnected exception: $exception"
631       if $ENV{DBIC_DBIRETRY_DEBUG};
632     $self->_populate_dbh;
633   }
634 }
635
636 =head2 disconnect
637
638 Our C<disconnect> method also performs a rollback first if the
639 database is not in C<AutoCommit> mode.
640
641 =cut
642
643 sub disconnect {
644   my ($self) = @_;
645
646   if( $self->_dbh ) {
647     my @actions;
648
649     push @actions, ( $self->on_disconnect_call || () );
650     push @actions, $self->_parse_connect_do ('on_disconnect_do');
651
652     $self->_do_connection_actions(disconnect_call_ => $_) for @actions;
653
654     $self->_dbh->rollback unless $self->_dbh_autocommit;
655     $self->_dbh->disconnect;
656     $self->_dbh(undef);
657     $self->{_dbh_gen}++;
658   }
659 }
660
661 =head2 with_deferred_fk_checks
662
663 =over 4
664
665 =item Arguments: C<$coderef>
666
667 =item Return Value: The return value of $coderef
668
669 =back
670
671 Storage specific method to run the code ref with FK checks deferred or
672 in MySQL's case disabled entirely.
673
674 =cut
675
676 # Storage subclasses should override this
677 sub with_deferred_fk_checks {
678   my ($self, $sub) = @_;
679
680   $sub->();
681 }
682
683 =head2 connected
684
685 =over
686
687 =item Arguments: none
688
689 =item Return Value: 1|0
690
691 =back
692
693 Verifies that the the current database handle is active and ready to execute
694 an SQL statement (i.e. the connection did not get stale, server is still
695 answering, etc.) This method is used internally by L</dbh>.
696
697 =cut
698
699 sub connected {
700   my $self = shift;
701   return 0 unless $self->_seems_connected;
702
703   #be on the safe side
704   local $self->_dbh->{RaiseError} = 1;
705
706   return $self->_ping;
707 }
708
709 sub _seems_connected {
710   my $self = shift;
711
712   my $dbh = $self->_dbh
713     or return 0;
714
715   if(defined $self->_conn_tid && $self->_conn_tid != threads->tid) {
716     $self->_dbh(undef);
717     $self->{_dbh_gen}++;
718     return 0;
719   }
720   else {
721     $self->_verify_pid;
722     return 0 if !$self->_dbh;
723   }
724
725   return $dbh->FETCH('Active');
726 }
727
728 sub _ping {
729   my $self = shift;
730
731   my $dbh = $self->_dbh or return 0;
732
733   return $dbh->ping;
734 }
735
736 # handle pid changes correctly
737 #  NOTE: assumes $self->_dbh is a valid $dbh
738 sub _verify_pid {
739   my ($self) = @_;
740
741   return if defined $self->_conn_pid && $self->_conn_pid == $$;
742
743   $self->_dbh->{InactiveDestroy} = 1;
744   $self->_dbh(undef);
745   $self->{_dbh_gen}++;
746
747   return;
748 }
749
750 sub ensure_connected {
751   my ($self) = @_;
752
753   unless ($self->connected) {
754     $self->_populate_dbh;
755   }
756 }
757
758 =head2 dbh
759
760 Returns a C<$dbh> - a data base handle of class L<DBI>. The returned handle
761 is guaranteed to be healthy by implicitly calling L</connected>, and if
762 necessary performing a reconnection before returning. Keep in mind that this
763 is very B<expensive> on some database engines. Consider using L<dbh_do>
764 instead.
765
766 =cut
767
768 sub dbh {
769   my ($self) = @_;
770
771   if (not $self->_dbh) {
772     $self->_populate_dbh;
773   } else {
774     $self->ensure_connected;
775   }
776   return $self->_dbh;
777 }
778
779 # this is the internal "get dbh or connect (don't check)" method
780 sub _get_dbh {
781   my $self = shift;
782   $self->_populate_dbh unless $self->_dbh;
783   return $self->_dbh;
784 }
785
786 sub _sql_maker_args {
787     my ($self) = @_;
788
789     return (
790       bindtype=>'columns',
791       array_datatypes => 1,
792       limit_dialect => $self->_get_dbh,
793       %{$self->_sql_maker_opts}
794     );
795 }
796
797 sub sql_maker {
798   my ($self) = @_;
799   unless ($self->_sql_maker) {
800     my $sql_maker_class = $self->sql_maker_class;
801     $self->ensure_class_loaded ($sql_maker_class);
802     $self->_sql_maker($sql_maker_class->new( $self->_sql_maker_args ));
803   }
804   return $self->_sql_maker;
805 }
806
807 sub _rebless {}
808
809 sub _populate_dbh {
810   my ($self) = @_;
811
812   my @info = @{$self->_dbi_connect_info || []};
813   $self->_dbh(undef); # in case ->connected failed we might get sent here
814   $self->_dbh($self->_connect(@info));
815
816   $self->_conn_pid($$);
817   $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
818
819   $self->_determine_driver;
820
821   # Always set the transaction depth on connect, since
822   #  there is no transaction in progress by definition
823   $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
824
825   $self->_run_connection_actions unless $self->{_in_determine_driver};
826 }
827
828 sub _run_connection_actions {
829   my $self = shift;
830   my @actions;
831
832   push @actions, ( $self->on_connect_call || () );
833   push @actions, $self->_parse_connect_do ('on_connect_do');
834
835   $self->_do_connection_actions(connect_call_ => $_) for @actions;
836 }
837
838 sub _determine_driver {
839   my ($self) = @_;
840
841   if ((not $self->_driver_determined) && (not $self->{_in_determine_driver})) {
842     my $started_unconnected = 0;
843     local $self->{_in_determine_driver} = 1;
844
845     if (ref($self) eq __PACKAGE__) {
846       my $driver;
847       if ($self->_dbh) { # we are connected
848         $driver = $self->_dbh->{Driver}{Name};
849       } else {
850         # try to use dsn to not require being connected, the driver may still
851         # force a connection in _rebless to determine version
852         ($driver) = $self->_dbi_connect_info->[0] =~ /dbi:([^:]+):/i;
853         $started_unconnected = 1;
854       }
855
856       my $storage_class = "DBIx::Class::Storage::DBI::${driver}";
857       if ($self->load_optional_class($storage_class)) {
858         mro::set_mro($storage_class, 'c3');
859         bless $self, $storage_class;
860         $self->_rebless();
861       }
862     }
863
864     $self->_driver_determined(1);
865
866     $self->_run_connection_actions
867         if $started_unconnected && defined $self->_dbh;
868   }
869 }
870
871 sub _do_connection_actions {
872   my $self          = shift;
873   my $method_prefix = shift;
874   my $call          = shift;
875
876   if (not ref($call)) {
877     my $method = $method_prefix . $call;
878     $self->$method(@_);
879   } elsif (ref($call) eq 'CODE') {
880     $self->$call(@_);
881   } elsif (ref($call) eq 'ARRAY') {
882     if (ref($call->[0]) ne 'ARRAY') {
883       $self->_do_connection_actions($method_prefix, $_) for @$call;
884     } else {
885       $self->_do_connection_actions($method_prefix, @$_) for @$call;
886     }
887   } else {
888     $self->throw_exception (sprintf ("Don't know how to process conection actions of type '%s'", ref($call)) );
889   }
890
891   return $self;
892 }
893
894 sub connect_call_do_sql {
895   my $self = shift;
896   $self->_do_query(@_);
897 }
898
899 sub disconnect_call_do_sql {
900   my $self = shift;
901   $self->_do_query(@_);
902 }
903
904 # override in db-specific backend when necessary
905 sub connect_call_datetime_setup { 1 }
906
907 sub _do_query {
908   my ($self, $action) = @_;
909
910   if (ref $action eq 'CODE') {
911     $action = $action->($self);
912     $self->_do_query($_) foreach @$action;
913   }
914   else {
915     # Most debuggers expect ($sql, @bind), so we need to exclude
916     # the attribute hash which is the second argument to $dbh->do
917     # furthermore the bind values are usually to be presented
918     # as named arrayref pairs, so wrap those here too
919     my @do_args = (ref $action eq 'ARRAY') ? (@$action) : ($action);
920     my $sql = shift @do_args;
921     my $attrs = shift @do_args;
922     my @bind = map { [ undef, $_ ] } @do_args;
923
924     $self->_query_start($sql, @bind);
925     $self->_dbh->do($sql, $attrs, @do_args);
926     $self->_query_end($sql, @bind);
927   }
928
929   return $self;
930 }
931
932 sub _connect {
933   my ($self, @info) = @_;
934
935   $self->throw_exception("You failed to provide any connection info")
936     if !@info;
937
938   my ($old_connect_via, $dbh);
939
940   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
941     $old_connect_via = $DBI::connect_via;
942     $DBI::connect_via = 'connect';
943   }
944
945   eval {
946     if(ref $info[0] eq 'CODE') {
947        $dbh = &{$info[0]}
948     }
949     else {
950        $dbh = DBI->connect(@info);
951     }
952
953     if($dbh && !$self->unsafe) {
954       my $weak_self = $self;
955       Scalar::Util::weaken($weak_self);
956       $dbh->{HandleError} = sub {
957           if ($weak_self) {
958             $weak_self->throw_exception("DBI Exception: $_[0]");
959           }
960           else {
961             croak ("DBI Exception: $_[0]");
962           }
963       };
964       $dbh->{ShowErrorStatement} = 1;
965       $dbh->{RaiseError} = 1;
966       $dbh->{PrintError} = 0;
967     }
968   };
969
970   $DBI::connect_via = $old_connect_via if $old_connect_via;
971
972   $self->throw_exception("DBI Connection failed: " . ($@||$DBI::errstr))
973     if !$dbh || $@;
974
975   $self->_dbh_autocommit($dbh->{AutoCommit});
976
977   $dbh;
978 }
979
980 sub svp_begin {
981   my ($self, $name) = @_;
982
983   $name = $self->_svp_generate_name
984     unless defined $name;
985
986   $self->throw_exception ("You can't use savepoints outside a transaction")
987     if $self->{transaction_depth} == 0;
988
989   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
990     unless $self->can('_svp_begin');
991
992   push @{ $self->{savepoints} }, $name;
993
994   $self->debugobj->svp_begin($name) if $self->debug;
995
996   return $self->_svp_begin($name);
997 }
998
999 sub svp_release {
1000   my ($self, $name) = @_;
1001
1002   $self->throw_exception ("You can't use savepoints outside a transaction")
1003     if $self->{transaction_depth} == 0;
1004
1005   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1006     unless $self->can('_svp_release');
1007
1008   if (defined $name) {
1009     $self->throw_exception ("Savepoint '$name' does not exist")
1010       unless grep { $_ eq $name } @{ $self->{savepoints} };
1011
1012     # Dig through the stack until we find the one we are releasing.  This keeps
1013     # the stack up to date.
1014     my $svp;
1015
1016     do { $svp = pop @{ $self->{savepoints} } } while $svp ne $name;
1017   } else {
1018     $name = pop @{ $self->{savepoints} };
1019   }
1020
1021   $self->debugobj->svp_release($name) if $self->debug;
1022
1023   return $self->_svp_release($name);
1024 }
1025
1026 sub svp_rollback {
1027   my ($self, $name) = @_;
1028
1029   $self->throw_exception ("You can't use savepoints outside a transaction")
1030     if $self->{transaction_depth} == 0;
1031
1032   $self->throw_exception ("Your Storage implementation doesn't support savepoints")
1033     unless $self->can('_svp_rollback');
1034
1035   if (defined $name) {
1036       # If they passed us a name, verify that it exists in the stack
1037       unless(grep({ $_ eq $name } @{ $self->{savepoints} })) {
1038           $self->throw_exception("Savepoint '$name' does not exist!");
1039       }
1040
1041       # Dig through the stack until we find the one we are releasing.  This keeps
1042       # the stack up to date.
1043       while(my $s = pop(@{ $self->{savepoints} })) {
1044           last if($s eq $name);
1045       }
1046       # Add the savepoint back to the stack, as a rollback doesn't remove the
1047       # named savepoint, only everything after it.
1048       push(@{ $self->{savepoints} }, $name);
1049   } else {
1050       # We'll assume they want to rollback to the last savepoint
1051       $name = $self->{savepoints}->[-1];
1052   }
1053
1054   $self->debugobj->svp_rollback($name) if $self->debug;
1055
1056   return $self->_svp_rollback($name);
1057 }
1058
1059 sub _svp_generate_name {
1060     my ($self) = @_;
1061
1062     return 'savepoint_'.scalar(@{ $self->{'savepoints'} });
1063 }
1064
1065 sub txn_begin {
1066   my $self = shift;
1067   if($self->{transaction_depth} == 0) {
1068     $self->debugobj->txn_begin()
1069       if $self->debug;
1070
1071     # being here implies we have AutoCommit => 1
1072     # if the user is utilizing txn_do - good for
1073     # him, otherwise we need to ensure that the
1074     # $dbh is healthy on BEGIN
1075     my $dbh_method = $self->{_in_dbh_do} ? '_dbh' : 'dbh';
1076     $self->$dbh_method->begin_work;
1077
1078   } elsif ($self->auto_savepoint) {
1079     $self->svp_begin;
1080   }
1081   $self->{transaction_depth}++;
1082 }
1083
1084 sub txn_commit {
1085   my $self = shift;
1086   if ($self->{transaction_depth} == 1) {
1087     my $dbh = $self->_dbh;
1088     $self->debugobj->txn_commit()
1089       if ($self->debug);
1090     $dbh->commit;
1091     $self->{transaction_depth} = 0
1092       if $self->_dbh_autocommit;
1093   }
1094   elsif($self->{transaction_depth} > 1) {
1095     $self->{transaction_depth}--;
1096     $self->svp_release
1097       if $self->auto_savepoint;
1098   }
1099 }
1100
1101 sub txn_rollback {
1102   my $self = shift;
1103   my $dbh = $self->_dbh;
1104   eval {
1105     if ($self->{transaction_depth} == 1) {
1106       $self->debugobj->txn_rollback()
1107         if ($self->debug);
1108       $self->{transaction_depth} = 0
1109         if $self->_dbh_autocommit;
1110       $dbh->rollback;
1111     }
1112     elsif($self->{transaction_depth} > 1) {
1113       $self->{transaction_depth}--;
1114       if ($self->auto_savepoint) {
1115         $self->svp_rollback;
1116         $self->svp_release;
1117       }
1118     }
1119     else {
1120       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
1121     }
1122   };
1123   if ($@) {
1124     my $error = $@;
1125     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
1126     $error =~ /$exception_class/ and $self->throw_exception($error);
1127     # ensure that a failed rollback resets the transaction depth
1128     $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
1129     $self->throw_exception($error);
1130   }
1131 }
1132
1133 # This used to be the top-half of _execute.  It was split out to make it
1134 #  easier to override in NoBindVars without duping the rest.  It takes up
1135 #  all of _execute's args, and emits $sql, @bind.
1136 sub _prep_for_execute {
1137   my ($self, $op, $extra_bind, $ident, $args) = @_;
1138
1139   if( Scalar::Util::blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1140     $ident = $ident->from();
1141   }
1142
1143   my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
1144
1145   unshift(@bind,
1146     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
1147       if $extra_bind;
1148   return ($sql, \@bind);
1149 }
1150
1151
1152 sub _fix_bind_params {
1153     my ($self, @bind) = @_;
1154
1155     ### Turn @bind from something like this:
1156     ###   ( [ "artist", 1 ], [ "cdid", 1, 3 ] )
1157     ### to this:
1158     ###   ( "'1'", "'1'", "'3'" )
1159     return
1160         map {
1161             if ( defined( $_ && $_->[1] ) ) {
1162                 map { qq{'$_'}; } @{$_}[ 1 .. $#$_ ];
1163             }
1164             else { q{'NULL'}; }
1165         } @bind;
1166 }
1167
1168 sub _query_start {
1169     my ( $self, $sql, @bind ) = @_;
1170
1171     if ( $self->debug ) {
1172         @bind = $self->_fix_bind_params(@bind);
1173
1174         $self->debugobj->query_start( $sql, @bind );
1175     }
1176 }
1177
1178 sub _query_end {
1179     my ( $self, $sql, @bind ) = @_;
1180
1181     if ( $self->debug ) {
1182         @bind = $self->_fix_bind_params(@bind);
1183         $self->debugobj->query_end( $sql, @bind );
1184     }
1185 }
1186
1187 sub _dbh_execute {
1188   my ($self, $dbh, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
1189
1190   my ($sql, $bind) = $self->_prep_for_execute($op, $extra_bind, $ident, \@args);
1191
1192   $self->_query_start( $sql, @$bind );
1193
1194   my $sth = $self->sth($sql,$op);
1195
1196   my $placeholder_index = 1;
1197
1198   foreach my $bound (@$bind) {
1199     my $attributes = {};
1200     my($column_name, @data) = @$bound;
1201
1202     if ($bind_attributes) {
1203       $attributes = $bind_attributes->{$column_name}
1204       if defined $bind_attributes->{$column_name};
1205     }
1206
1207     foreach my $data (@data) {
1208       my $ref = ref $data;
1209       $data = $ref && $ref ne 'ARRAY' ? ''.$data : $data; # stringify args (except arrayrefs)
1210
1211       $sth->bind_param($placeholder_index, $data, $attributes);
1212       $placeholder_index++;
1213     }
1214   }
1215
1216   # Can this fail without throwing an exception anyways???
1217   my $rv = $sth->execute();
1218   $self->throw_exception($sth->errstr) if !$rv;
1219
1220   $self->_query_end( $sql, @$bind );
1221
1222   return (wantarray ? ($rv, $sth, @$bind) : $rv);
1223 }
1224
1225 sub _execute {
1226     my $self = shift;
1227     $self->dbh_do('_dbh_execute', @_)
1228 }
1229
1230 sub insert {
1231   my ($self, $source, $to_insert) = @_;
1232
1233 # redispatch to insert method of storage we reblessed into, if necessary
1234   if (not $self->_driver_determined) {
1235     $self->_determine_driver;
1236     goto $self->can('insert');
1237   }
1238
1239   my $ident = $source->from;
1240   my $bind_attributes = $self->source_bind_attributes($source);
1241
1242   my $updated_cols = {};
1243
1244   foreach my $col ( $source->columns ) {
1245     if ( !defined $to_insert->{$col} ) {
1246       my $col_info = $source->column_info($col);
1247
1248       if ( $col_info->{auto_nextval} ) {
1249         $updated_cols->{$col} = $to_insert->{$col} = $self->_sequence_fetch(
1250           'nextval',
1251           $col_info->{sequence} ||
1252             $self->_dbh_get_autoinc_seq($self->_get_dbh, $source)
1253         );
1254       }
1255     }
1256   }
1257
1258   $self->_execute('insert' => [], $source, $bind_attributes, $to_insert);
1259
1260   return $updated_cols;
1261 }
1262
1263 ## Still not quite perfect, and EXPERIMENTAL
1264 ## Currently it is assumed that all values passed will be "normal", i.e. not
1265 ## scalar refs, or at least, all the same type as the first set, the statement is
1266 ## only prepped once.
1267 sub insert_bulk {
1268   my ($self, $source, $cols, $data) = @_;
1269   my %colvalues;
1270   my $table = $source->from;
1271   @colvalues{@$cols} = (0..$#$cols);
1272   my ($sql, @bind) = $self->sql_maker->insert($table, \%colvalues);
1273
1274   $self->_determine_driver;
1275
1276   $self->_query_start( $sql, @bind );
1277   my $sth = $self->sth($sql);
1278
1279 #  @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
1280
1281   ## This must be an arrayref, else nothing works!
1282   my $tuple_status = [];
1283
1284   ## Get the bind_attributes, if any exist
1285   my $bind_attributes = $self->source_bind_attributes($source);
1286
1287   ## Bind the values and execute
1288   my $placeholder_index = 1;
1289
1290   foreach my $bound (@bind) {
1291
1292     my $attributes = {};
1293     my ($column_name, $data_index) = @$bound;
1294
1295     if( $bind_attributes ) {
1296       $attributes = $bind_attributes->{$column_name}
1297       if defined $bind_attributes->{$column_name};
1298     }
1299
1300     my @data = map { $_->[$data_index] } @$data;
1301
1302     $sth->bind_param_array( $placeholder_index, [@data], $attributes );
1303     $placeholder_index++;
1304   }
1305   my $rv = eval { $sth->execute_array({ArrayTupleStatus => $tuple_status}) };
1306   if (my $err = $@) {
1307     my $i = 0;
1308     ++$i while $i <= $#$tuple_status && !ref $tuple_status->[$i];
1309
1310     $self->throw_exception($sth->errstr || "Unexpected populate error: $err")
1311       if ($i > $#$tuple_status);
1312
1313     require Data::Dumper;
1314     local $Data::Dumper::Terse = 1;
1315     local $Data::Dumper::Indent = 1;
1316     local $Data::Dumper::Useqq = 1;
1317     local $Data::Dumper::Quotekeys = 0;
1318
1319     $self->throw_exception(sprintf "%s for populate slice:\n%s",
1320       $tuple_status->[$i][1],
1321       Data::Dumper::Dumper(
1322         { map { $cols->[$_] => $data->[$i][$_] } (0 .. $#$cols) }
1323       ),
1324     );
1325   }
1326   $self->throw_exception($sth->errstr) if !$rv;
1327
1328   $self->_query_end( $sql, @bind );
1329   return (wantarray ? ($rv, $sth, @bind) : $rv);
1330 }
1331
1332 sub update {
1333   my $self = shift @_;
1334   my $source = shift @_;
1335   $self->_determine_driver;
1336   my $bind_attributes = $self->source_bind_attributes($source);
1337
1338   return $self->_execute('update' => [], $source, $bind_attributes, @_);
1339 }
1340
1341
1342 sub delete {
1343   my $self = shift @_;
1344   my $source = shift @_;
1345   $self->_determine_driver;
1346   my $bind_attrs = $self->source_bind_attributes($source);
1347
1348   return $self->_execute('delete' => [], $source, $bind_attrs, @_);
1349 }
1350
1351 # We were sent here because the $rs contains a complex search
1352 # which will require a subquery to select the correct rows
1353 # (i.e. joined or limited resultsets)
1354 #
1355 # Genarating a single PK column subquery is trivial and supported
1356 # by all RDBMS. However if we have a multicolumn PK, things get ugly.
1357 # Look at _multipk_update_delete()
1358 sub _subq_update_delete {
1359   my $self = shift;
1360   my ($rs, $op, $values) = @_;
1361
1362   my $rsrc = $rs->result_source;
1363
1364   # we already check this, but double check naively just in case. Should be removed soon
1365   my $sel = $rs->_resolved_attrs->{select};
1366   $sel = [ $sel ] unless ref $sel eq 'ARRAY';
1367   my @pcols = $rsrc->primary_columns;
1368   if (@$sel != @pcols) {
1369     $self->throw_exception (
1370       'Subquery update/delete can not be called on resultsets selecting a'
1371      .' number of columns different than the number of primary keys'
1372     );
1373   }
1374
1375   if (@pcols == 1) {
1376     return $self->$op (
1377       $rsrc,
1378       $op eq 'update' ? $values : (),
1379       { $pcols[0] => { -in => $rs->as_query } },
1380     );
1381   }
1382
1383   else {
1384     return $self->_multipk_update_delete (@_);
1385   }
1386 }
1387
1388 # ANSI SQL does not provide a reliable way to perform a multicol-PK
1389 # resultset update/delete involving subqueries. So by default resort
1390 # to simple (and inefficient) delete_all style per-row opearations,
1391 # while allowing specific storages to override this with a faster
1392 # implementation.
1393 #
1394 sub _multipk_update_delete {
1395   return shift->_per_row_update_delete (@_);
1396 }
1397
1398 # This is the default loop used to delete/update rows for multi PK
1399 # resultsets, and used by mysql exclusively (because it can't do anything
1400 # else).
1401 #
1402 # We do not use $row->$op style queries, because resultset update/delete
1403 # is not expected to cascade (this is what delete_all/update_all is for).
1404 #
1405 # There should be no race conditions as the entire operation is rolled
1406 # in a transaction.
1407 #
1408 sub _per_row_update_delete {
1409   my $self = shift;
1410   my ($rs, $op, $values) = @_;
1411
1412   my $rsrc = $rs->result_source;
1413   my @pcols = $rsrc->primary_columns;
1414
1415   my $guard = $self->txn_scope_guard;
1416
1417   # emulate the return value of $sth->execute for non-selects
1418   my $row_cnt = '0E0';
1419
1420   my $subrs_cur = $rs->cursor;
1421   while (my @pks = $subrs_cur->next) {
1422
1423     my $cond;
1424     for my $i (0.. $#pcols) {
1425       $cond->{$pcols[$i]} = $pks[$i];
1426     }
1427
1428     $self->$op (
1429       $rsrc,
1430       $op eq 'update' ? $values : (),
1431       $cond,
1432     );
1433
1434     $row_cnt++;
1435   }
1436
1437   $guard->commit;
1438
1439   return $row_cnt;
1440 }
1441
1442 sub _select {
1443   my $self = shift;
1444
1445   # localization is neccessary as
1446   # 1) there is no infrastructure to pass this around before SQLA2
1447   # 2) _select_args sets it and _prep_for_execute consumes it
1448   my $sql_maker = $self->sql_maker;
1449   local $sql_maker->{_dbic_rs_attrs};
1450
1451   return $self->_execute($self->_select_args(@_));
1452 }
1453
1454 sub _select_args_to_query {
1455   my $self = shift;
1456
1457   # localization is neccessary as
1458   # 1) there is no infrastructure to pass this around before SQLA2
1459   # 2) _select_args sets it and _prep_for_execute consumes it
1460   my $sql_maker = $self->sql_maker;
1461   local $sql_maker->{_dbic_rs_attrs};
1462
1463   # my ($op, $bind, $ident, $bind_attrs, $select, $cond, $order, $rows, $offset)
1464   #  = $self->_select_args($ident, $select, $cond, $attrs);
1465   my ($op, $bind, $ident, $bind_attrs, @args) =
1466     $self->_select_args(@_);
1467
1468   # my ($sql, $prepared_bind) = $self->_prep_for_execute($op, $bind, $ident, [ $select, $cond, $order, $rows, $offset ]);
1469   my ($sql, $prepared_bind) = $self->_prep_for_execute($op, $bind, $ident, \@args);
1470   $prepared_bind ||= [];
1471
1472   return wantarray
1473     ? ($sql, $prepared_bind, $bind_attrs)
1474     : \[ "($sql)", @$prepared_bind ]
1475   ;
1476 }
1477
1478 sub _select_args {
1479   my ($self, $ident, $select, $where, $attrs) = @_;
1480
1481   my ($alias2source, $rs_alias) = $self->_resolve_ident_sources ($ident);
1482
1483   my $sql_maker = $self->sql_maker;
1484   $sql_maker->{_dbic_rs_attrs} = {
1485     %$attrs,
1486     select => $select,
1487     from => $ident,
1488     where => $where,
1489     $rs_alias
1490       ? ( _source_handle => $alias2source->{$rs_alias}->handle )
1491       : ()
1492     ,
1493   };
1494
1495   # calculate bind_attrs before possible $ident mangling
1496   my $bind_attrs = {};
1497   for my $alias (keys %$alias2source) {
1498     my $bindtypes = $self->source_bind_attributes ($alias2source->{$alias}) || {};
1499     for my $col (keys %$bindtypes) {
1500
1501       my $fqcn = join ('.', $alias, $col);
1502       $bind_attrs->{$fqcn} = $bindtypes->{$col} if $bindtypes->{$col};
1503
1504       # Unqialified column names are nice, but at the same time can be
1505       # rather ambiguous. What we do here is basically go along with
1506       # the loop, adding an unqualified column slot to $bind_attrs,
1507       # alongside the fully qualified name. As soon as we encounter
1508       # another column by that name (which would imply another table)
1509       # we unset the unqualified slot and never add any info to it
1510       # to avoid erroneous type binding. If this happens the users
1511       # only choice will be to fully qualify his column name
1512
1513       if (exists $bind_attrs->{$col}) {
1514         $bind_attrs->{$col} = {};
1515       }
1516       else {
1517         $bind_attrs->{$col} = $bind_attrs->{$fqcn};
1518       }
1519     }
1520   }
1521
1522   # adjust limits
1523   if (
1524     $attrs->{software_limit}
1525       ||
1526     $sql_maker->_default_limit_syntax eq "GenericSubQ"
1527   ) {
1528     $attrs->{software_limit} = 1;
1529   }
1530   else {
1531     $self->throw_exception("rows attribute must be positive if present")
1532       if (defined($attrs->{rows}) && !($attrs->{rows} > 0));
1533
1534     # MySQL actually recommends this approach.  I cringe.
1535     $attrs->{rows} = 2**48 if not defined $attrs->{rows} and defined $attrs->{offset};
1536   }
1537
1538   my @limit;
1539
1540   # see if we need to tear the prefetch apart (either limited has_many or grouped prefetch)
1541   # otherwise delegate the limiting to the storage, unless software limit was requested
1542   if (
1543     ( $attrs->{rows} && keys %{$attrs->{collapse}} )
1544        ||
1545     ( $attrs->{group_by} && @{$attrs->{group_by}} &&
1546       $attrs->{_prefetch_select} && @{$attrs->{_prefetch_select}} )
1547   ) {
1548     ($ident, $select, $where, $attrs)
1549       = $self->_adjust_select_args_for_complex_prefetch ($ident, $select, $where, $attrs);
1550   }
1551   elsif (! $attrs->{software_limit} ) {
1552     push @limit, $attrs->{rows}, $attrs->{offset};
1553   }
1554
1555 ###
1556   # This would be the point to deflate anything found in $where
1557   # (and leave $attrs->{bind} intact). Problem is - inflators historically
1558   # expect a row object. And all we have is a resultsource (it is trivial
1559   # to extract deflator coderefs via $alias2source above).
1560   #
1561   # I don't see a way forward other than changing the way deflators are
1562   # invoked, and that's just bad...
1563 ###
1564
1565   my $order = { map
1566     { $attrs->{$_} ? ( $_ => $attrs->{$_} ) : ()  }
1567     (qw/order_by group_by having/ )
1568   };
1569
1570   return ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $where, $order, @limit);
1571 }
1572
1573 #
1574 # This is the code producing joined subqueries like:
1575 # SELECT me.*, other.* FROM ( SELECT me.* FROM ... ) JOIN other ON ... 
1576 #
1577 sub _adjust_select_args_for_complex_prefetch {
1578   my ($self, $from, $select, $where, $attrs) = @_;
1579
1580   $self->throw_exception ('Nothing to prefetch... how did we get here?!')
1581     if not @{$attrs->{_prefetch_select}};
1582
1583   $self->throw_exception ('Complex prefetches are not supported on resultsets with a custom from attribute')
1584     if (ref $from ne 'ARRAY' || ref $from->[0] ne 'HASH' || ref $from->[1] ne 'ARRAY');
1585
1586
1587   # generate inner/outer attribute lists, remove stuff that doesn't apply
1588   my $outer_attrs = { %$attrs };
1589   delete $outer_attrs->{$_} for qw/where bind rows offset group_by having/;
1590
1591   my $inner_attrs = { %$attrs };
1592   delete $inner_attrs->{$_} for qw/for collapse _prefetch_select _collapse_order_by select as/;
1593
1594
1595   # bring over all non-collapse-induced order_by into the inner query (if any)
1596   # the outer one will have to keep them all
1597   delete $inner_attrs->{order_by};
1598   if (my $ord_cnt = @{$outer_attrs->{order_by}} - @{$outer_attrs->{_collapse_order_by}} ) {
1599     $inner_attrs->{order_by} = [
1600       @{$outer_attrs->{order_by}}[ 0 .. $ord_cnt - 1]
1601     ];
1602   }
1603
1604
1605   # generate the inner/outer select lists
1606   # for inside we consider only stuff *not* brought in by the prefetch
1607   # on the outside we substitute any function for its alias
1608   my $outer_select = [ @$select ];
1609   my $inner_select = [];
1610   for my $i (0 .. ( @$outer_select - @{$outer_attrs->{_prefetch_select}} - 1) ) {
1611     my $sel = $outer_select->[$i];
1612
1613     if (ref $sel eq 'HASH' ) {
1614       $sel->{-as} ||= $attrs->{as}[$i];
1615       $outer_select->[$i] = join ('.', $attrs->{alias}, ($sel->{-as} || "inner_column_$i") );
1616     }
1617
1618     push @$inner_select, $sel;
1619   }
1620
1621   # normalize a copy of $from, so it will be easier to work with further
1622   # down (i.e. promote the initial hashref to an AoH)
1623   $from = [ @$from ];
1624   $from->[0] = [ $from->[0] ];
1625   my %original_join_info = map { $_->[0]{-alias} => $_->[0] } (@$from);
1626
1627
1628   # decide which parts of the join will remain in either part of
1629   # the outer/inner query
1630
1631   # First we compose a list of which aliases are used in restrictions
1632   # (i.e. conditions/order/grouping/etc). Since we do not have
1633   # introspectable SQLA, we fall back to ugly scanning of raw SQL for
1634   # WHERE, and for pieces of ORDER BY in order to determine which aliases
1635   # need to appear in the resulting sql.
1636   # It may not be very efficient, but it's a reasonable stop-gap
1637   # Also unqualified column names will not be considered, but more often
1638   # than not this is actually ok
1639   #
1640   # In the same loop we enumerate part of the selection aliases, as
1641   # it requires the same sqla hack for the time being
1642   my ($restrict_aliases, $select_aliases, $prefetch_aliases);
1643   {
1644     # produce stuff unquoted, so it can be scanned
1645     my $sql_maker = $self->sql_maker;
1646     local $sql_maker->{quote_char};
1647     my $sep = $self->_sql_maker_opts->{name_sep} || '.';
1648     $sep = "\Q$sep\E";
1649
1650     my $non_prefetch_select_sql = $sql_maker->_recurse_fields ($inner_select);
1651     my $prefetch_select_sql = $sql_maker->_recurse_fields ($outer_attrs->{_prefetch_select});
1652     my $where_sql = $sql_maker->where ($where);
1653     my $group_by_sql = $sql_maker->_order_by({
1654       map { $_ => $inner_attrs->{$_} } qw/group_by having/
1655     });
1656     my @non_prefetch_order_by_chunks = (map
1657       { ref $_ ? $_->[0] : $_ }
1658       $sql_maker->_order_by_chunks ($inner_attrs->{order_by})
1659     );
1660
1661
1662     for my $alias (keys %original_join_info) {
1663       my $seen_re = qr/\b $alias $sep/x;
1664
1665       for my $piece ($where_sql, $group_by_sql, @non_prefetch_order_by_chunks ) {
1666         if ($piece =~ $seen_re) {
1667           $restrict_aliases->{$alias} = 1;
1668         }
1669       }
1670
1671       if ($non_prefetch_select_sql =~ $seen_re) {
1672           $select_aliases->{$alias} = 1;
1673       }
1674
1675       if ($prefetch_select_sql =~ $seen_re) {
1676           $prefetch_aliases->{$alias} = 1;
1677       }
1678
1679     }
1680   }
1681
1682   # Add any non-left joins to the restriction list (such joins are indeed restrictions)
1683   for my $j (values %original_join_info) {
1684     my $alias = $j->{-alias} or next;
1685     $restrict_aliases->{$alias} = 1 if (
1686       (not $j->{-join_type})
1687         or
1688       ($j->{-join_type} !~ /^left (?: \s+ outer)? $/xi)
1689     );
1690   }
1691
1692   # mark all join parents as mentioned
1693   # (e.g.  join => { cds => 'tracks' } - tracks will need to bring cds too )
1694   for my $collection ($restrict_aliases, $select_aliases) {
1695     for my $alias (keys %$collection) {
1696       $collection->{$_} = 1
1697         for (@{ $original_join_info{$alias}{-join_path} || [] });
1698     }
1699   }
1700
1701   # construct the inner $from for the subquery
1702   my %inner_joins = (map { %{$_ || {}} } ($restrict_aliases, $select_aliases) );
1703   my @inner_from;
1704   for my $j (@$from) {
1705     push @inner_from, $j if $inner_joins{$j->[0]{-alias}};
1706   }
1707
1708   # if a multi-type join was needed in the subquery ("multi" is indicated by
1709   # presence in {collapse}) - add a group_by to simulate the collapse in the subq
1710   unless ($inner_attrs->{group_by}) {
1711     for my $alias (keys %inner_joins) {
1712
1713       # the dot comes from some weirdness in collapse
1714       # remove after the rewrite
1715       if ($attrs->{collapse}{".$alias"}) {
1716         $inner_attrs->{group_by} ||= $inner_select;
1717         last;
1718       }
1719     }
1720   }
1721
1722   # demote the inner_from head
1723   $inner_from[0] = $inner_from[0][0];
1724
1725   # generate the subquery
1726   my $subq = $self->_select_args_to_query (
1727     \@inner_from,
1728     $inner_select,
1729     $where,
1730     $inner_attrs,
1731   );
1732
1733   my $subq_joinspec = {
1734     -alias => $attrs->{alias},
1735     -source_handle => $inner_from[0]{-source_handle},
1736     $attrs->{alias} => $subq,
1737   };
1738
1739   # Generate the outer from - this is relatively easy (really just replace
1740   # the join slot with the subquery), with a major caveat - we can not
1741   # join anything that is non-selecting (not part of the prefetch), but at
1742   # the same time is a multi-type relationship, as it will explode the result.
1743   #
1744   # There are two possibilities here
1745   # - either the join is non-restricting, in which case we simply throw it away
1746   # - it is part of the restrictions, in which case we need to collapse the outer
1747   #   result by tackling yet another group_by to the outside of the query
1748
1749   # so first generate the outer_from, up to the substitution point
1750   my @outer_from;
1751   while (my $j = shift @$from) {
1752     if ($j->[0]{-alias} eq $attrs->{alias}) { # time to swap
1753       push @outer_from, [
1754         $subq_joinspec,
1755         @{$j}[1 .. $#$j],
1756       ];
1757       last; # we'll take care of what's left in $from below
1758     }
1759     else {
1760       push @outer_from, $j;
1761     }
1762   }
1763
1764   # see what's left - throw away if not selecting/restricting
1765   # also throw in a group_by if restricting to guard against
1766   # cross-join explosions
1767   #
1768   while (my $j = shift @$from) {
1769     my $alias = $j->[0]{-alias};
1770
1771     if ($select_aliases->{$alias} || $prefetch_aliases->{$alias}) {
1772       push @outer_from, $j;
1773     }
1774     elsif ($restrict_aliases->{$alias}) {
1775       push @outer_from, $j;
1776
1777       # FIXME - this should be obviated by SQLA2, as I'll be able to 
1778       # have restrict_inner and restrict_outer... or something to that
1779       # effect... I think...
1780
1781       # FIXME2 - I can't find a clean way to determine if a particular join
1782       # is a multi - instead I am just treating everything as a potential
1783       # explosive join (ribasushi)
1784       #
1785       # if (my $handle = $j->[0]{-source_handle}) {
1786       #   my $rsrc = $handle->resolve;
1787       #   ... need to bail out of the following if this is not a multi,
1788       #       as it will be much easier on the db ...
1789
1790           $outer_attrs->{group_by} ||= $outer_select;
1791       # }
1792     }
1793   }
1794
1795   # demote the outer_from head
1796   $outer_from[0] = $outer_from[0][0];
1797
1798   # This is totally horrific - the $where ends up in both the inner and outer query
1799   # Unfortunately not much can be done until SQLA2 introspection arrives, and even
1800   # then if where conditions apply to the *right* side of the prefetch, you may have
1801   # to both filter the inner select (e.g. to apply a limit) and then have to re-filter
1802   # the outer select to exclude joins you didin't want in the first place
1803   #
1804   # OTOH it can be seen as a plus: <ash> (notes that this query would make a DBA cry ;)
1805   return (\@outer_from, $outer_select, $where, $outer_attrs);
1806 }
1807
1808 sub _resolve_ident_sources {
1809   my ($self, $ident) = @_;
1810
1811   my $alias2source = {};
1812   my $rs_alias;
1813
1814   # the reason this is so contrived is that $ident may be a {from}
1815   # structure, specifying multiple tables to join
1816   if ( Scalar::Util::blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1817     # this is compat mode for insert/update/delete which do not deal with aliases
1818     $alias2source->{me} = $ident;
1819     $rs_alias = 'me';
1820   }
1821   elsif (ref $ident eq 'ARRAY') {
1822
1823     for (@$ident) {
1824       my $tabinfo;
1825       if (ref $_ eq 'HASH') {
1826         $tabinfo = $_;
1827         $rs_alias = $tabinfo->{-alias};
1828       }
1829       if (ref $_ eq 'ARRAY' and ref $_->[0] eq 'HASH') {
1830         $tabinfo = $_->[0];
1831       }
1832
1833       $alias2source->{$tabinfo->{-alias}} = $tabinfo->{-source_handle}->resolve
1834         if ($tabinfo->{-source_handle});
1835     }
1836   }
1837
1838   return ($alias2source, $rs_alias);
1839 }
1840
1841 # Takes $ident, \@column_names
1842 #
1843 # returns { $column_name => \%column_info, ... }
1844 # also note: this adds -result_source => $rsrc to the column info
1845 #
1846 # usage:
1847 #   my $col_sources = $self->_resolve_column_info($ident, @column_names);
1848 sub _resolve_column_info {
1849   my ($self, $ident, $colnames) = @_;
1850   my ($alias2src, $root_alias) = $self->_resolve_ident_sources($ident);
1851
1852   my $sep = $self->_sql_maker_opts->{name_sep} || '.';
1853   $sep = "\Q$sep\E";
1854
1855   my (%return, %seen_cols);
1856
1857   # compile a global list of column names, to be able to properly
1858   # disambiguate unqualified column names (if at all possible)
1859   for my $alias (keys %$alias2src) {
1860     my $rsrc = $alias2src->{$alias};
1861     for my $colname ($rsrc->columns) {
1862       push @{$seen_cols{$colname}}, $alias;
1863     }
1864   }
1865
1866   COLUMN:
1867   foreach my $col (@$colnames) {
1868     my ($alias, $colname) = $col =~ m/^ (?: ([^$sep]+) $sep)? (.+) $/x;
1869
1870     unless ($alias) {
1871       # see if the column was seen exactly once (so we know which rsrc it came from)
1872       if ($seen_cols{$colname} and @{$seen_cols{$colname}} == 1) {
1873         $alias = $seen_cols{$colname}[0];
1874       }
1875       else {
1876         next COLUMN;
1877       }
1878     }
1879
1880     my $rsrc = $alias2src->{$alias};
1881     $return{$col} = $rsrc && {
1882       %{$rsrc->column_info($colname)},
1883       -result_source => $rsrc,
1884       -source_alias => $alias,
1885     };
1886   }
1887
1888   return \%return;
1889 }
1890
1891 # Returns a counting SELECT for a simple count
1892 # query. Abstracted so that a storage could override
1893 # this to { count => 'firstcol' } or whatever makes
1894 # sense as a performance optimization
1895 sub _count_select {
1896   #my ($self, $source, $rs_attrs) = @_;
1897   return { count => '*' };
1898 }
1899
1900 # Returns a SELECT which will end up in the subselect
1901 # There may or may not be a group_by, as the subquery
1902 # might have been called to accomodate a limit
1903 #
1904 # Most databases would be happy with whatever ends up
1905 # here, but some choke in various ways.
1906 #
1907 sub _subq_count_select {
1908   my ($self, $source, $rs_attrs) = @_;
1909   return $rs_attrs->{group_by} if $rs_attrs->{group_by};
1910
1911   my @pcols = map { join '.', $rs_attrs->{alias}, $_ } ($source->primary_columns);
1912   return @pcols ? \@pcols : [ 1 ];
1913 }
1914
1915
1916 sub source_bind_attributes {
1917   my ($self, $source) = @_;
1918
1919   my $bind_attributes;
1920   foreach my $column ($source->columns) {
1921
1922     my $data_type = $source->column_info($column)->{data_type} || '';
1923     $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
1924      if $data_type;
1925   }
1926
1927   return $bind_attributes;
1928 }
1929
1930 =head2 select
1931
1932 =over 4
1933
1934 =item Arguments: $ident, $select, $condition, $attrs
1935
1936 =back
1937
1938 Handle a SQL select statement.
1939
1940 =cut
1941
1942 sub select {
1943   my $self = shift;
1944   my ($ident, $select, $condition, $attrs) = @_;
1945   return $self->cursor_class->new($self, \@_, $attrs);
1946 }
1947
1948 sub select_single {
1949   my $self = shift;
1950   my ($rv, $sth, @bind) = $self->_select(@_);
1951   my @row = $sth->fetchrow_array;
1952   my @nextrow = $sth->fetchrow_array if @row;
1953   if(@row && @nextrow) {
1954     carp "Query returned more than one row.  SQL that returns multiple rows is DEPRECATED for ->find and ->single";
1955   }
1956   # Need to call finish() to work round broken DBDs
1957   $sth->finish();
1958   return @row;
1959 }
1960
1961 =head2 sth
1962
1963 =over 4
1964
1965 =item Arguments: $sql
1966
1967 =back
1968
1969 Returns a L<DBI> sth (statement handle) for the supplied SQL.
1970
1971 =cut
1972
1973 sub _dbh_sth {
1974   my ($self, $dbh, $sql) = @_;
1975
1976   # 3 is the if_active parameter which avoids active sth re-use
1977   my $sth = $self->disable_sth_caching
1978     ? $dbh->prepare($sql)
1979     : $dbh->prepare_cached($sql, {}, 3);
1980
1981   # XXX You would think RaiseError would make this impossible,
1982   #  but apparently that's not true :(
1983   $self->throw_exception($dbh->errstr) if !$sth;
1984
1985   $sth;
1986 }
1987
1988 sub sth {
1989   my ($self, $sql) = @_;
1990   $self->dbh_do('_dbh_sth', $sql);
1991 }
1992
1993 sub _dbh_columns_info_for {
1994   my ($self, $dbh, $table) = @_;
1995
1996   if ($dbh->can('column_info')) {
1997     my %result;
1998     eval {
1999       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
2000       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
2001       $sth->execute();
2002       while ( my $info = $sth->fetchrow_hashref() ){
2003         my %column_info;
2004         $column_info{data_type}   = $info->{TYPE_NAME};
2005         $column_info{size}      = $info->{COLUMN_SIZE};
2006         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
2007         $column_info{default_value} = $info->{COLUMN_DEF};
2008         my $col_name = $info->{COLUMN_NAME};
2009         $col_name =~ s/^\"(.*)\"$/$1/;
2010
2011         $result{$col_name} = \%column_info;
2012       }
2013     };
2014     return \%result if !$@ && scalar keys %result;
2015   }
2016
2017   my %result;
2018   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
2019   $sth->execute;
2020   my @columns = @{$sth->{NAME_lc}};
2021   for my $i ( 0 .. $#columns ){
2022     my %column_info;
2023     $column_info{data_type} = $sth->{TYPE}->[$i];
2024     $column_info{size} = $sth->{PRECISION}->[$i];
2025     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
2026
2027     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
2028       $column_info{data_type} = $1;
2029       $column_info{size}    = $2;
2030     }
2031
2032     $result{$columns[$i]} = \%column_info;
2033   }
2034   $sth->finish;
2035
2036   foreach my $col (keys %result) {
2037     my $colinfo = $result{$col};
2038     my $type_num = $colinfo->{data_type};
2039     my $type_name;
2040     if(defined $type_num && $dbh->can('type_info')) {
2041       my $type_info = $dbh->type_info($type_num);
2042       $type_name = $type_info->{TYPE_NAME} if $type_info;
2043       $colinfo->{data_type} = $type_name if $type_name;
2044     }
2045   }
2046
2047   return \%result;
2048 }
2049
2050 sub columns_info_for {
2051   my ($self, $table) = @_;
2052   $self->dbh_do('_dbh_columns_info_for', $table);
2053 }
2054
2055 =head2 last_insert_id
2056
2057 Return the row id of the last insert.
2058
2059 =cut
2060
2061 sub _dbh_last_insert_id {
2062     # All Storage's need to register their own _dbh_last_insert_id
2063     # the old SQLite-based method was highly inappropriate
2064
2065     my $self = shift;
2066     my $class = ref $self;
2067     $self->throw_exception (<<EOE);
2068
2069 No _dbh_last_insert_id() method found in $class.
2070 Since the method of obtaining the autoincrement id of the last insert
2071 operation varies greatly between different databases, this method must be
2072 individually implemented for every storage class.
2073 EOE
2074 }
2075
2076 sub last_insert_id {
2077   my $self = shift;
2078   $self->dbh_do('_dbh_last_insert_id', @_);
2079 }
2080
2081 =head2 sqlt_type
2082
2083 Returns the database driver name.
2084
2085 =cut
2086
2087 sub sqlt_type { shift->_get_dbh->{Driver}->{Name} }
2088
2089 =head2 bind_attribute_by_data_type
2090
2091 Given a datatype from column info, returns a database specific bind
2092 attribute for C<< $dbh->bind_param($val,$attribute) >> or nothing if we will
2093 let the database planner just handle it.
2094
2095 Generally only needed for special case column types, like bytea in postgres.
2096
2097 =cut
2098
2099 sub bind_attribute_by_data_type {
2100     return;
2101 }
2102
2103 =head2 is_datatype_numeric
2104
2105 Given a datatype from column_info, returns a boolean value indicating if
2106 the current RDBMS considers it a numeric value. This controls how
2107 L<DBIx::Class::Row/set_column> decides whether to mark the column as
2108 dirty - when the datatype is deemed numeric a C<< != >> comparison will
2109 be performed instead of the usual C<eq>.
2110
2111 =cut
2112
2113 sub is_datatype_numeric {
2114   my ($self, $dt) = @_;
2115
2116   return 0 unless $dt;
2117
2118   return $dt =~ /^ (?:
2119     numeric | int(?:eger)? | (?:tiny|small|medium|big)int | dec(?:imal)? | real | float | double (?: \s+ precision)? | (?:big)?serial
2120   ) $/ix;
2121 }
2122
2123
2124 =head2 create_ddl_dir (EXPERIMENTAL)
2125
2126 =over 4
2127
2128 =item Arguments: $schema \@databases, $version, $directory, $preversion, \%sqlt_args
2129
2130 =back
2131
2132 Creates a SQL file based on the Schema, for each of the specified
2133 database engines in C<\@databases> in the given directory.
2134 (note: specify L<SQL::Translator> names, not L<DBI> driver names).
2135
2136 Given a previous version number, this will also create a file containing
2137 the ALTER TABLE statements to transform the previous schema into the
2138 current one. Note that these statements may contain C<DROP TABLE> or
2139 C<DROP COLUMN> statements that can potentially destroy data.
2140
2141 The file names are created using the C<ddl_filename> method below, please
2142 override this method in your schema if you would like a different file
2143 name format. For the ALTER file, the same format is used, replacing
2144 $version in the name with "$preversion-$version".
2145
2146 See L<SQL::Translator/METHODS> for a list of values for C<\%sqlt_args>.
2147 The most common value for this would be C<< { add_drop_table => 1 } >>
2148 to have the SQL produced include a C<DROP TABLE> statement for each table
2149 created. For quoting purposes supply C<quote_table_names> and
2150 C<quote_field_names>.
2151
2152 If no arguments are passed, then the following default values are assumed:
2153
2154 =over 4
2155
2156 =item databases  - ['MySQL', 'SQLite', 'PostgreSQL']
2157
2158 =item version    - $schema->schema_version
2159
2160 =item directory  - './'
2161
2162 =item preversion - <none>
2163
2164 =back
2165
2166 By default, C<\%sqlt_args> will have
2167
2168  { add_drop_table => 1, ignore_constraint_names => 1, ignore_index_names => 1 }
2169
2170 merged with the hash passed in. To disable any of those features, pass in a
2171 hashref like the following
2172
2173  { ignore_constraint_names => 0, # ... other options }
2174
2175
2176 Note that this feature is currently EXPERIMENTAL and may not work correctly
2177 across all databases, or fully handle complex relationships.
2178
2179 WARNING: Please check all SQL files created, before applying them.
2180
2181 =cut
2182
2183 sub create_ddl_dir {
2184   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
2185
2186   if(!$dir || !-d $dir) {
2187     carp "No directory given, using ./\n";
2188     $dir = "./";
2189   }
2190   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
2191   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
2192
2193   my $schema_version = $schema->schema_version || '1.x';
2194   $version ||= $schema_version;
2195
2196   $sqltargs = {
2197     add_drop_table => 1,
2198     ignore_constraint_names => 1,
2199     ignore_index_names => 1,
2200     %{$sqltargs || {}}
2201   };
2202
2203   $self->throw_exception(q{Can't create a ddl file without SQL::Translator 0.09003: '}
2204       . $self->_check_sqlt_message . q{'})
2205           if !$self->_check_sqlt_version;
2206
2207   my $sqlt = SQL::Translator->new( $sqltargs );
2208
2209   $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
2210   my $sqlt_schema = $sqlt->translate({ data => $schema })
2211     or $self->throw_exception ($sqlt->error);
2212
2213   foreach my $db (@$databases) {
2214     $sqlt->reset();
2215     $sqlt->{schema} = $sqlt_schema;
2216     $sqlt->producer($db);
2217
2218     my $file;
2219     my $filename = $schema->ddl_filename($db, $version, $dir);
2220     if (-e $filename && ($version eq $schema_version )) {
2221       # if we are dumping the current version, overwrite the DDL
2222       carp "Overwriting existing DDL file - $filename";
2223       unlink($filename);
2224     }
2225
2226     my $output = $sqlt->translate;
2227     if(!$output) {
2228       carp("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
2229       next;
2230     }
2231     if(!open($file, ">$filename")) {
2232       $self->throw_exception("Can't open $filename for writing ($!)");
2233       next;
2234     }
2235     print $file $output;
2236     close($file);
2237
2238     next unless ($preversion);
2239
2240     require SQL::Translator::Diff;
2241
2242     my $prefilename = $schema->ddl_filename($db, $preversion, $dir);
2243     if(!-e $prefilename) {
2244       carp("No previous schema file found ($prefilename)");
2245       next;
2246     }
2247
2248     my $difffile = $schema->ddl_filename($db, $version, $dir, $preversion);
2249     if(-e $difffile) {
2250       carp("Overwriting existing diff file - $difffile");
2251       unlink($difffile);
2252     }
2253
2254     my $source_schema;
2255     {
2256       my $t = SQL::Translator->new($sqltargs);
2257       $t->debug( 0 );
2258       $t->trace( 0 );
2259
2260       $t->parser( $db )
2261         or $self->throw_exception ($t->error);
2262
2263       my $out = $t->translate( $prefilename )
2264         or $self->throw_exception ($t->error);
2265
2266       $source_schema = $t->schema;
2267
2268       $source_schema->name( $prefilename )
2269         unless ( $source_schema->name );
2270     }
2271
2272     # The "new" style of producers have sane normalization and can support
2273     # diffing a SQL file against a DBIC->SQLT schema. Old style ones don't
2274     # And we have to diff parsed SQL against parsed SQL.
2275     my $dest_schema = $sqlt_schema;
2276
2277     unless ( "SQL::Translator::Producer::$db"->can('preprocess_schema') ) {
2278       my $t = SQL::Translator->new($sqltargs);
2279       $t->debug( 0 );
2280       $t->trace( 0 );
2281
2282       $t->parser( $db )
2283         or $self->throw_exception ($t->error);
2284
2285       my $out = $t->translate( $filename )
2286         or $self->throw_exception ($t->error);
2287
2288       $dest_schema = $t->schema;
2289
2290       $dest_schema->name( $filename )
2291         unless $dest_schema->name;
2292     }
2293
2294     my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
2295                                                   $dest_schema,   $db,
2296                                                   $sqltargs
2297                                                  );
2298     if(!open $file, ">$difffile") {
2299       $self->throw_exception("Can't write to $difffile ($!)");
2300       next;
2301     }
2302     print $file $diff;
2303     close($file);
2304   }
2305 }
2306
2307 =head2 deployment_statements
2308
2309 =over 4
2310
2311 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
2312
2313 =back
2314
2315 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
2316
2317 The L<SQL::Translator> (not L<DBI>) database driver name can be explicitly
2318 provided in C<$type>, otherwise the result of L</sqlt_type> is used as default.
2319
2320 C<$directory> is used to return statements from files in a previously created
2321 L</create_ddl_dir> directory and is optional. The filenames are constructed
2322 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
2323
2324 If no C<$directory> is specified then the statements are constructed on the
2325 fly using L<SQL::Translator> and C<$version> is ignored.
2326
2327 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
2328
2329 =cut
2330
2331 sub deployment_statements {
2332   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
2333   $type ||= $self->sqlt_type;
2334   $version ||= $schema->schema_version || '1.x';
2335   $dir ||= './';
2336   my $filename = $schema->ddl_filename($type, $version, $dir);
2337   if(-f $filename)
2338   {
2339       my $file;
2340       open($file, "<$filename")
2341         or $self->throw_exception("Can't open $filename ($!)");
2342       my @rows = <$file>;
2343       close($file);
2344       return join('', @rows);
2345   }
2346
2347   $self->throw_exception(q{Can't deploy without SQL::Translator 0.09003: '}
2348       . $self->_check_sqlt_message . q{'})
2349           if !$self->_check_sqlt_version;
2350
2351   # sources needs to be a parser arg, but for simplicty allow at top level
2352   # coming in
2353   $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
2354       if exists $sqltargs->{sources};
2355
2356   my $tr = SQL::Translator->new(
2357     producer => "SQL::Translator::Producer::${type}",
2358     %$sqltargs,
2359     parser => 'SQL::Translator::Parser::DBIx::Class',
2360     data => $schema,
2361   );
2362   return $tr->translate;
2363 }
2364
2365 sub deploy {
2366   my ($self, $schema, $type, $sqltargs, $dir) = @_;
2367   my $deploy = sub {
2368     my $line = shift;
2369     return if($line =~ /^--/);
2370     return if(!$line);
2371     # next if($line =~ /^DROP/m);
2372     return if($line =~ /^BEGIN TRANSACTION/m);
2373     return if($line =~ /^COMMIT/m);
2374     return if $line =~ /^\s+$/; # skip whitespace only
2375     $self->_query_start($line);
2376     eval {
2377       # do a dbh_do cycle here, as we need some error checking in
2378       # place (even though we will ignore errors)
2379       $self->dbh_do (sub { $_[1]->do($line) });
2380     };
2381     if ($@) {
2382       carp qq{$@ (running "${line}")};
2383     }
2384     $self->_query_end($line);
2385   };
2386   my @statements = $self->deployment_statements($schema, $type, undef, $dir, { %{ $sqltargs || {} }, no_comments => 1 } );
2387   if (@statements > 1) {
2388     foreach my $statement (@statements) {
2389       $deploy->( $statement );
2390     }
2391   }
2392   elsif (@statements == 1) {
2393     foreach my $line ( split(";\n", $statements[0])) {
2394       $deploy->( $line );
2395     }
2396   }
2397 }
2398
2399 =head2 datetime_parser
2400
2401 Returns the datetime parser class
2402
2403 =cut
2404
2405 sub datetime_parser {
2406   my $self = shift;
2407   return $self->{datetime_parser} ||= do {
2408     $self->_populate_dbh unless $self->_dbh;
2409     $self->build_datetime_parser(@_);
2410   };
2411 }
2412
2413 =head2 datetime_parser_type
2414
2415 Defines (returns) the datetime parser class - currently hardwired to
2416 L<DateTime::Format::MySQL>
2417
2418 =cut
2419
2420 sub datetime_parser_type { "DateTime::Format::MySQL"; }
2421
2422 =head2 build_datetime_parser
2423
2424 See L</datetime_parser>
2425
2426 =cut
2427
2428 sub build_datetime_parser {
2429   my $self = shift;
2430   my $type = $self->datetime_parser_type(@_);
2431   eval "use ${type}";
2432   $self->throw_exception("Couldn't load ${type}: $@") if $@;
2433   return $type;
2434 }
2435
2436 {
2437     my $_check_sqlt_version; # private
2438     my $_check_sqlt_message; # private
2439     sub _check_sqlt_version {
2440         return $_check_sqlt_version if defined $_check_sqlt_version;
2441         eval 'use SQL::Translator "0.09003"';
2442         $_check_sqlt_message = $@ || '';
2443         $_check_sqlt_version = !$@;
2444     }
2445
2446     sub _check_sqlt_message {
2447         _check_sqlt_version if !defined $_check_sqlt_message;
2448         $_check_sqlt_message;
2449     }
2450 }
2451
2452 =head2 is_replicating
2453
2454 A boolean that reports if a particular L<DBIx::Class::Storage::DBI> is set to
2455 replicate from a master database.  Default is undef, which is the result
2456 returned by databases that don't support replication.
2457
2458 =cut
2459
2460 sub is_replicating {
2461     return;
2462
2463 }
2464
2465 =head2 lag_behind_master
2466
2467 Returns a number that represents a certain amount of lag behind a master db
2468 when a given storage is replicating.  The number is database dependent, but
2469 starts at zero and increases with the amount of lag. Default in undef
2470
2471 =cut
2472
2473 sub lag_behind_master {
2474     return;
2475 }
2476
2477 sub DESTROY {
2478   my $self = shift;
2479   $self->_verify_pid if $self->_dbh;
2480
2481   # some databases need this to stop spewing warnings
2482   if (my $dbh = $self->_dbh) {
2483     eval { $dbh->disconnect };
2484   }
2485
2486   $self->_dbh(undef);
2487 }
2488
2489 1;
2490
2491 =head1 USAGE NOTES
2492
2493 =head2 DBIx::Class and AutoCommit
2494
2495 DBIx::Class can do some wonderful magic with handling exceptions,
2496 disconnections, and transactions when you use C<< AutoCommit => 1 >>
2497 (the default) combined with C<txn_do> for transaction support.
2498
2499 If you set C<< AutoCommit => 0 >> in your connect info, then you are always
2500 in an assumed transaction between commits, and you're telling us you'd
2501 like to manage that manually.  A lot of the magic protections offered by
2502 this module will go away.  We can't protect you from exceptions due to database
2503 disconnects because we don't know anything about how to restart your
2504 transactions.  You're on your own for handling all sorts of exceptional
2505 cases if you choose the C<< AutoCommit => 0 >> path, just as you would
2506 be with raw DBI.
2507
2508
2509 =head1 AUTHORS
2510
2511 Matt S. Trout <mst@shadowcatsystems.co.uk>
2512
2513 Andy Grundman <andy@hybridized.org>
2514
2515 =head1 LICENSE
2516
2517 You may distribute this code under the same terms as Perl itself.
2518
2519 =cut