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'   => sub { 1; },
   8         fallback => 1;
   9 use Carp::Clan qw/^DBIx::Class/;
  10 use Data::Page;
  11 use Storable;
  12 use DBIx::Class::ResultSetColumn;
  13 use base qw/DBIx::Class/;
  14
  15 __PACKAGE__->load_components(qw/AccessorGroup/);
  16 __PACKAGE__->mk_group_accessors('simple' => qw/result_source result_class/);
  17
  18 =head1 NAME
  19
  20 DBIx::Class::ResultSet - Responsible for fetching and creating resultset.
  21
  22 =head1 SYNOPSIS
  23
  24   my $rs   = $schema->resultset('User')->search(registered => 1);
  25   my @rows = $schema->resultset('CD')->search(year => 2005);
  26
  27 =head1 DESCRIPTION
  28
  29 The resultset is also known as an iterator. It is responsible for handling
  30 queries that may return an arbitrary number of rows, e.g. via L</search>
  31 or a C<has_many> relationship.
  32
  33 In the examples below, the following table classes are used:
  34
  35   package MyApp::Schema::Artist;
  36   use base qw/DBIx::Class/;
  37   __PACKAGE__->load_components(qw/Core/);
  38   __PACKAGE__->table('artist');
  39   __PACKAGE__->add_columns(qw/artistid name/);
  40   __PACKAGE__->set_primary_key('artistid');
  41   __PACKAGE__->has_many(cds => 'MyApp::Schema::CD');
  42   1;
  43
  44   package MyApp::Schema::CD;
  45   use base qw/DBIx::Class/;
  46   __PACKAGE__->load_components(qw/Core/);
  47   __PACKAGE__->table('cd');
  48   __PACKAGE__->add_columns(qw/cdid artist title year/);
  49   __PACKAGE__->set_primary_key('cdid');
  50   __PACKAGE__->belongs_to(artist => 'MyApp::Schema::Artist');
  51   1;
  52
  53 =head1 METHODS
  54
  55 =head2 new
  56
  57 =over 4
  58
  59 =item Arguments: $source, \%$attrs
  60
  61 =item Return Value: $rs
  62
  63 =back
  64
  65 The resultset constructor. Takes a source object (usually a
  66 L<DBIx::Class::ResultSourceProxy::Table>) and an attribute hash (see
  67 L</ATTRIBUTES> below).  Does not perform any queries -- these are
  68 executed as needed by the other methods.
  69
  70 Generally you won't need to construct a resultset manually.  You'll
  71 automatically get one from e.g. a L</search> called in scalar context:
  72
  73   my $rs = $schema->resultset('CD')->search({ title => '100th Window' });
  74
  75 IMPORTANT: If called on an object, proxies to new_result instead so
  76
  77   my $cd = $schema->resultset('CD')->new({ title => 'Spoon' });
  78
  79 will return a CD object, not a ResultSet.
  80
  81 =cut
  82
  83 sub new {
  84   my $class = shift;
  85   return $class->new_result(@_) if ref $class;
  86
  87   my ($source, $attrs) = @_;
  88   #weaken $source;
  89
  90   if ($attrs->{page}) {
  91     $attrs->{rows} ||= 10;
  92     $attrs->{offset} ||= 0;
  93     $attrs->{offset} += ($attrs->{rows} * ($attrs->{page} - 1));
  94   }
  95
  96   $attrs->{alias} ||= 'me';
  97
  98   my $self = {
  99     result_source => $source,
 100     result_class => $attrs->{result_class} || $source->result_class,
 101     cond => $attrs->{where},
 102     count => undef,
 103     pager => undef,
 104     attrs => $attrs
 105   };
 106
 107   bless $self, $class;
 108
 109   return $self;
 110 }
 111
 112 =head2 search
 113
 114 =over 4
 115
 116 =item Arguments: $cond, \%attrs?
 117
 118 =item Return Value: $resultset (scalar context), @row_objs (list context)
 119
 120 =back
 121
 122   my @cds    = $cd_rs->search({ year => 2001 }); # "... WHERE year = 2001"
 123   my $new_rs = $cd_rs->search({ year => 2005 });
 124
 125   my $new_rs = $cd_rs->search([ { year => 2005 }, { year => 2004 } ]);
 126                  # year = 2005 OR year = 2004
 127
 128 If you need to pass in additional attributes but no additional condition,
 129 call it as C<search(undef, \%attrs)>.
 130
 131   # "SELECT name, artistid FROM $artist_table"
 132   my @all_artists = $schema->resultset('Artist')->search(undef, {
 133     columns => [qw/name artistid/],
 134   });
 135
 136 For a list of attributes that can be passed to C<search>, see L</ATTRIBUTES>. For more examples of using this function, see L<Searching|DBIx::Class::Manual::Cookbook/Searching>.
 137
 138 =cut
 139
 140 sub search {
 141   my $self = shift;
 142   my $rs = $self->search_rs( @_ );
 143   return (wantarray ? $rs->all : $rs);
 144 }
 145
 146 =head2 search_rs
 147
 148 =over 4
 149
 150 =item Arguments: $cond, \%attrs?
 151
 152 =item Return Value: $resultset
 153
 154 =back
 155
 156 This method does the same exact thing as search() except it will
 157 always return a resultset, even in list context.
 158
 159 =cut
 160
 161 sub search_rs {
 162   my $self = shift;
 163
 164   my $rows;
 165
 166   unless (@_) {                 # no search, effectively just a clone
 167     $rows = $self->get_cache;
 168   }
 169
 170   my $attrs = {};
 171   $attrs = pop(@_) if @_ > 1 and ref $_[$#_] eq 'HASH';
 172   my $our_attrs = { %{$self->{attrs}} };
 173   my $having = delete $our_attrs->{having};
 174
 175   # merge new attrs into inherited
 176   foreach my $key (qw/join prefetch/) {
 177     next unless exists $attrs->{$key};
 178     $our_attrs->{$key} = $self->_merge_attr($our_attrs->{$key}, delete $attrs->{$key});
 179   }
 180
 181   my $new_attrs = { %{$our_attrs}, %{$attrs} };
 182   my $where = (@_
 183     ? (
 184         (@_ == 1 || ref $_[0] eq "HASH")
 185           ? shift
 186           : (
 187               (@_ % 2)
 188                 ? $self->throw_exception("Odd number of arguments to search")
 189                 : {@_}
 190              )
 191       )
 192     : undef
 193   );
 194
 195   if (defined $where) {
 196     $new_attrs->{where} = (
 197       defined $new_attrs->{where}
 198         ? { '-and' => [
 199               map {
 200                 ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_
 201               } $where, $new_attrs->{where}
 202             ]
 203           }
 204         : $where);
 205   }
 206
 207   if (defined $having) {
 208     $new_attrs->{having} = (
 209       defined $new_attrs->{having}
 210         ? { '-and' => [
 211               map {
 212                 ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_
 213               } $having, $new_attrs->{having}
 214             ]
 215           }
 216         : $having);
 217   }
 218
 219   my $rs = (ref $self)->new($self->result_source, $new_attrs);
 220   if ($rows) {
 221     $rs->set_cache($rows);
 222   }
 223   return $rs;
 224 }
 225
 226 =head2 search_literal
 227
 228 =over 4
 229
 230 =item Arguments: $sql_fragment, @bind_values
 231
 232 =item Return Value: $resultset (scalar context), @row_objs (list context)
 233
 234 =back
 235
 236   my @cds   = $cd_rs->search_literal('year = ? AND title = ?', qw/2001 Reload/);
 237   my $newrs = $artist_rs->search_literal('name = ?', 'Metallica');
 238
 239 Pass a literal chunk of SQL to be added to the conditional part of the
 240 resultset query.
 241
 242 =cut
 243
 244 sub search_literal {
 245   my ($self, $cond, @vals) = @_;
 246   my $attrs = (ref $vals[$#vals] eq 'HASH' ? { %{ pop(@vals) } } : {});
 247   $attrs->{bind} = [ @{$self->{attrs}{bind}||[]}, @vals ];
 248   return $self->search(\$cond, $attrs);
 249 }
 250
 251 =head2 find
 252
 253 =over 4
 254
 255 =item Arguments: @values | \%cols, \%attrs?
 256
 257 =item Return Value: $row_object
 258
 259 =back
 260
 261 Finds a row based on its primary key or unique constraint. For example, to find
 262 a row by its primary key:
 263
 264   my $cd = $schema->resultset('CD')->find(5);
 265
 266 You can also find a row by a specific unique constraint using the C<key>
 267 attribute. For example:
 268
 269   my $cd = $schema->resultset('CD')->find('Massive Attack', 'Mezzanine', {
 270     key => 'cd_artist_title'
 271   });
 272
 273 Additionally, you can specify the columns explicitly by name:
 274
 275   my $cd = $schema->resultset('CD')->find(
 276     {
 277       artist => 'Massive Attack',
 278       title  => 'Mezzanine',
 279     },
 280     { key => 'cd_artist_title' }
 281   );
 282
 283 If the C<key> is specified as C<primary>, it searches only on the primary key.
 284
 285 If no C<key> is specified, it searches on all unique constraints defined on the
 286 source, including the primary key.
 287
 288 If your table does not have a primary key, you B<must> provide a value for the
 289 C<key> attribute matching one of the unique constraints on the source.
 290
 291 See also L</find_or_create> and L</update_or_create>. For information on how to
 292 declare unique constraints, see
 293 L<DBIx::Class::ResultSource/add_unique_constraint>.
 294
 295 =cut
 296
 297 sub find {
 298   my $self = shift;
 299   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
 300
 301   # Default to the primary key, but allow a specific key
 302   my @cols = exists $attrs->{key}
 303     ? $self->result_source->unique_constraint_columns($attrs->{key})
 304     : $self->result_source->primary_columns;
 305   $self->throw_exception(
 306     "Can't find unless a primary key is defined or unique constraint is specified"
 307   ) unless @cols;
 308
 309   # Parse out a hashref from input
 310   my $input_query;
 311   if (ref $_[0] eq 'HASH') {
 312     $input_query = { %{$_[0]} };
 313   }
 314   elsif (@_ == @cols) {
 315     $input_query = {};
 316     @{$input_query}{@cols} = @_;
 317   }
 318   else {
 319     # Compatibility: Allow e.g. find(id => $value)
 320     carp "Find by key => value deprecated; please use a hashref instead";
 321     $input_query = {@_};
 322   }
 323
 324   my @unique_queries = $self->_unique_queries($input_query, $attrs);
 325
 326   # Build the final query: Default to the disjunction of the unique queries,
 327   # but allow the input query in case the ResultSet defines the query or the
 328   # user is abusing find
 329   my $alias = exists $attrs->{alias} ? $attrs->{alias} : $self->{attrs}{alias};
 330   my $query = @unique_queries
 331     ? [ map { $self->_add_alias($_, $alias) } @unique_queries ]
 332     : $self->_add_alias($input_query, $alias);
 333
 334   # Run the query
 335   if (keys %$attrs) {
 336     my $rs = $self->search($query, $attrs);
 337     return keys %{$rs->_resolved_attrs->{collapse}} ? $rs->next : $rs->single;
 338   }
 339   else {
 340     return keys %{$self->_resolved_attrs->{collapse}}
 341       ? $self->search($query)->next
 342       : $self->single($query);
 343   }
 344 }
 345
 346 # _add_alias
 347 #
 348 # Add the specified alias to the specified query hash. A copy is made so the
 349 # original query is not modified.
 350
 351 sub _add_alias {
 352   my ($self, $query, $alias) = @_;
 353
 354   my %aliased = %$query;
 355   foreach my $col (grep { ! m/\./ } keys %aliased) {
 356     $aliased{"$alias.$col"} = delete $aliased{$col};
 357   }
 358
 359   return \%aliased;
 360 }
 361
 362 # _unique_queries
 363 #
 364 # Build a list of queries which satisfy unique constraints.
 365
 366 sub _unique_queries {
 367   my ($self, $query, $attrs) = @_;
 368
 369   my @constraint_names = exists $attrs->{key}
 370     ? ($attrs->{key})
 371     : $self->result_source->unique_constraint_names;
 372
 373   my @unique_queries;
 374   foreach my $name (@constraint_names) {
 375     my @unique_cols = $self->result_source->unique_constraint_columns($name);
 376     my $unique_query = $self->_build_unique_query($query, \@unique_cols);
 377
 378     my $num_query = scalar keys %$unique_query;
 379     next unless $num_query;
 380
 381     # XXX: Assuming quite a bit about $self->{attrs}{where}
 382     my $num_cols = scalar @unique_cols;
 383     my $num_where = exists $self->{attrs}{where}
 384       ? scalar keys %{ $self->{attrs}{where} }
 385       : 0;
 386     push @unique_queries, $unique_query
 387       if $num_query + $num_where == $num_cols;
 388   }
 389
 390   return @unique_queries;
 391 }
 392
 393 # _build_unique_query
 394 #
 395 # Constrain the specified query hash based on the specified column names.
 396
 397 sub _build_unique_query {
 398   my ($self, $query, $unique_cols) = @_;
 399
 400   return {
 401     map  { $_ => $query->{$_} }
 402     grep { exists $query->{$_} }
 403       @$unique_cols
 404   };
 405 }
 406
 407 =head2 search_related
 408
 409 =over 4
 410
 411 =item Arguments: $rel, $cond, \%attrs?
 412
 413 =item Return Value: $new_resultset
 414
 415 =back
 416
 417   $new_rs = $cd_rs->search_related('artist', {
 418     name => 'Emo-R-Us',
 419   });
 420
 421 Searches the specified relationship, optionally specifying a condition and
 422 attributes for matching records. See L</ATTRIBUTES> for more information.
 423
 424 =cut
 425
 426 sub search_related {
 427   return shift->related_resultset(shift)->search(@_);
 428 }
 429
 430 =head2 cursor
 431
 432 =over 4
 433
 434 =item Arguments: none
 435
 436 =item Return Value: $cursor
 437
 438 =back
 439
 440 Returns a storage-driven cursor to the given resultset. See
 441 L<DBIx::Class::Cursor> for more information.
 442
 443 =cut
 444
 445 sub cursor {
 446   my ($self) = @_;
 447
 448   my $attrs = { %{$self->_resolved_attrs} };
 449   return $self->{cursor}
 450     ||= $self->result_source->storage->select($attrs->{from}, $attrs->{select},
 451           $attrs->{where},$attrs);
 452 }
 453
 454 =head2 single
 455
 456 =over 4
 457
 458 =item Arguments: $cond?
 459
 460 =item Return Value: $row_object?
 461
 462 =back
 463
 464   my $cd = $schema->resultset('CD')->single({ year => 2001 });
 465
 466 Inflates the first result without creating a cursor if the resultset has
 467 any records in it; if not returns nothing. Used by L</find> as an optimisation.
 468
 469 Can optionally take an additional condition *only* - this is a fast-code-path
 470 method; if you need to add extra joins or similar call ->search and then
 471 ->single without a condition on the $rs returned from that.
 472
 473 =cut
 474
 475 sub single {
 476   my ($self, $where) = @_;
 477   my $attrs = { %{$self->_resolved_attrs} };
 478   if ($where) {
 479     if (defined $attrs->{where}) {
 480       $attrs->{where} = {
 481         '-and' =>
 482             [ map { ref $_ eq 'ARRAY' ? [ -or => $_ ] : $_ }
 483                $where, delete $attrs->{where} ]
 484       };
 485     } else {
 486       $attrs->{where} = $where;
 487     }
 488   }
 489
 490 #  XXX: Disabled since it doesn't infer uniqueness in all cases
 491 #  unless ($self->_is_unique_query($attrs->{where})) {
 492 #    carp "Query not guaranteed to return a single row"
 493 #      . "; please declare your unique constraints or use search instead";
 494 #  }
 495
 496   my @data = $self->result_source->storage->select_single(
 497     $attrs->{from}, $attrs->{select},
 498     $attrs->{where}, $attrs
 499   );
 500
 501   return (@data ? $self->_construct_object(@data) : ());
 502 }
 503
 504 # _is_unique_query
 505 #
 506 # Try to determine if the specified query is guaranteed to be unique, based on
 507 # the declared unique constraints.
 508
 509 sub _is_unique_query {
 510   my ($self, $query) = @_;
 511
 512   my $collapsed = $self->_collapse_query($query);
 513   my $alias = $self->{attrs}{alias};
 514
 515   foreach my $name ($self->result_source->unique_constraint_names) {
 516     my @unique_cols = map {
 517       "$alias.$_"
 518     } $self->result_source->unique_constraint_columns($name);
 519
 520     # Count the values for each unique column
 521     my %seen = map { $_ => 0 } @unique_cols;
 522
 523     foreach my $key (keys %$collapsed) {
 524       my $aliased = $key =~ /\./ ? $key : "$alias.$key";
 525       next unless exists $seen{$aliased};  # Additional constraints are okay
 526       $seen{$aliased} = scalar keys %{ $collapsed->{$key} };
 527     }
 528
 529     # If we get 0 or more than 1 value for a column, it's not necessarily unique
 530     return 1 unless grep { $_ != 1 } values %seen;
 531   }
 532
 533   return 0;
 534 }
 535
 536 # _collapse_query
 537 #
 538 # Recursively collapse the query, accumulating values for each column.
 539
 540 sub _collapse_query {
 541   my ($self, $query, $collapsed) = @_;
 542
 543   $collapsed ||= {};
 544
 545   if (ref $query eq 'ARRAY') {
 546     foreach my $subquery (@$query) {
 547       next unless ref $subquery;  # -or
 548 #      warn "ARRAY: " . Dumper $subquery;
 549       $collapsed = $self->_collapse_query($subquery, $collapsed);
 550     }
 551   }
 552   elsif (ref $query eq 'HASH') {
 553     if (keys %$query and (keys %$query)[0] eq '-and') {
 554       foreach my $subquery (@{$query->{-and}}) {
 555 #        warn "HASH: " . Dumper $subquery;
 556         $collapsed = $self->_collapse_query($subquery, $collapsed);
 557       }
 558     }
 559     else {
 560 #      warn "LEAF: " . Dumper $query;
 561       foreach my $col (keys %$query) {
 562         my $value = $query->{$col};
 563         $collapsed->{$col}{$value}++;
 564       }
 565     }
 566   }
 567
 568   return $collapsed;
 569 }
 570
 571 =head2 get_column
 572
 573 =over 4
 574
 575 =item Arguments: $cond?
 576
 577 =item Return Value: $resultsetcolumn
 578
 579 =back
 580
 581   my $max_length = $rs->get_column('length')->max;
 582
 583 Returns a L<DBIx::Class::ResultSetColumn> instance for a column of the ResultSet.
 584
 585 =cut
 586
 587 sub get_column {
 588   my ($self, $column) = @_;
 589   my $new = DBIx::Class::ResultSetColumn->new($self, $column);
 590   return $new;
 591 }
 592
 593 =head2 search_like
 594
 595 =over 4
 596
 597 =item Arguments: $cond, \%attrs?
 598
 599 =item Return Value: $resultset (scalar context), @row_objs (list context)
 600
 601 =back
 602
 603   # WHERE title LIKE '%blue%'
 604   $cd_rs = $rs->search_like({ title => '%blue%'});
 605
 606 Performs a search, but uses C<LIKE> instead of C<=> as the condition. Note
 607 that this is simply a convenience method. You most likely want to use
 608 L</search> with specific operators.
 609
 610 For more information, see L<DBIx::Class::Manual::Cookbook>.
 611
 612 =cut
 613
 614 sub search_like {
 615   my $class = shift;
 616   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
 617   my $query = ref $_[0] eq 'HASH' ? { %{shift()} }: {@_};
 618   $query->{$_} = { 'like' => $query->{$_} } for keys %$query;
 619   return $class->search($query, { %$attrs });
 620 }
 621
 622 =head2 slice
 623
 624 =over 4
 625
 626 =item Arguments: $first, $last
 627
 628 =item Return Value: $resultset (scalar context), @row_objs (list context)
 629
 630 =back
 631
 632 Returns a resultset or object list representing a subset of elements from the
 633 resultset slice is called on. Indexes are from 0, i.e., to get the first
 634 three records, call:
 635
 636   my ($one, $two, $three) = $rs->slice(0, 2);
 637
 638 =cut
 639
 640 sub slice {
 641   my ($self, $min, $max) = @_;
 642   my $attrs = {}; # = { %{ $self->{attrs} || {} } };
 643   $attrs->{offset} = $self->{attrs}{offset} || 0;
 644   $attrs->{offset} += $min;
 645   $attrs->{rows} = ($max ? ($max - $min + 1) : 1);
 646   return $self->search(undef(), $attrs);
 647   #my $slice = (ref $self)->new($self->result_source, $attrs);
 648   #return (wantarray ? $slice->all : $slice);
 649 }
 650
 651 =head2 next
 652
 653 =over 4
 654
 655 =item Arguments: none
 656
 657 =item Return Value: $result?
 658
 659 =back
 660
 661 Returns the next element in the resultset (C<undef> is there is none).
 662
 663 Can be used to efficiently iterate over records in the resultset:
 664
 665   my $rs = $schema->resultset('CD')->search;
 666   while (my $cd = $rs->next) {
 667     print $cd->title;
 668   }
 669
 670 Note that you need to store the resultset object, and call C<next> on it.
 671 Calling C<< resultset('Table')->next >> repeatedly will always return the
 672 first record from the resultset.
 673
 674 =cut
 675
 676 sub next {
 677   my ($self) = @_;
 678   if (my $cache = $self->get_cache) {
 679     $self->{all_cache_position} ||= 0;
 680     return $cache->[$self->{all_cache_position}++];
 681   }
 682   if ($self->{attrs}{cache}) {
 683     $self->{all_cache_position} = 1;
 684     return ($self->all)[0];
 685   }
 686   my @row = (
 687     exists $self->{stashed_row}
 688       ? @{delete $self->{stashed_row}}
 689       : $self->cursor->next
 690   );
 691   return unless (@row);
 692   return $self->_construct_object(@row);
 693 }
 694
 695 sub _construct_object {
 696   my ($self, @row) = @_;
 697   my $info = $self->_collapse_result($self->{_attrs}{as}, \@row);
 698   my $new = $self->result_class->inflate_result($self->result_source, @$info);
 699   $new = $self->{_attrs}{record_filter}->($new)
 700     if exists $self->{_attrs}{record_filter};
 701   return $new;
 702 }
 703
 704 sub _collapse_result {
 705   my ($self, $as, $row, $prefix) = @_;
 706
 707   my %const;
 708   my @copy = @$row;
 709
 710   foreach my $this_as (@$as) {
 711     my $val = shift @copy;
 712     if (defined $prefix) {
 713       if ($this_as =~ m/^\Q${prefix}.\E(.+)$/) {
 714         my $remain = $1;
 715         $remain =~ /^(?:(.*)\.)?([^.]+)$/;
 716         $const{$1||''}{$2} = $val;
 717       }
 718     } else {
 719       $this_as =~ /^(?:(.*)\.)?([^.]+)$/;
 720       $const{$1||''}{$2} = $val;
 721     }
 722   }
 723
 724   my $alias = $self->{attrs}{alias};
 725   my $info = [ {}, {} ];
 726   foreach my $key (keys %const) {
 727     if (length $key && $key ne $alias) {
 728       my $target = $info;
 729       my @parts = split(/\./, $key);
 730       foreach my $p (@parts) {
 731         $target = $target->[1]->{$p} ||= [];
 732       }
 733       $target->[0] = $const{$key};
 734     } else {
 735       $info->[0] = $const{$key};
 736     }
 737   }
 738
 739   my @collapse;
 740   if (defined $prefix) {
 741     @collapse = map {
 742         m/^\Q${prefix}.\E(.+)$/ ? ($1) : ()
 743     } keys %{$self->{_attrs}{collapse}}
 744   } else {
 745     @collapse = keys %{$self->{_attrs}{collapse}};
 746   };
 747
 748   if (@collapse) {
 749     my ($c) = sort { length $a <=> length $b } @collapse;
 750     my $target = $info;
 751     foreach my $p (split(/\./, $c)) {
 752       $target = $target->[1]->{$p} ||= [];
 753     }
 754     my $c_prefix = (defined($prefix) ? "${prefix}.${c}" : $c);
 755     my @co_key = @{$self->{_attrs}{collapse}{$c_prefix}};
 756     my $tree = $self->_collapse_result($as, $row, $c_prefix);
 757     my %co_check = map { ($_, $tree->[0]->{$_}); } @co_key;
 758     my (@final, @raw);
 759
 760     while (
 761       !(
 762         grep {
 763           !defined($tree->[0]->{$_}) || $co_check{$_} ne $tree->[0]->{$_}
 764         } @co_key
 765         )
 766     ) {
 767       push(@final, $tree);
 768       last unless (@raw = $self->cursor->next);
 769       $row = $self->{stashed_row} = \@raw;
 770       $tree = $self->_collapse_result($as, $row, $c_prefix);
 771     }
 772     @$target = (@final ? @final : [ {}, {} ]);
 773       # single empty result to indicate an empty prefetched has_many
 774   }
 775
 776   #print "final info: " . Dumper($info);
 777   return $info;
 778 }
 779
 780 =head2 result_source
 781
 782 =over 4
 783
 784 =item Arguments: $result_source?
 785
 786 =item Return Value: $result_source
 787
 788 =back
 789
 790 An accessor for the primary ResultSource object from which this ResultSet
 791 is derived.
 792
 793 =head2 result_class
 794
 795 =over 4
 796
 797 =item Arguments: $result_class?
 798
 799 =item Return Value: $result_class
 800
 801 =back
 802
 803 An accessor for the class to use when creating row objects. Defaults to
 804 C<< result_source->result_class >> - which in most cases is the name of the
 805 L<"table"|DBIx::Class::Manual::Glossary/"ResultSource"> class.
 806
 807 =cut
 808
 809
 810 =head2 count
 811
 812 =over 4
 813
 814 =item Arguments: $cond, \%attrs??
 815
 816 =item Return Value: $count
 817
 818 =back
 819
 820 Performs an SQL C<COUNT> with the same query as the resultset was built
 821 with to find the number of elements. If passed arguments, does a search
 822 on the resultset and counts the results of that.
 823
 824 Note: When using C<count> with C<group_by>, L<DBIX::Class> emulates C<GROUP BY>
 825 using C<COUNT( DISTINCT( columns ) )>. Some databases (notably SQLite) do
 826 not support C<DISTINCT> with multiple columns. If you are using such a
 827 database, you should only use columns from the main table in your C<group_by>
 828 clause.
 829
 830 =cut
 831
 832 sub count {
 833   my $self = shift;
 834   return $self->search(@_)->count if @_ and defined $_[0];
 835   return scalar @{ $self->get_cache } if $self->get_cache;
 836   my $count = $self->_count;
 837   return 0 unless $count;
 838
 839   $count -= $self->{attrs}{offset} if $self->{attrs}{offset};
 840   $count = $self->{attrs}{rows} if
 841     $self->{attrs}{rows} and $self->{attrs}{rows} < $count;
 842   return $count;
 843 }
 844
 845 sub _count { # Separated out so pager can get the full count
 846   my $self = shift;
 847   my $select = { count => '*' };
 848
 849   my $attrs = { %{$self->_resolved_attrs} };
 850   if (my $group_by = delete $attrs->{group_by}) {
 851     delete $attrs->{having};
 852     my @distinct = (ref $group_by ?  @$group_by : ($group_by));
 853     # todo: try CONCAT for multi-column pk
 854     my @pk = $self->result_source->primary_columns;
 855     if (@pk == 1) {
 856       my $alias = $attrs->{alias};
 857       foreach my $column (@distinct) {
 858         if ($column =~ qr/^(?:\Q${alias}.\E)?$pk[0]$/) {
 859           @distinct = ($column);
 860           last;
 861         }
 862       }
 863     }
 864
 865     $select = { count => { distinct => \@distinct } };
 866   }
 867
 868   $attrs->{select} = $select;
 869   $attrs->{as} = [qw/count/];
 870
 871   # offset, order by and page are not needed to count. record_filter is cdbi
 872   delete $attrs->{$_} for qw/rows offset order_by page pager record_filter/;
 873
 874   my $tmp_rs = (ref $self)->new($self->result_source, $attrs);
 875   my ($count) = $tmp_rs->cursor->next;
 876   return $count;
 877 }
 878
 879 =head2 count_literal
 880
 881 =over 4
 882
 883 =item Arguments: $sql_fragment, @bind_values
 884
 885 =item Return Value: $count
 886
 887 =back
 888
 889 Counts the results in a literal query. Equivalent to calling L</search_literal>
 890 with the passed arguments, then L</count>.
 891
 892 =cut
 893
 894 sub count_literal { shift->search_literal(@_)->count; }
 895
 896 =head2 all
 897
 898 =over 4
 899
 900 =item Arguments: none
 901
 902 =item Return Value: @objects
 903
 904 =back
 905
 906 Returns all elements in the resultset. Called implicitly if the resultset
 907 is returned in list context.
 908
 909 =cut
 910
 911 sub all {
 912   my ($self) = @_;
 913   return @{ $self->get_cache } if $self->get_cache;
 914
 915   my @obj;
 916
 917   # TODO: don't call resolve here
 918   if (keys %{$self->_resolved_attrs->{collapse}}) {
 919 #  if ($self->{attrs}{prefetch}) {
 920       # Using $self->cursor->all is really just an optimisation.
 921       # If we're collapsing has_many prefetches it probably makes
 922       # very little difference, and this is cleaner than hacking
 923       # _construct_object to survive the approach
 924     my @row = $self->cursor->next;
 925     while (@row) {
 926       push(@obj, $self->_construct_object(@row));
 927       @row = (exists $self->{stashed_row}
 928                ? @{delete $self->{stashed_row}}
 929                : $self->cursor->next);
 930     }
 931   } else {
 932     @obj = map { $self->_construct_object(@$_) } $self->cursor->all;
 933   }
 934
 935   $self->set_cache(\@obj) if $self->{attrs}{cache};
 936   return @obj;
 937 }
 938
 939 =head2 reset
 940
 941 =over 4
 942
 943 =item Arguments: none
 944
 945 =item Return Value: $self
 946
 947 =back
 948
 949 Resets the resultset's cursor, so you can iterate through the elements again.
 950
 951 =cut
 952
 953 sub reset {
 954   my ($self) = @_;
 955   delete $self->{_attrs} if exists $self->{_attrs};
 956   $self->{all_cache_position} = 0;
 957   $self->cursor->reset;
 958   return $self;
 959 }
 960
 961 =head2 first
 962
 963 =over 4
 964
 965 =item Arguments: none
 966
 967 =item Return Value: $object?
 968
 969 =back
 970
 971 Resets the resultset and returns an object for the first result (if the
 972 resultset returns anything).
 973
 974 =cut
 975
 976 sub first {
 977   return $_[0]->reset->next;
 978 }
 979
 980 # _cond_for_update_delete
 981 #
 982 # update/delete require the condition to be modified to handle
 983 # the differing SQL syntax available.  This transforms the $self->{cond}
 984 # appropriately, returning the new condition.
 985
 986 sub _cond_for_update_delete {
 987   my ($self) = @_;
 988   my $cond = {};
 989
 990   # No-op. No condition, we're updating/deleting everything
 991   return $cond unless ref $self->{cond};
 992
 993   if (ref $self->{cond} eq 'ARRAY') {
 994     $cond = [
 995       map {
 996         my %hash;
 997         foreach my $key (keys %{$_}) {
 998           $key =~ /([^.]+)$/;
 999           $hash{$1} = $_->{$key};
1000         }
1001         \%hash;
1002       } @{$self->{cond}}
1003     ];
1004   }
1005   elsif (ref $self->{cond} eq 'HASH') {
1006     if ((keys %{$self->{cond}})[0] eq '-and') {
1007       $cond->{-and} = [];
1008
1009       my @cond = @{$self->{cond}{-and}};
1010       for (my $i = 0; $i < @cond; $i++) {
1011         my $entry = $cond[$i];
1012
1013         my %hash;
1014         if (ref $entry eq 'HASH') {
1015           foreach my $key (keys %{$entry}) {
1016             $key =~ /([^.]+)$/;
1017             $hash{$1} = $entry->{$key};
1018           }
1019         }
1020         else {
1021           $entry =~ /([^.]+)$/;
1022           $hash{$1} = $cond[++$i];
1023         }
1024
1025         push @{$cond->{-and}}, \%hash;
1026       }
1027     }
1028     else {
1029       foreach my $key (keys %{$self->{cond}}) {
1030         $key =~ /([^.]+)$/;
1031         $cond->{$1} = $self->{cond}{$key};
1032       }
1033     }
1034   }
1035   else {
1036     $self->throw_exception(
1037       "Can't update/delete on resultset with condition unless hash or array"
1038     );
1039   }
1040
1041   return $cond;
1042 }
1043
1044
1045 =head2 update
1046
1047 =over 4
1048
1049 =item Arguments: \%values
1050
1051 =item Return Value: $storage_rv
1052
1053 =back
1054
1055 Sets the specified columns in the resultset to the supplied values in a
1056 single query. Return value will be true if the update succeeded or false
1057 if no records were updated; exact type of success value is storage-dependent.
1058
1059 =cut
1060
1061 sub update {
1062   my ($self, $values) = @_;
1063   $self->throw_exception("Values for update must be a hash")
1064     unless ref $values eq 'HASH';
1065
1066   my $cond = $self->_cond_for_update_delete;
1067
1068   return $self->result_source->storage->update(
1069     $self->result_source->from, $values, $cond
1070   );
1071 }
1072
1073 =head2 update_all
1074
1075 =over 4
1076
1077 =item Arguments: \%values
1078
1079 =item Return Value: 1
1080
1081 =back
1082
1083 Fetches all objects and updates them one at a time. Note that C<update_all>
1084 will run DBIC cascade triggers, while L</update> will not.
1085
1086 =cut
1087
1088 sub update_all {
1089   my ($self, $values) = @_;
1090   $self->throw_exception("Values for update must be a hash")
1091     unless ref $values eq 'HASH';
1092   foreach my $obj ($self->all) {
1093     $obj->set_columns($values)->update;
1094   }
1095   return 1;
1096 }
1097
1098 =head2 delete
1099
1100 =over 4
1101
1102 =item Arguments: none
1103
1104 =item Return Value: 1
1105
1106 =back
1107
1108 Deletes the contents of the resultset from its result source. Note that this
1109 will not run DBIC cascade triggers. See L</delete_all> if you need triggers
1110 to run.
1111
1112 =cut
1113
1114 sub delete {
1115   my ($self) = @_;
1116
1117   my $cond = $self->_cond_for_update_delete;
1118
1119   $self->result_source->storage->delete($self->result_source->from, $cond);
1120   return 1;
1121 }
1122
1123 =head2 delete_all
1124
1125 =over 4
1126
1127 =item Arguments: none
1128
1129 =item Return Value: 1
1130
1131 =back
1132
1133 Fetches all objects and deletes them one at a time. Note that C<delete_all>
1134 will run DBIC cascade triggers, while L</delete> will not.
1135
1136 =cut
1137
1138 sub delete_all {
1139   my ($self) = @_;
1140   $_->delete for $self->all;
1141   return 1;
1142 }
1143
1144 =head2 pager
1145
1146 =over 4
1147
1148 =item Arguments: none
1149
1150 =item Return Value: $pager
1151
1152 =back
1153
1154 Return Value a L<Data::Page> object for the current resultset. Only makes
1155 sense for queries with a C<page> attribute.
1156
1157 =cut
1158
1159 sub pager {
1160   my ($self) = @_;
1161   my $attrs = $self->{attrs};
1162   $self->throw_exception("Can't create pager for non-paged rs")
1163     unless $self->{attrs}{page};
1164   $attrs->{rows} ||= 10;
1165   return $self->{pager} ||= Data::Page->new(
1166     $self->_count, $attrs->{rows}, $self->{attrs}{page});
1167 }
1168
1169 =head2 page
1170
1171 =over 4
1172
1173 =item Arguments: $page_number
1174
1175 =item Return Value: $rs
1176
1177 =back
1178
1179 Returns a resultset for the $page_number page of the resultset on which page
1180 is called, where each page contains a number of rows equal to the 'rows'
1181 attribute set on the resultset (10 by default).
1182
1183 =cut
1184
1185 sub page {
1186   my ($self, $page) = @_;
1187   return (ref $self)->new($self->result_source, { %{$self->{attrs}}, page => $page });
1188 }
1189
1190 =head2 new_result
1191
1192 =over 4
1193
1194 =item Arguments: \%vals
1195
1196 =item Return Value: $object
1197
1198 =back
1199
1200 Creates an object in the resultset's result class and returns it.
1201
1202 =cut
1203
1204 sub new_result {
1205   my ($self, $values) = @_;
1206   $self->throw_exception( "new_result needs a hash" )
1207     unless (ref $values eq 'HASH');
1208   $self->throw_exception(
1209     "Can't abstract implicit construct, condition not a hash"
1210   ) if ($self->{cond} && !(ref $self->{cond} eq 'HASH'));
1211
1212   my $alias = $self->{attrs}{alias};
1213   my %new = (
1214     %{ $self->_remove_alias($values, $alias) },
1215     %{ $self->_remove_alias($self->{cond}, $alias) },
1216   );
1217
1218   my $obj = $self->result_class->new(\%new);
1219   $obj->result_source($self->result_source) if $obj->can('result_source');
1220   return $obj;
1221 }
1222
1223 # _remove_alias
1224 #
1225 # Remove the specified alias from the specified query hash. A copy is made so
1226 # the original query is not modified.
1227
1228 sub _remove_alias {
1229   my ($self, $query, $alias) = @_;
1230
1231   my %unaliased = %{ $query || {} };
1232   foreach my $key (keys %unaliased) {
1233     $unaliased{$1} = delete $unaliased{$key}
1234       if $key =~ m/^(?:\Q$alias\E\.)?([^.]+)$/;
1235   }
1236
1237   return \%unaliased;
1238 }
1239
1240 =head2 find_or_new
1241
1242 =over 4
1243
1244 =item Arguments: \%vals, \%attrs?
1245
1246 =item Return Value: $object
1247
1248 =back
1249
1250 Find an existing record from this resultset. If none exists, instantiate a new
1251 result object and return it. The object will not be saved into your storage
1252 until you call L<DBIx::Class::Row/insert> on it.
1253
1254 If you want objects to be saved immediately, use L</find_or_create> instead.
1255
1256 =cut
1257
1258 sub find_or_new {
1259   my $self     = shift;
1260   my $attrs    = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
1261   my $hash     = ref $_[0] eq 'HASH' ? shift : {@_};
1262   my $exists   = $self->find($hash, $attrs);
1263   return defined $exists ? $exists : $self->new_result($hash);
1264 }
1265
1266 =head2 create
1267
1268 =over 4
1269
1270 =item Arguments: \%vals
1271
1272 =item Return Value: $object
1273
1274 =back
1275
1276 Inserts a record into the resultset and returns the object representing it.
1277
1278 Effectively a shortcut for C<< ->new_result(\%vals)->insert >>.
1279
1280 =cut
1281
1282 sub create {
1283   my ($self, $attrs) = @_;
1284   $self->throw_exception( "create needs a hashref" )
1285     unless ref $attrs eq 'HASH';
1286   return $self->new_result($attrs)->insert;
1287 }
1288
1289 =head2 find_or_create
1290
1291 =over 4
1292
1293 =item Arguments: \%vals, \%attrs?
1294
1295 =item Return Value: $object
1296
1297 =back
1298
1299   $class->find_or_create({ key => $val, ... });
1300
1301 Tries to find a record based on its primary key or unique constraint; if none
1302 is found, creates one and returns that instead.
1303
1304   my $cd = $schema->resultset('CD')->find_or_create({
1305     cdid   => 5,
1306     artist => 'Massive Attack',
1307     title  => 'Mezzanine',
1308     year   => 2005,
1309   });
1310
1311 Also takes an optional C<key> attribute, to search by a specific key or unique
1312 constraint. For example:
1313
1314   my $cd = $schema->resultset('CD')->find_or_create(
1315     {
1316       artist => 'Massive Attack',
1317       title  => 'Mezzanine',
1318     },
1319     { key => 'cd_artist_title' }
1320   );
1321
1322 See also L</find> and L</update_or_create>. For information on how to declare
1323 unique constraints, see L<DBIx::Class::ResultSource/add_unique_constraint>.
1324
1325 =cut
1326
1327 sub find_or_create {
1328   my $self     = shift;
1329   my $attrs    = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
1330   my $hash     = ref $_[0] eq 'HASH' ? shift : {@_};
1331   my $exists   = $self->find($hash, $attrs);
1332   return defined $exists ? $exists : $self->create($hash);
1333 }
1334
1335 =head2 update_or_create
1336
1337 =over 4
1338
1339 =item Arguments: \%col_values, { key => $unique_constraint }?
1340
1341 =item Return Value: $object
1342
1343 =back
1344
1345   $class->update_or_create({ col => $val, ... });
1346
1347 First, searches for an existing row matching one of the unique constraints
1348 (including the primary key) on the source of this resultset. If a row is
1349 found, updates it with the other given column values. Otherwise, creates a new
1350 row.
1351
1352 Takes an optional C<key> attribute to search on a specific unique constraint.
1353 For example:
1354
1355   # In your application
1356   my $cd = $schema->resultset('CD')->update_or_create(
1357     {
1358       artist => 'Massive Attack',
1359       title  => 'Mezzanine',
1360       year   => 1998,
1361     },
1362     { key => 'cd_artist_title' }
1363   );
1364
1365 If no C<key> is specified, it searches on all unique constraints defined on the
1366 source, including the primary key.
1367
1368 If the C<key> is specified as C<primary>, it searches only on the primary key.
1369
1370 See also L</find> and L</find_or_create>. For information on how to declare
1371 unique constraints, see L<DBIx::Class::ResultSource/add_unique_constraint>.
1372
1373 =cut
1374
1375 sub update_or_create {
1376   my $self = shift;
1377   my $attrs = (@_ > 1 && ref $_[$#_] eq 'HASH' ? pop(@_) : {});
1378   my $cond = ref $_[0] eq 'HASH' ? shift : {@_};
1379
1380   my $row = $self->find($cond, $attrs);
1381   if (defined $row) {
1382     $row->update($cond);
1383     return $row;
1384   }
1385
1386   return $self->create($cond);
1387 }
1388
1389 =head2 get_cache
1390
1391 =over 4
1392
1393 =item Arguments: none
1394
1395 =item Return Value: \@cache_objects?
1396
1397 =back
1398
1399 Gets the contents of the cache for the resultset, if the cache is set.
1400
1401 =cut
1402
1403 sub get_cache {
1404   shift->{all_cache};
1405 }
1406
1407 =head2 set_cache
1408
1409 =over 4
1410
1411 =item Arguments: \@cache_objects
1412
1413 =item Return Value: \@cache_objects
1414
1415 =back
1416
1417 Sets the contents of the cache for the resultset. Expects an arrayref
1418 of objects of the same class as those produced by the resultset. Note that
1419 if the cache is set the resultset will return the cached objects rather
1420 than re-querying the database even if the cache attr is not set.
1421
1422 =cut
1423
1424 sub set_cache {
1425   my ( $self, $data ) = @_;
1426   $self->throw_exception("set_cache requires an arrayref")
1427       if defined($data) && (ref $data ne 'ARRAY');
1428   $self->{all_cache} = $data;
1429 }
1430
1431 =head2 clear_cache
1432
1433 =over 4
1434
1435 =item Arguments: none
1436
1437 =item Return Value: []
1438
1439 =back
1440
1441 Clears the cache for the resultset.
1442
1443 =cut
1444
1445 sub clear_cache {
1446   shift->set_cache(undef);
1447 }
1448
1449 =head2 related_resultset
1450
1451 =over 4
1452
1453 =item Arguments: $relationship_name
1454
1455 =item Return Value: $resultset
1456
1457 =back
1458
1459 Returns a related resultset for the supplied relationship name.
1460
1461   $artist_rs = $schema->resultset('CD')->related_resultset('Artist');
1462
1463 =cut
1464
1465 sub related_resultset {
1466   my ($self, $rel) = @_;
1467
1468   $self->{related_resultsets} ||= {};
1469   return $self->{related_resultsets}{$rel} ||= do {
1470     my $rel_obj = $self->result_source->relationship_info($rel);
1471
1472     $self->throw_exception(
1473       "search_related: result source '" . $self->result_source->name .
1474         "' has no such relationship $rel")
1475       unless $rel_obj;
1476
1477     my ($from,$seen) = $self->_resolve_from($rel);
1478
1479     my $join_count = $seen->{$rel};
1480     my $alias = ($join_count > 1 ? join('_', $rel, $join_count) : $rel);
1481
1482     $self->result_source->schema->resultset($rel_obj->{class})->search_rs(
1483       undef, {
1484         %{$self->{attrs}||{}},
1485         join => undef,
1486         prefetch => undef,
1487         select => undef,
1488         as => undef,
1489         alias => $alias,
1490         where => $self->{cond},
1491         seen_join => $seen,
1492         from => $from,
1493     });
1494   };
1495 }
1496
1497 sub _resolve_from {
1498   my ($self, $extra_join) = @_;
1499   my $source = $self->result_source;
1500   my $attrs = $self->{attrs};
1501
1502   my $from = $attrs->{from}
1503     || [ { $attrs->{alias} => $source->from } ];
1504
1505   my $seen = { %{$attrs->{seen_join}||{}} };
1506
1507   my $join = ($attrs->{join}
1508                ? [ $attrs->{join}, $extra_join ]
1509                : $extra_join);
1510   $from = [
1511     @$from,
1512     ($join ? $source->resolve_join($join, $attrs->{alias}, $seen) : ()),
1513   ];
1514
1515   return ($from,$seen);
1516 }
1517
1518 sub _resolved_attrs {
1519   my $self = shift;
1520   return $self->{_attrs} if $self->{_attrs};
1521
1522   my $attrs = { %{$self->{attrs}||{}} };
1523   my $source = $self->{result_source};
1524   my $alias = $attrs->{alias};
1525
1526   $attrs->{columns} ||= delete $attrs->{cols} if exists $attrs->{cols};
1527   if ($attrs->{columns}) {
1528     delete $attrs->{as};
1529   } elsif (!$attrs->{select}) {
1530     $attrs->{columns} = [ $source->columns ];
1531   }
1532
1533   $attrs->{select} =
1534     ($attrs->{select}
1535       ? (ref $attrs->{select} eq 'ARRAY'
1536           ? [ @{$attrs->{select}} ]
1537           : [ $attrs->{select} ])
1538       : [ map { m/\./ ? $_ : "${alias}.$_" } @{delete $attrs->{columns}} ]
1539     );
1540   $attrs->{as} =
1541     ($attrs->{as}
1542       ? (ref $attrs->{as} eq 'ARRAY'
1543           ? [ @{$attrs->{as}} ]
1544           : [ $attrs->{as} ])
1545       : [ map { m/^\Q${alias}.\E(.+)$/ ? $1 : $_ } @{$attrs->{select}} ]
1546     );
1547
1548   my $adds;
1549   if ($adds = delete $attrs->{include_columns}) {
1550     $adds = [$adds] unless ref $adds eq 'ARRAY';
1551     push(@{$attrs->{select}}, @$adds);
1552     push(@{$attrs->{as}}, map { m/([^.]+)$/; $1 } @$adds);
1553   }
1554   if ($adds = delete $attrs->{'+select'}) {
1555     $adds = [$adds] unless ref $adds eq 'ARRAY';
1556     push(@{$attrs->{select}},
1557            map { /\./ || ref $_ ? $_ : "${alias}.$_" } @$adds);
1558   }
1559   if (my $adds = delete $attrs->{'+as'}) {
1560     $adds = [$adds] unless ref $adds eq 'ARRAY';
1561     push(@{$attrs->{as}}, @$adds);
1562   }
1563
1564   $attrs->{from} ||= [ { 'me' => $source->from } ];
1565
1566   if (exists $attrs->{join} || exists $attrs->{prefetch}) {
1567     my $join = delete $attrs->{join} || {};
1568
1569     if (defined $attrs->{prefetch}) {
1570       $join = $self->_merge_attr(
1571         $join, $attrs->{prefetch}
1572       );
1573     }
1574
1575     $attrs->{from} =   # have to copy here to avoid corrupting the original
1576       [
1577         @{$attrs->{from}},
1578         $source->resolve_join($join, $alias, { %{$attrs->{seen_join}||{}} })
1579       ];
1580   }
1581
1582   $attrs->{group_by} ||= $attrs->{select} if delete $attrs->{distinct};
1583   if ($attrs->{order_by}) {
1584     $attrs->{order_by} = (ref($attrs->{order_by}) eq 'ARRAY'
1585                            ? [ @{$attrs->{order_by}} ]
1586                            : [ $attrs->{order_by} ]);
1587   } else {
1588     $attrs->{order_by} = [];
1589   }
1590
1591   my $collapse = $attrs->{collapse} || {};
1592   if (my $prefetch = delete $attrs->{prefetch}) {
1593     $prefetch = $self->_merge_attr({}, $prefetch);
1594     my @pre_order;
1595     my $seen = $attrs->{seen_join} || {};
1596     foreach my $p (ref $prefetch eq 'ARRAY' ? @$prefetch : ($prefetch)) {
1597       # bring joins back to level of current class
1598       my @prefetch = $source->resolve_prefetch(
1599         $p, $alias, $seen, \@pre_order, $collapse
1600       );
1601       push(@{$attrs->{select}}, map { $_->[0] } @prefetch);
1602       push(@{$attrs->{as}}, map { $_->[1] } @prefetch);
1603     }
1604     push(@{$attrs->{order_by}}, @pre_order);
1605   }
1606   $attrs->{collapse} = $collapse;
1607
1608   return $self->{_attrs} = $attrs;
1609 }
1610
1611 sub _merge_attr {
1612   my ($self, $a, $b) = @_;
1613   return $b unless defined($a);
1614   return $a unless defined($b);
1615
1616   if (ref $b eq 'HASH' && ref $a eq 'HASH') {
1617     foreach my $key (keys %{$b}) {
1618       if (exists $a->{$key}) {
1619         $a->{$key} = $self->_merge_attr($a->{$key}, $b->{$key});
1620       } else {
1621         $a->{$key} = $b->{$key};
1622       }
1623     }
1624     return $a;
1625   } else {
1626     $a = [$a] unless ref $a eq 'ARRAY';
1627     $b = [$b] unless ref $b eq 'ARRAY';
1628
1629     my $hash = {};
1630     my @array;
1631     foreach my $x ($a, $b) {
1632       foreach my $element (@{$x}) {
1633         if (ref $element eq 'HASH') {
1634           $hash = $self->_merge_attr($hash, $element);
1635         } elsif (ref $element eq 'ARRAY') {
1636           push(@array, @{$element});
1637         } else {
1638           push(@array, $element) unless $b == $x
1639             && grep { $_ eq $element } @array;
1640         }
1641       }
1642     }
1643
1644     @array = grep { !exists $hash->{$_} } @array;
1645
1646     return keys %{$hash}
1647       ? ( scalar(@array)
1648             ? [$hash, @array]
1649             : $hash
1650         )
1651       : \@array;
1652   }
1653 }
1654
1655 =head2 throw_exception
1656
1657 See L<DBIx::Class::Schema/throw_exception> for details.
1658
1659 =cut
1660
1661 sub throw_exception {
1662   my $self=shift;
1663   $self->result_source->schema->throw_exception(@_);
1664 }
1665
1666 # XXX: FIXME: Attributes docs need clearing up
1667
1668 =head1 ATTRIBUTES
1669
1670 The resultset takes various attributes that modify its behavior. Here's an
1671 overview of them:
1672
1673 =head2 order_by
1674
1675 =over 4
1676
1677 =item Value: ($order_by | \@order_by)
1678
1679 =back
1680
1681 Which column(s) to order the results by. This is currently passed
1682 through directly to SQL, so you can give e.g. C<year DESC> for a
1683 descending order on the column `year'.
1684
1685 Please note that if you have quoting enabled (see
1686 L<DBIx::Class::Storage/quote_char>) you will need to do C<\'year DESC' > to
1687 specify an order. (The scalar ref causes it to be passed as raw sql to the DB,
1688 so you will need to manually quote things as appropriate.)
1689
1690 =head2 columns
1691
1692 =over 4
1693
1694 =item Value: \@columns
1695
1696 =back
1697
1698 Shortcut to request a particular set of columns to be retrieved.  Adds
1699 C<me.> onto the start of any column without a C<.> in it and sets C<select>
1700 from that, then auto-populates C<as> from C<select> as normal. (You may also
1701 use the C<cols> attribute, as in earlier versions of DBIC.)
1702
1703 =head2 include_columns
1704
1705 =over 4
1706
1707 =item Value: \@columns
1708
1709 =back
1710
1711 Shortcut to include additional columns in the returned results - for example
1712
1713   $schema->resultset('CD')->search(undef, {
1714     include_columns => ['artist.name'],
1715     join => ['artist']
1716   });
1717
1718 would return all CDs and include a 'name' column to the information
1719 passed to object inflation
1720
1721 =head2 select
1722
1723 =over 4
1724
1725 =item Value: \@select_columns
1726
1727 =back
1728
1729 Indicates which columns should be selected from the storage. You can use
1730 column names, or in the case of RDBMS back ends, function or stored procedure
1731 names:
1732
1733   $rs = $schema->resultset('Employee')->search(undef, {
1734     select => [
1735       'name',
1736       { count => 'employeeid' },
1737       { sum => 'salary' }
1738     ]
1739   });
1740
1741 When you use function/stored procedure names and do not supply an C<as>
1742 attribute, the column names returned are storage-dependent. E.g. MySQL would
1743 return a column named C<count(employeeid)> in the above example.
1744
1745 =head2 +select
1746
1747 =over 4
1748
1749 Indicates additional columns to be selected from storage.  Works the same as
1750 L<select> but adds columns to the selection.
1751
1752 =back
1753
1754 =head2 +as
1755
1756 =over 4
1757
1758 Indicates additional column names for those added via L<+select>.
1759
1760 =back
1761
1762 =head2 as
1763
1764 =over 4
1765
1766 =item Value: \@inflation_names
1767
1768 =back
1769
1770 Indicates column names for object inflation. This is used in conjunction with
1771 C<select>, usually when C<select> contains one or more function or stored
1772 procedure names:
1773
1774   $rs = $schema->resultset('Employee')->search(undef, {
1775     select => [
1776       'name',
1777       { count => 'employeeid' }
1778     ],
1779     as => ['name', 'employee_count'],
1780   });
1781
1782   my $employee = $rs->first(); # get the first Employee
1783
1784 If the object against which the search is performed already has an accessor
1785 matching a column name specified in C<as>, the value can be retrieved using
1786 the accessor as normal:
1787
1788   my $name = $employee->name();
1789
1790 If on the other hand an accessor does not exist in the object, you need to
1791 use C<get_column> instead:
1792
1793   my $employee_count = $employee->get_column('employee_count');
1794
1795 You can create your own accessors if required - see
1796 L<DBIx::Class::Manual::Cookbook> for details.
1797
1798 Please note: This will NOT insert an C<AS employee_count> into the SQL
1799 statement produced, it is used for internal access only. Thus
1800 attempting to use the accessor in an C<order_by> clause or similar
1801 will fail miserably.
1802
1803 To get around this limitation, you can supply literal SQL to your
1804 C<select> attibute that contains the C<AS alias> text, eg:
1805
1806   select => [\'myfield AS alias']
1807
1808 =head2 join
1809
1810 =over 4
1811
1812 =item Value: ($rel_name | \@rel_names | \%rel_names)
1813
1814 =back
1815
1816 Contains a list of relationships that should be joined for this query.  For
1817 example:
1818
1819   # Get CDs by Nine Inch Nails
1820   my $rs = $schema->resultset('CD')->search(
1821     { 'artist.name' => 'Nine Inch Nails' },
1822     { join => 'artist' }
1823   );
1824
1825 Can also contain a hash reference to refer to the other relation's relations.
1826 For example:
1827
1828   package MyApp::Schema::Track;
1829   use base qw/DBIx::Class/;
1830   __PACKAGE__->table('track');
1831   __PACKAGE__->add_columns(qw/trackid cd position title/);
1832   __PACKAGE__->set_primary_key('trackid');
1833   __PACKAGE__->belongs_to(cd => 'MyApp::Schema::CD');
1834   1;
1835
1836   # In your application
1837   my $rs = $schema->resultset('Artist')->search(
1838     { 'track.title' => 'Teardrop' },
1839     {
1840       join     => { cd => 'track' },
1841       order_by => 'artist.name',
1842     }
1843   );
1844
1845 You need to use the relationship (not the table) name in  conditions,
1846 because they are aliased as such. The current table is aliased as "me", so
1847 you need to use me.column_name in order to avoid ambiguity. For example:
1848
1849   # Get CDs from 1984 with a 'Foo' track
1850   my $rs = $schema->resultset('CD')->search(
1851     {
1852       'me.year' => 1984,
1853       'tracks.name' => 'Foo'
1854     },
1855     { join => 'tracks' }
1856   );
1857
1858 If the same join is supplied twice, it will be aliased to <rel>_2 (and
1859 similarly for a third time). For e.g.
1860
1861   my $rs = $schema->resultset('Artist')->search({
1862     'cds.title'   => 'Down to Earth',
1863     'cds_2.title' => 'Popular',
1864   }, {
1865     join => [ qw/cds cds/ ],
1866   });
1867
1868 will return a set of all artists that have both a cd with title 'Down
1869 to Earth' and a cd with title 'Popular'.
1870
1871 If you want to fetch related objects from other tables as well, see C<prefetch>
1872 below.
1873
1874 =head2 prefetch
1875
1876 =over 4
1877
1878 =item Value: ($rel_name | \@rel_names | \%rel_names)
1879
1880 =back
1881
1882 Contains one or more relationships that should be fetched along with the main
1883 query (when they are accessed afterwards they will have already been
1884 "prefetched").  This is useful for when you know you will need the related
1885 objects, because it saves at least one query:
1886
1887   my $rs = $schema->resultset('Tag')->search(
1888     undef,
1889     {
1890       prefetch => {
1891         cd => 'artist'
1892       }
1893     }
1894   );
1895
1896 The initial search results in SQL like the following:
1897
1898   SELECT tag.*, cd.*, artist.* FROM tag
1899   JOIN cd ON tag.cd = cd.cdid
1900   JOIN artist ON cd.artist = artist.artistid
1901
1902 L<DBIx::Class> has no need to go back to the database when we access the
1903 C<cd> or C<artist> relationships, which saves us two SQL statements in this
1904 case.
1905
1906 Simple prefetches will be joined automatically, so there is no need
1907 for a C<join> attribute in the above search. If you're prefetching to
1908 depth (e.g. { cd => { artist => 'label' } or similar), you'll need to
1909 specify the join as well.
1910
1911 C<prefetch> can be used with the following relationship types: C<belongs_to>,
1912 C<has_one> (or if you're using C<add_relationship>, any relationship declared
1913 with an accessor type of 'single' or 'filter').
1914
1915 =head2 page
1916
1917 =over 4
1918
1919 =item Value: $page
1920
1921 =back
1922
1923 Makes the resultset paged and specifies the page to retrieve. Effectively
1924 identical to creating a non-pages resultset and then calling ->page($page)
1925 on it.
1926
1927 If L<rows> attribute is not specified it defualts to 10 rows per page.
1928
1929 =head2 rows
1930
1931 =over 4
1932
1933 =item Value: $rows
1934
1935 =back
1936
1937 Specifes the maximum number of rows for direct retrieval or the number of
1938 rows per page if the page attribute or method is used.
1939
1940 =head2 offset
1941
1942 =over 4
1943
1944 =item Value: $offset
1945
1946 =back
1947
1948 Specifies the (zero-based) row number for the  first row to be returned, or the
1949 of the first row of the first page if paging is used.
1950
1951 =head2 group_by
1952
1953 =over 4
1954
1955 =item Value: \@columns
1956
1957 =back
1958
1959 A arrayref of columns to group by. Can include columns of joined tables.
1960
1961   group_by => [qw/ column1 column2 ... /]
1962
1963 =head2 having
1964
1965 =over 4
1966
1967 =item Value: $condition
1968
1969 =back
1970
1971 HAVING is a select statement attribute that is applied between GROUP BY and
1972 ORDER BY. It is applied to the after the grouping calculations have been
1973 done.
1974
1975   having => { 'count(employee)' => { '>=', 100 } }
1976
1977 =head2 distinct
1978
1979 =over 4
1980
1981 =item Value: (0 | 1)
1982
1983 =back
1984
1985 Set to 1 to group by all columns.
1986
1987 =head2 where
1988
1989 =over 4
1990
1991 Adds to the WHERE clause.
1992
1993   # only return rows WHERE deleted IS NULL for all searches
1994   __PACKAGE__->resultset_attributes({ where => { deleted => undef } }); )
1995
1996 Can be overridden by passing C<{ where => undef }> as an attribute
1997 to a resulset.
1998
1999 =back
2000
2001 =head2 cache
2002
2003 Set to 1 to cache search results. This prevents extra SQL queries if you
2004 revisit rows in your ResultSet:
2005
2006   my $resultset = $schema->resultset('Artist')->search( undef, { cache => 1 } );
2007
2008   while( my $artist = $resultset->next ) {
2009     ... do stuff ...
2010   }
2011
2012   $rs->first; # without cache, this would issue a query
2013
2014 By default, searches are not cached.
2015
2016 For more examples of using these attributes, see
2017 L<DBIx::Class::Manual::Cookbook>.
2018
2019 =head2 from
2020
2021 =over 4
2022
2023 =item Value: \@from_clause
2024
2025 =back
2026
2027 The C<from> attribute gives you manual control over the C<FROM> clause of SQL
2028 statements generated by L<DBIx::Class>, allowing you to express custom C<JOIN>
2029 clauses.
2030
2031 NOTE: Use this on your own risk.  This allows you to shoot off your foot!
2032
2033 C<join> will usually do what you need and it is strongly recommended that you
2034 avoid using C<from> unless you cannot achieve the desired result using C<join>.
2035 And we really do mean "cannot", not just tried and failed. Attempting to use
2036 this because you're having problems with C<join> is like trying to use x86
2037 ASM because you've got a syntax error in your C. Trust us on this.
2038
2039 Now, if you're still really, really sure you need to use this (and if you're
2040 not 100% sure, ask the mailing list first), here's an explanation of how this
2041 works.
2042
2043 The syntax is as follows -
2044
2045   [
2046     { <alias1> => <table1> },
2047     [
2048       { <alias2> => <table2>, -join_type => 'inner|left|right' },
2049       [], # nested JOIN (optional)
2050       { <table1.column1> => <table2.column2>, ... (more conditions) },
2051     ],
2052     # More of the above [ ] may follow for additional joins
2053   ]
2054
2055   <table1> <alias1>
2056   JOIN
2057     <table2> <alias2>
2058     [JOIN ...]
2059   ON <table1.column1> = <table2.column2>
2060   <more joins may follow>
2061
2062 An easy way to follow the examples below is to remember the following:
2063
2064     Anything inside "[]" is a JOIN
2065     Anything inside "{}" is a condition for the enclosing JOIN
2066
2067 The following examples utilize a "person" table in a family tree application.
2068 In order to express parent->child relationships, this table is self-joined:
2069
2070     # Person->belongs_to('father' => 'Person');
2071     # Person->belongs_to('mother' => 'Person');
2072
2073 C<from> can be used to nest joins. Here we return all children with a father,
2074 then search against all mothers of those children:
2075
2076   $rs = $schema->resultset('Person')->search(
2077       undef,
2078       {
2079           alias => 'mother', # alias columns in accordance with "from"
2080           from => [
2081               { mother => 'person' },
2082               [
2083                   [
2084                       { child => 'person' },
2085                       [
2086                           { father => 'person' },
2087                           { 'father.person_id' => 'child.father_id' }
2088                       ]
2089                   ],
2090                   { 'mother.person_id' => 'child.mother_id' }
2091               ],
2092           ]
2093       },
2094   );
2095
2096   # Equivalent SQL:
2097   # SELECT mother.* FROM person mother
2098   # JOIN (
2099   #   person child
2100   #   JOIN person father
2101   #   ON ( father.person_id = child.father_id )
2102   # )
2103   # ON ( mother.person_id = child.mother_id )
2104
2105 The type of any join can be controlled manually. To search against only people
2106 with a father in the person table, we could explicitly use C<INNER JOIN>:
2107
2108     $rs = $schema->resultset('Person')->search(
2109         undef,
2110         {
2111             alias => 'child', # alias columns in accordance with "from"
2112             from => [
2113                 { child => 'person' },
2114                 [
2115                     { father => 'person', -join_type => 'inner' },
2116                     { 'father.id' => 'child.father_id' }
2117                 ],
2118             ]
2119         },
2120     );
2121
2122     # Equivalent SQL:
2123     # SELECT child.* FROM person child
2124     # INNER JOIN person father ON child.father_id = father.id
2125
2126 =cut
2127
2128 1;