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