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