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