lib/DBIx/Class/ResultSet.pm

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