Add an internal unique colset finder and relax complex $rs update/delete code

As long as there is something to correlate the subquery by we are good

diff --git a/Changes b/Changes
index a6b9cf7..016cb09 100644 (file)
--- a/Changes
+++ b/Changes
@@ -4,11 +4,14 @@ Revision history for DBIx::Class
         - Issue a warning when DateTime objects are passed to ->search
         - Fast populate() in void context is now even more efficient by
           going directly through execute_for_fetch bypassing execute_array
-
-    * Fixes
         - Fix update()/delete() on complex resultsets to no longer fall back
           to silly row-by-row deletion, construct a massive OR statement
           instead
+        - Allow complex update/delete operations on sources without a
+          primary key, as long as they have at least one non-nullable
+          unique constraint
+
+    * Fixes
         - Fix SkipFirst and FirstSkip limit dialects (Informix and Firebird)
         - A number of corner case fixes of void context populate() with \[]
         - Fix corner case of forked children disconnecting the parents DBI

diff --git a/lib/DBIx/Class/ResultSet.pm b/lib/DBIx/Class/ResultSet.pm
index c8c89c5..0061dd7 100644 (file)
--- a/lib/DBIx/Class/ResultSet.pm
+++ b/lib/DBIx/Class/ResultSet.pm
@@ -1541,10 +1541,15 @@ sub _count_subq_rs {
   # extra selectors do not go in the subquery and there is no point of ordering it, nor locking it
   delete @{$sub_attrs}{qw/collapse columns as select _prefetch_selector_range order_by for/};
 
-  # if we multi-prefetch we group_by primary keys only as this is what we would
+  # if we multi-prefetch we group_by something unique, as this is what we would
   # get out of the rs via ->next/->all. We *DO WANT* to clobber old group_by regardless
   if ( keys %{$attrs->{collapse}}  ) {
-    $sub_attrs->{group_by} = [ map { "$attrs->{alias}.$_" } ($rsrc->_pri_cols) ]
+    $sub_attrs->{group_by} = [ map { "$attrs->{alias}.$_" } @{
+      $rsrc->_identifying_column_set || $self->throw_exception(
+        'Unable to construct a unique group_by criteria properly collapsing the '
+      . 'has_many prefetch before count()'
+      );
+    } ]
   }
 
   # Calculate subquery selector
@@ -1795,20 +1800,26 @@ sub _rs_update_delete {
   }
 
   # we got this far - means it is time to wrap a subquery
-  my $pcols = [ $rsrc->_pri_cols ];
+  my $idcols = $rsrc->_identifying_column_set || $self->throw_exception(
+    sprintf(
+      "Unable to perform complex resultset %s() without an identifying set of columns on source '%s'",
+      $op,
+      $rsrc->source_name,
+    )
+  );
   my $existing_group_by = delete $attrs->{group_by};
 
   # make a new $rs selecting only the PKs (that's all we really need for the subq)
   delete $attrs->{$_} for qw/collapse _collapse_order_by select _prefetch_selector_range as/;
-  $attrs->{columns} = [ map { "$attrs->{alias}.$_" } @$pcols ];
+  $attrs->{columns} = [ map { "$attrs->{alias}.$_" } @$idcols ];
   $attrs->{group_by} = \ '';  # FIXME - this is an evil hack, it causes the optimiser to kick in and throw away the LEFT joins
   my $subrs = (ref $self)->new($rsrc, $attrs);
 
-  if (@$pcols == 1) {
+  if (@$idcols == 1) {
     return $storage->$op (
       $rsrc,
       $op eq 'update' ? $values : (),
-      { $pcols->[0] => { -in => $subrs->as_query } },
+      { $idcols->[0] => { -in => $subrs->as_query } },
     );
   }
   elsif ($storage->_use_multicolumn_in) {
@@ -1816,7 +1827,7 @@ sub _rs_update_delete {
     my $sql_maker = $storage->sql_maker;
     my ($sql, @bind) = @${$subrs->as_query};
     $sql = sprintf ('(%s) IN %s', # the as_query already comes with a set of parenthesis
-      join (', ', map { $sql_maker->_quote ($_) } @$pcols),
+      join (', ', map { $sql_maker->_quote ($_) } @$idcols),
       $sql,
     );
 
@@ -1864,8 +1875,8 @@ sub _rs_update_delete {
     my @op_condition;
     for my $row ($subrs->search({}, { group_by => $subq_group_by })->cursor->all) {
       push @op_condition, { map
-        { $pcols->[$_] => $row->[$_] }
-        (0 .. $#$pcols)
+        { $idcols->[$_] => $row->[$_] }
+        (0 .. $#$idcols)
       };
     }
 

diff --git a/lib/DBIx/Class/ResultSource.pm b/lib/DBIx/Class/ResultSource.pm
index c523745..47ecc87 100644 (file)
--- a/lib/DBIx/Class/ResultSource.pm
+++ b/lib/DBIx/Class/ResultSource.pm
@@ -1448,6 +1448,32 @@ sub _compare_relationship_keys {
   ;
 }
 
+# optionally takes either an arrayref of column names, or a hashref of already
+# retrieved colinfos
+# returns an arrayref of column names of the shortest unique constraint
+# (matching some of the input if any), giving preference to the PK
+sub _identifying_column_set {
+  my ($self, $cols) = @_;
+
+  my %unique = $self->unique_constraints;
+  my $colinfos = ref $cols eq 'HASH' ? $cols : $self->columns_info($cols||());
+
+  # always prefer the PK first, and then shortest constraints first
+  USET:
+  for my $set (delete $unique{primary}, sort { @$a <=> @$b } (values %unique) ) {
+    next unless $set && @$set;
+
+    for (@$set) {
+      next USET unless ($colinfos->{$_} && !$colinfos->{$_}{is_nullable} );
+    }
+
+    # copy so we can mangle it at will
+    return [ @$set ];
+  }
+
+  return undef;
+}
+
 # Returns the {from} structure used to express JOIN conditions
 sub _resolve_join {
   my ($self, $join, $alias, $seen, $jpath, $parent_force_left) = @_;
@@ -1723,7 +1749,6 @@ sub _resolve_condition {
 # array of column names for each of those relationships. Column names are
 # prefixed relative to the current source, in accordance with where they appear
 # in the supplied relationships.
-
 sub _resolve_prefetch {
   my ($self, $pre, $alias, $alias_map, $order, $collapse, $pref_path) = @_;
   $pref_path ||= [];

diff --git a/lib/DBIx/Class/Storage/DBIHacks.pm b/lib/DBIx/Class/Storage/DBIHacks.pm
index 8272496..276cefd 100644 (file)
--- a/lib/DBIx/Class/Storage/DBIHacks.pm
+++ b/lib/DBIx/Class/Storage/DBIHacks.pm
@@ -516,7 +516,11 @@ sub _resolve_column_info {
       },
       -result_source => $rsrc,
       -source_alias => $source_alias,
+      -fq_colname => $col eq $colname ? "$source_alias.$col" : $col,
+      -colname => $colname,
     };
+
+    $return{"$source_alias.$colname"} = $return{$col} if $col eq $colname;
   }
 
   return \%return;