::DBI::Replicated - don't build pool/balancer from connect_info unless necessary
[dbsrgits/DBIx-Class.git] / lib / DBIx / Class / Storage / DBI.pm
index b31ac27..e4f5c49 100644 (file)
@@ -38,10 +38,10 @@ package # Hide from PAUSE
 
 use base qw/SQL::Abstract::Limit/;
 
-# This prevents the caching of $dbh in S::A::L, I believe
 sub new {
   my $self = shift->SUPER::new(@_);
 
+  # This prevents the caching of $dbh in S::A::L, I believe
   # If limit_dialect is a ref (like a $dbh), go ahead and replace
   #   it with what it resolves to:
   $self->{limit_dialect} = $self->_find_syntax($self->{limit_dialect})
@@ -50,6 +50,61 @@ sub new {
   $self;
 }
 
+
+
+# Some databases (sqlite) do not handle multiple parenthesis
+# around in/between arguments. A tentative x IN ( ( 1, 2 ,3) )
+# is interpreted as x IN 1 or something similar.
+#
+# Since we currently do not have access to the SQLA AST, resort
+# to barbaric mutilation of any SQL supplied in literal form
+
+sub _strip_outer_paren {
+  my ($self, $arg) = @_;
+
+  return $self->_SWITCH_refkind ($arg, {
+    ARRAYREFREF => sub {
+      $$arg->[0] = __strip_outer_paren ($$arg->[0]);
+      return $arg;
+    },
+    SCALARREF => sub {
+      return \__strip_outer_paren( $$arg );
+    },
+    FALLBACK => sub {
+      return $arg
+    },
+  });
+}
+
+sub __strip_outer_paren {
+  my $sql = shift;
+
+  if ($sql and not ref $sql) {
+    while ($sql =~ /^ \s* \( (.*) \) \s* $/x ) {
+      $sql = $1;
+    }
+  }
+
+  return $sql;
+}
+
+sub _where_field_IN {
+  my ($self, $lhs, $op, $rhs) = @_;
+  $rhs = $self->_strip_outer_paren ($rhs);
+  return $self->SUPER::_where_field_IN ($lhs, $op, $rhs);
+}
+
+sub _where_field_BETWEEN {
+  my ($self, $lhs, $op, $rhs) = @_;
+  $rhs = $self->_strip_outer_paren ($rhs);
+  return $self->SUPER::_where_field_BETWEEN ($lhs, $op, $rhs);
+}
+
+
+
+# DB2 is the only remaining DB using this. Even though we are not sure if
+# RowNumberOver is still needed here (should be part of SQLA) leave the 
+# code in place
 sub _RowNumberOver {
   my ($self, $sql, $order, $rows, $offset ) = @_;
 
@@ -57,7 +112,7 @@ sub _RowNumberOver {
   my $last = $rows + $offset;
   my ( $order_by ) = $self->_order_by( $order );
 
-  $sql = <<"";
+  $sql = <<"SQL";
 SELECT * FROM
 (
    SELECT Q1.*, ROW_NUMBER() OVER( ) AS ROW_NUM FROM (
@@ -67,6 +122,8 @@ SELECT * FROM
 ) Q2
 WHERE ROW_NUM BETWEEN $offset AND $last
 
+SQL
+
   return $sql;
 }
 
@@ -76,17 +133,26 @@ WHERE ROW_NUM BETWEEN $offset AND $last
 use Scalar::Util 'blessed';
 sub _find_syntax {
   my ($self, $syntax) = @_;
-  my $dbhname = blessed($syntax) ?  $syntax->{Driver}{Name} : $syntax;
+  
+  # DB2 is the only remaining DB using this. Even though we are not sure if
+  # RowNumberOver is still needed here (should be part of SQLA) leave the 
+  # code in place
+  my $dbhname = blessed($syntax) ? $syntax->{Driver}{Name} : $syntax;
   if(ref($self) && $dbhname && $dbhname eq 'DB2') {
     return 'RowNumberOver';
   }
-
+  
   $self->{_cached_syntax} ||= $self->SUPER::_find_syntax($syntax);
 }
 
 sub select {
   my ($self, $table, $fields, $where, $order, @rest) = @_;
-  $table = $self->_quote($table) unless ref($table);
+  if (ref $table eq 'SCALAR') {
+    $table = $$table;
+  }
+  elsif (not ref $table) {
+    $table = $self->_quote($table);
+  }
   local $self->{rownum_hack_count} = 1
     if (defined $rest[0] && $self->{limit_dialect} eq 'RowNum');
   @rest = (-1) unless defined $rest[0];
@@ -157,6 +223,13 @@ sub _recurse_fields {
         .'( '.$self->_recurse_fields($fields->{$func}).' )';
     }
   }
+  # Is the second check absolutely necessary?
+  elsif ( $ref eq 'REF' and ref($$fields) eq 'ARRAY' ) {
+    return $self->_bind_to_sql( $fields );
+  }
+  else {
+    Carp::croak($ref . qq{ unexpected in _recurse_fields()})
+  }
 }
 
 sub _order_by {
@@ -177,6 +250,9 @@ sub _order_by {
     if (defined $_[0]->{order_by}) {
       $ret .= $self->_order_by($_[0]->{order_by});
     }
+    if (grep { $_ =~ /^-(desc|asc)/i } keys %{$_[0]}) {
+      return $self->SUPER::_order_by($_[0]);
+    }
   } elsif (ref $_[0] eq 'SCALAR') {
     $ret = $self->_sqlcase(' order by ').${ $_[0] };
   } elsif (ref $_[0] eq 'ARRAY' && @{$_[0]}) {
@@ -237,15 +313,25 @@ sub _recurse_from {
     } else {
       push(@sqlf, $self->_make_as($to));
     }
-    push(@sqlf, ' ON (', $self->_join_condition($on), ')');
+    push(@sqlf, ' ON ', $self->_join_condition($on));
   }
   return join('', @sqlf);
 }
 
+sub _bind_to_sql {
+  my $self = shift;
+  my $arr  = shift;
+  my $sql = shift @$$arr;
+  $sql =~ s/\?/$self->_quote((shift @$$arr)->[1])/eg;
+  return $sql
+}
+
 sub _make_as {
   my ($self, $from) = @_;
-  return join(' ', map { (ref $_ eq 'SCALAR' ? $$_ : $self->_quote($_)) }
-                     reverse each %{$self->_skip_options($from)});
+  return join(' ', map { (ref $_ eq 'SCALAR' ? $$_ 
+                        : ref $_ eq 'REF'    ? $self->_bind_to_sql($_) 
+                        : $self->_quote($_)) 
+                       } reverse each %{$self->_skip_options($from)});
 }
 
 sub _skip_options {
@@ -875,7 +961,7 @@ sub dbh {
 sub _sql_maker_args {
     my ($self) = @_;
     
-    return ( bindtype=>'columns', limit_dialect => $self->dbh, %{$self->_sql_maker_opts} );
+    return ( bindtype=>'columns', array_datatypes => 1, limit_dialect => $self->dbh, %{$self->_sql_maker_opts} );
 }
 
 sub sql_maker {
@@ -906,11 +992,11 @@ sub _populate_dbh {
     }
   }
 
-  my $connection_do = $self->on_connect_do;
-  $self->_do_connection_actions($connection_do) if ref($connection_do);
-
   $self->_conn_pid($$);
   $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
+
+  my $connection_do = $self->on_connect_do;
+  $self->_do_connection_actions($connection_do) if ref($connection_do);
 }
 
 sub _do_connection_actions {
@@ -921,7 +1007,7 @@ sub _do_connection_actions {
     $self->_do_query($_) foreach @$connection_do;
   }
   elsif (ref $connection_do eq 'CODE') {
-    $connection_do->();
+    $connection_do->($self);
   }
 
   return $self;
@@ -935,10 +1021,18 @@ sub _do_query {
     $self->_do_query($_) foreach @$action;
   }
   else {
-    my @to_run = (ref $action eq 'ARRAY') ? (@$action) : ($action);
-    $self->_query_start(@to_run);
-    $self->_dbh->do(@to_run);
-    $self->_query_end(@to_run);
+    # Most debuggers expect ($sql, @bind), so we need to exclude
+    # the attribute hash which is the second argument to $dbh->do
+    # furthermore the bind values are usually to be presented
+    # as named arrayref pairs, so wrap those here too
+    my @do_args = (ref $action eq 'ARRAY') ? (@$action) : ($action);
+    my $sql = shift @do_args;
+    my $attrs = shift @do_args;
+    my @bind = map { [ undef, $_ ] } @do_args;
+
+    $self->_query_start($sql, @bind);
+    $self->_dbh->do($sql, $attrs, @do_args);
+    $self->_query_end($sql, @bind);
   }
 
   return $self;
@@ -1148,11 +1242,15 @@ sub txn_rollback {
 sub _prep_for_execute {
   my ($self, $op, $extra_bind, $ident, $args) = @_;
 
+  if( blessed($ident) && $ident->isa("DBIx::Class::ResultSource") ) {
+    $ident = $ident->from();
+  }
+
   my ($sql, @bind) = $self->sql_maker->$op($ident, @$args);
+
   unshift(@bind,
     map { ref $_ eq 'ARRAY' ? $_ : [ '!!dummy', $_ ] } @$extra_bind)
       if $extra_bind;
-
   return ($sql, \@bind);
 }
 
@@ -1193,10 +1291,6 @@ sub _query_end {
 
 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);
 
@@ -1216,7 +1310,8 @@ sub _dbh_execute {
     }
 
     foreach my $data (@data) {
-      $data = ref $data ? ''.$data : $data; # stringify args
+      my $ref = ref $data;
+      $data = $ref && $ref ne 'ARRAY' ? ''.$data : $data; # stringify args (except arrayrefs)
 
       $sth->bind_param($placeholder_index, $data, $attributes);
       $placeholder_index++;
@@ -1243,20 +1338,22 @@ sub insert {
   my $ident = $source->from; 
   my $bind_attributes = $self->source_bind_attributes($source);
 
+  my $updated_cols = {};
+
   $self->ensure_connected;
   foreach my $col ( $source->columns ) {
     if ( !defined $to_insert->{$col} ) {
       my $col_info = $source->column_info($col);
 
       if ( $col_info->{auto_nextval} ) {
-        $to_insert->{$col} = $self->_sequence_fetch( 'nextval', $col_info->{sequence} || $self->_dbh_get_autoinc_seq($self->dbh, $source) );
+        $updated_cols->{$col} = $to_insert->{$col} = $self->_sequence_fetch( 'nextval', $col_info->{sequence} || $self->_dbh_get_autoinc_seq($self->dbh, $source) );
       }
     }
   }
 
   $self->_execute('insert' => [], $source, $bind_attributes, $to_insert);
 
-  return $to_insert;
+  return $updated_cols;
 }
 
 ## Still not quite perfect, and EXPERIMENTAL
@@ -1276,13 +1373,7 @@ sub insert_bulk {
 #  @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);
@@ -1331,20 +1422,19 @@ sub delete {
 }
 
 sub _select {
+  my $self = shift;
+  my $sql_maker = $self->sql_maker;
+  local $sql_maker->{for};
+  return $self->_execute($self->_select_args(@_));
+}
+
+sub _select_args {
   my ($self, $ident, $select, $condition, $attrs) = @_;
   my $order = $attrs->{order_by};
 
-  if (ref $condition eq 'SCALAR') {
-    my $unwrap = ${$condition};
-    if ($unwrap =~ s/ORDER BY (.*)$//i) {
-      $order = $1;
-      $condition = \$unwrap;
-    }
-  }
-
   my $for = delete $attrs->{for};
   my $sql_maker = $self->sql_maker;
-  local $sql_maker->{for} = $for;
+  $sql_maker->{for} = $for;
 
   if (exists $attrs->{group_by} || $attrs->{having}) {
     $order = {
@@ -1366,8 +1456,7 @@ sub _select {
     $attrs->{rows} = 2**48 if not defined $attrs->{rows} and defined $attrs->{offset};
     push @args, $attrs->{rows}, $attrs->{offset};
   }
-
-  return $self->_execute(@args);
+  return @args;
 }
 
 sub source_bind_attributes {
@@ -1516,9 +1605,18 @@ 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');
+    # All Storage's need to register their own _dbh_last_insert_id
+    # the old SQLite-based method was highly inappropriate
+
+    my $self = shift;
+    my $class = ref $self;
+    $self->throw_exception (<<EOE);
+
+No _dbh_last_insert_id() method found in $class.
+Since the method of obtaining the autoincrement id of the last insert
+operation varies greatly between different databases, this method must be
+individually implemented for every storage class.
+EOE
 }
 
 sub last_insert_id {
@@ -1590,7 +1688,7 @@ sub create_ddl_dir {
     %{$sqltargs || {}}
   };
 
-  $self->throw_exception(q{Can't create a ddl file without SQL::Translator 0.09: '}
+  $self->throw_exception(q{Can't create a ddl file without SQL::Translator 0.09003: '}
       . $self->_check_sqlt_message . q{'})
           if !$self->_check_sqlt_version;
 
@@ -1601,7 +1699,6 @@ sub create_ddl_dir {
 
   foreach my $db (@$databases) {
     $sqlt->reset();
-    $sqlt = $self->configure_sqlt($sqlt, $db);
     $sqlt->{schema} = $sqlt_schema;
     $sqlt->producer($db);
 
@@ -1647,7 +1744,6 @@ sub create_ddl_dir {
       $t->debug( 0 );
       $t->trace( 0 );
       $t->parser( $db )                       or die $t->error;
-      $t = $self->configure_sqlt($t, $db);
       my $out = $t->translate( $prefilename ) or die $t->error;
       $source_schema = $t->schema;
       unless ( $source_schema->name ) {
@@ -1665,7 +1761,6 @@ sub create_ddl_dir {
       $t->debug( 0 );
       $t->trace( 0 );
       $t->parser( $db )                    or die $t->error;
-      $t = $self->configure_sqlt($t, $db);
       my $out = $t->translate( $filename ) or die $t->error;
       $dest_schema = $t->schema;
       $dest_schema->name( $filename )
@@ -1685,17 +1780,6 @@ sub create_ddl_dir {
   }
 }
 
-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
 
 =over 4
@@ -1726,7 +1810,7 @@ sub deployment_statements {
   $type ||= $self->sqlt_type;
   $version ||= $schema->schema_version || '1.x';
   $dir ||= './';
-  my $filename = $schema->ddl_filename($type, $dir, $version);
+  my $filename = $schema->ddl_filename($type, $version, $dir);
   if(-f $filename)
   {
       my $file;
@@ -1737,7 +1821,7 @@ sub deployment_statements {
       return join('', @rows);
   }
 
-  $self->throw_exception(q{Can't deploy without SQL::Translator 0.09: '}
+  $self->throw_exception(q{Can't deploy without SQL::Translator 0.09003: '}
       . $self->_check_sqlt_message . q{'})
           if !$self->_check_sqlt_version;
 
@@ -1757,22 +1841,32 @@ sub deployment_statements {
 
 sub deploy {
   my ($self, $schema, $type, $sqltargs, $dir) = @_;
-  foreach my $statement ( $self->deployment_statements($schema, $type, undef, $dir, { no_comments => 1, %{ $sqltargs || {} } } ) ) {
-    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->_query_start($line);
-      eval {
-        $self->dbh->do($line); # shouldn't be using ->dbh ?
-      };
-      if ($@) {
-        warn qq{$@ (running "${line}")};
-      }
-      $self->_query_end($line);
+  my $deploy = sub {
+    my $line = shift;
+    return if($line =~ /^--/);
+    return if(!$line);
+    # next if($line =~ /^DROP/m);
+    return if($line =~ /^BEGIN TRANSACTION/m);
+    return if($line =~ /^COMMIT/m);
+    return if $line =~ /^\s+$/; # skip whitespace only
+    $self->_query_start($line);
+    eval {
+      $self->dbh->do($line); # shouldn't be using ->dbh ?
+    };
+    if ($@) {
+      warn qq{$@ (running "${line}")};
+    }
+    $self->_query_end($line);
+  };
+  my @statements = $self->deployment_statements($schema, $type, undef, $dir, { no_comments => 1, %{ $sqltargs || {} } } );
+  if (@statements > 1) {
+    foreach my $statement (@statements) {
+      $deploy->( $statement );
+    }
+  }
+  elsif (@statements == 1) {
+    foreach my $line ( split(";\n", $statements[0])) {
+      $deploy->( $line );
     }
   }
 }
@@ -1819,7 +1913,7 @@ sub build_datetime_parser {
     my $_check_sqlt_message; # private
     sub _check_sqlt_version {
         return $_check_sqlt_version if defined $_check_sqlt_version;
-        eval 'use SQL::Translator "0.09"';
+        eval 'use SQL::Translator "0.09003"';
         $_check_sqlt_message = $@ || '';
         $_check_sqlt_version = !$@;
     }