cleanup cursor class handling
[dbsrgits/DBIx-Class.git] / lib / DBIx / Class / Storage / DBI.pm
index 8d2956b..5be636f 100644 (file)
@@ -3,20 +3,22 @@ package DBIx::Class::Storage::DBI;
 
 use base 'DBIx::Class::Storage';
 
-use strict;
+use strict;    
 use warnings;
 use DBI;
 use SQL::Abstract::Limit;
 use DBIx::Class::Storage::DBI::Cursor;
 use DBIx::Class::Storage::Statistics;
-use IO::File;
+use Scalar::Util qw/blessed weaken/;
 
-__PACKAGE__->mk_group_accessors(
-  'simple' =>
-    qw/_connect_info _dbh _sql_maker _sql_maker_opts _conn_pid _conn_tid
-       cursor on_connect_do transaction_depth/
+__PACKAGE__->mk_group_accessors('simple' =>
+    qw/_connect_info _dbi_connect_info _dbh _sql_maker _sql_maker_opts
+       _conn_pid _conn_tid disable_sth_caching on_connect_do
+       transaction_depth unsafe _dbh_autocommit/
 );
 
+__PACKAGE__->cursor_class('DBIx::Class::Storage::DBI::Cursor');
+
 BEGIN {
 
 package DBIC::SQL::Abstract; # Would merge upstream, but nate doesn't reply :(
@@ -58,9 +60,10 @@ WHERE ROW_NUM BETWEEN $offset AND $last
 
 # While we're at it, this should make LIMIT queries more efficient,
 #  without digging into things too deeply
+use Scalar::Util 'blessed';
 sub _find_syntax {
   my ($self, $syntax) = @_;
-  my $dbhname = eval { $syntax->{Driver}->{Name}} || '';
+  my $dbhname = blessed($syntax) ?  $syntax->{Driver}{Name} : $syntax;
   if(ref($self) && $dbhname && $dbhname eq 'DB2') {
     return 'RowNumberOver';
   }
@@ -114,7 +117,7 @@ sub _emulate_limit {
 }
 
 sub _recurse_fields {
-  my ($self, $fields) = @_;
+  my ($self, $fields, $params) = @_;
   my $ref = ref $fields;
   return $self->_quote($fields) unless $ref;
   return $$fields if $ref eq 'SCALAR';
@@ -122,10 +125,10 @@ sub _recurse_fields {
   if ($ref eq 'ARRAY') {
     return join(', ', map {
       $self->_recurse_fields($_)
-      .(exists $self->{rownum_hack_count}
-         ? ' AS col'.$self->{rownum_hack_count}++
-         : '')
-     } @$fields);
+        .(exists $self->{rownum_hack_count} && !($params && $params->{no_rownum_hack})
+          ? ' AS col'.$self->{rownum_hack_count}++
+          : '')
+      } @$fields);
   } elsif ($ref eq 'HASH') {
     foreach my $func (keys %$fields) {
       return $self->_sqlcase($func)
@@ -141,7 +144,7 @@ sub _order_by {
   if (ref $_[0] eq 'HASH') {
     if (defined $_[0]->{group_by}) {
       $ret = $self->_sqlcase(' group by ')
-               .$self->_recurse_fields($_[0]->{group_by});
+        .$self->_recurse_fields($_[0]->{group_by}, { no_rownum_hack => 1 });
     }
     if (defined $_[0]->{having}) {
       my $frag;
@@ -236,9 +239,18 @@ sub _join_condition {
   if (ref $cond eq 'HASH') {
     my %j;
     for (keys %$cond) {
-      my $x = '= '.$self->_quote($cond->{$_}); $j{$_} = \$x;
+      my $v = $cond->{$_};
+      if (ref $v) {
+        # XXX no throw_exception() in this package and croak() fails with strange results
+        Carp::croak(ref($v) . qq{ reference arguments are not supported in JOINS - try using \"..." instead'})
+            if ref($v) ne 'SCALAR';
+        $j{$_} = $v;
+      }
+      else {
+        my $x = '= '.$self->_quote($v); $j{$_} = \$x;
+      }
     };
-    return $self->_recurse_where(\%j);
+    return scalar($self->_recurse_where(\%j));
   } elsif (ref $cond eq 'ARRAY') {
     return join(' OR ', map { $self->_join_condition($_) } @$cond);
   } else {
@@ -301,9 +313,10 @@ documents DBI-specific methods and behaviors.
 sub new {
   my $new = shift->next::method(@_);
 
-  $new->cursor("DBIx::Class::Storage::DBI::Cursor");
   $new->transaction_depth(0);
   $new->_sql_maker_opts({});
+  $new->{_in_dbh_do} = 0;
+  $new->{_dbh_gen} = 0;
 
   $new;
 }
@@ -318,7 +331,13 @@ C<connect_info> here.
 
 The arrayref can either contain the same set of arguments one would
 normally pass to L<DBI/connect>, or a lone code reference which returns
-a connected database handle.
+a connected database handle.  Please note that the L<DBI> docs
+recommend that you always explicitly set C<AutoCommit> to either
+C<0> or C<1>.   L<DBIx::Class> further recommends that it be set
+to C<1>, and that you perform transactions via our L</txn_do>
+method.  L<DBIx::Class> will set it to C<1> if you do not do explicitly
+set it to zero.  This is the default for most DBDs.  See below for more
+details.
 
 In either case, if the final argument in your connect_info happens
 to be a hashref, C<connect_info> will look there for several
@@ -332,6 +351,11 @@ This can be set to an arrayref of literal sql statements, which will
 be executed immediately after making the connection to the database
 every time we [re-]connect.
 
+=item disable_sth_caching
+
+If set to a true value, this option will disable the caching of
+statement handles via L<DBI/prepare_cached>.
+
 =item limit_dialect 
 
 Sets the limit dialect. This is useful for JDBC-bridge among others
@@ -356,6 +380,22 @@ This only needs to be used in conjunction with L<quote_char>, and is used to
 specify the charecter that seperates elements (schemas, tables, columns) from 
 each other. In most cases this is simply a C<.>.
 
+=item unsafe
+
+This Storage driver normally installs its own C<HandleError>, sets
+C<RaiseError> and C<ShowErrorStatement> on, and sets C<PrintError> off on
+all database handles, including those supplied by a coderef.  It does this
+so that it can have consistent and useful error behavior.
+
+If you set this option to a true value, Storage will not do its usual
+modifications to the database handle's attributes, and instead relies on
+the settings in your connect_info DBI options (or the values you set in
+your connection coderef, in the case that you are connecting via coderef).
+
+Note that your custom settings can cause Storage to malfunction,
+especially if you set a C<HandleError> handler that suppresses exceptions
+and/or disable C<RaiseError>.
+
 =back
 
 These options can be mixed in with your other L<DBI> connection attributes,
@@ -366,11 +406,20 @@ Every time C<connect_info> is invoked, any previous settings for
 these options will be cleared before setting the new ones, regardless of
 whether any options are specified in the new C<connect_info>.
 
-Important note:  DBIC expects the returned database handle provided by 
-a subref argument to have RaiseError set on it.  If it doesn't, things
-might not work very well, YMMV.  If you don't use a subref, DBIC will
-force this setting for you anyways.  Setting HandleError to anything
-other than simple exception object wrapper might cause problems too.
+Another Important Note:
+
+DBIC can do some wonderful magic with handling exceptions,
+disconnections, and transactions when you use C<AutoCommit =&gt; 1>
+combined with C<txn_do> for transaction support.
+
+If you set C<AutoCommit =&gt; 0> in your connect info, then you are always
+in an assumed transaction between commits, and you're telling us you'd
+like to manage that manually.  A lot of DBIC's magic protections
+go away.  We can't protect you from exceptions due to database
+disconnects because we don't know anything about how to restart your
+transactions.  You're on your own for handling all sorts of exceptional
+cases if you choose the C<AutoCommit =&gt 0> path, just as you would
+be with raw DBI.
 
 Examples:
 
@@ -386,7 +435,7 @@ Examples:
       'dbi:Pg:dbname=foo',
       'postgres',
       'my_pg_password',
-      { AutoCommit => 0 },
+      { AutoCommit => 1 },
       { quote_char => q{"}, name_sep => q{.} },
     ]
   );
@@ -397,7 +446,7 @@ Examples:
       'dbi:Pg:dbname=foo',
       'postgres',
       'my_pg_password',
-      { AutoCommit => 0, quote_char => q{"}, name_sep => q{.} },
+      { AutoCommit => 1, quote_char => q{"}, name_sep => q{.} },
     ]
   );
 
@@ -409,6 +458,7 @@ Examples:
           quote_char => q{`},
           name_sep => q{@},
           on_connect_do => ['SET search_path TO myschema,otherschema,public'],
+          disable_sth_caching => 1,
       },
     ]
   );
@@ -424,24 +474,32 @@ sub connect_info {
   #  the new set of options
   $self->_sql_maker(undef);
   $self->_sql_maker_opts({});
+  $self->_connect_info([@$info_arg]); # copy for _connect_info
 
-  my $info = [ @$info_arg ]; # copy because we can alter it
-  my $last_info = $info->[-1];
+  my $dbi_info = [@$info_arg]; # copy for _dbi_connect_info
+
+  my $last_info = $dbi_info->[-1];
   if(ref $last_info eq 'HASH') {
-    if(my $on_connect_do = delete $last_info->{on_connect_do}) {
-      $self->on_connect_do($on_connect_do);
+    $last_info = { %$last_info }; # so delete is non-destructive
+    for my $storage_opt (qw/on_connect_do disable_sth_caching unsafe/) {
+      if(my $value = delete $last_info->{$storage_opt}) {
+        $self->$storage_opt($value);
+      }
     }
     for my $sql_maker_opt (qw/limit_dialect quote_char name_sep/) {
       if(my $opt_val = delete $last_info->{$sql_maker_opt}) {
         $self->_sql_maker_opts->{$sql_maker_opt} = $opt_val;
       }
     }
+    # re-insert modified hashref
+    $dbi_info->[-1] = $last_info;
 
     # Get rid of any trailing empty hashref
-    pop(@$info) if !keys %$last_info;
+    pop(@$dbi_info) if !keys %$last_info;
   }
+  $self->_dbi_connect_info($dbi_info);
 
-  $self->_connect_info($info);
+  $self->_connect_info;
 }
 
 =head2 on_connect_do
@@ -452,19 +510,26 @@ This method is deprecated in favor of setting via L</connect_info>.
 
 Arguments: $subref, @extra_coderef_args?
 
-Execute the given subref with the underlying database handle as its
-first argument, using the new exception-based connection management.
+Execute the given subref using the new exception-based connection management.
+
+The first two arguments will be the storage object that C<dbh_do> was called
+on and a database handle to use.  Any additional arguments will be passed
+verbatim to the called subref as arguments 2 and onwards.
 
-Any additional arguments will be passed verbatim to the called subref
-as arguments 2 and onwards.
+Using this (instead of $self->_dbh or $self->dbh) ensures correct
+exception handling and reconnection (or failover in future subclasses).
+
+Your subref should have no side-effects outside of the database, as
+there is the potential for your subref to be partially double-executed
+if the database connection was stale/dysfunctional.
 
 Example:
 
   my @stuff = $schema->storage->dbh_do(
     sub {
-      my $dbh = shift;
-      my $cols = join(q{, }, @_);
-      shift->selectrow_array("SELECT $cols FROM foo")
+      my ($storage, $dbh, @cols) = @_;
+      my $cols = join(q{, }, @cols);
+      $dbh->selectrow_array("SELECT $cols FROM foo");
     },
     @column_list
   );
@@ -475,11 +540,14 @@ sub dbh_do {
   my $self = shift;
   my $coderef = shift;
 
-  return $coderef->($self->_dbh, @_) if $self->{_in_txn_do};
-
   ref $coderef eq 'CODE' or $self->throw_exception
     ('$coderef must be a CODE reference');
 
+  return $coderef->($self, $self->_dbh, @_) if $self->{_in_dbh_do}
+      || $self->{transaction_depth};
+
+  local $self->{_in_dbh_do} = 1;
+
   my @result;
   my $want_array = wantarray;
 
@@ -487,13 +555,13 @@ sub dbh_do {
     $self->_verify_pid if $self->_dbh;
     $self->_populate_dbh if !$self->_dbh;
     if($want_array) {
-        @result = $coderef->($self->_dbh, @_);
+        @result = $coderef->($self, $self->_dbh, @_);
     }
     elsif(defined $want_array) {
-        $result[0] = $coderef->($self->_dbh, @_);
+        $result[0] = $coderef->($self, $self->_dbh, @_);
     }
     else {
-        $coderef->($self->_dbh, @_);
+        $coderef->($self, $self->_dbh, @_);
     }
   };
 
@@ -505,12 +573,12 @@ sub dbh_do {
   # We were not connected - reconnect and retry, but let any
   #  exception fall right through this time
   $self->_populate_dbh;
-  $coderef->($self->_dbh, @_);
+  $coderef->($self, $self->_dbh, @_);
 }
 
 # This is basically a blend of dbh_do above and DBIx::Class::Storage::txn_do.
 # It also informs dbh_do to bypass itself while under the direction of txn_do,
-#  via $self->{_in_txn_do} (this saves some redundant eval and errorcheck, etc)
+#  via $self->{_in_dbh_do} (this saves some redundant eval and errorcheck, etc)
 sub txn_do {
   my $self = shift;
   my $coderef = shift;
@@ -518,56 +586,55 @@ sub txn_do {
   ref $coderef eq 'CODE' or $self->throw_exception
     ('$coderef must be a CODE reference');
 
-  local $self->{_in_txn_do} = 1;
+  return $coderef->(@_) if $self->{transaction_depth};
 
-  my $tried = 0;
+  local $self->{_in_dbh_do} = 1;
 
   my @result;
   my $want_array = wantarray;
 
-  START_TXN: eval {
-    $self->_verify_pid if $self->_dbh;
-    $self->_populate_dbh if !$self->_dbh;
-
-    $self->txn_begin;
-    if($want_array) {
-        @result = $coderef->(@_);
-    }
-    elsif(defined $want_array) {
-        $result[0] = $coderef->(@_);
-    }
-    else {
-        $coderef->(@_);
-    }
-    $self->txn_commit;
-  };
+  my $tried = 0;
+  while(1) {
+    eval {
+      $self->_verify_pid if $self->_dbh;
+      $self->_populate_dbh if !$self->_dbh;
 
-  my $exception = $@;
-  if(!$exception) { return $want_array ? @result : $result[0] }
+      $self->txn_begin;
+      if($want_array) {
+          @result = $coderef->(@_);
+      }
+      elsif(defined $want_array) {
+          $result[0] = $coderef->(@_);
+      }
+      else {
+          $coderef->(@_);
+      }
+      $self->txn_commit;
+    };
 
-  if($tried++ > 0 || $self->connected) {
-    eval { $self->txn_rollback };
-    my $rollback_exception = $@;
-    if($rollback_exception) {
-      my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
-      $self->throw_exception($exception)  # propagate nested rollback
-        if $rollback_exception =~ /$exception_class/;
-
-      $self->throw_exception(
-        "Transaction aborted: ${exception}. "
-        . "Rollback failed: ${rollback_exception}"
-      );
+    my $exception = $@;
+    if(!$exception) { return $want_array ? @result : $result[0] }
+
+    if($tried++ > 0 || $self->connected) {
+      eval { $self->txn_rollback };
+      my $rollback_exception = $@;
+      if($rollback_exception) {
+        my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
+        $self->throw_exception($exception)  # propagate nested rollback
+          if $rollback_exception =~ /$exception_class/;
+
+        $self->throw_exception(
+          "Transaction aborted: ${exception}. "
+          . "Rollback failed: ${rollback_exception}"
+        );
+      }
+      $self->throw_exception($exception)
     }
-    $self->throw_exception($exception)
-  }
-
-  # We were not connected, and was first try - reconnect and retry
-  # XXX I know, gotos are evil.  If you can find a better way
-  #  to write this that doesn't duplicate a lot of code/structure,
-  #  and behaves identically, feel free...
 
-  $self->_populate_dbh;
-  goto START_TXN;
+    # We were not connected, and was first try - reconnect and retry
+    # via the while loop
+    $self->_populate_dbh;
+  }
 }
 
 =head2 disconnect
@@ -581,9 +648,10 @@ sub disconnect {
   my ($self) = @_;
 
   if( $self->connected ) {
-    $self->_dbh->rollback unless $self->_dbh->{AutoCommit};
+    $self->_dbh->rollback unless $self->_dbh_autocommit;
     $self->_dbh->disconnect;
     $self->_dbh(undef);
+    $self->{_dbh_gen}++;
   }
 }
 
@@ -592,7 +660,9 @@ sub connected {
 
   if(my $dbh = $self->_dbh) {
       if(defined $self->_conn_tid && $self->_conn_tid != threads->tid) {
-          return $self->_dbh(undef);
+          $self->_dbh(undef);
+          $self->{_dbh_gen}++;
+          return;
       }
       else {
           $self->_verify_pid;
@@ -612,6 +682,7 @@ sub _verify_pid {
 
   $self->_dbh->{InactiveDestroy} = 1;
   $self->_dbh(undef);
+  $self->{_dbh_gen}++;
 
   return;
 }
@@ -640,7 +711,7 @@ sub dbh {
 sub _sql_maker_args {
     my ($self) = @_;
     
-    return ( limit_dialect => $self->dbh, %{$self->_sql_maker_opts} );
+    return ( bindtype=>'columns', limit_dialect => $self->dbh, %{$self->_sql_maker_opts} );
 }
 
 sub sql_maker {
@@ -653,9 +724,13 @@ sub sql_maker {
 
 sub _populate_dbh {
   my ($self) = @_;
-  my @info = @{$self->_connect_info || []};
+  my @info = @{$self->_dbi_connect_info || []};
   $self->_dbh($self->_connect(@info));
 
+  # Always set the transaction depth on connect, since
+  #  there is no transaction in progress by definition
+  $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
+
   if(ref $self eq 'DBIx::Class::Storage::DBI') {
     my $driver = $self->_dbh->{Driver}->{Name};
     if ($self->load_optional_class("DBIx::Class::Storage::DBI::${driver}")) {
@@ -679,13 +754,13 @@ sub _connect {
   my ($self, @info) = @_;
 
   $self->throw_exception("You failed to provide any connection info")
-      if !@info;
+    if !@info;
 
   my ($old_connect_via, $dbh);
 
   if ($INC{'Apache/DBI.pm'} && $ENV{MOD_PERL}) {
-      $old_connect_via = $DBI::connect_via;
-      $DBI::connect_via = 'connect';
+    $old_connect_via = $DBI::connect_via;
+    $DBI::connect_via = 'connect';
   }
 
   eval {
@@ -694,141 +769,242 @@ sub _connect {
     }
     else {
        $dbh = DBI->connect(@info);
-       $dbh->{RaiseError} = 1;
-       $dbh->{PrintError} = 0;
+    }
+
+    if(!$self->unsafe) {
+      my $weak_self = $self;
+      weaken($weak_self);
+      $dbh->{HandleError} = sub {
+          $weak_self->throw_exception("DBI Exception: $_[0]")
+      };
+      $dbh->{ShowErrorStatement} = 1;
+      $dbh->{RaiseError} = 1;
+      $dbh->{PrintError} = 0;
     }
   };
 
   $DBI::connect_via = $old_connect_via if $old_connect_via;
 
-  if (!$dbh || $@) {
-    $self->throw_exception("DBI Connection failed: " . ($@ || $DBI::errstr));
-  }
+  $self->throw_exception("DBI Connection failed: " . ($@||$DBI::errstr))
+    if !$dbh || $@;
+
+  $self->_dbh_autocommit($dbh->{AutoCommit});
 
   $dbh;
 }
 
-sub __txn_begin {
-  my ($dbh, $self) = @_;
-  if ($dbh->{AutoCommit}) {
-    $self->debugobj->txn_begin()
-      if ($self->debug);
-    $dbh->begin_work;
-  }
-}
 
 sub txn_begin {
   my $self = shift;
-  $self->dbh_do(\&__txn_begin, $self)
-    if $self->{transaction_depth}++ == 0;
-}
-
-sub __txn_commit {
-  my ($dbh, $self) = @_;
-  if ($self->{transaction_depth} == 0) {
-    unless ($dbh->{AutoCommit}) {
-      $self->debugobj->txn_commit()
-        if ($self->debug);
-      $dbh->commit;
-    }
-  }
-  else {
-    if (--$self->{transaction_depth} == 0) {
-      $self->debugobj->txn_commit()
-        if ($self->debug);
-      $dbh->commit;
-    }
+  $self->ensure_connected();
+  if($self->{transaction_depth} == 0) {
+    $self->debugobj->txn_begin()
+      if $self->debug;
+    # this isn't ->_dbh-> because
+    #  we should reconnect on begin_work
+    #  for AutoCommit users
+    $self->dbh->begin_work;
   }
+  $self->{transaction_depth}++;
 }
 
 sub txn_commit {
   my $self = shift;
-  $self->dbh_do(\&__txn_commit, $self);
+  if ($self->{transaction_depth} == 1) {
+    my $dbh = $self->_dbh;
+    $self->debugobj->txn_commit()
+      if ($self->debug);
+    $dbh->commit;
+    $self->{transaction_depth} = 0
+      if $self->_dbh_autocommit;
+  }
+  elsif($self->{transaction_depth} > 1) {
+    $self->{transaction_depth}--
+  }
 }
 
-sub __txn_rollback {
-  my ($dbh, $self) = @_;
-  if ($self->{transaction_depth} == 0) {
-    unless ($dbh->{AutoCommit}) {
+sub txn_rollback {
+  my $self = shift;
+  my $dbh = $self->_dbh;
+  eval {
+    if ($self->{transaction_depth} == 1) {
       $self->debugobj->txn_rollback()
         if ($self->debug);
+      $self->{transaction_depth} = 0
+        if $self->_dbh_autocommit;
       $dbh->rollback;
     }
-  }
-  else {
-    if (--$self->{transaction_depth} == 0) {
-      $self->debugobj->txn_rollback()
-        if ($self->debug);
-      $dbh->rollback;
+    elsif($self->{transaction_depth} > 1) {
+      $self->{transaction_depth}--;
     }
     else {
       die DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION->new;
     }
-  }
-}
-
-sub txn_rollback {
-  my $self = shift;
-  eval { $self->dbh_do(\&__txn_rollback, $self) };
+  };
   if ($@) {
     my $error = $@;
     my $exception_class = "DBIx::Class::Storage::NESTED_ROLLBACK_EXCEPTION";
     $error =~ /$exception_class/ and $self->throw_exception($error);
-    $self->{transaction_depth} = 0;          # ensure that a failed rollback
-    $self->throw_exception($error);          # resets the transaction depth
+    # ensure that a failed rollback resets the transaction depth
+    $self->{transaction_depth} = $self->_dbh_autocommit ? 0 : 1;
+    $self->throw_exception($error);
   }
 }
 
-sub _execute {
-  my ($self, $op, $extra_bind, $ident, @args) = @_;
-  my ($sql, @bind) = $self->sql_maker->$op($ident, @args);
-  unshift(@bind, @$extra_bind) if $extra_bind;
+# This used to be the top-half of _execute.  It was split out to make it
+#  easier to override in NoBindVars without duping the rest.  It takes up
+#  all of _execute's args, and emits $sql, @bind.
+sub _prep_for_execute {
+  my ($self, $op, $extra_bind, $ident, $args) = @_;
+
+  my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
+  unshift(@bind,
+    map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
+      if $extra_bind;
+
+  return ($sql, \@bind);
+}
+
+sub _dbh_execute {
+  my ($self, $dbh, $op, $extra_bind, $ident, $bind_attributes, @args) = @_;
+  
+  if( blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
+    $ident = $ident->from();
+  }
+
+  my ($sql, $bind) = $self->_prep_for_execute($op, $extra_bind, $ident, \@args);
+
   if ($self->debug) {
-      my @debug_bind = map { defined $_ ? qq{'$_'} : q{'NULL'} } @bind;
+      my @debug_bind =
+        map { defined ($_ && $_->[1]) ? qq{'$_->[1]'} : q{'NULL'} } @$bind;
       $self->debugobj->query_start($sql, @debug_bind);
   }
-  my $sth = eval { $self->sth($sql,$op) };
 
-  if (!$sth || $@) {
-    $self->throw_exception(
-      'no sth generated via sql (' . ($@ || $self->_dbh->errstr) . "): $sql"
-    );
-  }
-  @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
-  my $rv;
-  if ($sth) {
-    my $time = time();
-    $rv = eval { $sth->execute(@bind) };
-
-    if ($@ || !$rv) {
-      $self->throw_exception("Error executing '$sql': ".($@ || $sth->errstr));
+  my $sth = $self->sth($sql,$op);
+
+  my $placeholder_index = 1; 
+
+  foreach my $bound (@$bind) {
+    my $attributes = {};
+    my($column_name, @data) = @$bound;
+
+    if ($bind_attributes) {
+      $attributes = $bind_attributes->{$column_name}
+      if defined $bind_attributes->{$column_name};
+    }
+
+    foreach my $data (@data) {
+      $data = ref $data ? ''.$data : $data; # stringify args
+
+      $sth->bind_param($placeholder_index, $data, $attributes);
+      $placeholder_index++;
     }
-  } else {
-    $self->throw_exception("'$sql' did not generate a statement.");
   }
+
+  # Can this fail without throwing an exception anyways???
+  my $rv = $sth->execute();
+  $self->throw_exception($sth->errstr) if !$rv;
+
   if ($self->debug) {
-      my @debug_bind = map { defined $_ ? qq{`$_'} : q{`NULL'} } @bind;
-      $self->debugobj->query_end($sql, @debug_bind);
+     my @debug_bind =
+       map { defined ($_ && $_->[1]) ? qq{'$_->[1]'} : q{'NULL'} } @$bind; 
+     $self->debugobj->query_end($sql, @debug_bind);
   }
-  return (wantarray ? ($rv, $sth, @bind) : $rv);
+
+  return (wantarray ? ($rv, $sth, @$bind) : $rv);
+}
+
+sub _execute {
+    my $self = shift;
+    $self->dbh_do($self->can('_dbh_execute'), @_)
 }
 
 sub insert {
-  my ($self, $ident, $to_insert) = @_;
-  $self->throw_exception(
-    "Couldn't insert ".join(', ',
-      map "$_ => $to_insert->{$_}", keys %$to_insert
-    )." into ${ident}"
-  ) unless ($self->_execute('insert' => [], $ident, $to_insert));
+  my ($self, $source, $to_insert) = @_;
+  
+  my $ident = $source->from; 
+  my $bind_attributes = $self->source_bind_attributes($source);
+
+  $self->_execute('insert' => [], $source, $bind_attributes, $to_insert);
+
   return $to_insert;
 }
 
+## Still not quite perfect, and EXPERIMENTAL
+## Currently it is assumed that all values passed will be "normal", i.e. not 
+## scalar refs, or at least, all the same type as the first set, the statement is
+## only prepped once.
+sub insert_bulk {
+  my ($self, $source, $cols, $data) = @_;
+  my %colvalues;
+  my $table = $source->from;
+  @colvalues{@$cols} = (0..$#$cols);
+  my ($sql, @bind) = $self->sql_maker->insert($table, \%colvalues);
+  
+  if ($self->debug) {
+      my @debug_bind = map { defined $_->[1] ? qq{$_->[1]} : q{'NULL'} } @bind;
+      $self->debugobj->query_start($sql, @debug_bind);
+  }
+  my $sth = $self->sth($sql);
+
+#  @bind = map { ref $_ ? ''.$_ : $_ } @bind; # stringify args
+
+  ## This must be an arrayref, else nothing works!
+  
+  my $tuple_status = [];
+  
+  ##use Data::Dumper;
+  ##print STDERR Dumper( $data, $sql, [@bind] );
+
+  my $time = time();
+
+  ## Get the bind_attributes, if any exist
+  my $bind_attributes = $self->source_bind_attributes($source);
+
+  ## Bind the values and execute
+  my $placeholder_index = 1; 
+
+  foreach my $bound (@bind) {
+
+    my $attributes = {};
+    my ($column_name, $data_index) = @$bound;
+
+    if( $bind_attributes ) {
+      $attributes = $bind_attributes->{$column_name}
+      if defined $bind_attributes->{$column_name};
+    }
+
+    my @data = map { $_->[$data_index] } @$data;
+
+    $sth->bind_param_array( $placeholder_index, [@data], $attributes );
+    $placeholder_index++;
+  }
+  my $rv = $sth->execute_array({ArrayTupleStatus => $tuple_status});
+  $self->throw_exception($sth->errstr) if !$rv;
+
+  if ($self->debug) {
+      my @debug_bind = map { defined $_ ? qq{`$_'} : q{`NULL'} } @bind;
+      $self->debugobj->query_end($sql, @debug_bind);
+  }
+  return (wantarray ? ($rv, $sth, @bind) : $rv);
+}
+
 sub update {
-  return shift->_execute('update' => [], @_);
+  my $self = shift @_;
+  my $source = shift @_;
+  my $bind_attributes = $self->source_bind_attributes($source);
+  
+  return $self->_execute('update' => [], $source, $bind_attributes, @_);
 }
 
+
 sub delete {
-  return shift->_execute('delete' => [], @_);
+  my $self = shift @_;
+  my $source = shift @_;
+  
+  my $bind_attrs = {}; ## If ever it's needed...
+  
+  return $self->_execute('delete' => [], $source, $bind_attrs, @_);
 }
 
 sub _select {
@@ -844,7 +1020,8 @@ sub _select {
       ($order ? (order_by => $order) : ())
     };
   }
-  my @args = ('select', $attrs->{bind}, $ident, $select, $condition, $order);
+  my $bind_attrs = {}; ## Future support
+  my @args = ('select', $attrs->{bind}, $ident, $bind_attrs, $select, $condition, $order);
   if ($attrs->{software_limit} ||
       $self->sql_maker->_default_limit_syntax eq "GenericSubQ") {
         $attrs->{software_limit} = 1;
@@ -856,10 +1033,36 @@ sub _select {
   return $self->_execute(@args);
 }
 
+sub source_bind_attributes {
+  my ($self, $source) = @_;
+  
+  my $bind_attributes;
+  foreach my $column ($source->columns) {
+  
+    my $data_type = $source->column_info($column)->{data_type} || '';
+    $bind_attributes->{$column} = $self->bind_attribute_by_data_type($data_type)
+     if $data_type;
+  }
+
+  return $bind_attributes;
+}
+
+=head2 select
+
+=over 4
+
+=item Arguments: $ident, $select, $condition, $attrs
+
+=back
+
+Handle a SQL select statement.
+
+=cut
+
 sub select {
   my $self = shift;
   my ($ident, $select, $condition, $attrs) = @_;
-  return $self->cursor->new($self, \@_, $attrs);
+  return $self->cursor_class->new($self, \@_, $attrs);
 }
 
 sub select_single {
@@ -873,24 +1076,38 @@ sub select_single {
 
 =head2 sth
 
+=over 4
+
+=item Arguments: $sql
+
+=back
+
 Returns a L<DBI> sth (statement handle) for the supplied SQL.
 
 =cut
 
-sub __sth {
-  my ($dbh, $sql) = @_;
+sub _dbh_sth {
+  my ($self, $dbh, $sql) = @_;
+
   # 3 is the if_active parameter which avoids active sth re-use
-  $dbh->prepare_cached($sql, {}, 3);
+  my $sth = $self->disable_sth_caching
+    ? $dbh->prepare($sql)
+    : $dbh->prepare_cached($sql, {}, 3);
+
+  # XXX You would think RaiseError would make this impossible,
+  #  but apparently that's not true :(
+  $self->throw_exception($dbh->errstr) if !$sth;
+
+  $sth;
 }
 
 sub sth {
   my ($self, $sql) = @_;
-  $self->dbh_do(\&__sth, $sql);
+  $self->dbh_do($self->can('_dbh_sth'), $sql);
 }
 
-
-sub __columns_info_for {
-  my ($dbh, $self, $table) = @_;
+sub _dbh_columns_info_for {
+  my ($self, $dbh, $table) = @_;
 
   if ($dbh->can('column_info')) {
     my %result;
@@ -914,18 +1131,12 @@ sub __columns_info_for {
   }
 
   my %result;
-  my $sth = $dbh->prepare("SELECT * FROM $table WHERE 1=0");
+  my $sth = $dbh->prepare($self->sql_maker->select($table, undef, \'1 = 0'));
   $sth->execute;
   my @columns = @{$sth->{NAME_lc}};
   for my $i ( 0 .. $#columns ){
     my %column_info;
-    my $type_num = $sth->{TYPE}->[$i];
-    my $type_name;
-    if(defined $type_num && $dbh->can('type_info')) {
-      my $type_info = $dbh->type_info($type_num);
-      $type_name = $type_info->{TYPE_NAME} if $type_info;
-    }
-    $column_info{data_type} = $type_name ? $type_name : $type_num;
+    $column_info{data_type} = $sth->{TYPE}->[$i];
     $column_info{size} = $sth->{PRECISION}->[$i];
     $column_info{is_nullable} = $sth->{NULLABLE}->[$i] ? 1 : 0;
 
@@ -936,13 +1147,25 @@ sub __columns_info_for {
 
     $result{$columns[$i]} = \%column_info;
   }
+  $sth->finish;
+
+  foreach my $col (keys %result) {
+    my $colinfo = $result{$col};
+    my $type_num = $colinfo->{data_type};
+    my $type_name;
+    if(defined $type_num && $dbh->can('type_info')) {
+      my $type_info = $dbh->type_info($type_num);
+      $type_name = $type_info->{TYPE_NAME} if $type_info;
+      $colinfo->{data_type} = $type_name if $type_name;
+    }
+  }
 
   return \%result;
 }
 
 sub columns_info_for {
   my ($self, $table) = @_;
-  $self->dbh_do(\&__columns_info_for, $self, $table);
+  $self->dbh_do($self->can('_dbh_columns_info_for'), $table);
 }
 
 =head2 last_insert_id
@@ -951,10 +1174,15 @@ Return the row id of the last insert.
 
 =cut
 
+sub _dbh_last_insert_id {
+    my ($self, $dbh, $source, $col) = @_;
+    # XXX This is a SQLite-ism as a default... is there a DBI-generic way?
+    $dbh->func('last_insert_rowid');
+}
+
 sub last_insert_id {
-  my ($self, $row) = @_;
-    
-  $self->dbh_do(sub { shift->func('last_insert_rowid') });
+  my $self = shift;
+  $self->dbh_do($self->can('_dbh_last_insert_id'), @_);
 }
 
 =head2 sqlt_type
@@ -963,27 +1191,38 @@ Returns the database driver name.
 
 =cut
 
-sub sqlt_type { shift->dbh_do(sub { shift->{Driver}->{Name} }) }
+sub sqlt_type { shift->dbh->{Driver}->{Name} }
+
+=head2 bind_attribute_by_data_type
 
-=head2 create_ddl_dir (EXPERIMENTAL)
+Given a datatype from column info, returns a database specific bind attribute for
+$dbh->bind_param($val,$attribute) or nothing if we will let the database planner
+just handle it.
+
+Generally only needed for special case column types, like bytea in postgres.
+
+=cut
+
+sub bind_attribute_by_data_type {
+    return;
+}
+
+=head2 create_ddl_dir
 
 =over 4
 
-=item Arguments: $schema \@databases, $version, $directory, $sqlt_args
+=item Arguments: $schema \@databases, $version, $directory, $preversion, $sqlt_args
 
 =back
 
-Creates an SQL file based on the Schema, for each of the specified
+Creates a SQL file based on the Schema, for each of the specified
 database types, in the given directory.
 
-Note that this feature is currently EXPERIMENTAL and may not work correctly
-across all databases, or fully handle complex relationships.
-
 =cut
 
 sub create_ddl_dir
 {
-  my ($self, $schema, $databases, $version, $dir, $sqltargs) = @_;
+  my ($self, $schema, $databases, $version, $dir, $preversion, $sqltargs) = @_;
 
   if(!$dir || !-d $dir)
   {
@@ -995,15 +1234,20 @@ sub create_ddl_dir
   $version ||= $schema->VERSION || '1.x';
   $sqltargs = { ( add_drop_table => 1 ), %{$sqltargs || {}} };
 
-  eval "use SQL::Translator";
-  $self->throw_exception("Can't deploy without SQL::Translator: $@") if $@;
+  $self->throw_exception(q{Can't create a ddl file without SQL::Translator 0.08: '}
+      . $self->_check_sqlt_message . q{'})
+          if !$self->_check_sqlt_version;
 
-  my $sqlt = SQL::Translator->new($sqltargs);
+  my $sqlt = SQL::Translator->new({
+#      debug => 1,
+      add_drop_table => 1,
+  });
   foreach my $db (@$databases)
   {
     $sqlt->reset();
     $sqlt->parser('SQL::Translator::Parser::DBIx::Class');
 #    $sqlt->parser_args({'DBIx::Class' => $schema);
+    $sqlt = $self->configure_sqlt($sqlt, $db);
     $sqlt->data($schema);
     $sqlt->producer($db);
 
@@ -1011,30 +1255,114 @@ sub create_ddl_dir
     my $filename = $schema->ddl_filename($db, $dir, $version);
     if(-e $filename)
     {
-      $self->throw_exception("$filename already exists, skipping $db");
+      warn("$filename already exists, skipping $db");
       next;
     }
-    open($file, ">$filename") 
-      or $self->throw_exception("Can't open $filename for writing ($!)");
+
     my $output = $sqlt->translate;
-#use Data::Dumper;
-#    print join(":", keys %{$schema->source_registrations});
-#    print Dumper($sqlt->schema);
     if(!$output)
     {
-      $self->throw_exception("Failed to translate to $db. (" . $sqlt->error . ")");
+      warn("Failed to translate to $db, skipping. (" . $sqlt->error . ")");
       next;
     }
+    if(!open($file, ">$filename"))
+    {
+        $self->throw_exception("Can't open $filename for writing ($!)");
+        next;
+    }
     print $file $output;
     close($file);
+
+    if($preversion)
+    {
+      require SQL::Translator::Diff;
+
+      my $prefilename = $schema->ddl_filename($db, $dir, $preversion);
+#      print "Previous version $prefilename\n";
+      if(!-e $prefilename)
+      {
+        warn("No previous schema file found ($prefilename)");
+        next;
+      }
+      #### We need to reparse the SQLite file we just wrote, so that 
+      ##   Diff doesnt get all confoosed, and Diff is *very* confused.
+      ##   FIXME: rip Diff to pieces!
+#      my $target_schema = $sqlt->schema;
+#      unless ( $target_schema->name ) {
+#        $target_schema->name( $filename );
+#      }
+      my @input;
+      push @input, {file => $prefilename, parser => $db};
+      push @input, {file => $filename, parser => $db};
+      my ( $source_schema, $source_db, $target_schema, $target_db ) = map {
+        my $file   = $_->{'file'};
+        my $parser = $_->{'parser'};
+
+        my $t = SQL::Translator->new;
+        $t->debug( 0 );
+        $t->trace( 0 );
+        $t->parser( $parser )            or die $t->error;
+        my $out = $t->translate( $file ) or die $t->error;
+        my $schema = $t->schema;
+        unless ( $schema->name ) {
+          $schema->name( $file );
+        }
+        ($schema, $parser);
+      } @input;
+
+      my $diff = SQL::Translator::Diff::schema_diff($source_schema, $db,
+                                                    $target_schema, $db,
+                                                    {}
+                                                   );
+      my $difffile = $schema->ddl_filename($db, $dir, $version, $preversion);
+      print STDERR "Diff: $difffile: $db, $dir, $version, $preversion \n";
+      if(-e $difffile)
+      {
+        warn("$difffile already exists, skipping");
+        next;
+      }
+      if(!open $file, ">$difffile")
+      { 
+        $self->throw_exception("Can't write to $difffile ($!)");
+        next;
+      }
+      print $file $diff;
+      close($file);
+    }
   }
+}
 
+sub configure_sqlt() {
+  my $self = shift;
+  my $tr = shift;
+  my $db = shift || $self->sqlt_type;
+  if ($db eq 'PostgreSQL') {
+    $tr->quote_table_names(0);
+    $tr->quote_field_names(0);
+  }
+  return $tr;
 }
 
 =head2 deployment_statements
 
-Create the statements for L</deploy> and
-L<DBIx::Class::Schema/deploy>.
+=over 4
+
+=item Arguments: $schema, $type, $version, $directory, $sqlt_args
+
+=back
+
+Returns the statements used by L</deploy> and L<DBIx::Class::Schema/deploy>.
+The database driver name is given by C<$type>, though the value from
+L</sqlt_type> is used if it is not specified.
+
+C<$directory> is used to return statements from files in a previously created
+L</create_ddl_dir> directory and is optional. The filenames are constructed
+from L<DBIx::Class::Schema/ddl_filename>, the schema name and the C<$version>.
+
+If no C<$directory> is specified then the statements are constructed on the
+fly using L<SQL::Translator> and C<$version> is ignored.
+
+See L<SQL::Translator/METHODS> for a list of values for C<$sqlt_args>.
 
 =cut
 
@@ -1045,48 +1373,56 @@ sub deployment_statements {
   $type ||= $self->sqlt_type;
   $version ||= $schema->VERSION || '1.x';
   $dir ||= './';
-  eval "use SQL::Translator";
-  if(!$@)
-  {
-    eval "use SQL::Translator::Parser::DBIx::Class;";
-    $self->throw_exception($@) if $@;
-    eval "use SQL::Translator::Producer::${type};";
-    $self->throw_exception($@) if $@;
-    my $tr = SQL::Translator->new(%$sqltargs);
-    SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
-    return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
-  }
-
   my $filename = $schema->ddl_filename($type, $dir, $version);
-  if(!-f $filename)
+  if(-f $filename)
   {
-#      $schema->create_ddl_dir([ $type ], $version, $dir, $sqltargs);
-      $self->throw_exception("No SQL::Translator, and no Schema file found, aborting deploy");
-      return;
+      my $file;
+      open($file, "<$filename") 
+        or $self->throw_exception("Can't open $filename ($!)");
+      my @rows = <$file>;
+      close($file);
+      return join('', @rows);
   }
-  my $file;
-  open($file, "<$filename") 
-      or $self->throw_exception("Can't open $filename ($!)");
-  my @rows = <$file>;
-  close($file);
 
-  return join('', @rows);
-  
+  $self->throw_exception(q{Can't deploy without SQL::Translator 0.08: '}
+      . $self->_check_sqlt_message . q{'})
+          if !$self->_check_sqlt_version;
+
+  require SQL::Translator::Parser::DBIx::Class;
+  eval qq{use SQL::Translator::Producer::${type}};
+  $self->throw_exception($@) if $@;
+
+  # sources needs to be a parser arg, but for simplicty allow at top level 
+  # coming in
+  $sqltargs->{parser_args}{sources} = delete $sqltargs->{sources}
+      if exists $sqltargs->{sources};
+
+  my $tr = SQL::Translator->new(%$sqltargs);
+  SQL::Translator::Parser::DBIx::Class::parse( $tr, $schema );
+  return "SQL::Translator::Producer::${type}"->can('produce')->($tr);
+
+  return;
+
 }
 
 sub deploy {
   my ($self, $schema, $type, $sqltargs, $dir) = @_;
   foreach my $statement ( $self->deployment_statements($schema, $type, undef, $dir, { no_comments => 1, %{ $sqltargs || {} } } ) ) {
-    for ( split(";\n", $statement)) {
-      next if($_ =~ /^--/);
-      next if(!$_);
-#      next if($_ =~ /^DROP/m);
-      next if($_ =~ /^BEGIN TRANSACTION/m);
-      next if($_ =~ /^COMMIT/m);
-      next if $_ =~ /^\s+$/; # skip whitespace only
-      $self->debugobj->query_start($_) if $self->debug;
-      $self->dbh->do($_) or warn "SQL was:\n $_"; # XXX exceptions?
-      $self->debugobj->query_end($_) if $self->debug;
+    foreach my $line ( split(";\n", $statement)) {
+      next if($line =~ /^--/);
+      next if(!$line);
+#      next if($line =~ /^DROP/m);
+      next if($line =~ /^BEGIN TRANSACTION/m);
+      next if($line =~ /^COMMIT/m);
+      next if $line =~ /^\s+$/; # skip whitespace only
+      $self->debugobj->query_start($line) if $self->debug;
+      eval {
+        $self->dbh->do($line); # shouldn't be using ->dbh ?
+      };
+      if ($@) {
+        warn qq{$@ (running "${line}")};
+      }
+      $self->debugobj->query_end($line) if $self->debug;
     }
   }
 }
@@ -1099,7 +1435,10 @@ Returns the datetime parser class
 
 sub datetime_parser {
   my $self = shift;
-  return $self->{datetime_parser} ||= $self->build_datetime_parser(@_);
+  return $self->{datetime_parser} ||= do {
+    $self->ensure_connected;
+    $self->build_datetime_parser(@_);
+  };
 }
 
 =head2 datetime_parser_type
@@ -1125,6 +1464,22 @@ sub build_datetime_parser {
   return $type;
 }
 
+{
+    my $_check_sqlt_version; # private
+    my $_check_sqlt_message; # private
+    sub _check_sqlt_version {
+        return $_check_sqlt_version if defined $_check_sqlt_version;
+        eval 'use SQL::Translator 0.08';
+        $_check_sqlt_message = $@ ? $@ : '';
+        $_check_sqlt_version = $@ ? 0 : 1;
+    }
+
+    sub _check_sqlt_message {
+        _check_sqlt_version if !defined $_check_sqlt_message;
+        $_check_sqlt_message;
+    }
+}
+
 sub DESTROY {
   my $self = shift;
   return if !$self->_dbh;
@@ -1180,4 +1535,3 @@ Andy Grundman <andy@hybridized.org>
 You may distribute this code under the same terms as Perl itself.
 
 =cut
-