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