lib/DBIx/Class/ResultSet.pm

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