lib/DBIx/Class/ResultSet.pm

   1 package DBIx::Class::ResultSet;
   2
   3 use strict;
   4 use warnings;
   5 use base qw/DBIx::Class/;
   6 use Carp::Clan qw/^DBIx::Class/;
   7 use DBIx::Class::Exception;
   8 use DBIx::Class::ResultSetColumn;
   9 use Scalar::Util qw/blessed weaken/;
  10 use Try::Tiny;
  11
  12 # not importing first() as it will clash with our own method
  13 use List::Util ();
  14
  15 use namespace::clean;
  16
  17
  18 BEGIN {
  19   # De-duplication in _merge_attr() is disabled, but left in for reference
  20   # (the merger is used for other things that ought not to be de-duped)
  21   *__HM_DEDUP = sub () { 0 };
  22 }
  23
  24 use overload
  25         '0+'     => "count",
  26         'bool'   => "_bool",
  27         fallback => 1;
  28
  29 __PACKAGE__->mk_group_accessors('simple' => qw/_result_class result_source/);
  30
  31 =head1 NAME
  32
  33 DBIx::Class::ResultSet - Represents a query used for fetching a set of results.
  34
  35 =head1 SYNOPSIS
  36
  37   my $users_rs   = $schema->resultset('User');
  38   while( $user = $users_rs->next) {
  39     print $user->username;
  40   }
  41
  42   my $registered_users_rs   = $schema->resultset('User')->search({ registered => 1 });
  43   my @cds_in_2005 = $schema->resultset('CD')->search({ year => 2005 })->all();
  44
  45 =head1 DESCRIPTION
  46
  47 A ResultSet is an object which stores a set of conditions representing
  48 a query. It is the backbone of DBIx::Class (i.e. the really
  49 important/useful bit).
  50
  51 No SQL is executed on the database when a ResultSet is created, it
  52 just stores all the conditions needed to create the query.
  53
  54 A basic ResultSet representing the data of an entire table is returned
  55 by calling C<resultset> on a L<DBIx::Class::Schema> and passing in a
  56 L<Source|DBIx::Class::Manual::Glossary/Source> name.
  57
  58   my $users_rs = $schema->resultset('User');
  59
  60 A new ResultSet is returned from calling L</search> on an existing
  61 ResultSet. The new one will contain all the conditions of the
  62 original, plus any new conditions added in the C<search> call.
  63
  64 A ResultSet also incorporates an implicit iterator. L</next> and L</reset>
  65 can be used to walk through all the L<DBIx::Class::Row>s the ResultSet
  66 represents.
  67
  68 The query that the ResultSet represents is B<only> executed against
  69 the database when these methods are called:
  70 L</find>, L</next>, L</all>, L</first>, L</single>, L</count>.
  71
  72 If a resultset is used in a numeric context it returns the L</count>.
  73 However, if it is used in a boolean context it is B<always> true.  So if
  74 you want to check if a resultset has any results, you must use C<if $rs
  75 != 0>.
  76
  77 =head1 EXAMPLES
  78
  79 =head2 Chaining resultsets
  80
  81 Let's say you've got a query that needs to be run to return some data
  82 to the user. But, you have an authorization system in place that
  83 prevents certain users from seeing certain information. So, you want
  84 to construct the basic query in one method, but add constraints to it in
  85 another.
  86
  87   sub get_data {
  88     my $self = shift;
  89     my $request = $self->get_request; # Get a request object somehow.
  90     my $schema = $self->get_schema;   # Get the DBIC schema object somehow.
  91
  92     my $cd_rs = $schema->resultset('CD')->search({
  93       title => $request->param('title'),
  94       year => $request->param('year'),
  95     });
  96
  97     $self->apply_security_policy( $cd_rs );
  98
  99     return $cd_rs->all();
 100   }
 101
 102   sub apply_security_policy {
 103     my $self = shift;
 104     my ($rs) = @_;
 105
 106     return $rs->search({
 107       subversive => 0,
 108     });
 109   }
 110
 111 =head3 Resolving conditions and attributes
 112
 113 When a resultset is chained from another resultset, conditions and
 114 attributes with the same keys need resolving.
 115
 116 L</join>, L</prefetch>, L</+select>, L</+as> attributes are merged
 117 into the existing ones from the original resultset.
 118
 119 The L</where> and L</having> attributes, and any search conditions, are
 120 merged with an SQL C<AND> to the existing condition from the original
 121 resultset.
 122
 123 All other attributes are overridden by any new ones supplied in the
 124 search attributes.
 125
 126 =head2 Multiple queries
 127
 128 Since a resultset just defines a query, you can do all sorts of
 129 things with it with the same object.
 130
 131   # Don't hit the DB yet.
 132   my $cd_rs = $schema->resultset('CD')->search({
 133     title => 'something',
 134     year => 2009,
 135   });
 136
 137   # Each of these hits the DB individually.
 138   my $count = $cd_rs->count;
 139   my $most_recent = $cd_rs->get_column('date_released')->max();
 140   my @records = $cd_rs->all;
 141
 142 And it's not just limited to SELECT statements.
 143
 144   $cd_rs->delete();
 145
 146 This is even cooler:
 147
 148   $cd_rs->create({ artist => 'Fred' });
 149
 150 Which is the same as:
 151
 152   $schema->resultset('CD')->create({
 153     title => 'something',
 154     year => 2009,
 155     artist => 'Fred'
 156   });
 157
 158 See: L</search>, L</count>, L</get_column>, L</all>, L</create>.
 159
 160 =head1 METHODS
 161
 162 =head2 new
 163
 164 =over 4
 165
 166 =item Arguments: $source, \%$attrs
 167
 168 =item Return Value: $rs
 169
 170 =back
 171
 172 The resultset constructor. Takes a source object (usually a
 173 L<DBIx::Class::ResultSourceProxy::Table>) and an attribute hash (see
 174 L</ATTRIBUTES> below).  Does not perform any queries -- these are
 175 executed as needed by the other methods.
 176
 177 Generally you won't need to construct a resultset manually.  You'll
 178 automatically get one from e.g. a L</search> called in scalar context:
 179
 180   my $rs = $schema->resultset('CD')->search({ title => '100th Window' });
 181
 182 IMPORTANT: If called on an object, proxies to new_result instead so
 183
 184   my $cd = $schema->resultset('CD')->new({ title => 'Spoon' });
 185
 186 will return a CD object, not a ResultSet.
 187
 188 =cut
 189
 190 sub new {
 191   my $class = shift;
 192   return $class->new_result(@_) if ref $class;
 193
 194   my ($source, $attrs) = @_;
 195   $source = $source->resolve
 196     if $source->isa('DBIx::Class::ResultSourceHandle');
 197   $attrs = { %{$attrs||{}} };
 198
 199   if ($attrs->{page}) {
 200     $attrs->{rows} ||= 10;
 201   }
 202
 203   $attrs->{alias} ||= 'me';
 204
 205   my $self = bless {
 206     result_source => $source,
 207     cond => $attrs->{where},
 208     pager => undef,
 209     attrs => $attrs,
 210   }, $class;
 211
 212   $self->result_class(
 213     $attrs->{result_class} || $source->result_class
 214   );
 215
 216   $self;
 217 }
 218
 219 =head2 search
 220
 221 =over 4
 222
 223 =item Arguments: $cond, \%attrs?
 224
 225 =item Return Value: $resultset (scalar context), @row_objs (list context)
 226
 227 =back
 228
 229   my @cds    = $cd_rs->search({ year => 2001 }); # "... WHERE year = 2001"
 230   my $new_rs = $cd_rs->search({ year => 2005 });
 231
 232   my $new_rs = $cd_rs->search([ { year => 2005 }, { year => 2004 } ]);
 233                  # year = 2005 OR year = 2004
 234
 235 If you need to pass in additional attributes but no additional condition,
 236 call it as C<search(undef, \%attrs)>.
 237
 238   # "SELECT name, artistid FROM $artist_table"
 239   my @all_artists = $schema->resultset('Artist')->search(undef, {
 240     columns => [qw/name artistid/],
 241   });
 242
 243 For a list of attributes that can be passed to C<search>, see
 244 L</ATTRIBUTES>. For more examples of using this function, see
 245 L<Searching|DBIx::Class::Manual::Cookbook/Searching>. For a complete
 246 documentation for the first argument, see L<SQL::Abstract>.
 247
 248 For more help on using joins with search, see L<DBIx::Class::Manual::Joining>.
 249
 250 =head3 CAVEAT
 251
 252 Note that L</search> does not process/deflate any of the values passed in the
 253 L<SQL::Abstract>-compatible search condition structure. This is unlike other
 254 condition-bound methods L</new>, L</create> and L</find>. The user must ensure
 255 manually that any value passed to this method will stringify to something the
 256 RDBMS knows how to deal with. A notable example is the handling of L<DateTime>
 257 objects, for more info see:
 258 L<DBIx::Class::Manual::Cookbook/Formatting_DateTime_objects_in_queries>.
 259
 260 =cut
 261
 262 sub search {
 263   my $self = shift;
 264   my $rs = $self->search_rs( @_ );
 265
 266   if (wantarray) {
 267     return $rs->all;
 268   }
 269   elsif (defined wantarray) {
 270     return $rs;
 271   }
 272   else {
 273     # we can be called by a relationship helper, which in
 274     # turn may be called in void context due to some braindead
 275     # overload or whatever else the user decided to be clever
 276     # at this particular day. Thus limit the exception to
 277     # external code calls only
 278     $self->throw_exception ('->search is *not* a mutator, calling it in void context makes no sense')
 279       if (caller)[0] !~ /^\QDBIx::Class::/;
 280
 281     return ();
 282   }
 283 }
 284
 285 =head2 search_rs
 286
 287 =over 4
 288
 289 =item Arguments: $cond, \%attrs?
 290
 291 =item Return Value: $resultset
 292
 293 =back
 294
 295 This method does the same exact thing as search() except it will
 296 always return a resultset, even in list context.
 297
 298 =cut
 299
 300 my $callsites_warned;
 301 sub search_rs {
 302   my $self = shift;
 303
 304   # Special-case handling for (undef, undef).
 305   if ( @_ == 2 && !defined $_[1] && !defined $_[0] ) {
 306     @_ = ();
 307   }
 308
 309   my $call_attrs = {};
 310   if (@_ > 1) {
 311     if (ref $_[-1] eq 'HASH') {
 312       # copy for _normalize_selection
 313       $call_attrs = { %{ pop @_ } };
 314     }
 315     elsif (! defined $_[-1] ) {
 316       pop @_;   # search({}, undef)
 317     }
 318   }
 319
 320   # see if we can keep the cache (no $rs changes)
 321   my $cache;
 322   my %safe = (alias => 1, cache => 1);
 323   if ( ! List::Util::first { !$safe{$_} } keys %$call_attrs and (
 324     ! defined $_[0]
 325       or
 326     ref $_[0] eq 'HASH' && ! keys %{$_[0]}
 327       or
 328     ref $_[0] eq 'ARRAY' && ! @{$_[0]}
 329   )) {
 330     $cache = $self->get_cache;
 331   }
 332
 333   my $rsrc = $self->result_source;
 334
 335   my $old_attrs = { %{$self->{attrs}} };
 336   my $old_having = delete $old_attrs->{having};
 337   my $old_where = delete $old_attrs->{where};
 338
 339   my $new_attrs = { %$old_attrs };
 340
 341   # take care of call attrs (only if anything is changing)
 342   if (keys %$call_attrs) {
 343
 344     $self->throw_exception ('_trailing_select is not a public attribute - do not use it in search()')
 345       if ( exists $call_attrs->{_trailing_select} or exists $call_attrs->{'+_trailing_select'} );
 346
 347     my @selector_attrs = qw/select as columns cols +select +as +columns include_columns _trailing_select +_trailing_select/;
 348
 349     # Normalize the selector list (operates on the passed-in attr structure)
 350     # Need to do it on every chain instead of only once on _resolved_attrs, in
 351     # order to separate 'as'-ed from blind 'select's
 352     $self->_normalize_selection ($call_attrs);
 353
 354     # start with blind overwriting merge, exclude selector attrs
 355     $new_attrs = { %{$old_attrs}, %{$call_attrs} };
 356     delete @{$new_attrs}{@selector_attrs};
 357
 358     # reset the current selector list if new selectors are supplied
 359     if (List::Util::first { exists $call_attrs->{$_} } qw/columns cols select as/) {
 360       delete @{$old_attrs}{@selector_attrs};
 361     }
 362
 363     for (@selector_attrs) {
 364       $new_attrs->{$_} = $self->_merge_attr($old_attrs->{$_}, $call_attrs->{$_})
 365         if ( exists $old_attrs->{$_} or exists $call_attrs->{$_} );
 366     }
 367
 368     # older deprecated name, use only if {columns} is not there
 369     if (my $c = delete $new_attrs->{cols}) {
 370       if ($new_attrs->{columns}) {
 371         carp "Resultset specifies both the 'columns' and the legacy 'cols' attributes - ignoring 'cols'";
 372       }
 373       else {
 374         $new_attrs->{columns} = $c;
 375       }
 376     }
 377
 378
 379     # join/prefetch use their own crazy merging heuristics
 380     foreach my $key (qw/join prefetch/) {
 381       $new_attrs->{$key} = $self->_merge_joinpref_attr($old_attrs->{$key}, $call_attrs->{$key})
 382         if exists $call_attrs->{$key};
 383     }
 384
 385     # stack binds together
 386     $new_attrs->{bind} = [ @{ $old_attrs->{bind} || [] }, @{ $call_attrs->{bind} || [] } ];
 387   }
 388
 389
 390   # rip apart the rest of @_, parse a condition
 391   my $call_cond = do {
 392
 393     if (ref $_[0] eq 'HASH') {
 394       (keys %{$_[0]}) ? $_[0] : undef
 395     }
 396     elsif (@_ == 1) {
 397       $_[0]
 398     }
 399     elsif (@_ % 2) {
 400       $self->throw_exception('Odd number of arguments to search')
 401     }
 402     else {
 403       +{ @_ }
 404     }
 405
 406   } if @_;
 407
 408   if( @_ > 1 and ! $rsrc->result_class->isa('DBIx::Class::CDBICompat') ) {
 409     # determine callsite obeying Carp::Clan rules (fucking ugly but don't have better ideas)
 410     my $callsite = do {
 411       my $w;
 412       local $SIG{__WARN__} = sub { $w = shift };
 413       carp;
 414       $w
 415     };
 416     carp 'search( %condition ) is deprecated, use search( \%condition ) instead'
 417       unless $callsites_warned->{$callsite}++;
 418   }
 419
 420   for ($old_where, $call_cond) {
 421     if (defined $_) {
 422       $new_attrs->{where} = $self->_stack_cond (
 423         $_, $new_attrs->{where}
 424       );
 425     }
 426   }
 427
 428   if (defined $old_having) {
 429     $new_attrs->{having} = $self->_stack_cond (
 430       $old_having, $new_attrs->{having}
 431     )
 432   }
 433
 434   my $rs = (ref $self)->new($rsrc, $new_attrs);
 435
 436   $rs->set_cache($cache) if ($cache);
 437
 438   return $rs;
 439 }
 440
 441 sub _normalize_selection {
 442   my ($self, $attrs) = @_;
 443
 444   # legacy syntax
 445   $attrs->{'+columns'} = $self->_merge_attr($attrs->{'+columns'}, delete $attrs->{include_columns})
 446     if exists $attrs->{include_columns};
 447
 448   # Keep the X vs +X separation until _resolved_attrs time - this allows to
 449   # delay the decision on whether to use a default select list ($rsrc->columns)
 450   # allowing stuff like the remove_columns helper to work
 451   #
 452   # select/as +select/+as pairs need special handling - the amount of select/as
 453   # elements in each pair does *not* have to be equal (think multicolumn
 454   # selectors like distinct(foo, bar) ). If the selector is bare (no 'as'
 455   # supplied at all) - try to infer the alias, either from the -as parameter
 456   # of the selector spec, or use the parameter whole if it looks like a column
 457   # name (ugly legacy heuristic). If all fails - leave the selector bare (which
 458   # is ok as well), but transport it over a separate attribute to make sure it is
 459   # the last thing in the select list, thus unable to throw off the corresponding
 460   # 'as' chain
 461   for my $pref ('', '+') {
 462
 463     my ($sel, $as) = map {
 464       my $key = "${pref}${_}";
 465
 466       my $val = [ ref $attrs->{$key} eq 'ARRAY'
 467         ? @{$attrs->{$key}}
 468         : $attrs->{$key} || ()
 469       ];
 470       delete $attrs->{$key};
 471       $val;
 472     } qw/select as/;
 473
 474     if (! @$as and ! @$sel ) {
 475       next;
 476     }
 477     elsif (@$as and ! @$sel) {
 478       $self->throw_exception(
 479         "Unable to handle ${pref}as specification (@$as) without a corresponding ${pref}select"
 480       );
 481     }
 482     elsif( ! @$as ) {
 483       # no as part supplied at all - try to deduce
 484       # if any @$as has been supplied we assume the user knows what (s)he is doing
 485       # and blindly keep stacking up pieces
 486       my (@new_sel, @new_trailing);
 487       for (@$sel) {
 488         if ( ref $_ eq 'HASH' and exists $_->{-as} ) {
 489           push @$as, $_->{-as};
 490           push @new_sel, $_;
 491         }
 492         # assume any plain no-space, no-parenthesis string to be a column spec
 493         # FIXME - this is retarded but is necessary to support shit like 'count(foo)'
 494         elsif ( ! ref $_ and $_ =~ /^ [^\s\(\)]+ $/x) {
 495           push @$as, $_;
 496           push @new_sel, $_;
 497         }
 498         # if all else fails - shove the selection to the trailing stack and move on
 499         else {
 500           push @new_trailing, $_;
 501         }
 502       }
 503
 504       @$sel = @new_sel;
 505       $attrs->{"${pref}_trailing_select"} = $self->_merge_attr($attrs->{"${pref}_trailing_select"}, \@new_trailing)
 506         if @new_trailing;
 507     }
 508     elsif (@$as < @$sel) {
 509       $self->throw_exception(
 510         "Unable to handle an ${pref}as specification (@$as) with less elements than the corresponding ${pref}select"
 511       );
 512     }
 513
 514     # now see what the result for this pair looks like:
 515     if (@$as == @$sel) {
 516
 517       # if balanced - treat as a columns entry
 518       $attrs->{"${pref}columns"} = $self->_merge_attr(
 519         $attrs->{"${pref}columns"},
 520         [ map { +{ $as->[$_] => $sel->[$_] } } ( 0 .. $#$as ) ]
 521       );
 522     }
 523     else {
 524       # unbalanced - shove in select/as, not subject to deduplication in _resolved_attrs
 525       $attrs->{"${pref}select"} = $self->_merge_attr($attrs->{"${pref}select"}, $sel);
 526       $attrs->{"${pref}as"} = $self->_merge_attr($attrs->{"${pref}as"}, $as);
 527     }
 528   }
 529
 530 }
 531
 532 sub _stack_cond {
 533   my ($self, $left, $right) = @_;
 534   if (defined $left xor defined $right) {
 535     return defined $left ? $left : $right;
 536   }
 537   elsif (defined $left) {
 538     return { -and => [ map
 539       { ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_ }
 540       ($left, $right)
 541     ]};
 542   }
 543
 544   return undef;
 545 }
 546
 547 =head2 search_literal
 548
 549 =over 4
 550
 551 =item Arguments: $sql_fragment, @bind_values
 552
 553 =item Return Value: $resultset (scalar context), @row_objs (list context)
 554
 555 =back
 556
 557   my @cds   = $cd_rs->search_literal('year = ? AND title = ?', qw/2001 Reload/);
 558   my $newrs = $artist_rs->search_literal('name = ?', 'Metallica');
 559
 560 Pass a literal chunk of SQL to be added to the conditional part of the
 561 resultset query.
 562
 563 CAVEAT: C<search_literal> is provided for Class::DBI compatibility and should
 564 only be used in that context. C<search_literal> is a convenience method.
 565 It is equivalent to calling $schema->search(\[]), but if you want to ensure
 566 columns are bound correctly, use C<search>.
 567
 568 Example of how to use C<search> instead of C<search_literal>
 569
 570   my @cds = $cd_rs->search_literal('cdid = ? AND (artist = ? OR artist = ?)', (2, 1, 2));
 571   my @cds = $cd_rs->search(\[ 'cdid = ? AND (artist = ? OR artist = ?)', [ 'cdid', 2 ], [ 'artist', 1 ], [ 'artist', 2 ] ]);
 572
 573
 574 See L<DBIx::Class::Manual::Cookbook/Searching> and
 575 L<DBIx::Class::Manual::FAQ/Searching> for searching techniques that do not
 576 require C<search_literal>.
 577
 578 =cut
 579
 580 sub search_literal {
 581   my ($self, $sql, @bind) = @_;
 582   my $attr;
 583   if ( @bind && ref($bind[-1]) eq 'HASH' ) {
 584     $attr = pop @bind;
 585   }
 586   return $self->search(\[ $sql, map [ __DUMMY__ => $_ ], @bind ], ($attr || () ));
 587 }
 588
 589 =head2 find
 590
 591 =over 4
 592
 593 =item Arguments: \%columns_values | @pk_values, \%attrs?
 594
 595 =item Return Value: $row_object | undef
 596
 597 =back
 598
 599 Finds and returns a single row based on supplied criteria. Takes either a
 600 hashref with the same format as L</create> (including inference of foreign
 601 keys from related objects), or a list of primary key values in the same
 602 order as the L<primary columns|DBIx::Class::ResultSource/primary_columns>
 603 declaration on the L</result_source>.
 604
 605 In either case an attempt is made to combine conditions already existing on
 606 the resultset with the condition passed to this method.
 607
 608 To aid with preparing the correct query for the storage you may supply the
 609 C<key> attribute, which is the name of a
 610 L<unique constraint|DBIx::Class::ResultSource/add_unique_constraint> (the
 611 unique constraint corresponding to the
 612 L<primary columns|DBIx::Class::ResultSource/primary_columns> is always named
 613 C<primary>). If the C<key> attribute has been supplied, and DBIC is unable
 614 to construct a query that satisfies the named unique constraint fully (
 615 non-NULL values for each column member of the constraint) an exception is
 616 thrown.
 617
 618 If no C<key> is specified, the search is carried over all unique constraints
 619 which are fully defined by the available condition.
 620
 621 If no such constraint is found, C<find> currently defaults to a simple
 622 C<< search->(\%column_values) >> which may or may not do what you expect.
 623 Note that this fallback behavior may be deprecated in further versions. If
 624 you need to search with arbitrary conditions - use L</search>. If the query
 625 resulting from this fallback produces more than one row, a warning to the
 626 effect is issued, though only the first row is constructed and returned as
 627 C<$row_object>.
 628
 629 In addition to C<key>, L</find> recognizes and applies standard
 630 L<resultset attributes|/ATTRIBUTES> in the same way as L</search> does.
 631
 632 Note that if you have extra concerns about the correctness of the resulting
 633 query you need to specify the C<key> attribute and supply the entire condition
 634 as an argument to find (since it is not always possible to perform the
 635 combination of the resultset condition with the supplied one, especially if
 636 the resultset condition contains literal sql).
 637
 638 For example, to find a row by its primary key:
 639
 640   my $cd = $schema->resultset('CD')->find(5);
 641
 642 You can also find a row by a specific unique constraint:
 643
 644   my $cd = $schema->resultset('CD')->find(
 645     {
 646       artist => 'Massive Attack',
 647       title  => 'Mezzanine',
 648     },
 649     { key => 'cd_artist_title' }
 650   );
 651
 652 See also L</find_or_create> and L</update_or_create>.
 653
 654 =cut
 655
 656 sub find {
 657   my $self = shift;
 658   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
 659
 660   my $rsrc = $self->result_source;
 661
 662   # Parse out the condition from input
 663   my $call_cond;
 664   if (ref $_[0] eq 'HASH') {
 665     $call_cond = { %{$_[0]} };
 666   }
 667   else {
 668     my $constraint = exists $attrs->{key} ? $attrs->{key} : 'primary';
 669     my @c_cols = $rsrc->unique_constraint_columns($constraint);
 670
 671     $self->throw_exception(
 672       "No constraint columns, maybe a malformed '$constraint' constraint?"
 673     ) unless @c_cols;
 674
 675     $self->throw_exception (
 676       'find() expects either a column/value hashref, or a list of values '
 677     . "corresponding to the columns of the specified unique constraint '$constraint'"
 678     ) unless @c_cols == @_;
 679
 680     $call_cond = {};
 681     @{$call_cond}{@c_cols} = @_;
 682   }
 683
 684   my %related;
 685   for my $key (keys %$call_cond) {
 686     if (
 687       my $keyref = ref($call_cond->{$key})
 688         and
 689       my $relinfo = $rsrc->relationship_info($key)
 690     ) {
 691       my $val = delete $call_cond->{$key};
 692
 693       next if $keyref eq 'ARRAY'; # has_many for multi_create
 694
 695       my $rel_q = $rsrc->_resolve_condition(
 696         $relinfo->{cond}, $val, $key
 697       );
 698       die "Can't handle complex relationship conditions in find" if ref($rel_q) ne 'HASH';
 699       @related{keys %$rel_q} = values %$rel_q;
 700     }
 701   }
 702
 703   # relationship conditions take precedence (?)
 704   @{$call_cond}{keys %related} = values %related;
 705
 706   my $alias = exists $attrs->{alias} ? $attrs->{alias} : $self->{attrs}{alias};
 707   my $final_cond;
 708   if (exists $attrs->{key}) {
 709     $final_cond = $self->_qualify_cond_columns (
 710
 711       $self->_build_unique_cond (
 712         $attrs->{key},
 713         $call_cond,
 714       ),
 715
 716       $alias,
 717     );
 718   }
 719   elsif ($self->{attrs}{accessor} and $self->{attrs}{accessor} eq 'single') {
 720     # This means that we got here after a merger of relationship conditions
 721     # in ::Relationship::Base::search_related (the row method), and furthermore
 722     # the relationship is of the 'single' type. This means that the condition
 723     # provided by the relationship (already attached to $self) is sufficient,
 724     # as there can be only one row in the database that would satisfy the
 725     # relationship
 726   }
 727   else {
 728     # no key was specified - fall down to heuristics mode:
 729     # run through all unique queries registered on the resultset, and
 730     # 'OR' all qualifying queries together
 731     my (@unique_queries, %seen_column_combinations);
 732     for my $c_name ($rsrc->unique_constraint_names) {
 733       next if $seen_column_combinations{
 734         join "\x00", sort $rsrc->unique_constraint_columns($c_name)
 735       }++;
 736
 737       push @unique_queries, try {
 738         $self->_build_unique_cond ($c_name, $call_cond, 'croak_on_nulls')
 739       } || ();
 740     }
 741
 742     $final_cond = @unique_queries
 743       ? [ map { $self->_qualify_cond_columns($_, $alias) } @unique_queries ]
 744       : $self->_non_unique_find_fallback ($call_cond, $attrs)
 745     ;
 746   }
 747
 748   # Run the query, passing the result_class since it should propagate for find
 749   my $rs = $self->search ($final_cond, {result_class => $self->result_class, %$attrs});
 750   if (keys %{$rs->_resolved_attrs->{collapse}}) {
 751     my $row = $rs->next;
 752     carp "Query returned more than one row" if $rs->next;
 753     return $row;
 754   }
 755   else {
 756     return $rs->single;
 757   }
 758 }
 759
 760 # This is a stop-gap method as agreed during the discussion on find() cleanup:
 761 # http://lists.scsys.co.uk/pipermail/dbix-class/2010-October/009535.html
 762 #
 763 # It is invoked when find() is called in legacy-mode with insufficiently-unique
 764 # condition. It is provided for overrides until a saner way forward is devised
 765 #
 766 # *NOTE* This is not a public method, and it's *GUARANTEED* to disappear down
 767 # the road. Please adjust your tests accordingly to catch this situation early
 768 # DBIx::Class::ResultSet->can('_non_unique_find_fallback') is reasonable
 769 #
 770 # The method will not be removed without an adequately complete replacement
 771 # for strict-mode enforcement
 772 sub _non_unique_find_fallback {
 773   my ($self, $cond, $attrs) = @_;
 774
 775   return $self->_qualify_cond_columns(
 776     $cond,
 777     exists $attrs->{alias}
 778       ? $attrs->{alias}
 779       : $self->{attrs}{alias}
 780   );
 781 }
 782
 783
 784 sub _qualify_cond_columns {
 785   my ($self, $cond, $alias) = @_;
 786
 787   my %aliased = %$cond;
 788   for (keys %aliased) {
 789     $aliased{"$alias.$_"} = delete $aliased{$_}
 790       if $_ !~ /\./;
 791   }
 792
 793   return \%aliased;
 794 }
 795
 796 my $callsites_warned_ucond;
 797 sub _build_unique_cond {
 798   my ($self, $constraint_name, $extra_cond, $croak_on_null) = @_;
 799
 800   my @c_cols = $self->result_source->unique_constraint_columns($constraint_name);
 801
 802   # combination may fail if $self->{cond} is non-trivial
 803   my ($final_cond) = try {
 804     $self->_merge_with_rscond ($extra_cond)
 805   } catch {
 806     +{ %$extra_cond }
 807   };
 808
 809   # trim out everything not in $columns
 810   $final_cond = { map {
 811     exists $final_cond->{$_}
 812       ? ( $_ => $final_cond->{$_} )
 813       : ()
 814   } @c_cols };
 815
 816   if (my @missing = grep
 817     { ! ($croak_on_null ? defined $final_cond->{$_} : exists $final_cond->{$_}) }
 818     (@c_cols)
 819   ) {
 820     $self->throw_exception( sprintf ( "Unable to satisfy requested constraint '%s', no values for column(s): %s",
 821       $constraint_name,
 822       join (', ', map { "'$_'" } @missing),
 823     ) );
 824   }
 825
 826   if (
 827     !$croak_on_null
 828       and
 829     !$ENV{DBIC_NULLABLE_KEY_NOWARN}
 830       and
 831     my @undefs = grep { ! defined $final_cond->{$_} } (keys %$final_cond)
 832   ) {
 833     my $callsite = do {
 834       my $w;
 835       local $SIG{__WARN__} = sub { $w = shift };
 836       carp;
 837       $w
 838     };
 839
 840     carp ( sprintf (
 841       "NULL/undef values supplied for requested unique constraint '%s' (NULL "
 842     . 'values in column(s): %s). This is almost certainly not what you wanted, '
 843     . 'though you can set DBIC_NULLABLE_KEY_NOWARN to disable this warning.',
 844       $constraint_name,
 845       join (', ', map { "'$_'" } @undefs),
 846     )) unless $callsites_warned_ucond->{$callsite}++;
 847   }
 848
 849   return $final_cond;
 850 }
 851
 852 =head2 search_related
 853
 854 =over 4
 855
 856 =item Arguments: $rel, $cond, \%attrs?
 857
 858 =item Return Value: $new_resultset
 859
 860 =back
 861
 862   $new_rs = $cd_rs->search_related('artist', {
 863     name => 'Emo-R-Us',
 864   });
 865
 866 Searches the specified relationship, optionally specifying a condition and
 867 attributes for matching records. See L</ATTRIBUTES> for more information.
 868
 869 =cut
 870
 871 sub search_related {
 872   return shift->related_resultset(shift)->search(@_);
 873 }
 874
 875 =head2 search_related_rs
 876
 877 This method works exactly the same as search_related, except that
 878 it guarantees a resultset, even in list context.
 879
 880 =cut
 881
 882 sub search_related_rs {
 883   return shift->related_resultset(shift)->search_rs(@_);
 884 }
 885
 886 =head2 cursor
 887
 888 =over 4
 889
 890 =item Arguments: none
 891
 892 =item Return Value: $cursor
 893
 894 =back
 895
 896 Returns a storage-driven cursor to the given resultset. See
 897 L<DBIx::Class::Cursor> for more information.
 898
 899 =cut
 900
 901 sub cursor {
 902   my ($self) = @_;
 903
 904   my $attrs = $self->_resolved_attrs_copy;
 905
 906   return $self->{cursor}
 907     ||= $self->result_source->storage->select($attrs->{from}, $attrs->{select},
 908           $attrs->{where},$attrs);
 909 }
 910
 911 =head2 single
 912
 913 =over 4
 914
 915 =item Arguments: $cond?
 916
 917 =item Return Value: $row_object | undef
 918
 919 =back
 920
 921   my $cd = $schema->resultset('CD')->single({ year => 2001 });
 922
 923 Inflates the first result without creating a cursor if the resultset has
 924 any records in it; if not returns C<undef>. Used by L</find> as a lean version
 925 of L</search>.
 926
 927 While this method can take an optional search condition (just like L</search>)
 928 being a fast-code-path it does not recognize search attributes. If you need to
 929 add extra joins or similar, call L</search> and then chain-call L</single> on the
 930 L<DBIx::Class::ResultSet> returned.
 931
 932 =over
 933
 934 =item B<Note>
 935
 936 As of 0.08100, this method enforces the assumption that the preceding
 937 query returns only one row. If more than one row is returned, you will receive
 938 a warning:
 939
 940   Query returned more than one row
 941
 942 In this case, you should be using L</next> or L</find> instead, or if you really
 943 know what you are doing, use the L</rows> attribute to explicitly limit the size
 944 of the resultset.
 945
 946 This method will also throw an exception if it is called on a resultset prefetching
 947 has_many, as such a prefetch implies fetching multiple rows from the database in
 948 order to assemble the resulting object.
 949
 950 =back
 951
 952 =cut
 953
 954 sub single {
 955   my ($self, $where) = @_;
 956   if(@_ > 2) {
 957       $self->throw_exception('single() only takes search conditions, no attributes. You want ->search( $cond, $attrs )->single()');
 958   }
 959
 960   my $attrs = $self->_resolved_attrs_copy;
 961
 962   if (keys %{$attrs->{collapse}}) {
 963     $self->throw_exception(
 964       'single() can not be used on resultsets prefetching has_many. Use find( \%cond ) or next() instead'
 965     );
 966   }
 967
 968   if ($where) {
 969     if (defined $attrs->{where}) {
 970       $attrs->{where} = {
 971         '-and' =>
 972             [ map { ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_ }
 973                $where, delete $attrs->{where} ]
 974       };
 975     } else {
 976       $attrs->{where} = $where;
 977     }
 978   }
 979
 980   my @data = $self->result_source->storage->select_single(
 981     $attrs->{from}, $attrs->{select},
 982     $attrs->{where}, $attrs
 983   );
 984
 985   return (@data ? ($self->_construct_object(@data))[0] : undef);
 986 }
 987
 988
 989 # _collapse_query
 990 #
 991 # Recursively collapse the query, accumulating values for each column.
 992
 993 sub _collapse_query {
 994   my ($self, $query, $collapsed) = @_;
 995
 996   $collapsed ||= {};
 997
 998   if (ref $query eq 'ARRAY') {
 999     foreach my $subquery (@$query) {
1000       next unless ref $subquery;  # -or
1001       $collapsed = $self->_collapse_query($subquery, $collapsed);
1002     }
1003   }
1004   elsif (ref $query eq 'HASH') {
1005     if (keys %$query and (keys %$query)[0] eq '-and') {
1006       foreach my $subquery (@{$query->{-and}}) {
1007         $collapsed = $self->_collapse_query($subquery, $collapsed);
1008       }
1009     }
1010     else {
1011       foreach my $col (keys %$query) {
1012         my $value = $query->{$col};
1013         $collapsed->{$col}{$value}++;
1014       }
1015     }
1016   }
1017
1018   return $collapsed;
1019 }
1020
1021 =head2 get_column
1022
1023 =over 4
1024
1025 =item Arguments: $cond?
1026
1027 =item Return Value: $resultsetcolumn
1028
1029 =back
1030
1031   my $max_length = $rs->get_column('length')->max;
1032
1033 Returns a L<DBIx::Class::ResultSetColumn> instance for a column of the ResultSet.
1034
1035 =cut
1036
1037 sub get_column {
1038   my ($self, $column) = @_;
1039   my $new = DBIx::Class::ResultSetColumn->new($self, $column);
1040   return $new;
1041 }
1042
1043 =head2 search_like
1044
1045 =over 4
1046
1047 =item Arguments: $cond, \%attrs?
1048
1049 =item Return Value: $resultset (scalar context), @row_objs (list context)
1050
1051 =back
1052
1053   # WHERE title LIKE '%blue%'
1054   $cd_rs = $rs->search_like({ title => '%blue%'});
1055
1056 Performs a search, but uses C<LIKE> instead of C<=> as the condition. Note
1057 that this is simply a convenience method retained for ex Class::DBI users.
1058 You most likely want to use L</search> with specific operators.
1059
1060 For more information, see L<DBIx::Class::Manual::Cookbook>.
1061
1062 This method is deprecated and will be removed in 0.09. Use L</search()>
1063 instead. An example conversion is:
1064
1065   ->search_like({ foo => 'bar' });
1066
1067   # Becomes
1068
1069   ->search({ foo => { like => 'bar' } });
1070
1071 =cut
1072
1073 sub search_like {
1074   my $class = shift;
1075   carp (
1076     'search_like() is deprecated and will be removed in DBIC version 0.09.'
1077    .' Instead use ->search({ x => { -like => "y%" } })'
1078    .' (note the outer pair of {}s - they are important!)'
1079   );
1080   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
1081   my $query = ref $_[0] eq 'HASH' ? { %{shift()} }: {@_};
1082   $query->{$_} = { 'like' => $query->{$_} } for keys %$query;
1083   return $class->search($query, { %$attrs });
1084 }
1085
1086 =head2 slice
1087
1088 =over 4
1089
1090 =item Arguments: $first, $last
1091
1092 =item Return Value: $resultset (scalar context), @row_objs (list context)
1093
1094 =back
1095
1096 Returns a resultset or object list representing a subset of elements from the
1097 resultset slice is called on. Indexes are from 0, i.e., to get the first
1098 three records, call:
1099
1100   my ($one, $two, $three) = $rs->slice(0, 2);
1101
1102 =cut
1103
1104 sub slice {
1105   my ($self, $min, $max) = @_;
1106   my $attrs = {}; # = { %{ $self->{attrs} || {} } };
1107   $attrs->{offset} = $self->{attrs}{offset} || 0;
1108   $attrs->{offset} += $min;
1109   $attrs->{rows} = ($max ? ($max - $min + 1) : 1);
1110   return $self->search(undef, $attrs);
1111   #my $slice = (ref $self)->new($self->result_source, $attrs);
1112   #return (wantarray ? $slice->all : $slice);
1113 }
1114
1115 =head2 next
1116
1117 =over 4
1118
1119 =item Arguments: none
1120
1121 =item Return Value: $result | undef
1122
1123 =back
1124
1125 Returns the next element in the resultset (C<undef> is there is none).
1126
1127 Can be used to efficiently iterate over records in the resultset:
1128
1129   my $rs = $schema->resultset('CD')->search;
1130   while (my $cd = $rs->next) {
1131     print $cd->title;
1132   }
1133
1134 Note that you need to store the resultset object, and call C<next> on it.
1135 Calling C<< resultset('Table')->next >> repeatedly will always return the
1136 first record from the resultset.
1137
1138 =cut
1139
1140 sub next {
1141   my ($self) = @_;
1142   if (my $cache = $self->get_cache) {
1143     $self->{all_cache_position} ||= 0;
1144     return $cache->[$self->{all_cache_position}++];
1145   }
1146   if ($self->{attrs}{cache}) {
1147     delete $self->{pager};
1148     $self->{all_cache_position} = 1;
1149     return ($self->all)[0];
1150   }
1151   if ($self->{stashed_objects}) {
1152     my $obj = shift(@{$self->{stashed_objects}});
1153     delete $self->{stashed_objects} unless @{$self->{stashed_objects}};
1154     return $obj;
1155   }
1156   my @row = (
1157     exists $self->{stashed_row}
1158       ? @{delete $self->{stashed_row}}
1159       : $self->cursor->next
1160   );
1161   return undef unless (@row);
1162   my ($row, @more) = $self->_construct_object(@row);
1163   $self->{stashed_objects} = \@more if @more;
1164   return $row;
1165 }
1166
1167 sub _construct_object {
1168   my ($self, @row) = @_;
1169
1170   my $info = $self->_collapse_result($self->{_attrs}{as}, \@row)
1171     or return ();
1172   my @new = $self->result_class->inflate_result($self->result_source, @$info);
1173   @new = $self->{_attrs}{record_filter}->(@new)
1174     if exists $self->{_attrs}{record_filter};
1175   return @new;
1176 }
1177
1178 sub _collapse_result {
1179   my ($self, $as_proto, $row) = @_;
1180
1181   my @copy = @$row;
1182
1183   # 'foo'         => [ undef, 'foo' ]
1184   # 'foo.bar'     => [ 'foo', 'bar' ]
1185   # 'foo.bar.baz' => [ 'foo.bar', 'baz' ]
1186
1187   my @construct_as = map { [ (/^(?:(.*)\.)?([^.]+)$/) ] } @$as_proto;
1188
1189   my %collapse = %{$self->{_attrs}{collapse}||{}};
1190
1191   my @pri_index;
1192
1193   # if we're doing collapsing (has_many prefetch) we need to grab records
1194   # until the PK changes, so fill @pri_index. if not, we leave it empty so
1195   # we know we don't have to bother.
1196
1197   # the reason for not using the collapse stuff directly is because if you
1198   # had for e.g. two artists in a row with no cds, the collapse info for
1199   # both would be NULL (undef) so you'd lose the second artist
1200
1201   # store just the index so we can check the array positions from the row
1202   # without having to contruct the full hash
1203
1204   if (keys %collapse) {
1205     my %pri = map { ($_ => 1) } $self->result_source->_pri_cols;
1206     foreach my $i (0 .. $#construct_as) {
1207       next if defined($construct_as[$i][0]); # only self table
1208       if (delete $pri{$construct_as[$i][1]}) {
1209         push(@pri_index, $i);
1210       }
1211       last unless keys %pri; # short circuit (Johnny Five Is Alive!)
1212     }
1213   }
1214
1215   # no need to do an if, it'll be empty if @pri_index is empty anyway
1216
1217   my %pri_vals = map { ($_ => $copy[$_]) } @pri_index;
1218
1219   my @const_rows;
1220
1221   do { # no need to check anything at the front, we always want the first row
1222
1223     my %const;
1224
1225     foreach my $this_as (@construct_as) {
1226       $const{$this_as->[0]||''}{$this_as->[1]} = shift(@copy);
1227     }
1228
1229     push(@const_rows, \%const);
1230
1231   } until ( # no pri_index => no collapse => drop straight out
1232       !@pri_index
1233     or
1234       do { # get another row, stash it, drop out if different PK
1235
1236         @copy = $self->cursor->next;
1237         $self->{stashed_row} = \@copy;
1238
1239         # last thing in do block, counts as true if anything doesn't match
1240
1241         # check xor defined first for NULL vs. NOT NULL then if one is
1242         # defined the other must be so check string equality
1243
1244         grep {
1245           (defined $pri_vals{$_} ^ defined $copy[$_])
1246           || (defined $pri_vals{$_} && ($pri_vals{$_} ne $copy[$_]))
1247         } @pri_index;
1248       }
1249   );
1250
1251   my $alias = $self->{attrs}{alias};
1252   my $info = [];
1253
1254   my %collapse_pos;
1255
1256   my @const_keys;
1257
1258   foreach my $const (@const_rows) {
1259     scalar @const_keys or do {
1260       @const_keys = sort { length($a) <=> length($b) } keys %$const;
1261     };
1262     foreach my $key (@const_keys) {
1263       if (length $key) {
1264         my $target = $info;
1265         my @parts = split(/\./, $key);
1266         my $cur = '';
1267         my $data = $const->{$key};
1268         foreach my $p (@parts) {
1269           $target = $target->[1]->{$p} ||= [];
1270           $cur .= ".${p}";
1271           if ($cur eq ".${key}" && (my @ckey = @{$collapse{$cur}||[]})) {
1272             # collapsing at this point and on final part
1273             my $pos = $collapse_pos{$cur};
1274             CK: foreach my $ck (@ckey) {
1275               if (!defined $pos->{$ck} || $pos->{$ck} ne $data->{$ck}) {
1276                 $collapse_pos{$cur} = $data;
1277                 delete @collapse_pos{ # clear all positioning for sub-entries
1278                   grep { m/^\Q${cur}.\E/ } keys %collapse_pos
1279                 };
1280                 push(@$target, []);
1281                 last CK;
1282               }
1283             }
1284           }
1285           if (exists $collapse{$cur}) {
1286             $target = $target->[-1];
1287           }
1288         }
1289         $target->[0] = $data;
1290       } else {
1291         $info->[0] = $const->{$key};
1292       }
1293     }
1294   }
1295
1296   return $info;
1297 }
1298
1299 =head2 result_source
1300
1301 =over 4
1302
1303 =item Arguments: $result_source?
1304
1305 =item Return Value: $result_source
1306
1307 =back
1308
1309 An accessor for the primary ResultSource object from which this ResultSet
1310 is derived.
1311
1312 =head2 result_class
1313
1314 =over 4
1315
1316 =item Arguments: $result_class?
1317
1318 =item Return Value: $result_class
1319
1320 =back
1321
1322 An accessor for the class to use when creating row objects. Defaults to
1323 C<< result_source->result_class >> - which in most cases is the name of the
1324 L<"table"|DBIx::Class::Manual::Glossary/"ResultSource"> class.
1325
1326 Note that changing the result_class will also remove any components
1327 that were originally loaded in the source class via
1328 L<DBIx::Class::ResultSource/load_components>. Any overloaded methods
1329 in the original source class will not run.
1330
1331 =cut
1332
1333 sub result_class {
1334   my ($self, $result_class) = @_;
1335   if ($result_class) {
1336     unless (ref $result_class) { # don't fire this for an object
1337       $self->ensure_class_loaded($result_class);
1338     }
1339     $self->_result_class($result_class);
1340     # THIS LINE WOULD BE A BUG - this accessor specifically exists to
1341     # permit the user to set result class on one result set only; it only
1342     # chains if provided to search()
1343     #$self->{attrs}{result_class} = $result_class if ref $self;
1344   }
1345   $self->_result_class;
1346 }
1347
1348 =head2 count
1349
1350 =over 4
1351
1352 =item Arguments: $cond, \%attrs??
1353
1354 =item Return Value: $count
1355
1356 =back
1357
1358 Performs an SQL C<COUNT> with the same query as the resultset was built
1359 with to find the number of elements. Passing arguments is equivalent to
1360 C<< $rs->search ($cond, \%attrs)->count >>
1361
1362 =cut
1363
1364 sub count {
1365   my $self = shift;
1366   return $self->search(@_)->count if @_ and defined $_[0];
1367   return scalar @{ $self->get_cache } if $self->get_cache;
1368
1369   my $attrs = $self->_resolved_attrs_copy;
1370
1371   # this is a little optimization - it is faster to do the limit
1372   # adjustments in software, instead of a subquery
1373   my $rows = delete $attrs->{rows};
1374   my $offset = delete $attrs->{offset};
1375
1376   my $crs;
1377   if ($self->_has_resolved_attr (qw/collapse group_by/)) {
1378     $crs = $self->_count_subq_rs ($attrs);
1379   }
1380   else {
1381     $crs = $self->_count_rs ($attrs);
1382   }
1383   my $count = $crs->next;
1384
1385   $count -= $offset if $offset;
1386   $count = $rows if $rows and $rows < $count;
1387   $count = 0 if ($count < 0);
1388
1389   return $count;
1390 }
1391
1392 =head2 count_rs
1393
1394 =over 4
1395
1396 =item Arguments: $cond, \%attrs??
1397
1398 =item Return Value: $count_rs
1399
1400 =back
1401
1402 Same as L</count> but returns a L<DBIx::Class::ResultSetColumn> object.
1403 This can be very handy for subqueries:
1404
1405   ->search( { amount => $some_rs->count_rs->as_query } )
1406
1407 As with regular resultsets the SQL query will be executed only after
1408 the resultset is accessed via L</next> or L</all>. That would return
1409 the same single value obtainable via L</count>.
1410
1411 =cut
1412
1413 sub count_rs {
1414   my $self = shift;
1415   return $self->search(@_)->count_rs if @_;
1416
1417   # this may look like a lack of abstraction (count() does about the same)
1418   # but in fact an _rs *must* use a subquery for the limits, as the
1419   # software based limiting can not be ported if this $rs is to be used
1420   # in a subquery itself (i.e. ->as_query)
1421   if ($self->_has_resolved_attr (qw/collapse group_by offset rows/)) {
1422     return $self->_count_subq_rs;
1423   }
1424   else {
1425     return $self->_count_rs;
1426   }
1427 }
1428
1429 #
1430 # returns a ResultSetColumn object tied to the count query
1431 #
1432 sub _count_rs {
1433   my ($self, $attrs) = @_;
1434
1435   my $rsrc = $self->result_source;
1436   $attrs ||= $self->_resolved_attrs;
1437
1438   my $tmp_attrs = { %$attrs };
1439   # take off any limits, record_filter is cdbi, and no point of ordering nor locking a count
1440   delete @{$tmp_attrs}{qw/rows offset order_by record_filter for/};
1441
1442   # overwrite the selector (supplied by the storage)
1443   $tmp_attrs->{select} = $rsrc->storage->_count_select ($rsrc, $attrs);
1444   $tmp_attrs->{as} = 'count';
1445   delete @{$tmp_attrs}{qw/columns _trailing_select/};
1446
1447   my $tmp_rs = $rsrc->resultset_class->new($rsrc, $tmp_attrs)->get_column ('count');
1448
1449   return $tmp_rs;
1450 }
1451
1452 #
1453 # same as above but uses a subquery
1454 #
1455 sub _count_subq_rs {
1456   my ($self, $attrs) = @_;
1457
1458   my $rsrc = $self->result_source;
1459   $attrs ||= $self->_resolved_attrs;
1460
1461   my $sub_attrs = { %$attrs };
1462   # extra selectors do not go in the subquery and there is no point of ordering it, nor locking it
1463   delete @{$sub_attrs}{qw/collapse columns as select _prefetch_selector_range _trailing_select order_by for/};
1464
1465   # if we multi-prefetch we group_by primary keys only as this is what we would
1466   # get out of the rs via ->next/->all. We *DO WANT* to clobber old group_by regardless
1467   if ( keys %{$attrs->{collapse}}  ) {
1468     $sub_attrs->{group_by} = [ map { "$attrs->{alias}.$_" } ($rsrc->_pri_cols) ]
1469   }
1470
1471   # Calculate subquery selector
1472   if (my $g = $sub_attrs->{group_by}) {
1473
1474     my $sql_maker = $rsrc->storage->sql_maker;
1475
1476     # necessary as the group_by may refer to aliased functions
1477     my $sel_index;
1478     for my $sel (@{$attrs->{select}}) {
1479       $sel_index->{$sel->{-as}} = $sel
1480         if (ref $sel eq 'HASH' and $sel->{-as});
1481     }
1482
1483     # anything from the original select mentioned on the group-by needs to make it to the inner selector
1484     # also look for named aggregates referred in the having clause
1485     # having often contains scalarrefs - thus parse it out entirely
1486     my @parts = @$g;
1487     if ($attrs->{having}) {
1488       local $sql_maker->{having_bind};
1489       local $sql_maker->{quote_char} = $sql_maker->{quote_char};
1490       local $sql_maker->{name_sep} = $sql_maker->{name_sep};
1491       unless (defined $sql_maker->{quote_char} and length $sql_maker->{quote_char}) {
1492         $sql_maker->{quote_char} = [ "\x00", "\xFF" ];
1493         # if we don't unset it we screw up retarded but unfortunately working
1494         # 'MAX(foo.bar)' => { '>', 3 }
1495         $sql_maker->{name_sep} = '';
1496       }
1497
1498       my ($lquote, $rquote, $sep) = map { quotemeta $_ } ($sql_maker->_quote_chars, $sql_maker->name_sep);
1499
1500       my $sql = $sql_maker->_parse_rs_attrs ({ having => $attrs->{having} });
1501
1502       # search for both a proper quoted qualified string, for a naive unquoted scalarref
1503       # and if all fails for an utterly naive quoted scalar-with-function
1504       while ($sql =~ /
1505         $rquote $sep $lquote (.+?) $rquote
1506           |
1507         [\s,] \w+ \. (\w+) [\s,]
1508           |
1509         [\s,] $lquote (.+?) $rquote [\s,]
1510       /gx) {
1511         push @parts, ($1 || $2 || $3);  # one of them matched if we got here
1512       }
1513     }
1514
1515     for (@parts) {
1516       my $colpiece = $sel_index->{$_} || $_;
1517
1518       # unqualify join-based group_by's. Arcane but possible query
1519       # also horrible horrible hack to alias a column (not a func.)
1520       # (probably need to introduce SQLA syntax)
1521       if ($colpiece =~ /\./ && $colpiece !~ /^$attrs->{alias}\./) {
1522         my $as = $colpiece;
1523         $as =~ s/\./__/;
1524         $colpiece = \ sprintf ('%s AS %s', map { $sql_maker->_quote ($_) } ($colpiece, $as) );
1525       }
1526       push @{$sub_attrs->{select}}, $colpiece;
1527     }
1528   }
1529   else {
1530     my @pcols = map { "$attrs->{alias}.$_" } ($rsrc->primary_columns);
1531     $sub_attrs->{select} = @pcols ? \@pcols : [ 1 ];
1532   }
1533
1534   return $rsrc->resultset_class
1535                ->new ($rsrc, $sub_attrs)
1536                 ->as_subselect_rs
1537                  ->search ({}, { columns => { count => $rsrc->storage->_count_select ($rsrc, $attrs) } })
1538                   ->get_column ('count');
1539 }
1540
1541 sub _bool {
1542   return 1;
1543 }
1544
1545 =head2 count_literal
1546
1547 =over 4
1548
1549 =item Arguments: $sql_fragment, @bind_values
1550
1551 =item Return Value: $count
1552
1553 =back
1554
1555 Counts the results in a literal query. Equivalent to calling L</search_literal>
1556 with the passed arguments, then L</count>.
1557
1558 =cut
1559
1560 sub count_literal { shift->search_literal(@_)->count; }
1561
1562 =head2 all
1563
1564 =over 4
1565
1566 =item Arguments: none
1567
1568 =item Return Value: @objects
1569
1570 =back
1571
1572 Returns all elements in the resultset. Called implicitly if the resultset
1573 is returned in list context.
1574
1575 =cut
1576
1577 sub all {
1578   my $self = shift;
1579   if(@_) {
1580       $self->throw_exception("all() doesn't take any arguments, you probably wanted ->search(...)->all()");
1581   }
1582
1583   return @{ $self->get_cache } if $self->get_cache;
1584
1585   my @obj;
1586
1587   if (keys %{$self->_resolved_attrs->{collapse}}) {
1588     # Using $self->cursor->all is really just an optimisation.
1589     # If we're collapsing has_many prefetches it probably makes
1590     # very little difference, and this is cleaner than hacking
1591     # _construct_object to survive the approach
1592     $self->cursor->reset;
1593     my @row = $self->cursor->next;
1594     while (@row) {
1595       push(@obj, $self->_construct_object(@row));
1596       @row = (exists $self->{stashed_row}
1597                ? @{delete $self->{stashed_row}}
1598                : $self->cursor->next);
1599     }
1600   } else {
1601     @obj = map { $self->_construct_object(@$_) } $self->cursor->all;
1602   }
1603
1604   $self->set_cache(\@obj) if $self->{attrs}{cache};
1605
1606   return @obj;
1607 }
1608
1609 =head2 reset
1610
1611 =over 4
1612
1613 =item Arguments: none
1614
1615 =item Return Value: $self
1616
1617 =back
1618
1619 Resets the resultset's cursor, so you can iterate through the elements again.
1620 Implicitly resets the storage cursor, so a subsequent L</next> will trigger
1621 another query.
1622
1623 =cut
1624
1625 sub reset {
1626   my ($self) = @_;
1627   delete $self->{_attrs} if exists $self->{_attrs};
1628   $self->{all_cache_position} = 0;
1629   $self->cursor->reset;
1630   return $self;
1631 }
1632
1633 =head2 first
1634
1635 =over 4
1636
1637 =item Arguments: none
1638
1639 =item Return Value: $object | undef
1640
1641 =back
1642
1643 Resets the resultset and returns an object for the first result (or C<undef>
1644 if the resultset is empty).
1645
1646 =cut
1647
1648 sub first {
1649   return $_[0]->reset->next;
1650 }
1651
1652
1653 # _rs_update_delete
1654 #
1655 # Determines whether and what type of subquery is required for the $rs operation.
1656 # If grouping is necessary either supplies its own, or verifies the current one
1657 # After all is done delegates to the proper storage method.
1658
1659 sub _rs_update_delete {
1660   my ($self, $op, $values) = @_;
1661
1662   my $rsrc = $self->result_source;
1663
1664   # if a condition exists we need to strip all table qualifiers
1665   # if this is not possible we'll force a subquery below
1666   my $cond = $rsrc->schema->storage->_strip_cond_qualifiers ($self->{cond});
1667
1668   my $needs_group_by_subq = $self->_has_resolved_attr (qw/collapse group_by -join/);
1669   my $needs_subq = $needs_group_by_subq || (not defined $cond) || $self->_has_resolved_attr(qw/rows offset/);
1670
1671   if ($needs_group_by_subq or $needs_subq) {
1672
1673     # make a new $rs selecting only the PKs (that's all we really need)
1674     my $attrs = $self->_resolved_attrs_copy;
1675
1676
1677     delete $attrs->{$_} for qw/collapse _collapse_order_by select _prefetch_selector_range as/;
1678     $attrs->{columns} = [ map { "$attrs->{alias}.$_" } ($self->result_source->_pri_cols) ];
1679
1680     if ($needs_group_by_subq) {
1681       # make sure no group_by was supplied, or if there is one - make sure it matches
1682       # the columns compiled above perfectly. Anything else can not be sanely executed
1683       # on most databases so croak right then and there
1684
1685       if (my $g = $attrs->{group_by}) {
1686         my @current_group_by = map
1687           { $_ =~ /\./ ? $_ : "$attrs->{alias}.$_" }
1688           @$g
1689         ;
1690
1691         if (
1692           join ("\x00", sort @current_group_by)
1693             ne
1694           join ("\x00", sort @{$attrs->{columns}} )
1695         ) {
1696           $self->throw_exception (
1697             "You have just attempted a $op operation on a resultset which does group_by"
1698             . ' on columns other than the primary keys, while DBIC internally needs to retrieve'
1699             . ' the primary keys in a subselect. All sane RDBMS engines do not support this'
1700             . ' kind of queries. Please retry the operation with a modified group_by or'
1701             . ' without using one at all.'
1702           );
1703         }
1704       }
1705       else {
1706         $attrs->{group_by} = $attrs->{columns};
1707       }
1708     }
1709
1710     my $subrs = (ref $self)->new($rsrc, $attrs);
1711     return $self->result_source->storage->_subq_update_delete($subrs, $op, $values);
1712   }
1713   else {
1714     return $rsrc->storage->$op(
1715       $rsrc,
1716       $op eq 'update' ? $values : (),
1717       $cond,
1718     );
1719   }
1720 }
1721
1722 =head2 update
1723
1724 =over 4
1725
1726 =item Arguments: \%values
1727
1728 =item Return Value: $storage_rv
1729
1730 =back
1731
1732 Sets the specified columns in the resultset to the supplied values in a
1733 single query. Note that this will not run any accessor/set_column/update
1734 triggers, nor will it update any row object instances derived from this
1735 resultset (this includes the contents of the L<resultset cache|/set_cache>
1736 if any). See L</update_all> if you need to execute any on-update
1737 triggers or cascades defined either by you or a
1738 L<result component|DBIx::Class::Manual::Component/WHAT_IS_A_COMPONENT>.
1739
1740 The return value is a pass through of what the underlying
1741 storage backend returned, and may vary. See L<DBI/execute> for the most
1742 common case.
1743
1744 =head3 CAVEAT
1745
1746 Note that L</update> does not process/deflate any of the values passed in.
1747 This is unlike the corresponding L<DBIx::Class::Row/update>. The user must
1748 ensure manually that any value passed to this method will stringify to
1749 something the RDBMS knows how to deal with. A notable example is the
1750 handling of L<DateTime> objects, for more info see:
1751 L<DBIx::Class::Manual::Cookbook/Formatting_DateTime_objects_in_queries>.
1752
1753 =cut
1754
1755 sub update {
1756   my ($self, $values) = @_;
1757   $self->throw_exception('Values for update must be a hash')
1758     unless ref $values eq 'HASH';
1759
1760   return $self->_rs_update_delete ('update', $values);
1761 }
1762
1763 =head2 update_all
1764
1765 =over 4
1766
1767 =item Arguments: \%values
1768
1769 =item Return Value: 1
1770
1771 =back
1772
1773 Fetches all objects and updates them one at a time via
1774 L<DBIx::Class::Row/update>. Note that C<update_all> will run DBIC defined
1775 triggers, while L</update> will not.
1776
1777 =cut
1778
1779 sub update_all {
1780   my ($self, $values) = @_;
1781   $self->throw_exception('Values for update_all must be a hash')
1782     unless ref $values eq 'HASH';
1783
1784   my $guard = $self->result_source->schema->txn_scope_guard;
1785   $_->update($values) for $self->all;
1786   $guard->commit;
1787   return 1;
1788 }
1789
1790 =head2 delete
1791
1792 =over 4
1793
1794 =item Arguments: none
1795
1796 =item Return Value: $storage_rv
1797
1798 =back
1799
1800 Deletes the rows matching this resultset in a single query. Note that this
1801 will not run any delete triggers, nor will it alter the
1802 L<in_storage|DBIx::Class::Row/in_storage> status of any row object instances
1803 derived from this resultset (this includes the contents of the
1804 L<resultset cache|/set_cache> if any). See L</delete_all> if you need to
1805 execute any on-delete triggers or cascades defined either by you or a
1806 L<result component|DBIx::Class::Manual::Component/WHAT_IS_A_COMPONENT>.
1807
1808 The return value is a pass through of what the underlying storage backend
1809 returned, and may vary. See L<DBI/execute> for the most common case.
1810
1811 =cut
1812
1813 sub delete {
1814   my $self = shift;
1815   $self->throw_exception('delete does not accept any arguments')
1816     if @_;
1817
1818   return $self->_rs_update_delete ('delete');
1819 }
1820
1821 =head2 delete_all
1822
1823 =over 4
1824
1825 =item Arguments: none
1826
1827 =item Return Value: 1
1828
1829 =back
1830
1831 Fetches all objects and deletes them one at a time via
1832 L<DBIx::Class::Row/delete>. Note that C<delete_all> will run DBIC defined
1833 triggers, while L</delete> will not.
1834
1835 =cut
1836
1837 sub delete_all {
1838   my $self = shift;
1839   $self->throw_exception('delete_all does not accept any arguments')
1840     if @_;
1841
1842   my $guard = $self->result_source->schema->txn_scope_guard;
1843   $_->delete for $self->all;
1844   $guard->commit;
1845   return 1;
1846 }
1847
1848 =head2 populate
1849
1850 =over 4
1851
1852 =item Arguments: \@data;
1853
1854 =back
1855
1856 Accepts either an arrayref of hashrefs or alternatively an arrayref of arrayrefs.
1857 For the arrayref of hashrefs style each hashref should be a structure suitable
1858 forsubmitting to a $resultset->create(...) method.
1859
1860 In void context, C<insert_bulk> in L<DBIx::Class::Storage::DBI> is used
1861 to insert the data, as this is a faster method.
1862
1863 Otherwise, each set of data is inserted into the database using
1864 L<DBIx::Class::ResultSet/create>, and the resulting objects are
1865 accumulated into an array. The array itself, or an array reference
1866 is returned depending on scalar or list context.
1867
1868 Example:  Assuming an Artist Class that has many CDs Classes relating:
1869
1870   my $Artist_rs = $schema->resultset("Artist");
1871
1872   ## Void Context Example
1873   $Artist_rs->populate([
1874      { artistid => 4, name => 'Manufactured Crap', cds => [
1875         { title => 'My First CD', year => 2006 },
1876         { title => 'Yet More Tweeny-Pop crap', year => 2007 },
1877       ],
1878      },
1879      { artistid => 5, name => 'Angsty-Whiny Girl', cds => [
1880         { title => 'My parents sold me to a record company', year => 2005 },
1881         { title => 'Why Am I So Ugly?', year => 2006 },
1882         { title => 'I Got Surgery and am now Popular', year => 2007 }
1883       ],
1884      },
1885   ]);
1886
1887   ## Array Context Example
1888   my ($ArtistOne, $ArtistTwo, $ArtistThree) = $Artist_rs->populate([
1889     { name => "Artist One"},
1890     { name => "Artist Two"},
1891     { name => "Artist Three", cds=> [
1892     { title => "First CD", year => 2007},
1893     { title => "Second CD", year => 2008},
1894   ]}
1895   ]);
1896
1897   print $ArtistOne->name; ## response is 'Artist One'
1898   print $ArtistThree->cds->count ## reponse is '2'
1899
1900 For the arrayref of arrayrefs style,  the first element should be a list of the
1901 fieldsnames to which the remaining elements are rows being inserted.  For
1902 example:
1903
1904   $Arstist_rs->populate([
1905     [qw/artistid name/],
1906     [100, 'A Formally Unknown Singer'],
1907     [101, 'A singer that jumped the shark two albums ago'],
1908     [102, 'An actually cool singer'],
1909   ]);
1910
1911 Please note an important effect on your data when choosing between void and
1912 wantarray context. Since void context goes straight to C<insert_bulk> in
1913 L<DBIx::Class::Storage::DBI> this will skip any component that is overriding
1914 C<insert>.  So if you are using something like L<DBIx-Class-UUIDColumns> to
1915 create primary keys for you, you will find that your PKs are empty.  In this
1916 case you will have to use the wantarray context in order to create those
1917 values.
1918
1919 =cut
1920
1921 sub populate {
1922   my $self = shift;
1923
1924   # cruft placed in standalone method
1925   my $data = $self->_normalize_populate_args(@_);
1926
1927   if(defined wantarray) {
1928     my @created;
1929     foreach my $item (@$data) {
1930       push(@created, $self->create($item));
1931     }
1932     return wantarray ? @created : \@created;
1933   }
1934   else {
1935     my $first = $data->[0];
1936
1937     # if a column is a registered relationship, and is a non-blessed hash/array, consider
1938     # it relationship data
1939     my (@rels, @columns);
1940     my $rsrc = $self->result_source;
1941     my $rels = { map { $_ => $rsrc->relationship_info($_) } $rsrc->relationships };
1942     for (keys %$first) {
1943       my $ref = ref $first->{$_};
1944       $rels->{$_} && ($ref eq 'ARRAY' or $ref eq 'HASH')
1945         ? push @rels, $_
1946         : push @columns, $_
1947       ;
1948     }
1949
1950     my @pks = $rsrc->primary_columns;
1951
1952     ## do the belongs_to relationships
1953     foreach my $index (0..$#$data) {
1954
1955       # delegate to create() for any dataset without primary keys with specified relationships
1956       if (grep { !defined $data->[$index]->{$_} } @pks ) {
1957         for my $r (@rels) {
1958           if (grep { ref $data->[$index]{$r} eq $_ } qw/HASH ARRAY/) {  # a related set must be a HASH or AoH
1959             my @ret = $self->populate($data);
1960             return;
1961           }
1962         }
1963       }
1964
1965       foreach my $rel (@rels) {
1966         next unless ref $data->[$index]->{$rel} eq "HASH";
1967         my $result = $self->related_resultset($rel)->create($data->[$index]->{$rel});
1968         my ($reverse_relname, $reverse_relinfo) = %{$rsrc->reverse_relationship_info($rel)};
1969         my $related = $result->result_source->_resolve_condition(
1970           $reverse_relinfo->{cond},
1971           $self,
1972           $result,
1973         );
1974
1975         delete $data->[$index]->{$rel};
1976         $data->[$index] = {%{$data->[$index]}, %$related};
1977
1978         push @columns, keys %$related if $index == 0;
1979       }
1980     }
1981
1982     ## inherit the data locked in the conditions of the resultset
1983     my ($rs_data) = $self->_merge_with_rscond({});
1984     delete @{$rs_data}{@columns};
1985     my @inherit_cols = keys %$rs_data;
1986     my @inherit_data = values %$rs_data;
1987
1988     ## do bulk insert on current row
1989     $rsrc->storage->insert_bulk(
1990       $rsrc,
1991       [@columns, @inherit_cols],
1992       [ map { [ @$_{@columns}, @inherit_data ] } @$data ],
1993     );
1994
1995     ## do the has_many relationships
1996     foreach my $item (@$data) {
1997
1998       my $main_row;
1999
2000       foreach my $rel (@rels) {
2001         next unless ref $item->{$rel} eq "ARRAY" && @{ $item->{$rel} };
2002
2003         $main_row ||= $self->new_result({map { $_ => $item->{$_} } @pks});
2004
2005         my $child = $main_row->$rel;
2006
2007         my $related = $child->result_source->_resolve_condition(
2008           $rels->{$rel}{cond},
2009           $child,
2010           $main_row,
2011         );
2012
2013         my @rows_to_add = ref $item->{$rel} eq 'ARRAY' ? @{$item->{$rel}} : ($item->{$rel});
2014         my @populate = map { {%$_, %$related} } @rows_to_add;
2015
2016         $child->populate( \@populate );
2017       }
2018     }
2019   }
2020 }
2021
2022
2023 # populate() argumnets went over several incarnations
2024 # What we ultimately support is AoH
2025 sub _normalize_populate_args {
2026   my ($self, $arg) = @_;
2027
2028   if (ref $arg eq 'ARRAY') {
2029     if (ref $arg->[0] eq 'HASH') {
2030       return $arg;
2031     }
2032     elsif (ref $arg->[0] eq 'ARRAY') {
2033       my @ret;
2034       my @colnames = @{$arg->[0]};
2035       foreach my $values (@{$arg}[1 .. $#$arg]) {
2036         push @ret, { map { $colnames[$_] => $values->[$_] } (0 .. $#colnames) };
2037       }
2038       return \@ret;
2039     }
2040   }
2041
2042   $self->throw_exception('Populate expects an arrayref of hashrefs or arrayref of arrayrefs');
2043 }
2044
2045 =head2 pager
2046
2047 =over 4
2048
2049 =item Arguments: none
2050
2051 =item Return Value: $pager
2052
2053 =back
2054
2055 Return Value a L<Data::Page> object for the current resultset. Only makes
2056 sense for queries with a C<page> attribute.
2057
2058 To get the full count of entries for a paged resultset, call
2059 C<total_entries> on the L<Data::Page> object.
2060
2061 =cut
2062
2063 # make a wizard good for both a scalar and a hashref
2064 my $mk_lazy_count_wizard = sub {
2065   require Variable::Magic;
2066
2067   my $stash = { total_rs => shift };
2068   my $slot = shift; # only used by the hashref magic
2069
2070   my $magic = Variable::Magic::wizard (
2071     data => sub { $stash },
2072
2073     (!$slot)
2074     ? (
2075       # the scalar magic
2076       get => sub {
2077         # set value lazily, and dispell for good
2078         ${$_[0]} = $_[1]{total_rs}->count;
2079         Variable::Magic::dispell (${$_[0]}, $_[1]{magic_selfref});
2080         return 1;
2081       },
2082       set => sub {
2083         # an explicit set implies dispell as well
2084         # the unless() is to work around "fun and giggles" below
2085         Variable::Magic::dispell (${$_[0]}, $_[1]{magic_selfref})
2086           unless (caller(2))[3] eq 'DBIx::Class::ResultSet::pager';
2087         return 1;
2088       },
2089     )
2090     : (
2091       # the uvar magic
2092       fetch => sub {
2093         if ($_[2] eq $slot and !$_[1]{inactive}) {
2094           my $cnt = $_[1]{total_rs}->count;
2095           $_[0]->{$slot} = $cnt;
2096
2097           # attempting to dispell in a fetch handle (works in store), seems
2098           # to invariable segfault on 5.10, 5.12, 5.13 :(
2099           # so use an inactivator instead
2100           #Variable::Magic::dispell (%{$_[0]}, $_[1]{magic_selfref});
2101           $_[1]{inactive}++;
2102         }
2103         return 1;
2104       },
2105       store => sub {
2106         if (! $_[1]{inactive} and $_[2] eq $slot) {
2107           #Variable::Magic::dispell (%{$_[0]}, $_[1]{magic_selfref});
2108           $_[1]{inactive}++
2109             unless (caller(2))[3] eq 'DBIx::Class::ResultSet::pager';
2110         }
2111         return 1;
2112       },
2113     ),
2114   );
2115
2116   $stash->{magic_selfref} = $magic;
2117   weaken ($stash->{magic_selfref}); # this fails on 5.8.1
2118
2119   return $magic;
2120 };
2121
2122 # the tie class for 5.8.1
2123 {
2124   package # hide from pause
2125     DBIx::Class::__DBIC_LAZY_RS_COUNT__;
2126   use base qw/Tie::Hash/;
2127
2128   sub FIRSTKEY { my $dummy = scalar keys %{$_[0]{data}}; each %{$_[0]{data}} }
2129   sub NEXTKEY  { each %{$_[0]{data}} }
2130   sub EXISTS   { exists $_[0]{data}{$_[1]} }
2131   sub DELETE   { delete $_[0]{data}{$_[1]} }
2132   sub CLEAR    { %{$_[0]{data}} = () }
2133   sub SCALAR   { scalar %{$_[0]{data}} }
2134
2135   sub TIEHASH {
2136     $_[1]{data} = {%{$_[1]{selfref}}};
2137     %{$_[1]{selfref}} = ();
2138     Scalar::Util::weaken ($_[1]{selfref});
2139     return bless ($_[1], $_[0]);
2140   };
2141
2142   sub FETCH {
2143     if ($_[1] eq $_[0]{slot}) {
2144       my $cnt = $_[0]{data}{$_[1]} = $_[0]{total_rs}->count;
2145       untie %{$_[0]{selfref}};
2146       %{$_[0]{selfref}} = %{$_[0]{data}};
2147       return $cnt;
2148     }
2149     else {
2150       $_[0]{data}{$_[1]};
2151     }
2152   }
2153
2154   sub STORE {
2155     $_[0]{data}{$_[1]} = $_[2];
2156     if ($_[1] eq $_[0]{slot}) {
2157       untie %{$_[0]{selfref}};
2158       %{$_[0]{selfref}} = %{$_[0]{data}};
2159     }
2160     $_[2];
2161   }
2162 }
2163
2164 sub pager {
2165   my ($self) = @_;
2166
2167   return $self->{pager} if $self->{pager};
2168
2169   if ($self->get_cache) {
2170     $self->throw_exception ('Pagers on cached resultsets are not supported');
2171   }
2172
2173   my $attrs = $self->{attrs};
2174   if (!defined $attrs->{page}) {
2175     $self->throw_exception("Can't create pager for non-paged rs");
2176   }
2177   elsif ($attrs->{page} <= 0) {
2178     $self->throw_exception('Invalid page number (page-numbers are 1-based)');
2179   }
2180   $attrs->{rows} ||= 10;
2181
2182   # throw away the paging flags and re-run the count (possibly
2183   # with a subselect) to get the real total count
2184   my $count_attrs = { %$attrs };
2185   delete $count_attrs->{$_} for qw/rows offset page pager/;
2186   my $total_rs = (ref $self)->new($self->result_source, $count_attrs);
2187
2188
2189 ### the following may seem awkward and dirty, but it's a thought-experiment
2190 ### necessary for future development of DBIx::DS. Do *NOT* change this code
2191 ### before talking to ribasushi/mst
2192
2193   require Data::Page;
2194   my $pager = Data::Page->new(
2195     0,  #start with an empty set
2196     $attrs->{rows},
2197     $self->{attrs}{page},
2198   );
2199
2200   my $data_slot = 'total_entries';
2201
2202   # Since we are interested in a cached value (once it's set - it's set), every
2203   # technique will detach from the magic-host once the time comes to fire the
2204   # ->count (or in the segfaulting case of >= 5.10 it will deactivate itself)
2205
2206   if ($] < 5.008003) {
2207     # 5.8.1 throws 'Modification of a read-only value attempted' when one tries
2208     # to weakref the magic container :(
2209     # tested on 5.8.1
2210     tie (%$pager, 'DBIx::Class::__DBIC_LAZY_RS_COUNT__',
2211       { slot => $data_slot, total_rs => $total_rs, selfref => $pager }
2212     );
2213   }
2214   elsif ($] < 5.010) {
2215     # We can use magic on the hash value slot. It's interesting that the magic is
2216     # attached to the hash-slot, and does *not* stop working once I do the dummy
2217     # assignments after the cast()
2218     # tested on 5.8.3 and 5.8.9
2219     my $magic = $mk_lazy_count_wizard->($total_rs);
2220     Variable::Magic::cast ( $pager->{$data_slot}, $magic );
2221
2222     # this is for fun and giggles
2223     $pager->{$data_slot} = -1;
2224     $pager->{$data_slot} = 0;
2225
2226     # this does not work for scalars, but works with
2227     # uvar magic below
2228     #my %vals = %$pager;
2229     #%$pager = ();
2230     #%{$pager} = %vals;
2231   }
2232   else {
2233     # And the uvar magic
2234     # works on 5.10.1, 5.12.1 and 5.13.4 in its current form,
2235     # however see the wizard maker for more notes
2236     my $magic = $mk_lazy_count_wizard->($total_rs, $data_slot);
2237     Variable::Magic::cast ( %$pager, $magic );
2238
2239     # still works
2240     $pager->{$data_slot} = -1;
2241     $pager->{$data_slot} = 0;
2242
2243     # this now works
2244     my %vals = %$pager;
2245     %$pager = ();
2246     %{$pager} = %vals;
2247   }
2248
2249   return $self->{pager} = $pager;
2250 }
2251
2252 =head2 page
2253
2254 =over 4
2255
2256 =item Arguments: $page_number
2257
2258 =item Return Value: $rs
2259
2260 =back
2261
2262 Returns a resultset for the $page_number page of the resultset on which page
2263 is called, where each page contains a number of rows equal to the 'rows'
2264 attribute set on the resultset (10 by default).
2265
2266 =cut
2267
2268 sub page {
2269   my ($self, $page) = @_;
2270   return (ref $self)->new($self->result_source, { %{$self->{attrs}}, page => $page });
2271 }
2272
2273 =head2 new_result
2274
2275 =over 4
2276
2277 =item Arguments: \%vals
2278
2279 =item Return Value: $rowobject
2280
2281 =back
2282
2283 Creates a new row object in the resultset's result class and returns
2284 it. The row is not inserted into the database at this point, call
2285 L<DBIx::Class::Row/insert> to do that. Calling L<DBIx::Class::Row/in_storage>
2286 will tell you whether the row object has been inserted or not.
2287
2288 Passes the hashref of input on to L<DBIx::Class::Row/new>.
2289
2290 =cut
2291
2292 sub new_result {
2293   my ($self, $values) = @_;
2294   $self->throw_exception( "new_result needs a hash" )
2295     unless (ref $values eq 'HASH');
2296
2297   my ($merged_cond, $cols_from_relations) = $self->_merge_with_rscond($values);
2298
2299   my %new = (
2300     %$merged_cond,
2301     @$cols_from_relations
2302       ? (-cols_from_relations => $cols_from_relations)
2303       : (),
2304     -result_source => $self->result_source, # DO NOT REMOVE THIS, REQUIRED
2305   );
2306
2307   return $self->result_class->new(\%new);
2308 }
2309
2310 # _merge_with_rscond
2311 #
2312 # Takes a simple hash of K/V data and returns its copy merged with the
2313 # condition already present on the resultset. Additionally returns an
2314 # arrayref of value/condition names, which were inferred from related
2315 # objects (this is needed for in-memory related objects)
2316 sub _merge_with_rscond {
2317   my ($self, $data) = @_;
2318
2319   my (%new_data, @cols_from_relations);
2320
2321   my $alias = $self->{attrs}{alias};
2322
2323   if (! defined $self->{cond}) {
2324     # just massage $data below
2325   }
2326   elsif ($self->{cond} eq $DBIx::Class::ResultSource::UNRESOLVABLE_CONDITION) {
2327     %new_data = %{ $self->{attrs}{related_objects} || {} };  # nothing might have been inserted yet
2328     @cols_from_relations = keys %new_data;
2329   }
2330   elsif (ref $self->{cond} ne 'HASH') {
2331     $self->throw_exception(
2332       "Can't abstract implicit construct, resultset condition not a hash"
2333     );
2334   }
2335   else {
2336     # precendence must be given to passed values over values inherited from
2337     # the cond, so the order here is important.
2338     my $collapsed_cond = $self->_collapse_cond($self->{cond});
2339     my %implied = %{$self->_remove_alias($collapsed_cond, $alias)};
2340
2341     while ( my($col, $value) = each %implied ) {
2342       my $vref = ref $value;
2343       if ($vref eq 'HASH' && keys(%$value) && (keys %$value)[0] eq '=') {
2344         $new_data{$col} = $value->{'='};
2345       }
2346       elsif( !$vref or $vref eq 'SCALAR' or blessed($value) ) {
2347         $new_data{$col} = $value;
2348       }
2349     }
2350   }
2351
2352   %new_data = (
2353     %new_data,
2354     %{ $self->_remove_alias($data, $alias) },
2355   );
2356
2357   return (\%new_data, \@cols_from_relations);
2358 }
2359
2360 # _has_resolved_attr
2361 #
2362 # determines if the resultset defines at least one
2363 # of the attributes supplied
2364 #
2365 # used to determine if a subquery is neccessary
2366 #
2367 # supports some virtual attributes:
2368 #   -join
2369 #     This will scan for any joins being present on the resultset.
2370 #     It is not a mere key-search but a deep inspection of {from}
2371 #
2372
2373 sub _has_resolved_attr {
2374   my ($self, @attr_names) = @_;
2375
2376   my $attrs = $self->_resolved_attrs;
2377
2378   my %extra_checks;
2379
2380   for my $n (@attr_names) {
2381     if (grep { $n eq $_ } (qw/-join/) ) {
2382       $extra_checks{$n}++;
2383       next;
2384     }
2385
2386     my $attr =  $attrs->{$n};
2387
2388     next if not defined $attr;
2389
2390     if (ref $attr eq 'HASH') {
2391       return 1 if keys %$attr;
2392     }
2393     elsif (ref $attr eq 'ARRAY') {
2394       return 1 if @$attr;
2395     }
2396     else {
2397       return 1 if $attr;
2398     }
2399   }
2400
2401   # a resolved join is expressed as a multi-level from
2402   return 1 if (
2403     $extra_checks{-join}
2404       and
2405     ref $attrs->{from} eq 'ARRAY'
2406       and
2407     @{$attrs->{from}} > 1
2408   );
2409
2410   return 0;
2411 }
2412
2413 # _collapse_cond
2414 #
2415 # Recursively collapse the condition.
2416
2417 sub _collapse_cond {
2418   my ($self, $cond, $collapsed) = @_;
2419
2420   $collapsed ||= {};
2421
2422   if (ref $cond eq 'ARRAY') {
2423     foreach my $subcond (@$cond) {
2424       next unless ref $subcond;  # -or
2425       $collapsed = $self->_collapse_cond($subcond, $collapsed);
2426     }
2427   }
2428   elsif (ref $cond eq 'HASH') {
2429     if (keys %$cond and (keys %$cond)[0] eq '-and') {
2430       foreach my $subcond (@{$cond->{-and}}) {
2431         $collapsed = $self->_collapse_cond($subcond, $collapsed);
2432       }
2433     }
2434     else {
2435       foreach my $col (keys %$cond) {
2436         my $value = $cond->{$col};
2437         $collapsed->{$col} = $value;
2438       }
2439     }
2440   }
2441
2442   return $collapsed;
2443 }
2444
2445 # _remove_alias
2446 #
2447 # Remove the specified alias from the specified query hash. A copy is made so
2448 # the original query is not modified.
2449
2450 sub _remove_alias {
2451   my ($self, $query, $alias) = @_;
2452
2453   my %orig = %{ $query || {} };
2454   my %unaliased;
2455
2456   foreach my $key (keys %orig) {
2457     if ($key !~ /\./) {
2458       $unaliased{$key} = $orig{$key};
2459       next;
2460     }
2461     $unaliased{$1} = $orig{$key}
2462       if $key =~ m/^(?:\Q$alias\E\.)?([^.]+)$/;
2463   }
2464
2465   return \%unaliased;
2466 }
2467
2468 =head2 as_query
2469
2470 =over 4
2471
2472 =item Arguments: none
2473
2474 =item Return Value: \[ $sql, @bind ]
2475
2476 =back
2477
2478 Returns the SQL query and bind vars associated with the invocant.
2479
2480 This is generally used as the RHS for a subquery.
2481
2482 =cut
2483
2484 sub as_query {
2485   my $self = shift;
2486
2487   my $attrs = $self->_resolved_attrs_copy;
2488
2489   # For future use:
2490   #
2491   # in list ctx:
2492   # my ($sql, \@bind, \%dbi_bind_attrs) = _select_args_to_query (...)
2493   # $sql also has no wrapping parenthesis in list ctx
2494   #
2495   my $sqlbind = $self->result_source->storage
2496     ->_select_args_to_query ($attrs->{from}, $attrs->{select}, $attrs->{where}, $attrs);
2497
2498   return $sqlbind;
2499 }
2500
2501 =head2 find_or_new
2502
2503 =over 4
2504
2505 =item Arguments: \%vals, \%attrs?
2506
2507 =item Return Value: $rowobject
2508
2509 =back
2510
2511   my $artist = $schema->resultset('Artist')->find_or_new(
2512     { artist => 'fred' }, { key => 'artists' });
2513
2514   $cd->cd_to_producer->find_or_new({ producer => $producer },
2515                                    { key => 'primary });
2516
2517 Find an existing record from this resultset using L</find>. if none exists,
2518 instantiate a new result object and return it. The object will not be saved
2519 into your storage until you call L<DBIx::Class::Row/insert> on it.
2520
2521 You most likely want this method when looking for existing rows using a unique
2522 constraint that is not the primary key, or looking for related rows.
2523
2524 If you want objects to be saved immediately, use L</find_or_create> instead.
2525
2526 B<Note>: Make sure to read the documentation of L</find> and understand the
2527 significance of the C<key> attribute, as its lack may skew your search, and
2528 subsequently result in spurious new objects.
2529
2530 B<Note>: Take care when using C<find_or_new> with a table having
2531 columns with default values that you intend to be automatically
2532 supplied by the database (e.g. an auto_increment primary key column).
2533 In normal usage, the value of such columns should NOT be included at
2534 all in the call to C<find_or_new>, even when set to C<undef>.
2535
2536 =cut
2537
2538 sub find_or_new {
2539   my $self     = shift;
2540   my $attrs    = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
2541   my $hash     = ref $_[0] eq 'HASH' ? shift : {@_};
2542   if (keys %$hash and my $row = $self->find($hash, $attrs) ) {
2543     return $row;
2544   }
2545   return $self->new_result($hash);
2546 }
2547
2548 =head2 create
2549
2550 =over 4
2551
2552 =item Arguments: \%vals
2553
2554 =item Return Value: a L<DBIx::Class::Row> $object
2555
2556 =back
2557
2558 Attempt to create a single new row or a row with multiple related rows
2559 in the table represented by the resultset (and related tables). This
2560 will not check for duplicate rows before inserting, use
2561 L</find_or_create> to do that.
2562
2563 To create one row for this resultset, pass a hashref of key/value
2564 pairs representing the columns of the table and the values you wish to
2565 store. If the appropriate relationships are set up, foreign key fields
2566 can also be passed an object representing the foreign row, and the
2567 value will be set to its primary key.
2568
2569 To create related objects, pass a hashref of related-object column values
2570 B<keyed on the relationship name>. If the relationship is of type C<multi>
2571 (L<DBIx::Class::Relationship/has_many>) - pass an arrayref of hashrefs.
2572 The process will correctly identify columns holding foreign keys, and will
2573 transparently populate them from the keys of the corresponding relation.
2574 This can be applied recursively, and will work correctly for a structure
2575 with an arbitrary depth and width, as long as the relationships actually
2576 exists and the correct column data has been supplied.
2577
2578
2579 Instead of hashrefs of plain related data (key/value pairs), you may
2580 also pass new or inserted objects. New objects (not inserted yet, see
2581 L</new>), will be inserted into their appropriate tables.
2582
2583 Effectively a shortcut for C<< ->new_result(\%vals)->insert >>.
2584
2585 Example of creating a new row.
2586
2587   $person_rs->create({
2588     name=>"Some Person",
2589     email=>"somebody@someplace.com"
2590   });
2591
2592 Example of creating a new row and also creating rows in a related C<has_many>
2593 or C<has_one> resultset.  Note Arrayref.
2594
2595   $artist_rs->create(
2596      { artistid => 4, name => 'Manufactured Crap', cds => [
2597         { title => 'My First CD', year => 2006 },
2598         { title => 'Yet More Tweeny-Pop crap', year => 2007 },
2599       ],
2600      },
2601   );
2602
2603 Example of creating a new row and also creating a row in a related
2604 C<belongs_to> resultset. Note Hashref.
2605
2606   $cd_rs->create({
2607     title=>"Music for Silly Walks",
2608     year=>2000,
2609     artist => {
2610       name=>"Silly Musician",
2611     }
2612   });
2613
2614 =over
2615
2616 =item WARNING
2617
2618 When subclassing ResultSet never attempt to override this method. Since
2619 it is a simple shortcut for C<< $self->new_result($attrs)->insert >>, a
2620 lot of the internals simply never call it, so your override will be
2621 bypassed more often than not. Override either L<new|DBIx::Class::Row/new>
2622 or L<insert|DBIx::Class::Row/insert> depending on how early in the
2623 L</create> process you need to intervene.
2624
2625 =back
2626
2627 =cut
2628
2629 sub create {
2630   my ($self, $attrs) = @_;
2631   $self->throw_exception( "create needs a hashref" )
2632     unless ref $attrs eq 'HASH';
2633   return $self->new_result($attrs)->insert;
2634 }
2635
2636 =head2 find_or_create
2637
2638 =over 4
2639
2640 =item Arguments: \%vals, \%attrs?
2641
2642 =item Return Value: $rowobject
2643
2644 =back
2645
2646   $cd->cd_to_producer->find_or_create({ producer => $producer },
2647                                       { key => 'primary' });
2648
2649 Tries to find a record based on its primary key or unique constraints; if none
2650 is found, creates one and returns that instead.
2651
2652   my $cd = $schema->resultset('CD')->find_or_create({
2653     cdid   => 5,
2654     artist => 'Massive Attack',
2655     title  => 'Mezzanine',
2656     year   => 2005,
2657   });
2658
2659 Also takes an optional C<key> attribute, to search by a specific key or unique
2660 constraint. For example:
2661
2662   my $cd = $schema->resultset('CD')->find_or_create(
2663     {
2664       artist => 'Massive Attack',
2665       title  => 'Mezzanine',
2666     },
2667     { key => 'cd_artist_title' }
2668   );
2669
2670 B<Note>: Make sure to read the documentation of L</find> and understand the
2671 significance of the C<key> attribute, as its lack may skew your search, and
2672 subsequently result in spurious row creation.
2673
2674 B<Note>: Because find_or_create() reads from the database and then
2675 possibly inserts based on the result, this method is subject to a race
2676 condition. Another process could create a record in the table after
2677 the find has completed and before the create has started. To avoid
2678 this problem, use find_or_create() inside a transaction.
2679
2680 B<Note>: Take care when using C<find_or_create> with a table having
2681 columns with default values that you intend to be automatically
2682 supplied by the database (e.g. an auto_increment primary key column).
2683 In normal usage, the value of such columns should NOT be included at
2684 all in the call to C<find_or_create>, even when set to C<undef>.
2685
2686 See also L</find> and L</update_or_create>. For information on how to declare
2687 unique constraints, see L<DBIx::Class::ResultSource/add_unique_constraint>.
2688
2689 =cut
2690
2691 sub find_or_create {
2692   my $self     = shift;
2693   my $attrs    = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
2694   my $hash     = ref $_[0] eq 'HASH' ? shift : {@_};
2695   if (keys %$hash and my $row = $self->find($hash, $attrs) ) {
2696     return $row;
2697   }
2698   return $self->create($hash);
2699 }
2700
2701 =head2 update_or_create
2702
2703 =over 4
2704
2705 =item Arguments: \%col_values, { key => $unique_constraint }?
2706
2707 =item Return Value: $row_object
2708
2709 =back
2710
2711   $resultset->update_or_create({ col => $val, ... });
2712
2713 Like L</find_or_create>, but if a row is found it is immediately updated via
2714 C<< $found_row->update (\%col_values) >>.
2715
2716
2717 Takes an optional C<key> attribute to search on a specific unique constraint.
2718 For example:
2719
2720   # In your application
2721   my $cd = $schema->resultset('CD')->update_or_create(
2722     {
2723       artist => 'Massive Attack',
2724       title  => 'Mezzanine',
2725       year   => 1998,
2726     },
2727     { key => 'cd_artist_title' }
2728   );
2729
2730   $cd->cd_to_producer->update_or_create({
2731     producer => $producer,
2732     name => 'harry',
2733   }, {
2734     key => 'primary',
2735   });
2736
2737 B<Note>: Make sure to read the documentation of L</find> and understand the
2738 significance of the C<key> attribute, as its lack may skew your search, and
2739 subsequently result in spurious row creation.
2740
2741 B<Note>: Take care when using C<update_or_create> with a table having
2742 columns with default values that you intend to be automatically
2743 supplied by the database (e.g. an auto_increment primary key column).
2744 In normal usage, the value of such columns should NOT be included at
2745 all in the call to C<update_or_create>, even when set to C<undef>.
2746
2747 See also L</find> and L</find_or_create>. For information on how to declare
2748 unique constraints, see L<DBIx::Class::ResultSource/add_unique_constraint>.
2749
2750 =cut
2751
2752 sub update_or_create {
2753   my $self = shift;
2754   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
2755   my $cond = ref $_[0] eq 'HASH' ? shift : {@_};
2756
2757   my $row = $self->find($cond, $attrs);
2758   if (defined $row) {
2759     $row->update($cond);
2760     return $row;
2761   }
2762
2763   return $self->create($cond);
2764 }
2765
2766 =head2 update_or_new
2767
2768 =over 4
2769
2770 =item Arguments: \%col_values, { key => $unique_constraint }?
2771
2772 =item Return Value: $rowobject
2773
2774 =back
2775
2776   $resultset->update_or_new({ col => $val, ... });
2777
2778 Like L</find_or_new> but if a row is found it is immediately updated via
2779 C<< $found_row->update (\%col_values) >>.
2780
2781 For example:
2782
2783   # In your application
2784   my $cd = $schema->resultset('CD')->update_or_new(
2785     {
2786       artist => 'Massive Attack',
2787       title  => 'Mezzanine',
2788       year   => 1998,
2789     },
2790     { key => 'cd_artist_title' }
2791   );
2792
2793   if ($cd->in_storage) {
2794       # the cd was updated
2795   }
2796   else {
2797       # the cd is not yet in the database, let's insert it
2798       $cd->insert;
2799   }
2800
2801 B<Note>: Make sure to read the documentation of L</find> and understand the
2802 significance of the C<key> attribute, as its lack may skew your search, and
2803 subsequently result in spurious new objects.
2804
2805 B<Note>: Take care when using C<update_or_new> with a table having
2806 columns with default values that you intend to be automatically
2807 supplied by the database (e.g. an auto_increment primary key column).
2808 In normal usage, the value of such columns should NOT be included at
2809 all in the call to C<update_or_new>, even when set to C<undef>.
2810
2811 See also L</find>, L</find_or_create> and L</find_or_new>.
2812
2813 =cut
2814
2815 sub update_or_new {
2816     my $self  = shift;
2817     my $attrs = ( @_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {} );
2818     my $cond  = ref $_[0] eq 'HASH' ? shift : {@_};
2819
2820     my $row = $self->find( $cond, $attrs );
2821     if ( defined $row ) {
2822         $row->update($cond);
2823         return $row;
2824     }
2825
2826     return $self->new_result($cond);
2827 }
2828
2829 =head2 get_cache
2830
2831 =over 4
2832
2833 =item Arguments: none
2834
2835 =item Return Value: \@cache_objects | undef
2836
2837 =back
2838
2839 Gets the contents of the cache for the resultset, if the cache is set.
2840
2841 The cache is populated either by using the L</prefetch> attribute to
2842 L</search> or by calling L</set_cache>.
2843
2844 =cut
2845
2846 sub get_cache {
2847   shift->{all_cache};
2848 }
2849
2850 =head2 set_cache
2851
2852 =over 4
2853
2854 =item Arguments: \@cache_objects
2855
2856 =item Return Value: \@cache_objects
2857
2858 =back
2859
2860 Sets the contents of the cache for the resultset. Expects an arrayref
2861 of objects of the same class as those produced by the resultset. Note that
2862 if the cache is set the resultset will return the cached objects rather
2863 than re-querying the database even if the cache attr is not set.
2864
2865 The contents of the cache can also be populated by using the
2866 L</prefetch> attribute to L</search>.
2867
2868 =cut
2869
2870 sub set_cache {
2871   my ( $self, $data ) = @_;
2872   $self->throw_exception("set_cache requires an arrayref")
2873       if defined($data) && (ref $data ne 'ARRAY');
2874   $self->{all_cache} = $data;
2875 }
2876
2877 =head2 clear_cache
2878
2879 =over 4
2880
2881 =item Arguments: none
2882
2883 =item Return Value: undef
2884
2885 =back
2886
2887 Clears the cache for the resultset.
2888
2889 =cut
2890
2891 sub clear_cache {
2892   shift->set_cache(undef);
2893 }
2894
2895 =head2 is_paged
2896
2897 =over 4
2898
2899 =item Arguments: none
2900
2901 =item Return Value: true, if the resultset has been paginated
2902
2903 =back
2904
2905 =cut
2906
2907 sub is_paged {
2908   my ($self) = @_;
2909   return !!$self->{attrs}{page};
2910 }
2911
2912 =head2 is_ordered
2913
2914 =over 4
2915
2916 =item Arguments: none
2917
2918 =item Return Value: true, if the resultset has been ordered with C<order_by>.
2919
2920 =back
2921
2922 =cut
2923
2924 sub is_ordered {
2925   my ($self) = @_;
2926   return scalar $self->result_source->storage->_extract_order_criteria($self->{attrs}{order_by});
2927 }
2928
2929 =head2 related_resultset
2930
2931 =over 4
2932
2933 =item Arguments: $relationship_name
2934
2935 =item Return Value: $resultset
2936
2937 =back
2938
2939 Returns a related resultset for the supplied relationship name.
2940
2941   $artist_rs = $schema->resultset('CD')->related_resultset('Artist');
2942
2943 =cut
2944
2945 sub related_resultset {
2946   my ($self, $rel) = @_;
2947
2948   $self->{related_resultsets} ||= {};
2949   return $self->{related_resultsets}{$rel} ||= do {
2950     my $rsrc = $self->result_source;
2951     my $rel_info = $rsrc->relationship_info($rel);
2952
2953     $self->throw_exception(
2954       "search_related: result source '" . $rsrc->source_name .
2955         "' has no such relationship $rel")
2956       unless $rel_info;
2957
2958     my $attrs = $self->_chain_relationship($rel);
2959
2960     my $join_count = $attrs->{seen_join}{$rel};
2961
2962     my $alias = $self->result_source->storage
2963         ->relname_to_table_alias($rel, $join_count);
2964
2965     # since this is search_related, and we already slid the select window inwards
2966     # (the select/as attrs were deleted in the beginning), we need to flip all
2967     # left joins to inner, so we get the expected results
2968     # read the comment on top of the actual function to see what this does
2969     $attrs->{from} = $rsrc->schema->storage->_inner_join_to_node ($attrs->{from}, $alias);
2970
2971
2972     #XXX - temp fix for result_class bug. There likely is a more elegant fix -groditi
2973     delete @{$attrs}{qw(result_class alias)};
2974
2975     my $new_cache;
2976
2977     if (my $cache = $self->get_cache) {
2978       if ($cache->[0] && $cache->[0]->related_resultset($rel)->get_cache) {
2979         $new_cache = [ map { @{$_->related_resultset($rel)->get_cache} }
2980                         @$cache ];
2981       }
2982     }
2983
2984     my $rel_source = $rsrc->related_source($rel);
2985
2986     my $new = do {
2987
2988       # The reason we do this now instead of passing the alias to the
2989       # search_rs below is that if you wrap/overload resultset on the
2990       # source you need to know what alias it's -going- to have for things
2991       # to work sanely (e.g. RestrictWithObject wants to be able to add
2992       # extra query restrictions, and these may need to be $alias.)
2993
2994       my $rel_attrs = $rel_source->resultset_attributes;
2995       local $rel_attrs->{alias} = $alias;
2996
2997       $rel_source->resultset
2998                  ->search_rs(
2999                      undef, {
3000                        %$attrs,
3001                        where => $attrs->{where},
3002                    });
3003     };
3004     $new->set_cache($new_cache) if $new_cache;
3005     $new;
3006   };
3007 }
3008
3009 =head2 current_source_alias
3010
3011 =over 4
3012
3013 =item Arguments: none
3014
3015 =item Return Value: $source_alias
3016
3017 =back
3018
3019 Returns the current table alias for the result source this resultset is built
3020 on, that will be used in the SQL query. Usually it is C<me>.
3021
3022 Currently the source alias that refers to the result set returned by a
3023 L</search>/L</find> family method depends on how you got to the resultset: it's
3024 C<me> by default, but eg. L</search_related> aliases it to the related result
3025 source name (and keeps C<me> referring to the original result set). The long
3026 term goal is to make L<DBIx::Class> always alias the current resultset as C<me>
3027 (and make this method unnecessary).
3028
3029 Thus it's currently necessary to use this method in predefined queries (see
3030 L<DBIx::Class::Manual::Cookbook/Predefined searches>) when referring to the
3031 source alias of the current result set:
3032
3033   # in a result set class
3034   sub modified_by {
3035     my ($self, $user) = @_;
3036
3037     my $me = $self->current_source_alias;
3038
3039     return $self->search(
3040       "$me.modified" => $user->id,
3041     );
3042   }
3043
3044 =cut
3045
3046 sub current_source_alias {
3047   my ($self) = @_;
3048
3049   return ($self->{attrs} || {})->{alias} || 'me';
3050 }
3051
3052 =head2 as_subselect_rs
3053
3054 =over 4
3055
3056 =item Arguments: none
3057
3058 =item Return Value: $resultset
3059
3060 =back
3061
3062 Act as a barrier to SQL symbols.  The resultset provided will be made into a
3063 "virtual view" by including it as a subquery within the from clause.  From this
3064 point on, any joined tables are inaccessible to ->search on the resultset (as if
3065 it were simply where-filtered without joins).  For example:
3066
3067  my $rs = $schema->resultset('Bar')->search({'x.name' => 'abc'},{ join => 'x' });
3068
3069  # 'x' now pollutes the query namespace
3070
3071  # So the following works as expected
3072  my $ok_rs = $rs->search({'x.other' => 1});
3073
3074  # But this doesn't: instead of finding a 'Bar' related to two x rows (abc and
3075  # def) we look for one row with contradictory terms and join in another table
3076  # (aliased 'x_2') which we never use
3077  my $broken_rs = $rs->search({'x.name' => 'def'});
3078
3079  my $rs2 = $rs->as_subselect_rs;
3080
3081  # doesn't work - 'x' is no longer accessible in $rs2, having been sealed away
3082  my $not_joined_rs = $rs2->search({'x.other' => 1});
3083
3084  # works as expected: finds a 'table' row related to two x rows (abc and def)
3085  my $correctly_joined_rs = $rs2->search({'x.name' => 'def'});
3086
3087 Another example of when one might use this would be to select a subset of
3088 columns in a group by clause:
3089
3090  my $rs = $schema->resultset('Bar')->search(undef, {
3091    group_by => [qw{ id foo_id baz_id }],
3092  })->as_subselect_rs->search(undef, {
3093    columns => [qw{ id foo_id }]
3094  });
3095
3096 In the above example normally columns would have to be equal to the group by,
3097 but because we isolated the group by into a subselect the above works.
3098
3099 =cut
3100
3101 sub as_subselect_rs {
3102   my $self = shift;
3103
3104   my $attrs = $self->_resolved_attrs;
3105
3106   my $fresh_rs = (ref $self)->new (
3107     $self->result_source
3108   );
3109
3110   # these pieces will be locked in the subquery
3111   delete $fresh_rs->{cond};
3112   delete @{$fresh_rs->{attrs}}{qw/where bind/};
3113
3114   return $fresh_rs->search( {}, {
3115     from => [{
3116       $attrs->{alias} => $self->as_query,
3117       -alias  => $attrs->{alias},
3118       -rsrc   => $self->result_source,
3119     }],
3120     alias => $attrs->{alias},
3121   });
3122 }
3123
3124 # This code is called by search_related, and makes sure there
3125 # is clear separation between the joins before, during, and
3126 # after the relationship. This information is needed later
3127 # in order to properly resolve prefetch aliases (any alias
3128 # with a relation_chain_depth less than the depth of the
3129 # current prefetch is not considered)
3130 #
3131 # The increments happen twice per join. An even number means a
3132 # relationship specified via a search_related, whereas an odd
3133 # number indicates a join/prefetch added via attributes
3134 #
3135 # Also this code will wrap the current resultset (the one we
3136 # chain to) in a subselect IFF it contains limiting attributes
3137 sub _chain_relationship {
3138   my ($self, $rel) = @_;
3139   my $source = $self->result_source;
3140   my $attrs = { %{$self->{attrs}||{}} };
3141
3142   # we need to take the prefetch the attrs into account before we
3143   # ->_resolve_join as otherwise they get lost - captainL
3144   my $join = $self->_merge_joinpref_attr( $attrs->{join}, $attrs->{prefetch} );
3145
3146   delete @{$attrs}{qw/join prefetch collapse group_by distinct select as columns +select +as +columns/};
3147
3148   my $seen = { %{ (delete $attrs->{seen_join}) || {} } };
3149
3150   my $from;
3151   my @force_subq_attrs = qw/offset rows group_by having/;
3152
3153   if (
3154     ($attrs->{from} && ref $attrs->{from} ne 'ARRAY')
3155       ||
3156     $self->_has_resolved_attr (@force_subq_attrs)
3157   ) {
3158     # Nuke the prefetch (if any) before the new $rs attrs
3159     # are resolved (prefetch is useless - we are wrapping
3160     # a subquery anyway).
3161     my $rs_copy = $self->search;
3162     $rs_copy->{attrs}{join} = $self->_merge_joinpref_attr (
3163       $rs_copy->{attrs}{join},
3164       delete $rs_copy->{attrs}{prefetch},
3165     );
3166
3167     $from = [{
3168       -rsrc   => $source,
3169       -alias  => $attrs->{alias},
3170       $attrs->{alias} => $rs_copy->as_query,
3171     }];
3172     delete @{$attrs}{@force_subq_attrs, qw/where bind/};
3173     $seen->{-relation_chain_depth} = 0;
3174   }
3175   elsif ($attrs->{from}) {  #shallow copy suffices
3176     $from = [ @{$attrs->{from}} ];
3177   }
3178   else {
3179     $from = [{
3180       -rsrc  => $source,
3181       -alias => $attrs->{alias},
3182       $attrs->{alias} => $source->from,
3183     }];
3184   }
3185
3186   my $jpath = ($seen->{-relation_chain_depth})
3187     ? $from->[-1][0]{-join_path}
3188     : [];
3189
3190   my @requested_joins = $source->_resolve_join(
3191     $join,
3192     $attrs->{alias},
3193     $seen,
3194     $jpath,
3195   );
3196
3197   push @$from, @requested_joins;
3198
3199   $seen->{-relation_chain_depth}++;
3200
3201   # if $self already had a join/prefetch specified on it, the requested
3202   # $rel might very well be already included. What we do in this case
3203   # is effectively a no-op (except that we bump up the chain_depth on
3204   # the join in question so we could tell it *is* the search_related)
3205   my $already_joined;
3206
3207   # we consider the last one thus reverse
3208   for my $j (reverse @requested_joins) {
3209     my ($last_j) = keys %{$j->[0]{-join_path}[-1]};
3210     if ($rel eq $last_j) {
3211       $j->[0]{-relation_chain_depth}++;
3212       $already_joined++;
3213       last;
3214     }
3215   }
3216
3217   unless ($already_joined) {
3218     push @$from, $source->_resolve_join(
3219       $rel,
3220       $attrs->{alias},
3221       $seen,
3222       $jpath,
3223     );
3224   }
3225
3226   $seen->{-relation_chain_depth}++;
3227
3228   return {%$attrs, from => $from, seen_join => $seen};
3229 }
3230
3231 # too many times we have to do $attrs = { %{$self->_resolved_attrs} }
3232 sub _resolved_attrs_copy {
3233   my $self = shift;
3234   return { %{$self->_resolved_attrs (@_)} };
3235 }
3236
3237 sub _resolved_attrs {
3238   my $self = shift;
3239   return $self->{_attrs} if $self->{_attrs};
3240
3241   my $attrs  = { %{ $self->{attrs} || {} } };
3242   my $source = $self->result_source;
3243   my $alias  = $attrs->{alias};
3244
3245   # one last pass of normalization
3246   $self->_normalize_selection($attrs);
3247
3248   # default selection list
3249   $attrs->{columns} = [ $source->columns ]
3250     unless List::Util::first { exists $attrs->{$_} } qw/columns cols select as _trailing_select/;
3251
3252   # merge selectors together
3253   for (qw/columns select as _trailing_select/) {
3254     $attrs->{$_} = $self->_merge_attr($attrs->{$_}, $attrs->{"+$_"})
3255       if $attrs->{$_} or $attrs->{"+$_"};
3256   }
3257
3258   # disassemble columns
3259   my (@sel, @as);
3260   if (my $cols = delete $attrs->{columns}) {
3261     for my $c (ref $cols eq 'ARRAY' ? @$cols : $cols) {
3262       if (ref $c eq 'HASH') {
3263         for my $as (keys %$c) {
3264           push @sel, $c->{$as};
3265           push @as, $as;
3266         }
3267       }
3268       else {
3269         push @sel, $c;
3270         push @as, $c;
3271       }
3272     }
3273   }
3274
3275   # when trying to weed off duplicates later do not go past this point -
3276   # everything added from here on is unbalanced "anyone's guess" stuff
3277   my $dedup_stop_idx = $#as;
3278
3279   push @as, @{ ref $attrs->{as} eq 'ARRAY' ? $attrs->{as} : [ $attrs->{as} ] }
3280     if $attrs->{as};
3281   push @sel, @{ ref $attrs->{select} eq 'ARRAY' ? $attrs->{select} : [ $attrs->{select} ] }
3282     if $attrs->{select};
3283
3284   # assume all unqualified selectors to apply to the current alias (legacy stuff)
3285   for (@sel) {
3286     $_ = (ref $_ or $_ =~ /\./) ? $_ : "$alias.$_";
3287   }
3288
3289   # disqualify all $alias.col as-bits (collapser mandated)
3290   for (@as) {
3291     $_ = ($_ =~ /^\Q$alias.\E(.+)$/) ? $1 : $_;
3292   }
3293
3294   # de-duplicate the result (remove *identical* select/as pairs)
3295   # and also die on duplicate {as} pointing to different {select}s
3296   # not using a c-style for as the condition is prone to shrinkage
3297   my $seen;
3298   my $i = 0;
3299   while ($i <= $dedup_stop_idx) {
3300     if ($seen->{"$sel[$i] \x00\x00 $as[$i]"}++) {
3301       splice @sel, $i, 1;
3302       splice @as, $i, 1;
3303       $dedup_stop_idx--;
3304     }
3305     elsif ($seen->{$as[$i]}++) {
3306       $self->throw_exception(
3307         "inflate_result() alias '$as[$i]' specified twice with different SQL-side {select}-ors"
3308       );
3309     }
3310     else {
3311       $i++;
3312     }
3313   }
3314
3315   $attrs->{select} = \@sel;
3316   $attrs->{as} = \@as;
3317
3318   $attrs->{from} ||= [{
3319     -rsrc   => $source,
3320     -alias  => $self->{attrs}{alias},
3321     $self->{attrs}{alias} => $source->from,
3322   }];
3323
3324   if ( $attrs->{join} || $attrs->{prefetch} ) {
3325
3326     $self->throw_exception ('join/prefetch can not be used with a custom {from}')
3327       if ref $attrs->{from} ne 'ARRAY';
3328
3329     my $join = (delete $attrs->{join}) || {};
3330
3331     if ( defined $attrs->{prefetch} ) {
3332       $join = $self->_merge_joinpref_attr( $join, $attrs->{prefetch} );
3333     }
3334
3335     $attrs->{from} =    # have to copy here to avoid corrupting the original
3336       [
3337         @{ $attrs->{from} },
3338         $source->_resolve_join(
3339           $join,
3340           $alias,
3341           { %{ $attrs->{seen_join} || {} } },
3342           ( $attrs->{seen_join} && keys %{$attrs->{seen_join}})
3343             ? $attrs->{from}[-1][0]{-join_path}
3344             : []
3345           ,
3346         )
3347       ];
3348   }
3349
3350   if ( defined $attrs->{order_by} ) {
3351     $attrs->{order_by} = (
3352       ref( $attrs->{order_by} ) eq 'ARRAY'
3353       ? [ @{ $attrs->{order_by} } ]
3354       : [ $attrs->{order_by} || () ]
3355     );
3356   }
3357
3358   if ($attrs->{group_by} and ref $attrs->{group_by} ne 'ARRAY') {
3359     $attrs->{group_by} = [ $attrs->{group_by} ];
3360   }
3361
3362   # generate the distinct induced group_by early, as prefetch will be carried via a
3363   # subquery (since a group_by is present)
3364   if (delete $attrs->{distinct}) {
3365     if ($attrs->{group_by}) {
3366       carp ("Useless use of distinct on a grouped resultset ('distinct' is ignored when a 'group_by' is present)");
3367     }
3368     else {
3369       # distinct affects only the main selection part, not what prefetch may
3370       # add below. However trailing is not yet a part of the selection as
3371       # prefetch must insert before it
3372       $attrs->{group_by} = $source->storage->_group_over_selection (
3373         $attrs->{from},
3374         [ @{$attrs->{select}||[]}, @{$attrs->{_trailing_select}||[]} ],
3375         $attrs->{order_by},
3376       );
3377     }
3378   }
3379
3380   $attrs->{collapse} ||= {};
3381   if ($attrs->{prefetch}) {
3382     my $prefetch = $self->_merge_joinpref_attr( {}, delete $attrs->{prefetch} );
3383
3384     my $prefetch_ordering = [];
3385
3386     # this is a separate structure (we don't look in {from} directly)
3387     # as the resolver needs to shift things off the lists to work
3388     # properly (identical-prefetches on different branches)
3389     my $join_map = {};
3390     if (ref $attrs->{from} eq 'ARRAY') {
3391
3392       my $start_depth = $attrs->{seen_join}{-relation_chain_depth} || 0;
3393
3394       for my $j ( @{$attrs->{from}}[1 .. $#{$attrs->{from}} ] ) {
3395         next unless $j->[0]{-alias};
3396         next unless $j->[0]{-join_path};
3397         next if ($j->[0]{-relation_chain_depth} || 0) < $start_depth;
3398
3399         my @jpath = map { keys %$_ } @{$j->[0]{-join_path}};
3400
3401         my $p = $join_map;
3402         $p = $p->{$_} ||= {} for @jpath[ ($start_depth/2) .. $#jpath]; #only even depths are actual jpath boundaries
3403         push @{$p->{-join_aliases} }, $j->[0]{-alias};
3404       }
3405     }
3406
3407     my @prefetch =
3408       $source->_resolve_prefetch( $prefetch, $alias, $join_map, $prefetch_ordering, $attrs->{collapse} );
3409
3410     # we need to somehow mark which columns came from prefetch
3411     if (@prefetch) {
3412       my $sel_end = $#{$attrs->{select}};
3413       $attrs->{_prefetch_selector_range} = [ $sel_end + 1, $sel_end + @prefetch ];
3414     }
3415
3416     push @{ $attrs->{select} }, (map { $_->[0] } @prefetch);
3417     push @{ $attrs->{as} }, (map { $_->[1] } @prefetch);
3418
3419     push( @{$attrs->{order_by}}, @$prefetch_ordering );
3420     $attrs->{_collapse_order_by} = \@$prefetch_ordering;
3421   }
3422
3423
3424   push @{ $attrs->{select} }, @{$attrs->{_trailing_select}}
3425     if $attrs->{_trailing_select};
3426
3427   # if both page and offset are specified, produce a combined offset
3428   # even though it doesn't make much sense, this is what pre 081xx has
3429   # been doing
3430   if (my $page = delete $attrs->{page}) {
3431     $attrs->{offset} =
3432       ($attrs->{rows} * ($page - 1))
3433             +
3434       ($attrs->{offset} || 0)
3435     ;
3436   }
3437
3438   return $self->{_attrs} = $attrs;
3439 }
3440
3441 sub _rollout_attr {
3442   my ($self, $attr) = @_;
3443
3444   if (ref $attr eq 'HASH') {
3445     return $self->_rollout_hash($attr);
3446   } elsif (ref $attr eq 'ARRAY') {
3447     return $self->_rollout_array($attr);
3448   } else {
3449     return [$attr];
3450   }
3451 }
3452
3453 sub _rollout_array {
3454   my ($self, $attr) = @_;
3455
3456   my @rolled_array;
3457   foreach my $element (@{$attr}) {
3458     if (ref $element eq 'HASH') {
3459       push( @rolled_array, @{ $self->_rollout_hash( $element ) } );
3460     } elsif (ref $element eq 'ARRAY') {
3461       #  XXX - should probably recurse here
3462       push( @rolled_array, @{$self->_rollout_array($element)} );
3463     } else {
3464       push( @rolled_array, $element );
3465     }
3466   }
3467   return \@rolled_array;
3468 }
3469
3470 sub _rollout_hash {
3471   my ($self, $attr) = @_;
3472
3473   my @rolled_array;
3474   foreach my $key (keys %{$attr}) {
3475     push( @rolled_array, { $key => $attr->{$key} } );
3476   }
3477   return \@rolled_array;
3478 }
3479
3480 sub _calculate_score {
3481   my ($self, $a, $b) = @_;
3482
3483   if (defined $a xor defined $b) {
3484     return 0;
3485   }
3486   elsif (not defined $a) {
3487     return 1;
3488   }
3489
3490   if (ref $b eq 'HASH') {
3491     my ($b_key) = keys %{$b};
3492     if (ref $a eq 'HASH') {
3493       my ($a_key) = keys %{$a};
3494       if ($a_key eq $b_key) {
3495         return (1 + $self->_calculate_score( $a->{$a_key}, $b->{$b_key} ));
3496       } else {
3497         return 0;
3498       }
3499     } else {
3500       return ($a eq $b_key) ? 1 : 0;
3501     }
3502   } else {
3503     if (ref $a eq 'HASH') {
3504       my ($a_key) = keys %{$a};
3505       return ($b eq $a_key) ? 1 : 0;
3506     } else {
3507       return ($b eq $a) ? 1 : 0;
3508     }
3509   }
3510 }
3511
3512 sub _merge_joinpref_attr {
3513   my ($self, $orig, $import) = @_;
3514
3515   return $import unless defined($orig);
3516   return $orig unless defined($import);
3517
3518   $orig = $self->_rollout_attr($orig);
3519   $import = $self->_rollout_attr($import);
3520
3521   my $seen_keys;
3522   foreach my $import_element ( @{$import} ) {
3523     # find best candidate from $orig to merge $b_element into
3524     my $best_candidate = { position => undef, score => 0 }; my $position = 0;
3525     foreach my $orig_element ( @{$orig} ) {
3526       my $score = $self->_calculate_score( $orig_element, $import_element );
3527       if ($score > $best_candidate->{score}) {
3528         $best_candidate->{position} = $position;
3529         $best_candidate->{score} = $score;
3530       }
3531       $position++;
3532     }
3533     my ($import_key) = ( ref $import_element eq 'HASH' ) ? keys %{$import_element} : ($import_element);
3534
3535     if ($best_candidate->{score} == 0 || exists $seen_keys->{$import_key}) {
3536       push( @{$orig}, $import_element );
3537     } else {
3538       my $orig_best = $orig->[$best_candidate->{position}];
3539       # merge orig_best and b_element together and replace original with merged
3540       if (ref $orig_best ne 'HASH') {
3541         $orig->[$best_candidate->{position}] = $import_element;
3542       } elsif (ref $import_element eq 'HASH') {
3543         my ($key) = keys %{$orig_best};
3544         $orig->[$best_candidate->{position}] = { $key => $self->_merge_joinpref_attr($orig_best->{$key}, $import_element->{$key}) };
3545       }
3546     }
3547     $seen_keys->{$import_key} = 1; # don't merge the same key twice
3548   }
3549
3550   return $orig;
3551 }
3552
3553 {
3554   my $hm;
3555
3556   sub _merge_attr {
3557     $hm ||= do {
3558       require Hash::Merge;
3559       my $hm = Hash::Merge->new;
3560
3561       $hm->specify_behavior({
3562         SCALAR => {
3563           SCALAR => sub {
3564             my ($defl, $defr) = map { defined $_ } (@_[0,1]);
3565
3566             if ($defl xor $defr) {
3567               return [ $defl ? $_[0] : $_[1] ];
3568             }
3569             elsif (! $defl) {
3570               return [];
3571             }
3572             elsif (__HM_DEDUP and $_[0] eq $_[1]) {
3573               return [ $_[0] ];
3574             }
3575             else {
3576               return [$_[0], $_[1]];
3577             }
3578           },
3579           ARRAY => sub {
3580             return $_[1] if !defined $_[0];
3581             return $_[1] if __HM_DEDUP and List::Util::first { $_ eq $_[0] } @{$_[1]};
3582             return [$_[0], @{$_[1]}]
3583           },
3584           HASH  => sub {
3585             return [] if !defined $_[0] and !keys %{$_[1]};
3586             return [ $_[1] ] if !defined $_[0];
3587             return [ $_[0] ] if !keys %{$_[1]};
3588             return [$_[0], $_[1]]
3589           },
3590         },
3591         ARRAY => {
3592           SCALAR => sub {
3593             return $_[0] if !defined $_[1];
3594             return $_[0] if __HM_DEDUP and List::Util::first { $_ eq $_[1] } @{$_[0]};
3595             return [@{$_[0]}, $_[1]]
3596           },
3597           ARRAY => sub {
3598             my @ret = @{$_[0]} or return $_[1];
3599             return [ @ret, @{$_[1]} ] unless __HM_DEDUP;
3600             my %idx = map { $_ => 1 } @ret;
3601             push @ret, grep { ! defined $idx{$_} } (@{$_[1]});
3602             \@ret;
3603           },
3604           HASH => sub {
3605             return [ $_[1] ] if ! @{$_[0]};
3606             return $_[0] if !keys %{$_[1]};
3607             return $_[0] if __HM_DEDUP and List::Util::first { $_ eq $_[1] } @{$_[0]};
3608             return [ @{$_[0]}, $_[1] ];
3609           },
3610         },
3611         HASH => {
3612           SCALAR => sub {
3613             return [] if !keys %{$_[0]} and !defined $_[1];
3614             return [ $_[0] ] if !defined $_[1];
3615             return [ $_[1] ] if !keys %{$_[0]};
3616             return [$_[0], $_[1]]
3617           },
3618           ARRAY => sub {
3619             return [] if !keys %{$_[0]} and !@{$_[1]};
3620             return [ $_[0] ] if !@{$_[1]};
3621             return $_[1] if !keys %{$_[0]};
3622             return $_[1] if __HM_DEDUP and List::Util::first { $_ eq $_[0] } @{$_[1]};
3623             return [ $_[0], @{$_[1]} ];
3624           },
3625           HASH => sub {
3626             return [] if !keys %{$_[0]} and !keys %{$_[1]};
3627             return [ $_[0] ] if !keys %{$_[1]};
3628             return [ $_[1] ] if !keys %{$_[0]};
3629             return [ $_[0] ] if $_[0] eq $_[1];
3630             return [ $_[0], $_[1] ];
3631           },
3632         }
3633       } => 'DBIC_RS_ATTR_MERGER');
3634       $hm;
3635     };
3636
3637     return $hm->merge ($_[1], $_[2]);
3638   }
3639 }
3640
3641 sub STORABLE_freeze {
3642   my ($self, $cloning) = @_;
3643   my $to_serialize = { %$self };
3644
3645   # A cursor in progress can't be serialized (and would make little sense anyway)
3646   delete $to_serialize->{cursor};
3647
3648   Storable::nfreeze($to_serialize);
3649 }
3650
3651 # need this hook for symmetry
3652 sub STORABLE_thaw {
3653   my ($self, $cloning, $serialized) = @_;
3654
3655   %$self = %{ Storable::thaw($serialized) };
3656
3657   $self;
3658 }
3659
3660
3661 =head2 throw_exception
3662
3663 See L<DBIx::Class::Schema/throw_exception> for details.
3664
3665 =cut
3666
3667 sub throw_exception {
3668   my $self=shift;
3669
3670   if (ref $self and my $rsrc = $self->result_source) {
3671     $rsrc->throw_exception(@_)
3672   }
3673   else {
3674     DBIx::Class::Exception->throw(@_);
3675   }
3676 }
3677
3678 # XXX: FIXME: Attributes docs need clearing up
3679
3680 =head1 ATTRIBUTES
3681
3682 Attributes are used to refine a ResultSet in various ways when
3683 searching for data. They can be passed to any method which takes an
3684 C<\%attrs> argument. See L</search>, L</search_rs>, L</find>,
3685 L</count>.
3686
3687 These are in no particular order:
3688
3689 =head2 order_by
3690
3691 =over 4
3692
3693 =item Value: ( $order_by | \@order_by | \%order_by )
3694
3695 =back
3696
3697 Which column(s) to order the results by.
3698
3699 [The full list of suitable values is documented in
3700 L<SQL::Abstract/"ORDER BY CLAUSES">; the following is a summary of
3701 common options.]
3702
3703 If a single column name, or an arrayref of names is supplied, the
3704 argument is passed through directly to SQL. The hashref syntax allows
3705 for connection-agnostic specification of ordering direction:
3706
3707  For descending order:
3708
3709   order_by => { -desc => [qw/col1 col2 col3/] }
3710
3711  For explicit ascending order:
3712
3713   order_by => { -asc => 'col' }
3714
3715 The old scalarref syntax (i.e. order_by => \'year DESC') is still
3716 supported, although you are strongly encouraged to use the hashref
3717 syntax as outlined above.
3718
3719 =head2 columns
3720
3721 =over 4
3722
3723 =item Value: \@columns
3724
3725 =back
3726
3727 Shortcut to request a particular set of columns to be retrieved. Each
3728 column spec may be a string (a table column name), or a hash (in which
3729 case the key is the C<as> value, and the value is used as the C<select>
3730 expression). Adds C<me.> onto the start of any column without a C<.> in
3731 it and sets C<select> from that, then auto-populates C<as> from
3732 C<select> as normal. (You may also use the C<cols> attribute, as in
3733 earlier versions of DBIC.)
3734
3735 Essentially C<columns> does the same as L</select> and L</as>.
3736
3737     columns => [ 'foo', { bar => 'baz' } ]
3738
3739 is the same as
3740
3741     select => [qw/foo baz/],
3742     as => [qw/foo bar/]
3743
3744 =head2 +columns
3745
3746 =over 4
3747
3748 =item Value: \@columns
3749
3750 =back
3751
3752 Indicates additional columns to be selected from storage. Works the same
3753 as L</columns> but adds columns to the selection. (You may also use the
3754 C<include_columns> attribute, as in earlier versions of DBIC). For
3755 example:-
3756
3757   $schema->resultset('CD')->search(undef, {
3758     '+columns' => ['artist.name'],
3759     join => ['artist']
3760   });
3761
3762 would return all CDs and include a 'name' column to the information
3763 passed to object inflation. Note that the 'artist' is the name of the
3764 column (or relationship) accessor, and 'name' is the name of the column
3765 accessor in the related table.
3766
3767 =head2 include_columns
3768
3769 =over 4
3770
3771 =item Value: \@columns
3772
3773 =back
3774
3775 Deprecated.  Acts as a synonym for L</+columns> for backward compatibility.
3776
3777 =head2 select
3778
3779 =over 4
3780
3781 =item Value: \@select_columns
3782
3783 =back
3784
3785 Indicates which columns should be selected from the storage. You can use
3786 column names, or in the case of RDBMS back ends, function or stored procedure
3787 names:
3788
3789   $rs = $schema->resultset('Employee')->search(undef, {
3790     select => [
3791       'name',
3792       { count => 'employeeid' },
3793       { max => { length => 'name' }, -as => 'longest_name' }
3794     ]
3795   });
3796
3797   # Equivalent SQL
3798   SELECT name, COUNT( employeeid ), MAX( LENGTH( name ) ) AS longest_name FROM employee
3799
3800 B<NOTE:> You will almost always need a corresponding L</as> attribute when you
3801 use L</select>, to instruct DBIx::Class how to store the result of the column.
3802 Also note that the L</as> attribute has nothing to do with the SQL-side 'AS'
3803 identifier aliasing. You can however alias a function, so you can use it in
3804 e.g. an C<ORDER BY> clause. This is done via the C<-as> B<select function
3805 attribute> supplied as shown in the example above.
3806
3807 =head2 +select
3808
3809 =over 4
3810
3811 Indicates additional columns to be selected from storage.  Works the same as
3812 L</select> but adds columns to the default selection, instead of specifying
3813 an explicit list.
3814
3815 =back
3816
3817 =head2 +as
3818
3819 =over 4
3820
3821 Indicates additional column names for those added via L</+select>. See L</as>.
3822
3823 =back
3824
3825 =head2 as
3826
3827 =over 4
3828
3829 =item Value: \@inflation_names
3830
3831 =back
3832
3833 Indicates column names for object inflation. That is L</as> indicates the
3834 slot name in which the column value will be stored within the
3835 L<Row|DBIx::Class::Row> object. The value will then be accessible via this
3836 identifier by the C<get_column> method (or via the object accessor B<if one
3837 with the same name already exists>) as shown below. The L</as> attribute has
3838 B<nothing to do> with the SQL-side C<AS>. See L</select> for details.
3839
3840   $rs = $schema->resultset('Employee')->search(undef, {
3841     select => [
3842       'name',
3843       { count => 'employeeid' },
3844       { max => { length => 'name' }, -as => 'longest_name' }
3845     ],
3846     as => [qw/
3847       name
3848       employee_count
3849       max_name_length
3850     /],
3851   });
3852
3853 If the object against which the search is performed already has an accessor
3854 matching a column name specified in C<as>, the value can be retrieved using
3855 the accessor as normal:
3856
3857   my $name = $employee->name();
3858
3859 If on the other hand an accessor does not exist in the object, you need to
3860 use C<get_column> instead:
3861
3862   my $employee_count = $employee->get_column('employee_count');
3863
3864 You can create your own accessors if required - see
3865 L<DBIx::Class::Manual::Cookbook> for details.
3866
3867 =head2 join
3868
3869 =over 4
3870
3871 =item Value: ($rel_name | \@rel_names | \%rel_names)
3872
3873 =back
3874
3875 Contains a list of relationships that should be joined for this query.  For
3876 example:
3877
3878   # Get CDs by Nine Inch Nails
3879   my $rs = $schema->resultset('CD')->search(
3880     { 'artist.name' => 'Nine Inch Nails' },
3881     { join => 'artist' }
3882   );
3883
3884 Can also contain a hash reference to refer to the other relation's relations.
3885 For example:
3886
3887   package MyApp::Schema::Track;
3888   use base qw/DBIx::Class/;
3889   __PACKAGE__->table('track');
3890   __PACKAGE__->add_columns(qw/trackid cd position title/);
3891   __PACKAGE__->set_primary_key('trackid');
3892   __PACKAGE__->belongs_to(cd => 'MyApp::Schema::CD');
3893   1;
3894
3895   # In your application
3896   my $rs = $schema->resultset('Artist')->search(
3897     { 'track.title' => 'Teardrop' },
3898     {
3899       join     => { cd => 'track' },
3900       order_by => 'artist.name',
3901     }
3902   );
3903
3904 You need to use the relationship (not the table) name in  conditions,
3905 because they are aliased as such. The current table is aliased as "me", so
3906 you need to use me.column_name in order to avoid ambiguity. For example:
3907
3908   # Get CDs from 1984 with a 'Foo' track
3909   my $rs = $schema->resultset('CD')->search(
3910     {
3911       'me.year' => 1984,
3912       'tracks.name' => 'Foo'
3913     },
3914     { join => 'tracks' }
3915   );
3916
3917 If the same join is supplied twice, it will be aliased to <rel>_2 (and
3918 similarly for a third time). For e.g.
3919
3920   my $rs = $schema->resultset('Artist')->search({
3921     'cds.title'   => 'Down to Earth',
3922     'cds_2.title' => 'Popular',
3923   }, {
3924     join => [ qw/cds cds/ ],
3925   });
3926
3927 will return a set of all artists that have both a cd with title 'Down
3928 to Earth' and a cd with title 'Popular'.
3929
3930 If you want to fetch related objects from other tables as well, see C<prefetch>
3931 below.
3932
3933 For more help on using joins with search, see L<DBIx::Class::Manual::Joining>.
3934
3935 =head2 prefetch
3936
3937 =over 4
3938
3939 =item Value: ($rel_name | \@rel_names | \%rel_names)
3940
3941 =back
3942
3943 Contains one or more relationships that should be fetched along with
3944 the main query (when they are accessed afterwards the data will
3945 already be available, without extra queries to the database).  This is
3946 useful for when you know you will need the related objects, because it
3947 saves at least one query:
3948
3949   my $rs = $schema->resultset('Tag')->search(
3950     undef,
3951     {
3952       prefetch => {
3953         cd => 'artist'
3954       }
3955     }
3956   );
3957
3958 The initial search results in SQL like the following:
3959
3960   SELECT tag.*, cd.*, artist.* FROM tag
3961   JOIN cd ON tag.cd = cd.cdid
3962   JOIN artist ON cd.artist = artist.artistid
3963
3964 L<DBIx::Class> has no need to go back to the database when we access the
3965 C<cd> or C<artist> relationships, which saves us two SQL statements in this
3966 case.
3967
3968 Simple prefetches will be joined automatically, so there is no need
3969 for a C<join> attribute in the above search.
3970
3971 C<prefetch> can be used with the following relationship types: C<belongs_to>,
3972 C<has_one> (or if you're using C<add_relationship>, any relationship declared
3973 with an accessor type of 'single' or 'filter'). A more complex example that
3974 prefetches an artists cds, the tracks on those cds, and the tags associated
3975 with that artist is given below (assuming many-to-many from artists to tags):
3976
3977  my $rs = $schema->resultset('Artist')->search(
3978    undef,
3979    {
3980      prefetch => [
3981        { cds => 'tracks' },
3982        { artist_tags => 'tags' }
3983      ]
3984    }
3985  );
3986
3987
3988 B<NOTE:> If you specify a C<prefetch> attribute, the C<join> and C<select>
3989 attributes will be ignored.
3990
3991 B<CAVEATs>: Prefetch does a lot of deep magic. As such, it may not behave
3992 exactly as you might expect.
3993
3994 =over 4
3995
3996 =item *
3997
3998 Prefetch uses the L</cache> to populate the prefetched relationships. This
3999 may or may not be what you want.
4000
4001 =item *
4002
4003 If you specify a condition on a prefetched relationship, ONLY those
4004 rows that match the prefetched condition will be fetched into that relationship.
4005 This means that adding prefetch to a search() B<may alter> what is returned by
4006 traversing a relationship. So, if you have C<< Artist->has_many(CDs) >> and you do
4007
4008   my $artist_rs = $schema->resultset('Artist')->search({
4009       'cds.year' => 2008,
4010   }, {
4011       join => 'cds',
4012   });
4013
4014   my $count = $artist_rs->first->cds->count;
4015
4016   my $artist_rs_prefetch = $artist_rs->search( {}, { prefetch => 'cds' } );
4017
4018   my $prefetch_count = $artist_rs_prefetch->first->cds->count;
4019
4020   cmp_ok( $count, '==', $prefetch_count, "Counts should be the same" );
4021
4022 that cmp_ok() may or may not pass depending on the datasets involved. This
4023 behavior may or may not survive the 0.09 transition.
4024
4025 =back
4026
4027 =head2 page
4028
4029 =over 4
4030
4031 =item Value: $page
4032
4033 =back
4034
4035 Makes the resultset paged and specifies the page to retrieve. Effectively
4036 identical to creating a non-pages resultset and then calling ->page($page)
4037 on it.
4038
4039 If L</rows> attribute is not specified it defaults to 10 rows per page.
4040
4041 When you have a paged resultset, L</count> will only return the number
4042 of rows in the page. To get the total, use the L</pager> and call
4043 C<total_entries> on it.
4044
4045 =head2 rows
4046
4047 =over 4
4048
4049 =item Value: $rows
4050
4051 =back
4052
4053 Specifies the maximum number of rows for direct retrieval or the number of
4054 rows per page if the page attribute or method is used.
4055
4056 =head2 offset
4057
4058 =over 4
4059
4060 =item Value: $offset
4061
4062 =back
4063
4064 Specifies the (zero-based) row number for the  first row to be returned, or the
4065 of the first row of the first page if paging is used.
4066
4067 =head2 group_by
4068
4069 =over 4
4070
4071 =item Value: \@columns
4072
4073 =back
4074
4075 A arrayref of columns to group by. Can include columns of joined tables.
4076
4077   group_by => [qw/ column1 column2 ... /]
4078
4079 =head2 having
4080
4081 =over 4
4082
4083 =item Value: $condition
4084
4085 =back
4086
4087 HAVING is a select statement attribute that is applied between GROUP BY and
4088 ORDER BY. It is applied to the after the grouping calculations have been
4089 done.
4090
4091   having => { 'count_employee' => { '>=', 100 } }
4092
4093 or with an in-place function in which case literal SQL is required:
4094
4095   having => \[ 'count(employee) >= ?', [ count => 100 ] ]
4096
4097 =head2 distinct
4098
4099 =over 4
4100
4101 =item Value: (0 | 1)
4102
4103 =back
4104
4105 Set to 1 to group by all columns. If the resultset already has a group_by
4106 attribute, this setting is ignored and an appropriate warning is issued.
4107
4108 =head2 where
4109
4110 =over 4
4111
4112 Adds to the WHERE clause.
4113
4114   # only return rows WHERE deleted IS NULL for all searches
4115   __PACKAGE__->resultset_attributes({ where => { deleted => undef } }); )
4116
4117 Can be overridden by passing C<< { where => undef } >> as an attribute
4118 to a resultset.
4119
4120 =back
4121
4122 =head2 cache
4123
4124 Set to 1 to cache search results. This prevents extra SQL queries if you
4125 revisit rows in your ResultSet:
4126
4127   my $resultset = $schema->resultset('Artist')->search( undef, { cache => 1 } );
4128
4129   while( my $artist = $resultset->next ) {
4130     ... do stuff ...
4131   }
4132
4133   $rs->first; # without cache, this would issue a query
4134
4135 By default, searches are not cached.
4136
4137 For more examples of using these attributes, see
4138 L<DBIx::Class::Manual::Cookbook>.
4139
4140 =head2 for
4141
4142 =over 4
4143
4144 =item Value: ( 'update' | 'shared' )
4145
4146 =back
4147
4148 Set to 'update' for a SELECT ... FOR UPDATE or 'shared' for a SELECT
4149 ... FOR SHARED.
4150
4151 =cut
4152
4153 1;