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