From: Peter Rabbitson <ribasushi@cpan.org>
Date: Sat, 25 Feb 2012 14:36:43 +0000 (+0100)
Subject: I think we are done here
X-Git-Url: http://git.shadowcat.co.uk/gitweb/gitweb.cgi?a=commitdiff_plain;h=4e9fc3f3;p=dbsrgits%2FDBIx-Class-Historic.git

I think we are done here
---

diff --git a/lib/DBIx/Class/ResultSet.pm b/lib/DBIx/Class/ResultSet.pm
index a2a461b..2c0207e 100644
--- a/lib/DBIx/Class/ResultSet.pm
+++ b/lib/DBIx/Class/ResultSet.pm
@@ -1058,8 +1058,9 @@ sub single {
     $attrs->{from}, $attrs->{select},
     $attrs->{where}, $attrs
   )];
-
-  return @$data ? $self->_construct_objects($data)->[0] : undef;
+  return undef unless @$data;
+  $self->{stashed_rows} = [ $data ];
+  $self->_construct_objects->[0];
 }
 
 
@@ -1245,83 +1246,109 @@ sub next {
 # order the result sensibly) OR until the cursor is exhausted (an
 # unordered collapsing resultset effectively triggers ->all)
 sub _construct_objects {
-  my ($self, $fetched_row, $fetch_all) = @_;
+  my ($self, $fetch_all) = @_;
 
+  my $rsrc = $self->result_source;
   my $attrs = $self->_resolved_attrs;
-  my $unordered = 0;  # will deal with this later
+  my $cursor = $self->cursor;
 
   # this will be used as both initial raw-row collector AND as a RV of
-  # _construct_objects. Not regrowing the   # array twice matters a lot...
+  # _construct_objects. Not regrowing the array twice matters a lot...
   # a suprising amount actually
-  my $rows;
-
-  # $fetch_all implies all() which means all stashes have been cleared
-  # and the cursor reset
+  my $rows = (delete $self->{stashed_rows}) || [];
   if ($fetch_all) {
-    # FIXME - we can do better, cursor->all (well a diff. method) should return a ref
-    $rows = [ $self->cursor->all ];
+    # FIXME - we can do better, cursor->next/all (well diff. methods) should return a ref
+    $rows = [ @$rows, $cursor->all ];
   }
-  elsif ($unordered) {
-    $rows = [
-      $fetched_row||(),
-      @{ delete $self->{stashed_rows} || []},
-      $self->cursor->all,
-    ];
+  elsif (!$attrs->{collapse}) {
+    push @$rows, do { my @r = $cursor->next; @r ? \@r : () }
+      unless @$rows;
   }
-  else {  # simple single object
-    $rows = [ $fetched_row || ( @{$self->{stashed_rows}||[]} ? shift @{$self->{stashed_rows}} : [$self->cursor->next] ) ];
+  else {
+    $attrs->{_ordered_for_collapse} ||= (!$attrs->{order_by}) ? undef : do {
+      my $st = $rsrc->schema->storage;
+      my @ord_cols = map
+        { $_->[0] }
+        ( $st->_extract_order_criteria($attrs->{order_by}) )
+      ;
+
+      my $colinfos = $st->_resolve_column_info($attrs->{from}, \@ord_cols);
+
+      for (0 .. $#ord_cols) {
+        if (
+          ! $colinfos->{$ord_cols[$_]}
+            or
+          $colinfos->{$ord_cols[$_]}{-result_source} != $rsrc
+        ) {
+          splice @ord_cols, $_;
+          last;
+        }
+      }
+
+      # since all we check here are the start of the order_by belonging to the
+      # top level $rsrc, the order stability check will fail unless the whole
+      # thing is ordered as we need it
+      (@ord_cols and $rsrc->_identifying_column_set({ map
+        { $colinfos->{$_}{-colname} => $colinfos->{$_} }
+        @ord_cols
+      })) ? 1 : 0;
+    };
+
+    if ($attrs->{_ordered_for_collapse}) {
+      push @$rows, do { my @r = $cursor->next; @r ? \@r : () };
+    }
+    # instead of looping over ->next, use ->all in stealth mode
+    elsif (! $cursor->{done}) {
+      push @$rows, $cursor->all;
+      $cursor->{done} = 1;
+      $fetch_all = 1;
+    }
   }
 
-  return undef unless @{$rows->[0]||[]};
+  return undef unless @$rows;
 
-  my $rsrc = $self->result_source;
   my $res_class = $self->result_class;
   my $inflator = $res_class->can ('inflate_result')
     or $self->throw_exception("Inflator $res_class does not provide an inflate_result() method");
 
-  # construct a much simpler array->hash folder for the one-table cases right here
-  if ($attrs->{_single_object_inflation} and ! $attrs->{collapse}) {
+  my $infmap = $attrs->{as};
+
+  if (!$attrs->{collapse} and $attrs->{_single_object_inflation}) {
+    # construct a much simpler array->hash folder for the one-table cases right here
+
     # FIXME SUBOPTIMAL this is a very very very hot spot
     # while rather optimal we can *still* do much better, by
     # building a smarter [Row|HRI]::inflate_result(), and
     # switch to feeding it data via a much leaner interface
     #
-    my $infmap = $attrs->{as};
-    my @as_idx = 0..$#$infmap;
-    for my $r (@$rows) {
-      $r = [{ map { $infmap->[$_] => $r->[$_] } @as_idx }]
+    # crude unscientific benchmarking indicated the shortcut eval is not worth it for
+    # this particular resultset size
+    if (@$rows < 60) {
+      my @as_idx = 0..$#$infmap;
+      for my $r (@$rows) {
+        $r = $inflator->($res_class, $rsrc, { map { $infmap->[$_] => $r->[$_] } @as_idx } );
+      }
+    }
+    else {
+      eval sprintf (
+        '$_ = $inflator->($res_class, $rsrc, { %s }) for @$rows',
+        join (', ', map { "\$infmap->[$_] => \$_->[$_]" } 0..$#$infmap )
+      );
     }
-
-    # FIXME - this seems to be faster than the hashmapper above, especially
-    # on more rows, but need a better bench-environment to confirm
-    #eval sprintf (
-    #  '$_ = [{ %s }] for @$rows',
-    #  join (', ', map { "\$infmap->[$_] => \$_->[$_]" } 0..$#$infmap )
-    #);
   }
   else {
-    push @$rows, @{$self->{stashed_rows}||[]};
-
-    my $perl = $rsrc->_mk_row_parser({
-      inflate_map => $attrs->{as},
+    ($self->{_row_parser} ||= eval sprintf 'sub { %s }', $rsrc->_mk_row_parser({
+      inflate_map => $infmap,
       selection => $attrs->{select},
       collapse => $attrs->{collapse},
-      unordered => $unordered,
-    });
+    }) or die $@)->($rows, $fetch_all ? () : (
+      sub { my @r = $cursor->next or return; \@r },
+      ($self->{stashed_rows} = []),
+    ));  # modify $rows in-place, shrinking/extending as necessary
 
-    (eval "sub { no warnings; no strict; $perl }")->( # disable of strictures seems to have some effect, weird
-      $rows,  # modify in-place, shrinking/extending as necessary
-      ($attrs->{collapse} and ! $fetch_all and ! $unordered)
-        ? (
-            sub { my @r = $self->cursor->next or return undef; \@r },
-            ($self->{stashed_rows} = []), # this is where we empty things and prepare for leftovers
-          )
-        : ()
-      ,
-    );
-  }
+    $_ = $inflator->($res_class, $rsrc, @$_) for @$rows;
 
-  $_ = $res_class->$inflator($rsrc, @$_) for @$rows;
+  }
 
   # CDBI compat stuff
   if ($attrs->{record_filter}) {
@@ -1628,7 +1655,7 @@ sub all {
 
   $self->cursor->reset;
 
-  my $objs = $self->_construct_objects(undef, 'fetch_all') || [];
+  my $objs = $self->_construct_objects('fetch_all') || [];
 
   $self->set_cache($objs) if $self->{attrs}{cache};
 
@@ -3425,6 +3452,8 @@ sub _resolved_attrs {
     push @{ $attrs->{as} }, (map { $_->[1] } @prefetch);
   }
 
+  $attrs->{_single_object_inflation} = ! List::Util::first { $_ =~ /\./ } @{$attrs->{as}};
+
   # run through the resulting joinstructure (starting from our current slot)
   # and unset collapse if proven unnesessary
   if ($attrs->{collapse} && ref $attrs->{from} eq 'ARRAY') {
@@ -3450,7 +3479,11 @@ sub _resolved_attrs {
     }
   }
 
-  $attrs->{_single_object_inflation} = ! List::Util::first { $_ =~ /\./ } @{$attrs->{as}};
+  if (! $attrs->{order_by} and $attrs->{collapse}) {
+    # default order for collapsing unless the user asked for something
+    $attrs->{order_by} = [ map { "$alias.$_" } $source->primary_columns ];
+    $attrs->{_ordered_for_collapse} = 1;
+  }
 
   # if both page and offset are specified, produce a combined offset
   # even though it doesn't make much sense, this is what pre 081xx has
@@ -3674,6 +3707,9 @@ sub STORABLE_freeze {
   # A cursor in progress can't be serialized (and would make little sense anyway)
   delete $to_serialize->{cursor};
 
+  # the parser can be regenerated
+  delete $to_serialize->{_row_parser};
+
   # nor is it sensical to store a not-yet-fired-count pager
   if ($to_serialize->{pager} and ref $to_serialize->{pager}{total_entries} eq 'CODE') {
     delete $to_serialize->{pager};
diff --git a/lib/DBIx/Class/ResultSource/RowParser.pm b/lib/DBIx/Class/ResultSource/RowParser.pm
index c9d4c20..d71ca30 100644
--- a/lib/DBIx/Class/ResultSource/RowParser.pm
+++ b/lib/DBIx/Class/ResultSource/RowParser.pm
@@ -1,4 +1,5 @@
-package DBIx::Class::ResultSource::RowParser;
+package # hide from the pauses
+  DBIx::Class::ResultSource::RowParser;
 
 use strict;
 use warnings;
@@ -32,7 +33,7 @@ sub _resolve_prefetch {
     map {
       $self->_resolve_prefetch($_, $alias, $alias_map, $order, [ @$pref_path ] ),
       $self->related_source($_)->_resolve_prefetch(
-               $pre->{$_}, "${alias}.$_", $alias_map, $order, [ @$pref_path, $_] )
+         $pre->{$_}, "${alias}.$_", $alias_map, $order, [ @$pref_path, $_] )
     } keys %$pre;
     return @ret;
   }
@@ -56,50 +57,9 @@ sub _resolve_prefetch {
       unless $rel_info;
 
     my $as_prefix = ($alias =~ /^.*?\.(.+)$/ ? $1.'.' : '');
-    my $rel_source = $self->related_source($pre);
-
-    if ($rel_info->{attrs}{accessor} && $rel_info->{attrs}{accessor} eq 'multi') {
-      $self->throw_exception(
-        "Can't prefetch has_many ${pre} (join cond too complex)")
-        unless ref($rel_info->{cond}) eq 'HASH';
-      my $dots = @{[$as_prefix =~ m/\./g]} + 1; # +1 to match the ".${as_prefix}"
-
-      #my @col = map { (/^self\.(.+)$/ ? ("${as_prefix}.$1") : ()); }
-      #              values %{$rel_info->{cond}};
-      my @key = map { (/^foreign\.(.+)$/ ? ($1) : ()); }
-                    keys %{$rel_info->{cond}};
-
-      push @$order, map { "${as}.$_" } @key;
-
-      if (my $rel_order = $rel_info->{attrs}{order_by}) {
-        # this is kludgy and incomplete, I am well aware
-        # but the parent method is going away entirely anyway
-        # so sod it
-        my $sql_maker = $self->storage->sql_maker;
-        my ($orig_ql, $orig_qr) = $sql_maker->_quote_chars;
-        my $sep = $sql_maker->name_sep;
-
-        # install our own quoter, so we can catch unqualified stuff
-        local $sql_maker->{quote_char} = ["\x00", "\xFF"];
-
-        my $quoted_prefix = "\x00${as}\xFF";
-
-        for my $chunk ( $sql_maker->_order_by_chunks ($rel_order) ) {
-          my @bind;
-          ($chunk, @bind) = @$chunk if ref $chunk;
-
-          $chunk = "${quoted_prefix}${sep}${chunk}"
-            unless $chunk =~ /\Q$sep/;
-
-          $chunk =~ s/\x00/$orig_ql/g;
-          $chunk =~ s/\xFF/$orig_qr/g;
-          push @$order, \[$chunk, @bind];
-        }
-      }
-    }
 
     return map { [ "${as}.$_", "${as_prefix}${pre}.$_", ] }
-      $rel_source->columns;
+      $self->related_source($pre)->columns;
   }
 }
 
@@ -155,11 +115,11 @@ sub _resolve_collapse {
         $_ =~ s/^ (?: foreign | self ) \.//x for ($f, $s);
         $relinfo->{$rel}{fk_map}{$s} = $f;
 
-        # need to know source from *our* pov, hnce $rel.
+        # need to know source from *our* pov, hence $rel.
         $my_cols->{$s} ||= { via_fk => "$rel.$f" } if (
           defined $rel_cols->{$rel}{$f} # in fact selected
             and
-          (! $node_idx_ref or $relinfo->{$rel}{is_inner}) # either top-level or an inner join
+          $relinfo->{$rel}{is_inner}
         );
       }
     }
@@ -194,14 +154,14 @@ sub _resolve_collapse {
   if (
     $my_cols
       and
-    my $uset = $self->_unique_column_set ($my_cols)
+    my $idset = $self->_identifying_column_set ({map { $_ => $my_cols->{$_}{colinfo} } keys %$my_cols})
   ) {
     # see if the resulting collapser relies on any implied columns,
     # and fix stuff up if this is the case
+    my @reduced_set = grep { ! $assumed_from_parent->{columns}{$_} } @$idset;
 
-    my $parent_collapser_used = defined delete @{$uset}{keys %{$assumed_from_parent->{columns}}};
     $collapse_map->{-node_id} = __unique_numlist(
-      $parent_collapser_used ? @{$parent_info->{collapse_on}} : (),
+      (@reduced_set != @$idset) ? @{$parent_info->{collapse_on}} : (),
       (map
         {
           my $fqc = join ('.',
@@ -211,7 +171,7 @@ sub _resolve_collapse {
 
           $as_fq_idx->{$fqc};
         }
-        keys %$uset
+        @reduced_set
       ),
     );
   }
@@ -309,26 +269,6 @@ sub _resolve_collapse {
   return $collapse_map;
 }
 
-sub _unique_column_set {
-  my ($self, $cols) = @_;
-
-  my %unique = $self->unique_constraints;
-
-  # always prefer the PK first, and then shortest constraints first
-  USET:
-  for my $set (delete $unique{primary}, sort { @$a <=> @$b } (values %unique) ) {
-    next unless $set && @$set;
-
-    for (@$set) {
-      next USET unless ($cols->{$_} && $cols->{$_}{colinfo} && !$cols->{$_}{colinfo}{is_nullable} );
-    }
-
-    return { map { $_ => 1 } @$set };
-  }
-
-  return undef;
-}
-
 # Takes an arrayref of {as} dbic column aliases and the collapse and select
 # attributes from the same $rs (the slector requirement is a temporary
 # workaround), and returns a coderef capable of:
@@ -389,8 +329,6 @@ sub _mk_row_parser {
 
   my ($parser_src);
   if ($args->{collapse}) {
-    # FIXME - deal with unorderedness
-    #    unordered => $unordered
 
     my $collapse_map = $self->_resolve_collapse (
       # FIXME
@@ -406,27 +344,21 @@ sub _mk_row_parser {
       }
     );
 
-    my $unrolled_top_branch_id_indexes = join (', ', @{$collapse_map->{-branch_id}});
+    my $top_branch_idx_list = join (', ', @{$collapse_map->{-branch_id}});
 
-    my ($sequenced_top_branch_id, $sequenced_top_node_id) = map
-      { join ('', map { "{'\xFF__IDVALPOS__${_}__\xFF'}" } @$_ ) }
-      $collapse_map->{-branch_id}, $collapse_map->{-node_id}
-    ;
+    my $top_node_id_path = join ('', map
+      { "{'\xFF__IDVALPOS__${_}__\xFF'}" }
+      @{$collapse_map->{-node_id}}
+    );
 
-    my $rolled_out_assemblers = __visit_infmap_collapse (
+    my $rel_assemblers = __visit_infmap_collapse (
       $inflate_index, $collapse_map
     );
-    my @sprintf_args = (
-      $unrolled_top_branch_id_indexes,
-      $sequenced_top_branch_id,
-      $sequenced_top_node_id,
-      $rolled_out_assemblers,
-    );
-
-    $parser_src = sprintf (<<'EOS', @sprintf_args);
 
+    $parser_src = sprintf (<<'EOS', $top_branch_idx_list, $top_node_id_path, $rel_assemblers);
 ### BEGIN STRING EVAL
-  my ($rows_pos, $result_pos, $cur_row, @cur_row_id_values, $is_new_res, @collapse_idx) = (0,0);
+
+  my ($rows_pos, $result_pos, $cur_row, @cur_row_ids, @collapse_idx, $is_new_res) = (0,0);
 
   # this loop is a bit arcane - the rationale is that the passed in
   # $_[0] will either have only one row (->next) or will have all
@@ -435,30 +367,21 @@ sub _mk_row_parser {
   # array, since the collapsed prefetch is smaller by definition.
   # At the end we cut the leftovers away and move on.
   while ($cur_row =
-    ( ( $rows_pos >= 0 and $_[0][$rows_pos++] ) or do { $rows_pos = -1; 0 } )
+    ( ( $rows_pos >= 0 and $_[0][$rows_pos++] ) or do { $rows_pos = -1; undef } )
       ||
     ($_[1] and $_[1]->())
   ) {
 
-    # FIXME
-    # optimize this away when we know we have no undefs in the collapse map
-    $cur_row_id_values[$_] = defined $cur_row->[$_] ? $cur_row->[$_] : "\xFF\xFFN\xFFU\xFFL\xFFL\xFF\xFF"
+    $cur_row_ids[$_] = defined $cur_row->[$_] ? $cur_row->[$_] : "\xFF\xFFN\xFFU\xFFL\xFFL\xFF\xFF"
       for (%1$s); # the top branch_id includes all id values
 
-    # check top branch for doubling via a has_many non-selecting join or something
-    # 0 is reserved for this (node indexes start from 1)
-    next if $collapse_idx[0]%2$s++;
-
-    $is_new_res = ! $collapse_idx[1]%3$s;
-
-    # lazify
-    # fire on ordered only
-#    if ($is_new_res = ! $collapse_idx[1]{$cur_row_id_values[2]}) {
-#    }
+    $is_new_res = ! $collapse_idx[1]%2$s and (
+      $_[1] and $result_pos and (unshift @{$_[2]}, $cur_row) and last
+    );
 
-    %4$s
+    %3$s
 
-    $_[0][$result_pos++] = $collapse_idx[1]%3$s
+    $_[0][$result_pos++] = $collapse_idx[1]%2$s
       if $is_new_res;
   }
 
@@ -468,7 +391,7 @@ EOS
 
     # change the quoted placeholders to unquoted alias-references
     $parser_src =~ s/ \' \xFF__VALPOS__(\d+)__\xFF \' /"\$cur_row->[$1]"/gex;
-    $parser_src =~ s/ \' \xFF__IDVALPOS__(\d+)__\xFF \' /"\$cur_row_id_values[$1]"/gex;
+    $parser_src =~ s/ \' \xFF__IDVALPOS__(\d+)__\xFF \' /"\$cur_row_ids[$1]"/gex;
   }
 
   else {
@@ -557,7 +480,7 @@ sub __visit_infmap_collapse {
 
   my $me_struct = keys %$my_cols
     ? __visit_dump([{ map { $_ => "\xFF__VALPOS__$my_cols->{$_}__\xFF" } (keys %$my_cols) }])
-    : 'undef'
+    : undef
   ;
   my $node_idx_ref = sprintf '$collapse_idx[%d]%s', $collapse_map->{-node_index}, $sequenced_node_id;
 
@@ -571,20 +494,20 @@ sub __visit_infmap_collapse {
     push @src, sprintf( '%s ||= %s;',
       $node_idx_ref,
       $me_struct,
-    );
+    ) if $me_struct;
   }
   elsif ($collapse_map->{-is_single}) {
-    push @src, sprintf ( '%s = %s ||= %s;',
+    push @src, sprintf ( '%s ||= %s%s;',
       $parent_idx_ref,
       $node_idx_ref,
-      $me_struct,
+      $me_struct ? " ||= $me_struct" : '',
     );
   }
   else {
-    push @src, sprintf('push @{%s}, %s = %s if !%s;',
+    push @src, sprintf('push @{%s}, %s%s unless %s;',
       $parent_idx_ref,
       $node_idx_ref,
-      $me_struct,
+      $me_struct ? " ||= $me_struct" : '',
       $node_idx_ref,
     );
   }
@@ -594,7 +517,8 @@ sub __visit_infmap_collapse {
 
   for my $rel (sort keys %$rel_cols) {
 
-    push @src, sprintf( '%s[1]{%s} ||= [];', $node_idx_ref, perlstring($rel) );
+    push @src, sprintf( '%s[1]{%s} ||= [];', $node_idx_ref, perlstring($rel) )
+      unless $collapse_map->{$rel}{-is_single};
 
     push @src,  __visit_infmap_collapse($rel_cols->{$rel}, $collapse_map->{$rel}, {
       node_idx => $collapse_map->{-node_index},
@@ -605,7 +529,7 @@ sub __visit_infmap_collapse {
 
     # FIXME SUBOPTIMAL - disabled to satisfy t/resultset/inflate_result_api.t
     #if ($collapse_map->{$rel}{-is_optional} and my @null_checks = map
-    #  { "(! defined '\xFF__VALPOS__${_}__\xFF')" }
+    #  { "(! defined '\xFF__IDVALPOS__${_}__\xFF')" }
     #  sort { $a <=> $b } grep
     #    { ! $known_defined->{$_} }
     #    @{$collapse_map->{$rel}{-node_id}}
diff --git a/t/90join_torture.t b/t/90join_torture.t
index aa8c3fb..ef5dec5 100644
--- a/t/90join_torture.t
+++ b/t/90join_torture.t
@@ -50,6 +50,7 @@ lives_ok (sub {
           ON producer_2.producerid = cd_to_producer_2.producer
         JOIN artist artist ON artist.artistid = me.artist
       WHERE ( ( producer.name = ? AND producer_2.name = ? ) )
+      ORDER BY me.cdid
     )',
     [
       [ { sqlt_datatype => 'varchar', dbic_colname => 'producer.name', sqlt_size => 100 }
diff --git a/t/prefetch/_internals.t b/t/prefetch/_internals.t
deleted file mode 100644
index c7f7dd9..0000000
--- a/t/prefetch/_internals.t
+++ /dev/null
@@ -1,415 +0,0 @@
-use strict;
-use warnings;
-
-use Test::More;
-use lib qw(t/lib);
-use DBICTest;
-use B::Deparse;
-
-
-my $schema = DBICTest->init_schema(no_deploy => 1);
-
-my ($as, $vals, @pairs);
-
-# artwork-artist deliberately mixed around
-@pairs = (
-  'artwork_to_artist.artist_id' => '2',
-
-  'cd_id' => '1',
-
-  'artwork_to_artist.artwork_cd_id' => '1',
-
-  'cd.artist' => '1',
-  'cd.cdid' => '1',
-  'cd.title' => 'Spoonful of bees',
-
-  'cd.artist.artistid' => '7',
-  'cd.artist.name' => 'Caterwauler McCrae',
-  'artwork_to_artist.artist.name' => 'xenowhinycide',
-);
-while (@pairs) {
-  push @$as, shift @pairs;
-  push @$vals, shift @pairs;
-}
-
-=begin
-
-my $parser = $schema->source ('Artwork')->_mk_row_parser({
-  inflate_map => $as,
-  collapse => 1,
-});
-
-is_deeply (
-  $parser->($vals),
-  [
-    {
-      cd_id => 1,
-    },
-
-    {
-      artwork_to_artist => [
-        {
-          artist_id => 2,
-          artwork_cd_id => 1,
-        },
-        {
-          artist => [
-            {
-              name => 'xenowhinycide',
-            },
-            undef,
-            [ 2, 1 ], # inherited from artwork_to_artist (child-parent definition)
-          ],
-        },
-        [ 2, 1 ]  # artwork_to_artist own data, in selection order
-      ],
-
-      cd => [
-        {
-          artist => 1,
-          cdid => 1,
-          title => 'Spoonful of bees',
-        },
-        {
-          artist => [
-            {
-              artistid => 7,
-              name => 'Caterwauler McCrae',
-            },
-            undef,
-            [ 7 ], # our own id
-          ]
-        },
-        [ 1 ], # our cdid fk
-      ]
-    },
-    [ 1 ], # our id
-  ],
-  'generated row parser works as expected',
-);
-
-#=begin
-
-undef $_ for ($as, $vals);
-@pairs = (
-  'name' => 'Caterwauler McCrae',
-  'cds.tracks.cd' => '3',
-  'cds.tracks.title' => 'Fowlin',
-  'cds.tracks.cd_single.title' => 'Awesome single',
-);
-while (@pairs) {
-  push @$as, shift @pairs;
-  push @$vals, shift @pairs;
-}
-$parser = $schema->source ('Artist')->_mk_row_parser($as);
-
-is_deeply (
-  $parser->($vals),
-  [
-    {
-      name => 'Caterwauler McCrae'
-    },
-    {
-      cds => [
-        {},
-        {
-          tracks => [
-            {
-              cd => 3,
-              title => 'Fowlin'
-            },
-            {
-              cd_single => [
-                {
-                  title => 'Awesome single',
-                },
-              ],
-            },
-          ]
-        }
-      ]
-    }
-  ],
-  'generated parser works as expected over missing joins (no collapse)',
-);
-
-=cut
-
-undef $_ for ($as, $vals);
-@pairs = (
-    'tracks.lyrics.lyric_versions.text'                => 'unique when combined with the lyric collapsable by the 1:1 tracks-parent',
-    'existing_single_track.cd.artist.artistid'         => 'artist_id (gives uniq. to its entire parent chain)',
-    'existing_single_track.cd.artist.cds.year'         => 'non-unique cds col (year)',
-    'year'                                             => 'non unique main year',
-    'genreid'                                          => 'non-unique/nullable main genid',
-    'tracks.title'                                     => 'non-unique title (missing multicol const. part)',
-    'existing_single_track.cd.artist.cds.cdid'         => 'cds unique id col to give uniquiness to ...tracks.title below',
-    'latest_cd'                                        => 'random function (not a colname)',
-    'existing_single_track.cd.artist.cds.tracks.title' => 'unique track title (when combined with ...cds.cdid above)',
-    'existing_single_track.cd.artist.cds.genreid'      => 'nullable cds col (genreid)',
-);
-while (@pairs) {
-  push @$as, shift @pairs;
-  push @$vals, shift @pairs;
-}
-
-is_deeply (
-  $schema->source ('CD')->_resolve_collapse ( { map { $as->[$_] => $_ } (0 .. $#$as) } ),
-  {
-    -node_index => 1,
-    -node_id => [ 1 ], # existing_single_track.cd.artist.artistid
-    -branch_id => [ 0, 1, 5, 6, 8 ],
-
-    existing_single_track => {
-      -node_index => 2,
-      -node_id => [ 1 ], # existing_single_track.cd.artist.artistid
-      -branch_id => [ 1, 6, 8 ],
-      -is_single => 1,
-
-      cd => {
-        -node_index => 3,
-        -node_id => [ 1 ], # existing_single_track.cd.artist.artistid
-        -branch_id => [ 1, 6, 8 ],
-        -is_single => 1,
-
-        artist => {
-          -node_index => 4,
-          -node_id => [ 1 ], # existing_single_track.cd.artist.artistid
-          -branch_id => [ 1, 6, 8 ],
-          -is_single => 1,
-
-          cds => {
-            -node_index => 5,
-            -node_id => [ 6 ], # existing_single_track.cd.artist.cds.cdid
-            -branch_id => [ 6, 8 ],
-            -is_optional => 1,
-
-            tracks => {
-              -node_index => 6,
-              -node_id => [ 6, 8 ], # existing_single_track.cd.artist.cds.cdid, existing_single_track.cd.artist.cds.tracks.title
-              -branch_id => [ 6, 8 ],
-              -is_optional => 1,
-            }
-          }
-        }
-      }
-    },
-    tracks => {
-      -node_index => 7,
-      -node_id => [ 1, 5 ], # existing_single_track.cd.artist.artistid, tracks.title
-      -branch_id => [ 0, 1, 5 ],
-      -is_optional => 1,
-
-      lyrics => {
-        -node_index => 8,
-        -node_id => [ 1, 5 ], # existing_single_track.cd.artist.artistid, tracks.title
-        -branch_id => [ 0, 1, 5 ],
-        -is_single => 1,
-        -is_optional => 1,
-
-        lyric_versions => {
-          -node_index => 9,
-          -node_id => [ 0, 1, 5 ], # tracks.lyrics.lyric_versions.text, existing_single_track.cd.artist.artistid, tracks.title
-          -branch_id => [ 0, 1, 5 ],
-          -is_optional => 1,
-        },
-      },
-    }
-  },
-  'Correct collapse map constructed',
-);
-
-done_testing;
-__END__
-=cut
-
-my $parser = $schema->source ('CD')->_mk_row_parser ({ inflate_map => $as, collapse => 1 });
-
-=begin
-
-is_deeply (
-  $parser->($vals),
-  [
-    {
-      latest_cd => 'random function (not a colname)',
-      year => 'non unique main year',
-      genreid => 'non-unique/nullable main genid'
-    },
-    {
-      existing_single_track => [
-        {},
-        {
-          cd => [
-            {},
-            {
-              artist => [
-                { artistid => 'artist_id (gives uniq. to its entire parent chain)' },
-                {
-                  cds => [
-                    {
-                      cdid => 'cds unique id col to give uniquiness to ...tracks.title below',
-                      year => 'non-unique cds col (year)',
-                      genreid => 'nullable cds col (genreid)'
-                    },
-                    {
-                      tracks => [
-                        {
-                          title => 'unique track title (when combined with ...cds.cdid above)'
-                        },
-                        undef,
-                        [
-                          'cds unique id col to give uniquiness to ...tracks.title below',
-                          'unique track title (when combined with ...cds.cdid above)',
-                        ],
-                      ]
-                    },
-                    [ 'cds unique id col to give uniquiness to ...tracks.title below' ],
-                  ]
-                },
-                [ 'artist_id (gives uniq. to its entire parent chain)' ],
-              ]
-            },
-            [ 'artist_id (gives uniq. to its entire parent chain)' ],
-          ]
-        },
-        [ 'artist_id (gives uniq. to its entire parent chain)' ],
-      ],
-      tracks => [
-        {
-          title => 'non-unique title (missing multicol const. part)'
-        },
-        {
-          lyrics => [
-            {},
-            {
-              lyric_versions => [
-                {
-                  text => 'unique when combined with the lyric collapsable by the 1:1 tracks-parent',
-                },
-                undef,
-                [
-                  'unique when combined with the lyric collapsable by the 1:1 tracks-parent',
-                  'artist_id (gives uniq. to its entire parent chain)',
-                  'non-unique title (missing multicol const. part)',
-                ],
-              ],
-            },
-            [
-              'artist_id (gives uniq. to its entire parent chain)',
-              'non-unique title (missing multicol const. part)',
-            ],
-          ],
-        },
-        [
-          'artist_id (gives uniq. to its entire parent chain)',
-          'non-unique title (missing multicol const. part)',
-        ],
-      ],
-    },
-    [ 'artist_id (gives uniq. to its entire parent chain)' ],
-  ],
-  'Proper row parser constructed',
-);
-
-=cut
-
-# For extra insanity test/showcase the parser's guts:
-my $deparser = B::Deparse->new;
-is (
-  $deparser->coderef2text ($parser),
-  $deparser->coderef2text ( sub { package DBIx::Class::ResultSource;
-    my $rows = [];
-    while (1) {
-      my $r = (shift @{$_[0]->{row_stash}}) || ($_[0]->{next_row} and $_[0]->{next_row}->()) || last;
-
-    }
-    return $rows
-
-
-    [
-      {
-        genreid => $_[0][4],
-        latest_cd => $_[0][7],
-        year => $_[0][3]
-      },
-      {
-
-        existing_single_track => [
-          {},
-          {
-            cd => [
-              {},
-              {
-                artist => [
-                  {
-                    artistid => $_[0][1]
-                  },
-                  {
-
-                    !defined($_[0][6]) ? () : (
-                    cds => [
-                      {
-                        cdid => $_[0][6],
-                        genreid => $_[0][9],
-                        year => $_[0][2]
-                      },
-                      {
-
-                        !defined($_[0][8]) ? () : (
-                        tracks => [
-                          {
-                            title => $_[0][8]
-                          },
-                          undef,
-                          [ $_[0][6], $_[0][8] ]
-                        ])
-
-                      },
-                      [ $_[0][6] ]
-                    ]),
-
-                  },
-                  [ $_[0][1] ],
-                ],
-              },
-              [ $_[0][1] ],
-            ],
-          },
-          [ $_[0][1] ],
-        ],
-
-        !defined($_[0][5]) ? () : (
-        tracks => [
-          {
-            title => $_[0][5],
-          },
-          {
-
-            lyrics => [
-              {},
-              {
-
-                !defined($_[0][0]) ? () : (
-                lyric_versions => [
-                  {
-                    text => $_[0][0]
-                  },
-                  undef,
-                  [ $_[0][0], $_[0][1], $_[0][5] ],
-                ]),
-
-              },
-              [ $_[0][1], $_[0][5] ],
-            ],
-
-          },
-          [ $_[0][1], $_[0][5] ],
-        ]),
-      },
-      [ $_[0][1] ],
-    ];
-  }),
-  'Deparsed version of the parser coderef looks correct',
-);
-
-done_testing;
diff --git a/t/prefetch/correlated.t b/t/prefetch/correlated.t
index 694cf0b..3506027 100644
--- a/t/prefetch/correlated.t
+++ b/t/prefetch/correlated.t
@@ -45,6 +45,7 @@ is_same_sql_bind(
       LEFT JOIN track tracks
         ON tracks.cd = me.cdid
     WHERE me.artist != ?
+    ORDER BY me.cdid
   )',
   [
 
@@ -116,6 +117,7 @@ is_same_sql_bind(
       LEFT JOIN track tracks
         ON tracks.cd = me.cdid
     WHERE me.artist != ?
+    ORDER BY me.cdid
   )',
   [
 
diff --git a/t/prefetch/grouped.t b/t/prefetch/grouped.t
index 27d3865..c50b7ef 100644
--- a/t/prefetch/grouped.t
+++ b/t/prefetch/grouped.t
@@ -294,8 +294,10 @@ for ($cd_rs->all) {
               FROM cd me
               JOIN artist artist ON artist.artistid = me.artist
             GROUP BY me.cdid, me.artist, me.title, me.year, me.genreid, me.single_track
+            ORDER BY me.cdid
           ) me
           JOIN artist artist ON artist.artistid = me.artist
+          ORDER BY me.cdid
       )',
       [],
     );
@@ -321,12 +323,14 @@ for ($cd_rs->all) {
               JOIN artist artist ON artist.artistid = me.artist
             WHERE ( tracks.title != ? )
             GROUP BY me.cdid, me.artist, me.title, me.year, me.genreid, me.single_track
+            ORDER BY me.cdid
           ) me
           LEFT JOIN track tracks ON tracks.cd = me.cdid
           JOIN artist artist ON artist.artistid = me.artist
         WHERE ( tracks.title != ? )
         GROUP BY me.cdid, me.artist, me.title, me.year, me.genreid, me.single_track,
                  artist.artistid, artist.name, artist.rank, artist.charfield
+        ORDER BY me.cdid
       )',
       [ map { [ { sqlt_datatype => 'varchar', sqlt_size => 100, dbic_colname => 'tracks.title' }
             => 'ugabuganoexist' ] } (1,2)
diff --git a/t/prefetch/incomplete.t b/t/prefetch/incomplete.t
index 5d4aee5..781c1e1 100644
--- a/t/prefetch/incomplete.t
+++ b/t/prefetch/incomplete.t
@@ -10,7 +10,7 @@ my $schema = DBICTest->init_schema();
 
 lives_ok(sub {
   # while cds.* will be selected anyway (prefetch currently forces the result of _resolve_prefetch)
-  # only the requested me.name column will be fetched.
+  # only the requested me.name/me.artistid columns will be fetched.
 
   # reference sql with select => [...]
   #   SELECT me.name, cds.title, me.artistid, cds.cdid, cds.artist, cds.title, cds.year, cds.genreid, cds.single_track FROM ...
diff --git a/t/prefetch/join_type.t b/t/prefetch/join_type.t
index 1698d6f..10a8783 100644
--- a/t/prefetch/join_type.t
+++ b/t/prefetch/join_type.t
@@ -38,6 +38,7 @@ is_same_sql_bind (
       JOIN artist artist ON artist.artistid = me.artist
       LEFT JOIN cd cds ON cds.artist = artist.artistid
       LEFT JOIN artist artist_2 ON artist_2.artistid = cds.artist
+    ORDER BY me.cdid
   )',
   [],
 );
diff --git a/t/prefetch/lazy_cursor.t b/t/prefetch/lazy_cursor.t
new file mode 100644
index 0000000..ef7d5ec
--- /dev/null
+++ b/t/prefetch/lazy_cursor.t
@@ -0,0 +1,73 @@
+use strict;
+use warnings;
+
+use Test::More;
+use Test::Exception;
+use lib qw(t/lib);
+use DBICTest;
+
+my $schema = DBICTest->init_schema();
+
+my $rs = $schema->resultset('Artist')->search({}, {
+  select => 'artistid',
+  prefetch => { cds => 'tracks' },
+});
+
+my $initial_artists_cnt = $rs->count;
+
+# create one extra artist with just one cd with just one track
+# and then an artist with nothing at all
+# the implicit order by me.artistid will get them back in correct order
+$rs->create({
+  name => 'foo',
+  cds => [{
+    year => 2012,
+    title => 'foocd',
+    tracks => [{
+      title => 'footrack',
+    }]
+  }],
+});
+$rs->create({ name => 'bar' });
+$rs->create({ name => 'baz' });
+
+# make sure we are reentrant, and also check with explicit order_by
+for (undef, undef, 'me.artistid') {
+  $rs = $rs->search({}, { order_by => $_ }) if $_;
+
+  for (1 .. $initial_artists_cnt) {
+    is ($rs->next->artistid, $_, 'Default fixture artists in order') || exit;
+  }
+
+  my $foo_artist = $rs->next;
+  is ($foo_artist->cds->next->tracks->next->title, 'footrack', 'Right track');
+
+  is (
+    [$rs->cursor->next]->[0],
+    $initial_artists_cnt + 3,
+    'Very last artist still on the cursor'
+  );
+
+  is_deeply ([$rs->cursor->next], [], 'Nothing else left');
+
+  is ($rs->next->artistid, $initial_artists_cnt + 2, 'Row stashed in resultset still accessible');
+  is ($rs->next, undef, 'Nothing left in resultset either');
+
+  $rs->reset;
+}
+
+$rs->next;
+
+my @objs = $rs->all;
+is (@objs, $initial_artists_cnt + 3, '->all resets everything correctly');
+is ( ($rs->cursor->next)[0], 1, 'Cursor auto-rewound after all()');
+is ($rs->{stashed_rows}, undef, 'Nothing else left in $rs stash');
+
+my $unordered_rs = $rs->search({}, { order_by => 'cds.title' });
+ok ($unordered_rs->next, 'got row 1');
+is_deeply ([$unordered_rs->cursor->next], [], 'Nothing left on cursor, eager slurp');
+ok ($unordered_rs->next, "got row $_")  for (2 .. $initial_artists_cnt + 3);
+is ($unordered_rs->next, undef, 'End of RS reached');
+is ($unordered_rs->next, undef, 'End of RS not lost');
+
+done_testing;
diff --git a/t/prefetch/manual.t b/t/prefetch/manual.t
index 72bde38..7a22245 100644
--- a/t/prefetch/manual.t
+++ b/t/prefetch/manual.t
@@ -203,4 +203,27 @@ TODO: {
 
 is ($rs->cursor->next, undef, 'cursor exhausted');
 
+TODO: {
+local $TODO = 'this does not work at all, need to promote rsattrs to an object on its own';
+# make sure has_many column redirection does not do weird stuff when collapse is requested
+for my $pref_args (
+  { prefetch => 'cds'},
+  { collapse => 1 }
+) {
+  for my $col_and_join_args (
+    { '+columns' => { 'cd_title' => 'cds_2.title' }, join => [ 'cds', 'cds' ] },
+    { '+columns' => { 'cd_title' => 'cds.title' }, join => 'cds', }
+  ) {
+
+    my $weird_rs = $schema->resultset('Artist')->search({}, {
+      %$col_and_join_args, %$pref_args,
+    });
+
+    for (qw/next all first/) {
+      throws_ok { $weird_rs->$_ } qr/not yet determined exception text/;
+    }
+  }
+}
+}
+
 done_testing;
diff --git a/t/prefetch/standard.t b/t/prefetch/standard.t
index 56781be..493b538 100644
--- a/t/prefetch/standard.t
+++ b/t/prefetch/standard.t
@@ -253,6 +253,11 @@ sub make_hash_struc {
     my $rs = shift;
 
     my $struc = {};
+    # all of these ought to work, but do not for some reason
+    # a noop cloning search() pollution?
+    #foreach my $art ( $rs->search({}, { order_by => 'me.artistid' })->all ) {
+    #foreach my $art ( $rs->search({}, {})->all ) {
+    #foreach my $art ( $rs->search()->all ) {
     foreach my $art ( $rs->all ) {
         foreach my $cd ( $art->cds ) {
             foreach my $track ( $cd->tracks ) {
diff --git a/t/prefetch/with_limit.t b/t/prefetch/with_limit.t
index 97dffcc..522324c 100644
--- a/t/prefetch/with_limit.t
+++ b/t/prefetch/with_limit.t
@@ -183,6 +183,7 @@ is_same_sql_bind (
           FROM cd me
           JOIN artist artist ON artist.artistid = me.artist
         WHERE ( ( artist.name = ? AND me.year = ? ) )
+        ORDER BY me.cdid
         LIMIT ?
       ) me
       LEFT JOIN track tracks
@@ -190,6 +191,7 @@ is_same_sql_bind (
       JOIN artist artist
         ON artist.artistid = me.artist
     WHERE ( ( artist.name = ? AND me.year = ? ) )
+    ORDER BY me.cdid
   )',
   [
     [ { sqlt_datatype => 'varchar', sqlt_size => 100, dbic_colname => 'artist.name' } => 'foo' ],
diff --git a/t/relationship/custom.t b/t/relationship/custom.t
index 543c7c0..98b8b45 100644
--- a/t/relationship/custom.t
+++ b/t/relationship/custom.t
@@ -139,9 +139,6 @@ is_deeply(
   '16 correct cds found'
 );
 
-TODO: {
-local $TODO = 'Prefetch on custom rels can not work until the collapse rewrite is finished '
-  . '(currently collapser requires a right-side (which is indeterministic) order-by)';
 lives_ok {
 
 my @all_artists_with_80_cds_pref = $schema->resultset("Artist")->search
@@ -154,7 +151,6 @@ is_deeply(
 );
 
 } 'prefetchy-fetchy-fetch';
-} # end of TODO
 
 
 # try to create_related a 80s cd
diff --git a/t/resultset/rowparser_internals.t b/t/resultset/rowparser_internals.t
new file mode 100644
index 0000000..5bcf939
--- /dev/null
+++ b/t/resultset/rowparser_internals.t
@@ -0,0 +1,301 @@
+use strict;
+use warnings;
+
+use Test::More;
+use lib qw(t/lib);
+use DBICTest;
+use B::Deparse;
+
+my $schema = DBICTest->init_schema(no_deploy => 1);
+my $infmap = [qw/single_track.cd.artist.name year/];
+
+is_same_src (
+  $schema->source ('CD')->_mk_row_parser({
+    inflate_map => $infmap,
+  }),
+  '$_ = [
+    { year => $_->[1] },
+    { single_track => [
+      undef,
+      { cd => [
+        undef,
+        { artist => [
+          { name  => $_->[0] },
+        ] },
+      ]},
+    ]},
+  ] for @{$_[0]}',
+  'Simple 1:1 descending non-collapsing parser',
+);
+
+$infmap = [qw/
+  single_track.cd.artist.artistid
+  year
+  single_track.cd.artist.cds.tracks.title
+  single_track.cd.artist.cds.cdid
+  title
+  artist
+/];
+is_same_src (
+  $schema->source ('CD')->_mk_row_parser({
+    inflate_map => $infmap,
+  }),
+  '$_ = [
+    { artist => $_->[5], title => $_->[4], year => $_->[1] },
+    { single_track => [
+      undef,
+      { cd => [
+        undef,
+        { artist => [
+          { artistid => $_->[0] },
+          { cds => [
+            { cdid => $_->[3] },
+            { tracks => [
+              { title => $_->[2] }
+            ] },
+          ] },
+        ] },
+      ] },
+    ] },
+  ] for @{$_[0]}',
+  '1:1 descending non-collapsing parser terminating with chained 1:M:M',
+);
+
+is_deeply (
+  $schema->source('CD')->_resolve_collapse({map { $infmap->[$_] => $_ } 0 .. $#$infmap}),
+  {
+    -node_index => 1,
+    -node_id => [ 4, 5 ],
+    -branch_id => [ 0, 2, 3, 4, 5 ],
+
+    single_track => {
+      -node_index => 2,
+      -node_id => [ 4, 5],
+      -branch_id => [ 0, 2, 3, 4, 5],
+      -is_optional => 1,
+      -is_single => 1,
+
+      cd => {
+        -node_index => 3,
+        -node_id => [ 4, 5 ],
+        -branch_id => [ 0, 2, 3, 4, 5 ],
+        -is_single => 1,
+
+        artist => {
+          -node_index => 4,
+          -node_id => [ 0 ],
+          -branch_id => [ 0, 2, 3 ],
+          -is_single => 1,
+
+          cds => {
+            -node_index => 5,
+            -node_id => [ 3 ],
+            -branch_id => [ 2, 3 ],
+            -is_optional => 1,
+
+            tracks => {
+              -node_index => 6,
+              -node_id => [ 2, 3 ],
+              -branch_id => [ 2, 3 ],
+              -is_optional => 1,
+            },
+          },
+        },
+      },
+    },
+  },
+  'Correct collapse map for 1:1 descending chain terminating with chained 1:M:M'
+);
+
+is_same_src (
+  $schema->source ('CD')->_mk_row_parser({
+    inflate_map => $infmap,
+    collapse => 1,
+  }),
+  ' my($rows_pos, $result_pos, $cur_row, @cur_row_ids, @collapse_idx, $is_new_res) = (0, 0);
+
+    while ($cur_row = (
+      ( $rows_pos >= 0 and $_[0][$rows_pos++] ) or do { $rows_pos = -1; undef } )
+        ||
+      ( $_[1] and $_[1]->() )
+    ) {
+
+      $cur_row_ids[$_] = defined $cur_row->[$_] ? $cur_row->[$_] : "\xFF\xFFN\xFFU\xFFL\xFFL\xFF\xFF"
+        for (0, 2, 3, 4, 5);
+
+      # a present cref implies lazy prefetch, implies a supplied stash in $_[2]
+      $_[1] and $result_pos and unshift(@{$_[2]}, $cur_row) and last
+        if $is_new_res = ! $collapse_idx[1]{$cur_row_ids[4]}{$cur_row_ids[5]};
+
+      $collapse_idx[1]{$cur_row_ids[4]}{$cur_row_ids[5]} ||= [{ artist => $cur_row->[5], title => $cur_row->[4], year => $cur_row->[1] }];
+      $collapse_idx[1]{$cur_row_ids[4]}{$cur_row_ids[5]}[1]{single_track} ||= $collapse_idx[2]{$cur_row_ids[4]}{$cur_row_ids[5]};
+      $collapse_idx[2]{$cur_row_ids[4]}{$cur_row_ids[5]}[1]{cd} ||= $collapse_idx[3]{$cur_row_ids[4]}{$cur_row_ids[5]};
+      $collapse_idx[3]{$cur_row_ids[4]}{$cur_row_ids[5]}[1]{artist} ||= $collapse_idx[4]{$cur_row_ids[0]} ||= [{ artistid => $cur_row->[0] }];
+
+      $collapse_idx[4]{$cur_row_ids[0]}[1]{cds} ||= [];
+      push @{$collapse_idx[4]{$cur_row_ids[0]}[1]{cds}}, $collapse_idx[5]{$cur_row_ids[3]} ||= [{ cdid => $cur_row->[3] }]
+        unless $collapse_idx[5]{$cur_row_ids[3]};
+
+      $collapse_idx[5]{$cur_row_ids[3]}[1]{tracks} ||= [];
+      push @{$collapse_idx[5]{$cur_row_ids[3]}[1]{tracks}}, $collapse_idx[6]{$cur_row_ids[2]}{$cur_row_ids[3]} ||= [{ title => $cur_row->[2] }]
+        unless $collapse_idx[6]{$cur_row_ids[2]}{$cur_row_ids[3]};
+
+      $_[0][$result_pos++] = $collapse_idx[1]{$cur_row_ids[4]}{$cur_row_ids[5]}
+        if $is_new_res;
+    }
+    splice @{$_[0]}, $result_pos;
+  ',
+  'Same 1:1 descending terminating with chained 1:M:M but with collapse',
+);
+
+$infmap = [qw/
+  tracks.lyrics.lyric_versions.text
+  existing_single_track.cd.artist.artistid
+  existing_single_track.cd.artist.cds.year
+  year
+  genreid
+  tracks.title
+  existing_single_track.cd.artist.cds.cdid
+  latest_cd
+  existing_single_track.cd.artist.cds.tracks.title
+  existing_single_track.cd.artist.cds.genreid
+/];
+
+is_deeply (
+  $schema->source('CD')->_resolve_collapse({map { $infmap->[$_] => $_ } 0 .. $#$infmap}),
+  {
+    -node_index => 1,
+    -node_id => [ 1 ], # existing_single_track.cd.artist.artistid
+    -branch_id => [ 0, 1, 5, 6, 8 ],
+
+    existing_single_track => {
+      -node_index => 2,
+      -node_id => [ 1 ], # existing_single_track.cd.artist.artistid
+      -branch_id => [ 1, 6, 8 ],
+      -is_single => 1,
+
+      cd => {
+        -node_index => 3,
+        -node_id => [ 1 ], # existing_single_track.cd.artist.artistid
+        -branch_id => [ 1, 6, 8 ],
+        -is_single => 1,
+
+        artist => {
+          -node_index => 4,
+          -node_id => [ 1 ], # existing_single_track.cd.artist.artistid
+          -branch_id => [ 1, 6, 8 ],
+          -is_single => 1,
+
+          cds => {
+            -node_index => 5,
+            -node_id => [ 6 ], # existing_single_track.cd.artist.cds.cdid
+            -branch_id => [ 6, 8 ],
+            -is_optional => 1,
+
+            tracks => {
+              -node_index => 6,
+              -node_id => [ 6, 8 ], # existing_single_track.cd.artist.cds.cdid, existing_single_track.cd.artist.cds.tracks.title
+              -branch_id => [ 6, 8 ],
+              -is_optional => 1,
+            }
+          }
+        }
+      }
+    },
+    tracks => {
+      -node_index => 7,
+      -node_id => [ 1, 5 ], # existing_single_track.cd.artist.artistid, tracks.title
+      -branch_id => [ 0, 1, 5 ],
+      -is_optional => 1,
+
+      lyrics => {
+        -node_index => 8,
+        -node_id => [ 1, 5 ], # existing_single_track.cd.artist.artistid, tracks.title
+        -branch_id => [ 0, 1, 5 ],
+        -is_single => 1,
+        -is_optional => 1,
+
+        lyric_versions => {
+          -node_index => 9,
+          -node_id => [ 0, 1, 5 ], # tracks.lyrics.lyric_versions.text, existing_single_track.cd.artist.artistid, tracks.title
+          -branch_id => [ 0, 1, 5 ],
+          -is_optional => 1,
+        },
+      },
+    }
+  },
+  'Correct collapse map constructed',
+);
+
+is_same_src (
+  $schema->source ('CD')->_mk_row_parser({
+    inflate_map => $infmap,
+    collapse => 1,
+  }),
+  ' my ($rows_pos, $result_pos, $cur_row, @cur_row_ids, @collapse_idx, $is_new_res) = (0,0);
+
+    while ($cur_row = (
+      ( $rows_pos >= 0 and $_[0][$rows_pos++] ) or do { $rows_pos = -1; undef } )
+        ||
+      ( $_[1] and $_[1]->() )
+    ) {
+
+      $cur_row_ids[$_] = defined $cur_row->[$_] ? $cur_row->[$_] : "\xFF\xFFN\xFFU\xFFL\xFFL\xFF\xFF"
+        for (0, 1, 5, 6, 8);
+
+      $is_new_res = ! $collapse_idx[1]{$cur_row_ids[1]} and (
+        $_[1] and $result_pos and (unshift @{$_[2]}, $cur_row) and last
+      );
+
+      $collapse_idx[1]{$cur_row_ids[1]} ||= [{ latest_cd => $cur_row->[7], year => $cur_row->[3], genreid => $cur_row->[4] }];
+
+      $collapse_idx[1]{$cur_row_ids[1]}[1]{existing_single_track} ||= $collapse_idx[2]{$cur_row_ids[1]};
+      $collapse_idx[2]{$cur_row_ids[1]}[1]{cd} ||= $collapse_idx[3]{$cur_row_ids[1]};
+      $collapse_idx[3]{$cur_row_ids[1]}[1]{artist} ||= $collapse_idx[4]{$cur_row_ids[1]} ||= [{ artistid => $cur_row->[1] }];
+
+      $collapse_idx[4]{$cur_row_ids[1]}[1]{cds} ||= [];
+      push @{ $collapse_idx[4]{$cur_row_ids[1]}[1]{cds} }, $collapse_idx[5]{$cur_row_ids[6]} ||= [{ cdid => $cur_row->[6], genreid => $cur_row->[9], year => $cur_row->[2] }]
+        unless $collapse_idx[5]{$cur_row_ids[6]};
+
+      $collapse_idx[5]{$cur_row_ids[6]}[1]{tracks} ||= [];
+      push @{ $collapse_idx[5]{$cur_row_ids[6]}[1]{tracks} }, $collapse_idx[6]{$cur_row_ids[6]}{$cur_row_ids[8]} ||= [{ title => $cur_row->[8] }]
+        unless $collapse_idx[6]{$cur_row_ids[6]}{$cur_row_ids[8]};
+
+      $collapse_idx[1]{$cur_row_ids[1]}[1]{tracks} ||= [];
+      push @{ $collapse_idx[1]{$cur_row_ids[1]}[1]{tracks} }, $collapse_idx[7]{$cur_row_ids[1]}{$cur_row_ids[5]} ||= [{ title => $cur_row->[5] }]
+        unless $collapse_idx[7]{$cur_row_ids[1]}{$cur_row_ids[5]};
+
+      $collapse_idx[7]{$cur_row_ids[1]}{$cur_row_ids[5]}[1]{lyrics} ||= $collapse_idx[8]{$cur_row_ids[1]}{$cur_row_ids[5] };
+
+      $collapse_idx[8]{$cur_row_ids[1]}{$cur_row_ids[5]}[1]{lyric_versions} ||= [];
+      push @{ $collapse_idx[8]{$cur_row_ids[1]}{$cur_row_ids[5]}[1]{lyric_versions} }, $collapse_idx[9]{$cur_row_ids[0]}{$cur_row_ids[1]}{$cur_row_ids[5]} ||= [{ text => $cur_row->[0] }]
+        unless $collapse_idx[9]{$cur_row_ids[0]}{$cur_row_ids[1]}{$cur_row_ids[5]};
+
+      $_[0][$result_pos++] = $collapse_idx[1]{$cur_row_ids[1]}
+        if $is_new_res;
+    }
+
+    splice @{$_[0]}, $result_pos;
+  ',
+  'Multiple has_many on multiple branches torture test',
+);
+
+done_testing;
+
+my $deparser;
+sub is_same_src {
+  $deparser ||= B::Deparse->new;
+  local $Test::Builder::Level = $Test::Builder::Level + 1;
+
+  my ($got, $expect) = map {
+    my $cref = eval "sub { $_ }" or do {
+      fail "Coderef does not compile!\n\n$@\n\n$_";
+      return undef;
+    };
+    $deparser->coderef2text($cref);
+  } @_[0,1];
+
+  is ($got, $expect, $_[2]||() )
+    or note ("Originals source:\n\n$_[0]\n\n$_[1]\n");
+}
+
diff --git a/t/sqlmaker/limit_dialects/rownum.t b/t/sqlmaker/limit_dialects/rownum.t
index 2f46599..9e896fe 100644
--- a/t/sqlmaker/limit_dialects/rownum.t
+++ b/t/sqlmaker/limit_dialects/rownum.t
@@ -146,7 +146,7 @@ for my $test_set (
         { id => 'foo.id' },
         { 'ends_with_me.id' => 'ends_with_me.id' },
       ],
-      order_by => [qw( artist title )],
+      order_by => [qw( year artist title )],
     }),
     sql => '(
       SELECT id, ends_with_me__id
@@ -156,7 +156,7 @@ for my $test_set (
           SELECT foo.id AS id, ends_with_me.id AS ends_with_me__id
             FROM cd me
           WHERE id = ?
-          ORDER BY artist, title
+          ORDER BY year, artist, title
         ) me
         WHERE ROWNUM <= ?
       ) me