Add better error reporting on bulk_insert (ash++)
[dbsrgits/DBIx-Class.git] / lib / DBIx / Class / Storage / DBI.pm
index 8df1894..0ae3564 100644 (file)
@@ -11,6 +11,7 @@ use DBIx::Class::SQLAHacks;
 use DBIx::Class::Storage::DBI::Cursor;
 use DBIx::Class::Storage::Statistics;
 use Scalar::Util qw/blessed weaken/;
+use List::Util();
 
 __PACKAGE__->mk_group_accessors('simple' =>
     qw/_connect_info _dbi_connect_info _dbh _sql_maker _sql_maker_opts
@@ -153,6 +154,10 @@ the database.  Its value may contain:
 
 =over
 
+=item a scalar
+
+This contains one SQL statement to execute.
+
 =item an array reference
 
 This contains SQL statements to execute in order.  Each element contains
@@ -610,35 +615,56 @@ sub _populate_dbh {
   my @info = @{$self->_dbi_connect_info || []};
   $self->_dbh($self->_connect(@info));
 
+  $self->_conn_pid($$);
+  $self->_conn_tid(threads->tid) if $INC{'threads.pm'};
+
+  $self->_determine_driver;
+
   # 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};
+  my $connection_do = $self->on_connect_do;
+  $self->_do_connection_actions($connection_do) if $connection_do;
+}
+
+sub _determine_driver {
+  my ($self) = @_;
+
+  if (ref $self eq 'DBIx::Class::Storage::DBI') {
+    my $driver;
+
+    if ($self->_dbh) { # we are connected
+      $driver = $self->_dbh->{Driver}{Name};
+    } else {
+      # try to use dsn to not require being connected, the driver may still
+      # force a connection in _rebless to determine version
+      ($driver) = $self->_dbi_connect_info->[0] =~ /dbi:([^:]+):/i;
+    }
+
     if ($self->load_optional_class("DBIx::Class::Storage::DBI::${driver}")) {
       bless $self, "DBIx::Class::Storage::DBI::${driver}";
       $self->_rebless();
     }
   }
-
-  $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 {
   my $self = shift;
   my $connection_do = shift;
 
-  if (ref $connection_do eq 'ARRAY') {
+  if (!ref $connection_do) {
+    $self->_do_query($connection_do);
+  }
+  elsif (ref $connection_do eq 'ARRAY') {
     $self->_do_query($_) foreach @$connection_do;
   }
   elsif (ref $connection_do eq 'CODE') {
     $connection_do->($self);
   }
+  else {
+    $self->throw_exception (sprintf ("Don't know how to process conection actions of type '%s'", ref $connection_do) );
+  }
 
   return $self;
 }
@@ -1026,7 +1052,27 @@ sub insert_bulk {
     $sth->bind_param_array( $placeholder_index, [@data], $attributes );
     $placeholder_index++;
   }
-  my $rv = $sth->execute_array({ArrayTupleStatus => $tuple_status});
+  my $rv = eval { $sth->execute_array({ArrayTupleStatus => $tuple_status}) };
+  if (my $err = $@) {
+    my $i = 0;
+    ++$i while $i <= $#$tuple_status && !ref $tuple_status->[$i];
+
+    $self->throw_exception($sth->errstr || "Unexpected populate error: $err")
+      if ($i > $#$tuple_status);
+
+    require Data::Dumper;
+    local $Data::Dumper::Terse = 1;
+    local $Data::Dumper::Indent = 1;
+    local $Data::Dumper::Useqq = 1;
+    local $Data::Dumper::Quotekeys = 0;
+
+    $self->throw_exception(sprintf "%s for populate slice:\n%s",
+      $tuple_status->[$i][1],
+      Data::Dumper::Dumper(
+        { map { $cols->[$_] => $data->[$i][$_] } (0 .. $#$cols) }
+      ),
+    );
+  }
   $self->throw_exception($sth->errstr) if !$rv;
 
   $self->_query_end( $sql, @bind );
@@ -1051,6 +1097,97 @@ sub delete {
   return $self->_execute('delete' => [], $source, $bind_attrs, @_);
 }
 
+# We were sent here because the $rs contains a complex search
+# which will require a subquery to select the correct rows
+# (i.e. joined or limited resultsets)
+#
+# Genarating a single PK column subquery is trivial and supported
+# by all RDBMS. However if we have a multicolumn PK, things get ugly.
+# Look at _multipk_update_delete()
+sub subq_update_delete {
+  my $self = shift;
+  my ($rs, $op, $values) = @_;
+
+  my $rsrc = $rs->result_source;
+
+  # we already check this, but double check naively just in case. Should be removed soon
+  my $sel = $rs->_resolved_attrs->{select};
+  $sel = [ $sel ] unless ref $sel eq 'ARRAY';
+  my @pcols = $rsrc->primary_columns;
+  if (@$sel != @pcols) {
+    $self->throw_exception (
+      'Subquery update/delete can not be called on resultsets selecting a'
+     .' number of columns different than the number of primary keys'
+    );
+  }
+
+  if (@pcols == 1) {
+    return $self->$op (
+      $rsrc,
+      $op eq 'update' ? $values : (),
+      { $pcols[0] => { -in => $rs->as_query } },
+    );
+  }
+
+  else {
+    return $self->_multipk_update_delete (@_);
+  }
+}
+
+# ANSI SQL does not provide a reliable way to perform a multicol-PK
+# resultset update/delete involving subqueries. So by default resort
+# to simple (and inefficient) delete_all style per-row opearations,
+# while allowing specific storages to override this with a faster
+# implementation.
+#
+sub _multipk_update_delete {
+  return shift->_per_row_update_delete (@_);
+}
+
+# This is the default loop used to delete/update rows for multi PK
+# resultsets, and used by mysql exclusively (because it can't do anything
+# else).
+#
+# We do not use $row->$op style queries, because resultset update/delete
+# is not expected to cascade (this is what delete_all/update_all is for).
+#
+# There should be no race conditions as the entire operation is rolled
+# in a transaction.
+#
+sub _per_row_update_delete {
+  my $self = shift;
+  my ($rs, $op, $values) = @_;
+
+  my $rsrc = $rs->result_source;
+  my @pcols = $rsrc->primary_columns;
+
+  my $guard = $self->txn_scope_guard;
+
+  # emulate the return value of $sth->execute for non-selects
+  my $row_cnt = '0E0';
+
+  my $subrs_cur = $rs->cursor;
+  while (my @pks = $subrs_cur->next) {
+
+    my $cond;
+    for my $i (0.. $#pcols) {
+      $cond->{$pcols[$i]} = $pks[$i];
+    }
+
+    $self->$op (
+      $rsrc,
+      $op eq 'update' ? $values : (),
+      $cond,
+    );
+
+    $row_cnt++;
+  }
+
+  $guard->commit;
+
+  return $row_cnt;
+}
+
 sub _select {
   my $self = shift;
   my $sql_maker = $self->sql_maker;
@@ -1066,17 +1203,20 @@ sub _select_args {
   my $sql_maker = $self->sql_maker;
   $sql_maker->{for} = $for;
 
-  if (exists $attrs->{group_by} || $attrs->{having}) {
+  my @in_order_attrs = qw/group_by having _virtual_order_by/;
+  if (List::Util::first { exists $attrs->{$_} } (@in_order_attrs) ) {
     $order = {
-      group_by => $attrs->{group_by},
-      having => $attrs->{having},
-      ($order ? (order_by => $order) : ())
+      ($order
+        ? (order_by => $order)
+        : ()
+      ),
+      ( map { $_ => $attrs->{$_} } (@in_order_attrs) )
     };
   }
   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") {
+      $sql_maker->_default_limit_syntax eq "GenericSubQ") {
         $attrs->{software_limit} = 1;
   } else {
     $self->throw_exception("rows attribute must be positive if present")