e08ff9ccd89569d1ed9b23ea5607e0eacab4a81e
[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 ('Complex prefetches are not supported on resultsets with a custom from attribute')
1581     if (ref $from ne 'ARRAY');
1582
1583   # copies for mangling
1584   $from = [ @$from ];
1585   $select = [ @$select ];
1586   $attrs = { %$attrs };
1587
1588   # separate attributes
1589   my $sub_attrs = { %$attrs };
1590   delete $attrs->{$_} for qw/where bind rows offset group_by having/;
1591   delete $sub_attrs->{$_} for qw/for collapse _prefetch_select _collapse_order_by select as/;
1592
1593   my $select_root_alias = $attrs->{alias};
1594   my $sql_maker = $self->sql_maker;
1595
1596   # create subquery select list - consider only stuff *not* brought in by the prefetch
1597   my $sub_select = [];
1598   my $sub_group_by;
1599   for my $i (0 .. @{$attrs->{select}} - @{$attrs->{_prefetch_select}} - 1) {
1600     my $sel = $attrs->{select}[$i];
1601
1602     # alias any functions to the dbic-side 'as' label
1603     # adjust the outer select accordingly
1604     if (ref $sel eq 'HASH' ) {
1605       $sel->{-as} ||= $attrs->{as}[$i];
1606       $select->[$i] = join ('.', $attrs->{alias}, ($sel->{-as} || "select_$i") );
1607     }
1608
1609     push @$sub_select, $sel;
1610   }
1611
1612   # bring over all non-collapse-induced order_by into the inner query (if any)
1613   # the outer one will have to keep them all
1614   delete $sub_attrs->{order_by};
1615   if (my $ord_cnt = @{$attrs->{order_by}} - @{$attrs->{_collapse_order_by}} ) {
1616     $sub_attrs->{order_by} = [
1617       @{$attrs->{order_by}}[ 0 .. $ord_cnt - 1]
1618     ];
1619   }
1620
1621   # mangle {from}, keep in mind that $from is "headless" from here on
1622   my $join_root = shift @$from;
1623
1624   my %inner_joins;
1625   my %join_info = map { $_->[0]{-alias} => $_->[0] } (@$from);
1626
1627   # in complex search_related chains $select_root_alias may *not* be
1628   # 'me' so always include it in the inner join
1629   $inner_joins{$select_root_alias} = 1 if ($join_root->{-alias} ne $select_root_alias);
1630
1631
1632   # decide which parts of the join will remain on the inside
1633   #
1634   # this is not a very viable optimisation, but it was written
1635   # before I realised this, so might as well remain. We can throw
1636   # away _any_ branches of the join tree that are:
1637   # 1) not mentioned in the condition/order
1638   # 2) left-join leaves (or left-join leaf chains)
1639   # Most of the join conditions will not satisfy this, but for real
1640   # complex queries some might, and we might make some RDBMS happy.
1641   #
1642   #
1643   # since we do not have introspectable SQLA, we fall back to ugly
1644   # scanning of raw SQL for WHERE, and for pieces of ORDER BY
1645   # in order to determine what goes into %inner_joins
1646   # It may not be very efficient, but it's a reasonable stop-gap
1647   {
1648     # produce stuff unquoted, so it can be scanned
1649     local $sql_maker->{quote_char};
1650     my $sep = $self->_sql_maker_opts->{name_sep} || '.';
1651     $sep = "\Q$sep\E";
1652
1653     my @order_by = (map
1654       { ref $_ ? $_->[0] : $_ }
1655       $sql_maker->_order_by_chunks ($sub_attrs->{order_by})
1656     );
1657
1658     my $where_sql = $sql_maker->where ($where);
1659     my $select_sql = $sql_maker->_recurse_fields ($sub_select);
1660
1661     # sort needed joins
1662     for my $alias (keys %join_info) {
1663
1664       # any table alias found on a column name in where or order_by
1665       # gets included in %inner_joins
1666       # Also any parent joins that are needed to reach this particular alias
1667       for my $piece ($select_sql, $where_sql, @order_by ) {
1668         if ($piece =~ /\b $alias $sep/x) {
1669           $inner_joins{$alias} = 1;
1670         }
1671       }
1672     }
1673   }
1674
1675   # scan for non-leaf/non-left joins and mark as needed
1676   # also mark all ancestor joins that are needed to reach this particular alias
1677   # (e.g.  join => { cds => 'tracks' } - tracks will bring cds too )
1678   #
1679   # traverse by the size of the -join_path i.e. reverse depth first
1680   for my $alias (sort { @{$join_info{$b}{-join_path}} <=> @{$join_info{$a}{-join_path}} } (keys %join_info) ) {
1681
1682     my $j = $join_info{$alias};
1683     $inner_joins{$alias} = 1 if (! $j->{-join_type} || ($j->{-join_type} !~ /^left$/i) );
1684
1685     if ($inner_joins{$alias}) {
1686       $inner_joins{$_} = 1 for (@{$j->{-join_path}});
1687     }
1688   }
1689
1690   # construct the inner $from for the subquery
1691   my $inner_from = [ $join_root ];
1692   for my $j (@$from) {
1693     push @$inner_from, $j if $inner_joins{$j->[0]{-alias}};
1694   }
1695
1696   # if a multi-type join was needed in the subquery ("multi" is indicated by
1697   # presence in {collapse}) - add a group_by to simulate the collapse in the subq
1698   unless ($sub_attrs->{group_by}) {
1699     for my $alias (keys %inner_joins) {
1700
1701       # the dot comes from some weirdness in collapse
1702       # remove after the rewrite
1703       if ($attrs->{collapse}{".$alias"}) {
1704         $sub_attrs->{group_by} ||= $sub_select;
1705         last;
1706       }
1707     }
1708   }
1709
1710   # generate the subquery
1711   my $subq = $self->_select_args_to_query (
1712     $inner_from,
1713     $sub_select,
1714     $where,
1715     $sub_attrs
1716   );
1717   my $subq_joinspec = {
1718     -alias => $select_root_alias,
1719     -source_handle => $join_root->{-source_handle},
1720     $select_root_alias => $subq,
1721   };
1722
1723   # Generate a new from (really just replace the join slot with the subquery)
1724   # Before we would start the outer chain from the subquery itself (i.e.
1725   # SELECT ... FROM (SELECT ... ) alias JOIN ..., but this turned out to be
1726   # a bad idea for search_related, as the root of the chain was effectively
1727   # lost (i.e. $artist_rs->search_related ('cds'... ) would result in alias
1728   # of 'cds', which would prevent from doing things like order_by artist.*)
1729   # See t/prefetch/via_search_related.t for a better idea
1730   my @outer_from;
1731   if ($join_root->{-alias} eq $select_root_alias) { # just swap the root part and we're done
1732     @outer_from = (
1733       $subq_joinspec,
1734       @$from,
1735     )
1736   }
1737   else {  # this is trickier
1738     @outer_from = ($join_root);
1739
1740     for my $j (@$from) {
1741       if ($j->[0]{-alias} eq $select_root_alias) {
1742         push @outer_from, [
1743           $subq_joinspec,
1744           @{$j}[1 .. $#$j],
1745         ];
1746       }
1747       else {
1748         push @outer_from, $j;
1749       }
1750     }
1751   }
1752
1753   # This is totally horrific - the $where ends up in both the inner and outer query
1754   # Unfortunately not much can be done until SQLA2 introspection arrives, and even
1755   # then if where conditions apply to the *right* side of the prefetch, you may have
1756   # to both filter the inner select (e.g. to apply a limit) and then have to re-filter
1757   # the outer select to exclude joins you didin't want in the first place
1758   #
1759   # OTOH it can be seen as a plus: <ash> (notes that this query would make a DBA cry ;)
1760   return (\@outer_from, $select, $where, $attrs);
1761 }
1762
1763 sub _resolve_ident_sources {
1764   my ($self, $ident) = @_;
1765
1766   my $alias2source = {};
1767   my $rs_alias;
1768
1769   # the reason this is so contrived is that $ident may be a {from}
1770   # structure, specifying multiple tables to join
1771   if ( Scalar::Util::blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
1772     # this is compat mode for insert/update/delete which do not deal with aliases
1773     $alias2source->{me} = $ident;
1774     $rs_alias = 'me';
1775   }
1776   elsif (ref $ident eq 'ARRAY') {
1777
1778     for (@$ident) {
1779       my $tabinfo;
1780       if (ref $_ eq 'HASH') {
1781         $tabinfo = $_;
1782         $rs_alias = $tabinfo->{-alias};
1783       }
1784       if (ref $_ eq 'ARRAY' and ref $_->[0] eq 'HASH') {
1785         $tabinfo = $_->[0];
1786       }
1787
1788       $alias2source->{$tabinfo->{-alias}} = $tabinfo->{-source_handle}->resolve
1789         if ($tabinfo->{-source_handle});
1790     }
1791   }
1792
1793   return ($alias2source, $rs_alias);
1794 }
1795
1796 # Takes $ident, \@column_names
1797 #
1798 # returns { $column_name => \%column_info, ... }
1799 # also note: this adds -result_source => $rsrc to the column info
1800 #
1801 # usage:
1802 #   my $col_sources = $self->_resolve_column_info($ident, @column_names);
1803 sub _resolve_column_info {
1804   my ($self, $ident, $colnames) = @_;
1805   my ($alias2src, $root_alias) = $self->_resolve_ident_sources($ident);
1806
1807   my $sep = $self->_sql_maker_opts->{name_sep} || '.';
1808   $sep = "\Q$sep\E";
1809
1810   my (%return, %seen_cols);
1811
1812   # compile a global list of column names, to be able to properly
1813   # disambiguate unqualified column names (if at all possible)
1814   for my $alias (keys %$alias2src) {
1815     my $rsrc = $alias2src->{$alias};
1816     for my $colname ($rsrc->columns) {
1817       push @{$seen_cols{$colname}}, $alias;
1818     }
1819   }
1820
1821   COLUMN:
1822   foreach my $col (@$colnames) {
1823     my ($alias, $colname) = $col =~ m/^ (?: ([^$sep]+) $sep)? (.+) $/x;
1824
1825     unless ($alias) {
1826       # see if the column was seen exactly once (so we know which rsrc it came from)
1827       if ($seen_cols{$colname} and @{$seen_cols{$colname}} == 1) {
1828         $alias = $seen_cols{$colname}[0];
1829       }
1830       else {
1831         next COLUMN;
1832       }
1833     }
1834
1835     my $rsrc = $alias2src->{$alias};
1836     $return{$col} = $rsrc && {
1837       %{$rsrc->column_info($colname)},
1838       -result_source => $rsrc,
1839       -source_alias => $alias,
1840     };
1841   }
1842
1843   return \%return;
1844 }
1845
1846 # Returns a counting SELECT for a simple count
1847 # query. Abstracted so that a storage could override
1848 # this to { count => 'firstcol' } or whatever makes
1849 # sense as a performance optimization
1850 sub _count_select {
1851   #my ($self, $source, $rs_attrs) = @_;
1852   return { count => '*' };
1853 }
1854
1855 # Returns a SELECT which will end up in the subselect
1856 # There may or may not be a group_by, as the subquery
1857 # might have been called to accomodate a limit
1858 #
1859 # Most databases would be happy with whatever ends up
1860 # here, but some choke in various ways.
1861 #
1862 sub _subq_count_select {
1863   my ($self, $source, $rs_attrs) = @_;
1864   return $rs_attrs->{group_by} if $rs_attrs->{group_by};
1865
1866   my @pcols = map { join '.', $rs_attrs->{alias}, $_ } ($source->primary_columns);
1867   return @pcols ? \@pcols : [ 1 ];
1868 }
1869
1870
1871 sub source_bind_attributes {
1872   my ($self, $source) = @_;
1873
1874   my $bind_attributes;
1875   foreach my $column ($source->columns) {
1876
1877     my $data_type = $source->column_info($column)->{data_type} || '';
1878     $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
1879      if $data_type;
1880   }
1881
1882   return $bind_attributes;
1883 }
1884
1885 =head2 select
1886
1887 =over 4
1888
1889 =item Arguments: $ident, $select, $condition, $attrs
1890
1891 =back
1892
1893 Handle a SQL select statement.
1894
1895 =cut
1896
1897 sub select {
1898   my $self = shift;
1899   my ($ident, $select, $condition, $attrs) = @_;
1900   return $self->cursor_class->new($self, \@_, $attrs);
1901 }
1902
1903 sub select_single {
1904   my $self = shift;
1905   my ($rv, $sth, @bind) = $self->_select(@_);
1906   my @row = $sth->fetchrow_array;
1907   my @nextrow = $sth->fetchrow_array if @row;
1908   if(@row && @nextrow) {
1909     carp "Query returned more than one row.  SQL that returns multiple rows is DEPRECATED for ->find and ->single";
1910   }
1911   # Need to call finish() to work round broken DBDs
1912   $sth->finish();
1913   return @row;
1914 }
1915
1916 =head2 sth
1917
1918 =over 4
1919
1920 =item Arguments: $sql
1921
1922 =back
1923
1924 Returns a L<DBI> sth (statement handle) for the supplied SQL.
1925
1926 =cut
1927
1928 sub _dbh_sth {
1929   my ($self, $dbh, $sql) = @_;
1930
1931   # 3 is the if_active parameter which avoids active sth re-use
1932   my $sth = $self->disable_sth_caching
1933     ? $dbh->prepare($sql)
1934     : $dbh->prepare_cached($sql, {}, 3);
1935
1936   # XXX You would think RaiseError would make this impossible,
1937   #  but apparently that's not true :(
1938   $self->throw_exception($dbh->errstr) if !$sth;
1939
1940   $sth;
1941 }
1942
1943 sub sth {
1944   my ($self, $sql) = @_;
1945   $self->dbh_do('_dbh_sth', $sql);
1946 }
1947
1948 sub _dbh_columns_info_for {
1949   my ($self, $dbh, $table) = @_;
1950
1951   if ($dbh->can('column_info')) {
1952     my %result;
1953     eval {
1954       my ($schema,$tab) = $table =~ /^(.+?)\.(.+)$/ ? ($1,$2) : (undef,$table);
1955       my $sth = $dbh->column_info( undef,$schema, $tab, '%' );
1956       $sth->execute();
1957       while ( my $info = $sth->fetchrow_hashref() ){
1958         my %column_info;
1959         $column_info{data_type}   = $info->{TYPE_NAME};
1960         $column_info{size}      = $info->{COLUMN_SIZE};
1961         $column_info{is_nullable}   = $info->{NULLABLE} ? 1 : 0;
1962         $column_info{default_value} = $info->{COLUMN_DEF};
1963         my $col_name = $info->{COLUMN_NAME};
1964         $col_name =~ s/^\"(.*)\"$/$1/;
1965
1966         $result{$col_name} = \%column_info;
1967       }
1968     };
1969     return \%result if !$@ && scalar keys %result;
1970   }
1971
1972   my %result;
1973   my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
1974   $sth->execute;
1975   my @columns = @{$sth->{NAME_lc}};
1976   for my $i ( 0 .. $#columns ){
1977     my %column_info;
1978     $column_info{data_type} = $sth->{TYPE}->[$i];
1979     $column_info{size} = $sth->{PRECISION}->[$i];
1980     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
1981
1982     if ($column_info{data_type} =~ m/^(.*?)\((.*?)\)$/) {
1983       $column_info{data_type} = $1;
1984       $column_info{size}    = $2;
1985     }
1986
1987     $result{$columns[$i]} = \%column_info;
1988   }
1989   $sth->finish;
1990
1991   foreach my $col (keys %result) {
1992     my $colinfo = $result{$col};
1993     my $type_num = $colinfo->{data_type};
1994     my $type_name;
1995     if(defined $type_num && $dbh->can('type_info')) {
1996       my $type_info = $dbh->type_info($type_num);
1997       $type_name = $type_info->{TYPE_NAME} if $type_info;
1998       $colinfo->{data_type} = $type_name if $type_name;
1999     }
2000   }
2001
2002   return \%result;
2003 }
2004
2005 sub columns_info_for {
2006   my ($self, $table) = @_;
2007   $self->dbh_do('_dbh_columns_info_for', $table);
2008 }
2009
2010 =head2 last_insert_id
2011
2012 Return the row id of the last insert.
2013
2014 =cut
2015
2016 sub _dbh_last_insert_id {
2017     # All Storage's need to register their own _dbh_last_insert_id
2018     # the old SQLite-based method was highly inappropriate
2019
2020     my $self = shift;
2021     my $class = ref $self;
2022     $self->throw_exception (<<EOE);
2023
2024 No _dbh_last_insert_id() method found in $class.
2025 Since the method of obtaining the autoincrement id of the last insert
2026 operation varies greatly between different databases, this method must be
2027 individually implemented for every storage class.
2028 EOE
2029 }
2030
2031 sub last_insert_id {
2032   my $self = shift;
2033   $self->dbh_do('_dbh_last_insert_id', @_);
2034 }
2035
2036 =head2 sqlt_type
2037
2038 Returns the database driver name.
2039
2040 =cut
2041
2042 sub sqlt_type { shift->_get_dbh->{Driver}->{Name} }
2043
2044 =head2 bind_attribute_by_data_type
2045
2046 Given a datatype from column info, returns a database specific bind
2047 attribute for C<< $dbh->bind_param($val,$attribute) >> or nothing if we will
2048 let the database planner just handle it.
2049
2050 Generally only needed for special case column types, like bytea in postgres.
2051
2052 =cut
2053
2054 sub bind_attribute_by_data_type {
2055     return;
2056 }
2057
2058 =head2 is_datatype_numeric
2059
2060 Given a datatype from column_info, returns a boolean value indicating if
2061 the current RDBMS considers it a numeric value. This controls how
2062 L<DBIx::Class::Row/set_column> decides whether to mark the column as
2063 dirty - when the datatype is deemed numeric a C<< != >> comparison will
2064 be performed instead of the usual C<eq>.
2065
2066 =cut
2067
2068 sub is_datatype_numeric {
2069   my ($self, $dt) = @_;
2070
2071   return 0 unless $dt;
2072
2073   return $dt =~ /^ (?:
2074     numeric | int(?:eger)? | (?:tiny|small|medium|big)int | dec(?:imal)? | real | float | double (?: \s+ precision)? | (?:big)?serial
2075   ) $/ix;
2076 }
2077
2078
2079 =head2 create_ddl_dir (EXPERIMENTAL)
2080
2081 =over 4
2082
2083 =item Arguments: $schema \@databases, $version, $directory, $preversion, \%sqlt_args
2084
2085 =back
2086
2087 Creates a SQL file based on the Schema, for each of the specified
2088 database engines in C<\@databases> in the given directory.
2089 (note: specify L<SQL::Translator> names, not L<DBI> driver names).
2090
2091 Given a previous version number, this will also create a file containing
2092 the ALTER TABLE statements to transform the previous schema into the
2093 current one. Note that these statements may contain C<DROP TABLE> or
2094 C<DROP COLUMN> statements that can potentially destroy data.
2095
2096 The file names are created using the C<ddl_filename> method below, please
2097 override this method in your schema if you would like a different file
2098 name format. For the ALTER file, the same format is used, replacing
2099 $version in the name with "$preversion-$version".
2100
2101 See L<SQL::Translator/METHODS> for a list of values for C<\%sqlt_args>.
2102 The most common value for this would be C<< { add_drop_table => 1 } >>
2103 to have the SQL produced include a C<DROP TABLE> statement for each table
2104 created. For quoting purposes supply C<quote_table_names> and
2105 C<quote_field_names>.
2106
2107 If no arguments are passed, then the following default values are assumed:
2108
2109 =over 4
2110
2111 =item databases  - ['MySQL', 'SQLite', 'PostgreSQL']
2112
2113 =item version    - $schema->schema_version
2114
2115 =item directory  - './'
2116
2117 =item preversion - <none>
2118
2119 =back
2120
2121 By default, C<\%sqlt_args> will have
2122
2123  { add_drop_table => 1, ignore_constraint_names => 1, ignore_index_names => 1 }
2124
2125 merged with the hash passed in. To disable any of those features, pass in a
2126 hashref like the following
2127
2128  { ignore_constraint_names => 0, # ... other options }
2129
2130
2131 Note that this feature is currently EXPERIMENTAL and may not work correctly
2132 across all databases, or fully handle complex relationships.
2133
2134 WARNING: Please check all SQL files created, before applying them.
2135
2136 =cut
2137
2138 sub create_ddl_dir {
2139   my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
2140
2141   if(!$dir || !-d $dir) {
2142     carp "No directory given, using ./\n";
2143     $dir = "./";
2144   }
2145   $databases ||= ['MySQL', 'SQLite', 'PostgreSQL'];
2146   $databases = [ $databases ] if(ref($databases) ne 'ARRAY');
2147
2148   my $schema_version = $schema->schema_version || '1.x';
2149   $version ||= $schema_version;
2150
2151   $sqltargs = {
2152     add_drop_table => 1,
2153     ignore_constraint_names => 1,
2154     ignore_index_names => 1,
2155     %{$sqltargs || {}}
2156   };
2157
2158   $self->throw_exception(q{Can't create a ddl file without SQL::Translator 0.09003: '}
2159       . $self->_check_sqlt_message . q{'})
2160           if !$self->_check_sqlt_version;
2161
2162   my $sqlt = SQL::Translator->new( $sqltargs );
2163
2164   $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
2165   my $sqlt_schema = $sqlt->translate({ data => $schema })
2166     or $self->throw_exception ($sqlt->error);
2167
2168   foreach my $db (@$databases) {
2169     $sqlt->reset();
2170     $sqlt->{schema} = $sqlt_schema;
2171     $sqlt->producer($db);
2172
2173     my $file;
2174     my $filename = $schema->ddl_filename($db, $version, $dir);
2175     if (-e $filename && ($version eq $schema_version )) {
2176       # if we are dumping the current version, overwrite the DDL
2177       carp "Overwriting existing DDL file - $filename";
2178       unlink($filename);
2179     }
2180
2181     my $output = $sqlt->translate;
2182     if(!$output) {
2183       carp("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
2184       next;
2185     }
2186     if(!open($file, ">$filename")) {
2187       $self->throw_exception("Can't open $filename for writing ($!)");
2188       next;
2189     }
2190     print $file $output;
2191     close($file);
2192
2193     next unless ($preversion);
2194
2195     require SQL::Translator::Diff;
2196
2197     my $prefilename = $schema->ddl_filename($db, $preversion, $dir);
2198     if(!-e $prefilename) {
2199       carp("No previous schema file found ($prefilename)");
2200       next;
2201     }
2202
2203     my $difffile = $schema->ddl_filename($db, $version, $dir, $preversion);
2204     if(-e $difffile) {
2205       carp("Overwriting existing diff file - $difffile");
2206       unlink($difffile);
2207     }
2208
2209     my $source_schema;
2210     {
2211       my $t = SQL::Translator->new($sqltargs);
2212       $t->debug( 0 );
2213       $t->trace( 0 );
2214
2215       $t->parser( $db )
2216         or $self->throw_exception ($t->error);
2217
2218       my $out = $t->translate( $prefilename )
2219         or $self->throw_exception ($t->error);
2220
2221       $source_schema = $t->schema;
2222
2223       $source_schema->name( $prefilename )
2224         unless ( $source_schema->name );
2225     }
2226
2227     # The "new" style of producers have sane normalization and can support
2228     # diffing a SQL file against a DBIC->SQLT schema. Old style ones don't
2229     # And we have to diff parsed SQL against parsed SQL.
2230     my $dest_schema = $sqlt_schema;
2231
2232     unless ( "SQL::Translator::Producer::$db"->can('preprocess_schema') ) {
2233       my $t = SQL::Translator->new($sqltargs);
2234       $t->debug( 0 );
2235       $t->trace( 0 );
2236
2237       $t->parser( $db )
2238         or $self->throw_exception ($t->error);
2239
2240       my $out = $t->translate( $filename )
2241         or $self->throw_exception ($t->error);
2242
2243       $dest_schema = $t->schema;
2244
2245       $dest_schema->name( $filename )
2246         unless $dest_schema->name;
2247     }
2248
2249     my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
2250                                                   $dest_schema,   $db,
2251                                                   $sqltargs
2252                                                  );
2253     if(!open $file, ">$difffile") {
2254       $self->throw_exception("Can't write to $difffile ($!)");
2255       next;
2256     }
2257     print $file $diff;
2258     close($file);
2259   }
2260 }
2261
2262 =head2 deployment_statements
2263
2264 =over 4
2265
2266 =item Arguments: $schema, $type, $version, $directory, $sqlt_args
2267
2268 =back
2269
2270 Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
2271
2272 The L<SQL::Translator> (not L<DBI>) database driver name can be explicitly
2273 provided in C<$type>, otherwise the result of L</sqlt_type> is used as default.
2274
2275 C<$directory> is used to return statements from files in a previously created
2276 L</create_ddl_dir> directory and is optional. The filenames are constructed
2277 from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
2278
2279 If no C<$directory> is specified then the statements are constructed on the
2280 fly using L<SQL::Translator> and C<$version> is ignored.
2281
2282 See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
2283
2284 =cut
2285
2286 sub deployment_statements {
2287   my ($self, $schema, $type, $version, $dir, $sqltargs) = @_;
2288   $type ||= $self->sqlt_type;
2289   $version ||= $schema->schema_version || '1.x';
2290   $dir ||= './';
2291   my $filename = $schema->ddl_filename($type, $version, $dir);
2292   if(-f $filename)
2293   {
2294       my $file;
2295       open($file, "<$filename")
2296         or $self->throw_exception("Can't open $filename ($!)");
2297       my @rows = <$file>;
2298       close($file);
2299       return join('', @rows);
2300   }
2301
2302   $self->throw_exception(q{Can't deploy without SQL::Translator 0.09003: '}
2303       . $self->_check_sqlt_message . q{'})
2304           if !$self->_check_sqlt_version;
2305
2306   # sources needs to be a parser arg, but for simplicty allow at top level
2307   # coming in
2308   $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
2309       if exists $sqltargs->{sources};
2310
2311   my $tr = SQL::Translator->new(
2312     producer => "SQL::Translator::Producer::${type}",
2313     %$sqltargs,
2314     parser => 'SQL::Translator::Parser::DBIx::Class',
2315     data => $schema,
2316   );
2317   return $tr->translate;
2318 }
2319
2320 sub deploy {
2321   my ($self, $schema, $type, $sqltargs, $dir) = @_;
2322   my $deploy = sub {
2323     my $line = shift;
2324     return if($line =~ /^--/);
2325     return if(!$line);
2326     # next if($line =~ /^DROP/m);
2327     return if($line =~ /^BEGIN TRANSACTION/m);
2328     return if($line =~ /^COMMIT/m);
2329     return if $line =~ /^\s+$/; # skip whitespace only
2330     $self->_query_start($line);
2331     eval {
2332       # do a dbh_do cycle here, as we need some error checking in
2333       # place (even though we will ignore errors)
2334       $self->dbh_do (sub { $_[1]->do($line) });
2335     };
2336     if ($@) {
2337       carp qq{$@ (running "${line}")};
2338     }
2339     $self->_query_end($line);
2340   };
2341   my @statements = $self->deployment_statements($schema, $type, undef, $dir, { %{ $sqltargs || {} }, no_comments => 1 } );
2342   if (@statements > 1) {
2343     foreach my $statement (@statements) {
2344       $deploy->( $statement );
2345     }
2346   }
2347   elsif (@statements == 1) {
2348     foreach my $line ( split(";\n", $statements[0])) {
2349       $deploy->( $line );
2350     }
2351   }
2352 }
2353
2354 =head2 datetime_parser
2355
2356 Returns the datetime parser class
2357
2358 =cut
2359
2360 sub datetime_parser {
2361   my $self = shift;
2362   return $self->{datetime_parser} ||= do {
2363     $self->_populate_dbh unless $self->_dbh;
2364     $self->build_datetime_parser(@_);
2365   };
2366 }
2367
2368 =head2 datetime_parser_type
2369
2370 Defines (returns) the datetime parser class - currently hardwired to
2371 L<DateTime::Format::MySQL>
2372
2373 =cut
2374
2375 sub datetime_parser_type { "DateTime::Format::MySQL"; }
2376
2377 =head2 build_datetime_parser
2378
2379 See L</datetime_parser>
2380
2381 =cut
2382
2383 sub build_datetime_parser {
2384   my $self = shift;
2385   my $type = $self->datetime_parser_type(@_);
2386   eval "use ${type}";
2387   $self->throw_exception("Couldn't load ${type}: $@") if $@;
2388   return $type;
2389 }
2390
2391 {
2392     my $_check_sqlt_version; # private
2393     my $_check_sqlt_message; # private
2394     sub _check_sqlt_version {
2395         return $_check_sqlt_version if defined $_check_sqlt_version;
2396         eval 'use SQL::Translator "0.09003"';
2397         $_check_sqlt_message = $@ || '';
2398         $_check_sqlt_version = !$@;
2399     }
2400
2401     sub _check_sqlt_message {
2402         _check_sqlt_version if !defined $_check_sqlt_message;
2403         $_check_sqlt_message;
2404     }
2405 }
2406
2407 =head2 is_replicating
2408
2409 A boolean that reports if a particular L<DBIx::Class::Storage::DBI> is set to
2410 replicate from a master database.  Default is undef, which is the result
2411 returned by databases that don't support replication.
2412
2413 =cut
2414
2415 sub is_replicating {
2416     return;
2417
2418 }
2419
2420 =head2 lag_behind_master
2421
2422 Returns a number that represents a certain amount of lag behind a master db
2423 when a given storage is replicating.  The number is database dependent, but
2424 starts at zero and increases with the amount of lag. Default in undef
2425
2426 =cut
2427
2428 sub lag_behind_master {
2429     return;
2430 }
2431
2432 sub DESTROY {
2433   my $self = shift;
2434   $self->_verify_pid if $self->_dbh;
2435
2436   # some databases need this to stop spewing warnings
2437   if (my $dbh = $self->_dbh) {
2438     eval { $dbh->disconnect };
2439   }
2440
2441   $self->_dbh(undef);
2442 }
2443
2444 1;
2445
2446 =head1 USAGE NOTES
2447
2448 =head2 DBIx::Class and AutoCommit
2449
2450 DBIx::Class can do some wonderful magic with handling exceptions,
2451 disconnections, and transactions when you use C<< AutoCommit => 1 >>
2452 (the default) combined with C<txn_do> for transaction support.
2453
2454 If you set C<< AutoCommit => 0 >> in your connect info, then you are always
2455 in an assumed transaction between commits, and you're telling us you'd
2456 like to manage that manually.  A lot of the magic protections offered by
2457 this module will go away.  We can't protect you from exceptions due to database
2458 disconnects because we don't know anything about how to restart your
2459 transactions.  You're on your own for handling all sorts of exceptional
2460 cases if you choose the C<< AutoCommit => 0 >> path, just as you would
2461 be with raw DBI.
2462
2463
2464 =head1 AUTHORS
2465
2466 Matt S. Trout <mst@shadowcatsystems.co.uk>
2467
2468 Andy Grundman <andy@hybridized.org>
2469
2470 =head1 LICENSE
2471
2472 You may distribute this code under the same terms as Perl itself.
2473
2474 =cut