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