From: Peter Rabbitson <ribasushi@cpan.org>
Date: Mon, 11 Feb 2013 16:07:16 +0000 (+0100)
Subject: Reintroduce conditional null-branch pruning and add direct-to-HRI option
X-Git-Tag: v0.08240~2
X-Git-Url: http://git.shadowcat.co.uk/gitweb/gitweb.cgi?a=commitdiff_plain;h=ce556881;p=dbsrgits%2FDBIx-Class.git

Reintroduce conditional null-branch pruning and add direct-to-HRI option

What we do here is heavy parameterization of the rowparser coderef generator.

The first change introduces pruning of "null" left joined branches. In the
case of the collapsing codepath things are easy - we already precalculated
definitive non-nullable column sets describing each node separately. All we
need to do is check that the *first* node-specific id is not-NULL, and we
are in business. In the case of non-collapsing parsers things get ugly - we
do not have a collapse map (it can't be calculated most of the time) and
we do not yet have lightweight "left-join-ness" analysis. So instead we treat
any branch with *all* of its values (current node + leaves) being NULL as
a "left joined" branch. This should eventually be fixed, because it prevents
reliable selection of branches with all-nullable columns.

To make matters even more complicated - we can not prune by default, because
this was not the case for older code (and the format of data fed to
inflate_result has been public API for a long time). Since DBIx::Class::Row
no longer does prunning internaly in its inflate_result we need to compromise
(it was in fact necessary to remove that naive pruning code to allow empty
intermediate objects). The pruning is engaged automatically on any
result class that inherits from DBIx::Class::Row, with the hope that any
overrides of inflate_result do superficial modifications to the data before
passing it on. If the class is entirely new (i.e. *not* part of the ::Row
inheritance chain) - we do not prune and return everything the way we did up
until 0.08206 or so.

There is currently no way to influence the pruning behavior from the user API
which is *probably* a problem, but I am punting it.

The second change introduces the hri_style flag, which alters the resulting
structure to be a direct HRI product (everything is mixed in one hashref,
relationships clobber identically named coluns). Thus we do not invoke the HRI
inflate_result() at all anymore (it is left in for the casual overrider), but
return things in a single pass. There aren't much more performance gains to
be had here, next step would be leaner ResultSet initialization.

The switch and decision-caching in ResultSet.pm that powers the above got
rather hairy, but I couldn't figure out a saner way to properly dispatch to
the necessary rowparser builder. The rest of the changes turned out
surprisingly clean and unobtrusive.
---

diff --git a/lib/DBIx/Class/ResultSet.pm b/lib/DBIx/Class/ResultSet.pm
index 70025db..460a233 100644
--- a/lib/DBIx/Class/ResultSet.pm
+++ b/lib/DBIx/Class/ResultSet.pm
@@ -1281,18 +1281,15 @@ sub _construct_objects {
   # 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...
   # a suprising amount actually
-  my $rows = (delete $self->{stashed_rows}) || [];
+  my $rows = delete $self->{stashed_rows};
+
   if ($fetch_all) {
     # FIXME SUBOPTIMAL - we can do better, cursor->next/all (well diff. methods) should return a ref
-    $rows = [ @$rows, $cursor->all ];
-  }
-  elsif (!$attrs->{collapse}) {
-    # FIXME SUBOPTIMAL - we can do better, cursor->next/all (well diff. methods) should return a ref
-    push @$rows, do { my @r = $cursor->next; @r ? \@r : () }
-      unless @$rows;
+    $rows = [ ($rows ? @$rows : ()), $cursor->all ];
   }
-  else {
-    $attrs->{_ordered_for_collapse} ||= (!$attrs->{order_by}) ? undef : do {
+  elsif( $attrs->{collapse} ) {
+
+    $attrs->{_ordered_for_collapse} = (!$attrs->{order_by}) ? 0 : do {
       my $st = $rsrc->schema->storage;
       my @ord_cols = map
         { $_->[0] }
@@ -1319,68 +1316,128 @@ sub _construct_objects {
         { $colinfos->{$_}{-colname} => $colinfos->{$_} }
         @ord_cols
       })) ? 1 : 0;
-    };
+    } unless defined $attrs->{_ordered_for_collapse};
 
-    if ($attrs->{_ordered_for_collapse}) {
-      push @$rows, do { my @r = $cursor->next; @r ? \@r : () };
-    }
-    # instead of looping over ->next, use ->all in stealth mode
-    # *without* calling a ->reset afterwards
-    # FIXME - encapsulation breach, got to be a better way
-    elsif (! $cursor->{_done}) {
-      push @$rows, $cursor->all;
-      $cursor->{_done} = 1;
+    if (! $attrs->{_ordered_for_collapse}) {
       $fetch_all = 1;
+
+      # instead of looping over ->next, use ->all in stealth mode
+      # *without* calling a ->reset afterwards
+      # FIXME - encapsulation breach, got to be a better way
+      if (! $cursor->{_done}) {
+        $rows = [ ($rows ? @$rows : ()), $cursor->all ];
+        $cursor->{_done} = 1;
+      }
     }
   }
 
-  return undef unless @$rows;
+  if (! $fetch_all and ! @{$rows||[]} ) {
+    # FIXME SUBOPTIMAL - we can do better, cursor->next/all (well diff. methods) should return a ref
+    if (scalar (my @r = $cursor->next) ) {
+      $rows = [ \@r ];
+    }
+  }
 
-  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");
+  return undef unless @{$rows||[]};
+
+  my @extra_collapser_args;
+  if ($attrs->{collapse} and ! $fetch_all ) {
+
+    @extra_collapser_args = (
+      # FIXME SUBOPTIMAL - we can do better, cursor->next/all (well diff. methods) should return a ref
+      sub { my @r = $cursor->next or return; \@r }, # how the collapser gets more rows
+      ($self->{stashed_rows} = []),                 # where does it stuff excess
+    );
+  }
+
+  # hotspot - skip the setter
+  my $res_class = $self->_result_class;
+
+  my $inflator_cref = $self->{_result_inflator}{cref} ||= do {
+    $res_class->can ('inflate_result')
+      or $self->throw_exception("Inflator $res_class does not provide an inflate_result() method");
+  };
 
   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
+  $self->{_result_inflator}{is_hri} = do { ( $inflator_cref == (
+    require DBIx::Class::ResultClass::HashRefInflator
+      &&
+    DBIx::Class::ResultClass::HashRefInflator->can('inflate_result')
+  ) ) ? 1 : 0
+  } unless defined $self->{_result_inflator}{is_hri};
 
+  if ($attrs->{_single_resultclass_inflation}) {
+    # construct a much simpler array->hash folder for the one-table cases right here
+    if ($self->{_result_inflator}{is_hri}) {
+      for my $r (@$rows) {
+        $r = { map { $infmap->[$_] => $r->[$_] } 0..$#$infmap };
+      }
+    }
     # 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
+    # building a smarter Row::inflate_result(), and
     # switch to feeding it data via a much leaner interface
     #
     # crude unscientific benchmarking indicated the shortcut eval is not worth it for
     # this particular resultset size
-    if (@$rows < 60) {
-      my @as_idx = 0..$#$infmap;
+    elsif (@$rows < 60) {
       for my $r (@$rows) {
-        $r = $inflator->($res_class, $rsrc, { map { $infmap->[$_] => $r->[$_] } @as_idx } );
+        $r = $inflator_cref->($res_class, $rsrc, { map { $infmap->[$_] => $r->[$_] } (0..$#$infmap) } );
       }
     }
     else {
       eval sprintf (
-        '$_ = $inflator->($res_class, $rsrc, { %s }) for @$rows',
+        '$_ = $inflator_cref->($res_class, $rsrc, { %s }) for @$rows',
         join (', ', map { "\$infmap->[$_] => \$_->[$_]" } 0..$#$infmap )
       );
     }
   }
-  else {
-    $self->{_row_parser} ||= eval sprintf 'sub { %s }', $rsrc->_mk_row_parser({
+  # Special-case multi-object HRI (we always prune)
+  elsif ($self->{_result_inflator}{is_hri}) {
+    ( $self->{_row_parser}{hri} ||= $rsrc->_mk_row_parser({
+      eval => 1,
       inflate_map => $infmap,
       selection => $attrs->{select},
       collapse => $attrs->{collapse},
       premultiplied => $attrs->{_main_source_premultiplied},
-    }) or die $@;
+      hri_style => 1,
+      prune_null_branches => 1,
+    }) )->($rows, @extra_collapser_args);
+  }
+  # Regular multi-object
+  else {
 
-    # modify $rows in-place, shrinking/extending as necessary
-    $self->{_row_parser}->($rows, $fetch_all ? () : (
-      # FIXME SUBOPTIMAL - we can do better, cursor->next/all (well diff. methods) should return a ref
-      sub { my @r = $cursor->next or return; \@r }, # how the collapser gets more rows
-      ($self->{stashed_rows} = []),                 # where does it stuff excess
-    ));
+    # The rationale is - if this is the ::Row inflator itself, or an around()
+    # we do prune, because we expect it.
+    # If not the case - let the user deal with the full output themselves
+    # Warn them while we are at it so we get a better idea what is out there
+    # on the DarkPan
+    $self->{_result_inflator}{prune_null_branches} = do {
+      $res_class->isa('DBIx::Class::Row')
+    } ? 1 : 0 unless defined $self->{_result_inflator}{prune_null_branches};
+
+    unless ($self->{_result_inflator}{prune_null_branches}) {
+      carp_once (
+        "ResultClass $res_class does not inherit from DBIx::Class::Row and "
+      . 'therefore its inflate_result() will receive the full prefetched data '
+      . 'tree, without any branch definedness checks. This is a compatibility '
+      . 'measure which will eventually disappear entirely. Please refer to '
+      . 't/resultset/inflate_result_api.t for an exhaustive description of the '
+      . 'upcoming changes'
+      );
+    }
 
-    $_ = $inflator->($res_class, $rsrc, @$_) for @$rows;
+    ( $self->{_row_parser}{classic}{$self->{_result_inflator}{prune_null_branches}} ||= $rsrc->_mk_row_parser({
+      eval => 1,
+      inflate_map => $infmap,
+      selection => $attrs->{select},
+      collapse => $attrs->{collapse},
+      premultiplied => $attrs->{_main_source_premultiplied},
+      prune_null_branches => $self->{_result_inflator}{prune_null_branches},
+    }) )->($rows, @extra_collapser_args);
+
+    $_ = $inflator_cref->($res_class, $rsrc, @$_) for @$rows;
   }
 
   # CDBI compat stuff
@@ -1428,6 +1485,7 @@ in the original source class will not run.
 sub result_class {
   my ($self, $result_class) = @_;
   if ($result_class) {
+
     unless (ref $result_class) { # don't fire this for an object
       $self->ensure_class_loaded($result_class);
     }
@@ -1436,6 +1494,8 @@ sub result_class {
     # permit the user to set result class on one result set only; it only
     # chains if provided to search()
     #$self->{attrs}{result_class} = $result_class if ref $self;
+
+    delete $self->{_result_inflator};
   }
   $self->_result_class;
 }
@@ -2996,7 +3056,6 @@ Returns a related resultset for the supplied relationship name.
 sub related_resultset {
   my ($self, $rel) = @_;
 
-  $self->{related_resultsets} ||= {};
   return $self->{related_resultsets}{$rel} ||= do {
     my $rsrc = $self->result_source;
     my $rel_info = $rsrc->relationship_info($rel);
@@ -3023,13 +3082,13 @@ sub related_resultset {
     #XXX - temp fix for result_class bug. There likely is a more elegant fix -groditi
     delete @{$attrs}{qw(result_class alias)};
 
-    my $new_cache;
+    my $related_cache;
 
     if (my $cache = $self->get_cache) {
-      if ($cache->[0] && $cache->[0]->related_resultset($rel)->get_cache) {
-        $new_cache = [ map { @{$_->related_resultset($rel)->get_cache||[]} }
-                        @$cache ];
-      }
+      $related_cache = [ map
+        { @{$_->related_resultset($rel)->get_cache||[]} }
+        @$cache
+      ];
     }
 
     my $rel_source = $rsrc->related_source($rel);
@@ -3052,7 +3111,7 @@ sub related_resultset {
                        where => $attrs->{where},
                    });
     };
-    $new->set_cache($new_cache) if $new_cache;
+    $new->set_cache($related_cache) if $related_cache;
     $new;
   };
 }
@@ -3500,7 +3559,7 @@ sub _resolved_attrs {
   }
 
   if ( ! List::Util::first { $_ =~ /\./ } @{$attrs->{as}} ) {
-    $attrs->{_single_object_inflation} = 1;
+    $attrs->{_single_resultclass_inflation} = 1;
     $attrs->{collapse} = 0;
   }
 
@@ -3774,7 +3833,7 @@ sub STORABLE_freeze {
 
   # A cursor in progress can't be serialized (and would make little sense anyway)
   # the parser can be regenerated (and can't be serialized)
-  delete @{$to_serialize}{qw/cursor _row_parser/};
+  delete @{$to_serialize}{qw/cursor _row_parser _result_inflator/};
 
   # 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') {
diff --git a/lib/DBIx/Class/ResultSource/RowParser.pm b/lib/DBIx/Class/ResultSource/RowParser.pm
index d4b75f2..065c36c 100644
--- a/lib/DBIx/Class/ResultSource/RowParser.pm
+++ b/lib/DBIx/Class/ResultSource/RowParser.pm
@@ -100,9 +100,13 @@ sub _mk_row_parser {
     ( 0 .. $#{$args->{inflate_map}} )
   };
 
+  my $src;
+
   if (! $args->{collapse} ) {
-    return assemble_simple_parser({
+    $src = assemble_simple_parser({
       val_index => $val_index,
+      hri_style => $args->{hri_style},
+      prune_null_branches => $args->{prune_null_branches},
     });
   }
   else {
@@ -122,11 +126,18 @@ sub _mk_row_parser {
       }
     });
 
-    return assemble_collapsing_parser({
+    $src = assemble_collapsing_parser({
       val_index => $val_index,
       collapse_map => $collapse_map,
+      hri_style => $args->{hri_style},
+      prune_null_branches => $args->{prune_null_branches},
     });
   }
+
+  return (! $args->{eval})
+    ? $src
+    : eval "sub { $src }" || die $@
+  ;
 }
 
 
diff --git a/lib/DBIx/Class/ResultSource/RowParser/Util.pm b/lib/DBIx/Class/ResultSource/RowParser/Util.pm
index a4e2eb5..7aa2b49 100644
--- a/lib/DBIx/Class/ResultSource/RowParser/Util.pm
+++ b/lib/DBIx/Class/ResultSource/RowParser/Util.pm
@@ -4,6 +4,7 @@ package # hide from the pauses
 use strict;
 use warnings;
 
+use List::Util 'first';
 use B 'perlstring';
 
 use base 'Exporter';
@@ -50,39 +51,42 @@ sub __visit_infmap_simple {
   my @relperl;
   for my $rel (sort keys %$rel_cols) {
 
-    # DISABLEPRUNE
-    #my $optional = $args->{is_optional};
-    #$optional ||= ($args->{rsrc}->relationship_info($rel)->{attrs}{join_type} || '') =~ /^left/i;
-
     push @relperl, join ' => ', perlstring($rel), __visit_infmap_simple({ %$args,
       val_index => $rel_cols->{$rel},
-      # DISABLEPRUNE
-      #non_top => 1,
-      #is_optional => $optional,
     });
 
-    # FIXME SUBOPTIMAL DISABLEPRUNE - disabled to satisfy t/resultset/inflate_result_api.t
-    #if ($optional and my @branch_null_checks = map
-    #  { "(! defined '\xFF__VALPOS__${_}__\xFF')" }
-    #  sort { $a <=> $b } values %{$rel_cols->{$rel}}
-    #) {
-    #  $relperl[-1] = sprintf ( '(%s) ? ( %s => [] ) : ( %s )',
-    #    join (' && ', @branch_null_checks ),
-    #    perlstring($rel),
-    #    $relperl[-1],
-    #  );
-    #}
+    if ($args->{prune_null_branches} and keys %$my_cols) {
+
+      my @branch_null_checks = map
+        { "( ! defined '\xFF__VALPOS__${_}__\xFF' )" }
+        sort { $a <=> $b } values %{$rel_cols->{$rel}}
+      ;
+
+      $relperl[-1] = sprintf ( '(%s) ? ( %s => %s ) : ( %s )',
+        join (' && ', @branch_null_checks ),
+        perlstring($rel),
+        $args->{hri_style} ? 'undef' : '[]',
+        $relperl[-1],
+      );
+    }
   }
 
-  my $me_struct = keys %$my_cols
-    ? __visit_dump({ map { $_ => "\xFF__VALPOS__$my_cols->{$_}__\xFF" } (keys %$my_cols) })
-    : 'undef'
-  ;
+  my $me_struct;
+  $me_struct = __visit_dump({ map { $_ => "\xFF__VALPOS__$my_cols->{$_}__\xFF" } (keys %$my_cols) })
+    if keys %$my_cols;
 
-  return sprintf '[%s]', join (',',
-    $me_struct,
-    @relperl ? sprintf ('{ %s }', join (',', @relperl)) : (),
-  );
+  if ($args->{hri_style}) {
+    $me_struct =~ s/^ \s* \{ | \} \s* $//gx
+      if $me_struct;
+
+    return sprintf '{ %s }', join (', ', $me_struct||(), @relperl);
+  }
+  else {
+    return sprintf '[%s]', join (',',
+      $me_struct || 'undef',
+      @relperl ? sprintf ('{ %s }', join (',', @relperl)) : (),
+    );
+  }
 }
 
 sub assemble_collapsing_parser {
@@ -100,7 +104,7 @@ sub assemble_collapsing_parser {
 
     my @path_parts = map { sprintf
       "( ( defined '\xFF__VALPOS__%d__\xFF' ) && (join qq(\xFF), '', %s, '') )",
-      $_->[0],  # checking just first is enough - one defined, all defined
+      $_->[0],  # checking just first is enough - one ID defined, all defined
       ( join ', ', map { "'\xFF__VALPOS__${_}__\xFF'" } @$_ ),
     } @variants;
 
@@ -126,10 +130,9 @@ sub assemble_collapsing_parser {
 
   my $list_of_idcols = join(', ', sort { $a <=> $b } keys %{ $stats->{idcols_seen} } );
 
-  my $parser_src = sprintf (<<'EOS', $list_of_idcols, $top_node_key, $top_node_key_assembler||'', $data_assemblers);
+  my $parser_src = sprintf (<<'EOS', $list_of_idcols, $top_node_key, $top_node_key_assembler||'', join( "\n", @{$data_assemblers||[]} ) );
 ### BEGIN LITERAL STRING EVAL
   my ($rows_pos, $result_pos, $cur_row_data, %%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
   # rows already pulled in (->all and/or unordered). Given that the
@@ -141,11 +144,10 @@ sub assemble_collapsing_parser {
       ||
     ($_[1] and $_[1]->())
   ) {
-
     # due to left joins some of the ids may be NULL/undef, and
     # won't play well when used as hash lookups
     # we also need to differentiate NULLs on per-row/per-col basis
-    #(otherwise folding of optional 1:1s will be greatly confused
+    # (otherwise folding of optional 1:1s will be greatly confused
     $cur_row_ids{$_} = defined $cur_row_data->[$_] ? $cur_row_data->[$_] : "\0NULL\xFF$rows_pos\xFF$_\0"
       for (%1$s);
 
@@ -182,7 +184,7 @@ sub __visit_infmap_collapse {
 
   my $cur_node_idx = ${ $args->{-node_idx_counter} ||= \do { my $x = 0} }++;
 
-  my ($my_cols, $rel_cols);
+  my ($my_cols, $rel_cols) = {};
   for ( keys %{$args->{val_index}} ) {
     if ($_ =~ /^ ([^\.]+) \. (.+) /x) {
       $rel_cols->{$1}{$2} = $args->{val_index}{$_};
@@ -192,79 +194,105 @@ sub __visit_infmap_collapse {
     }
   }
 
+
   my $node_key = $args->{collapse_map}->{-custom_node_key} || join ('', map
     { "{'\xFF__IDVALPOS__${_}__\xFF'}" }
     @{$args->{collapse_map}->{-identifying_columns}}
   );
 
-  my $me_struct = $my_cols
-    ? __visit_dump([{ map { $_ => "\xFF__VALPOS__$my_cols->{$_}__\xFF" } (keys %$my_cols) }])
-    : undef
-  ;
-  my $node_idx_slot = sprintf '$collapse_idx[%d]%s', $cur_node_idx, $node_key;
+  my $me_struct;
+
+  if ($args->{hri_style}) {
+    delete $my_cols->{$_} for grep { $rel_cols->{$_} } keys %$my_cols;
+  }
 
-  my $parent_attach_slot = sprintf( '$collapse_idx[%d]%s[1]{%s}',
-    @{$args}{qw/-parent_node_idx -parent_node_key/},
-    perlstring($args->{-node_relname}),
-  ) if $args->{-node_relname};
+  if (keys %$my_cols) {
+    $me_struct = __visit_dump({ map { $_ => "\xFF__VALPOS__$my_cols->{$_}__\xFF" } (keys %$my_cols) });
+    $me_struct = "[ $me_struct ]" unless $args->{hri_style};
+  }
+
+  my $node_idx_slot = sprintf '$collapse_idx[%d]%s', $cur_node_idx, $node_key;
 
   my @src;
+
   if ($cur_node_idx == 0) {
     push @src, sprintf( '%s ||= %s;',
       $node_idx_slot,
       $me_struct,
     ) if $me_struct;
   }
-  elsif ($args->{collapse_map}->{-is_single}) {
-    push @src, sprintf ( '%s ||= %s%s;',
-      $parent_attach_slot,
-      $node_idx_slot,
-      $me_struct ? " ||= $me_struct" : '',
-    );
-  }
   else {
-    push @src, sprintf('push @{%s}, %s%s unless %s;',
-      $parent_attach_slot,
-      $node_idx_slot,
-      $me_struct ? " ||= $me_struct" : '',
-      $node_idx_slot,
+    my $parent_attach_slot = sprintf( '$collapse_idx[%d]%s%s{%s}',
+      @{$args}{qw/-parent_node_idx -parent_node_key/},
+      $args->{hri_style} ? '' : '[1]',
+      perlstring($args->{-node_relname}),
     );
+
+    if ($args->{collapse_map}->{-is_single}) {
+      push @src, sprintf ( '%s ||= %s%s;',
+        $parent_attach_slot,
+        $node_idx_slot,
+        $me_struct ? " ||= $me_struct" : '',
+      );
+    }
+    else {
+      push @src, sprintf('(! %s) and push @{%s}, %s%s;',
+        $node_idx_slot,
+        $parent_attach_slot,
+        $node_idx_slot,
+        $me_struct ? " = $me_struct" : '',
+      );
+    }
   }
 
-  # DISABLEPRUNE
-  #my $known_defined = { %{ $parent_info->{known_defined} || {} } };
-  #$known_defined->{$_}++ for @{$args->{collapse_map}->{-identifying_columns}};
-  my $stats;
+  my $known_present_ids = { map { $_ => 1 } @{$args->{collapse_map}{-identifying_columns}} };
+  my ($stats, $rel_src);
+
   for my $rel (sort keys %$rel_cols) {
 
-#    push @src, sprintf(
-#      '%s[1]{%s} ||= [];', $node_idx_slot, perlstring($rel)
-#    ) unless $args->{collapse_map}->{$rel}{-is_single};
+    my $relinfo = $args->{collapse_map}{$rel};
+    if ($args->{collapse_map}{-is_optional}) {
+      $relinfo = { %$relinfo, -is_optional => 1 };
+    }
 
-    ($src[$#src + 1], $stats->{$rel}) = __visit_infmap_collapse({ %$args,
+    ($rel_src, $stats->{$rel}) = __visit_infmap_collapse({ %$args,
       val_index => $rel_cols->{$rel},
-      collapse_map => $args->{collapse_map}{$rel},
+      collapse_map => $relinfo,
       -parent_node_idx => $cur_node_idx,
       -parent_node_key => $node_key,
       -node_relname => $rel,
     });
 
-    # FIXME SUBOPTIMAL DISABLEPRUNE - disabled to satisfy t/resultset/inflate_result_api.t
-    #if ($args->{collapse_map}->{$rel}{-is_optional} and my @null_checks = map
-    #  { "(! defined '\xFF__IDVALPOS__${_}__\xFF')" }
-    #  sort { $a <=> $b } grep
-    #    { ! $known_defined->{$_} }
-    #    @{$args->{collapse_map}->{$rel}{-identifying_columns}}
-    #) {
-    #  $src[-1] = sprintf( '(%s) or %s',
-    #    join (' || ', @null_checks ),
-    #    $src[-1],
-    #  );
-    #}
+    my $rel_src_pos = $#src + 1;
+    push @src, @$rel_src;
+
+    if (
+      $args->{prune_null_branches}
+        and
+      $relinfo->{-is_optional}
+        and
+      defined ( my $first_distinct_child_idcol = first
+        { ! $known_present_ids->{$_} }
+        @{$relinfo->{-identifying_columns}}
+      )
+    ) {
+
+      $src[$rel_src_pos] = sprintf( '%s and %s',
+        "( defined '\xFF__VALPOS__${first_distinct_child_idcol}__\xFF' )",
+        $src[$rel_src_pos],
+      );
+
+      splice @src, $rel_src_pos + 1, 0, sprintf ( '%s%s{%s} ||= %s;',
+        $node_idx_slot,
+        $args->{hri_style} ? '' : '[1]',
+        perlstring($rel),
+        $args->{hri_style} && $relinfo->{-is_single} ? 'undef' : '[]',
+      );
+    }
   }
 
   return (
-    join("\n", @src),
+    \@src,
     {
       idcols_seen => {
         ( map { %{ $_->{idcols_seen} } } values %$stats ),
diff --git a/lib/DBIx/Class/Row.pm b/lib/DBIx/Class/Row.pm
index 3a4dda1..d88edc2 100644
--- a/lib/DBIx/Class/Row.pm
+++ b/lib/DBIx/Class/Row.pm
@@ -1182,40 +1182,30 @@ sub inflate_result {
     ref $class || $class
   ;
 
-  foreach my $pre (keys %{$prefetch||{}}) {
+  if ($prefetch) {
+    for my $pre ( keys %$prefetch ) {
 
-    my @pre_vals;
-    @pre_vals = (ref $prefetch->{$pre}[0] eq 'ARRAY')
-      ? @{$prefetch->{$pre}} : $prefetch->{$pre}
-    if @{$prefetch->{$pre}||[]};
+      my @pre_objects;
+      if (@{$prefetch->{$pre}||[]}) {
+        my $pre_source = $source->related_source($pre);
 
-    my $pre_source = $source->related_source($pre);
-
-    my $accessor = $source->relationship_info($pre)->{attrs}{accessor}
-      or $class->throw_exception("No accessor type declared for prefetched relationship '$pre'");
-
-    my @pre_objects;
-    for my $me_pref (@pre_vals) {
+        @pre_objects = map {
+          $pre_source->result_class->inflate_result( $pre_source, @$_ )
+        } ( ref $prefetch->{$pre}[0] eq 'ARRAY' ?  @{$prefetch->{$pre}} : $prefetch->{$pre} );
+      }
 
-      # FIXME SUBOPTIMAL - the new row parsers can very well optimize
-      # this away entirely, and *never* return such empty rows.
-      # For now we maintain inflate_result API backcompat, see
-      # t/resultset/inflate_result_api.t
-      next unless defined first { defined $_ } values %{$me_pref->[0]};
+      my $accessor = $source->relationship_info($pre)->{attrs}{accessor}
+        or $class->throw_exception("No accessor type declared for prefetched relationship '$pre'");
 
-      push @pre_objects, $pre_source->result_class->inflate_result(
-        $pre_source, @$me_pref
-      );
-    }
+      if ($accessor eq 'single') {
+        $new->{_relationship_data}{$pre} = $pre_objects[0];
+      }
+      elsif ($accessor eq 'filter') {
+        $new->{_inflated_column}{$pre} = $pre_objects[0];
+      }
 
-    if ($accessor eq 'single') {
-      $new->{_relationship_data}{$pre} = $pre_objects[0];
+      $new->related_resultset($pre)->set_cache(\@pre_objects);
     }
-    elsif ($accessor eq 'filter') {
-      $new->{_inflated_column}{$pre} = $pre_objects[0];
-    }
-
-    $new->related_resultset($pre)->set_cache(\@pre_objects);
   }
 
   $new->in_storage (1);
diff --git a/t/inflate/hri.t b/t/inflate/hri.t
index 1dca9c2..d027e26 100644
--- a/t/inflate/hri.t
+++ b/t/inflate/hri.t
@@ -34,7 +34,6 @@ my $schema = DBICTest->init_schema();
     is ($rs->result_class, 'DBICTest::CDSubclass', 'original class unchanged');
     is ($hri_rs->result_class, 'DBIx::Class::ResultClass::HashRefInflator', 'result_class accessor pre-set via attribute');
 
-
     my $datahashref1 = $hri_rs->next;
     is_deeply(
       [ sort keys %$datahashref1 ],
diff --git a/t/inflate/hri_torture.t b/t/inflate/hri_torture.t
new file mode 100644
index 0000000..04237f9
--- /dev/null
+++ b/t/inflate/hri_torture.t
@@ -0,0 +1,436 @@
+use strict;
+use warnings;
+
+use Test::More;
+use lib qw(t/lib);
+use DBICTest;
+
+# More tests like this in t/prefetch/manual.t
+
+my $schema = DBICTest->init_schema(no_populate => 1, quote_names => 1);
+$schema->resultset('Artist')->create({ name => 'JMJ', cds => [{
+  title => 'Magnetic Fields',
+  year => 1981,
+  genre => { name => 'electro' },
+  tracks => [
+    { title => 'm1' },
+    { title => 'm2' },
+    { title => 'm3' },
+    { title => 'm4' },
+  ],
+} ] });
+
+
+$schema->resultset('CD')->create({
+  title => 'Equinoxe',
+  year => 1978,
+  artist => { name => 'JMJ' },
+  genre => { name => 'electro' },
+  tracks => [
+    { title => 'e1' },
+    { title => 'e2' },
+    { title => 'e3' },
+  ],
+  single_track => {
+    title => 'o1',
+    cd => {
+      title => 'Oxygene',
+      year => 1976,
+      artist => { name => 'JMJ' },
+      tracks => [
+        { title => 'o2', position => 2},  # the position should not be needed here, bug in MC
+      ],
+    },
+  },
+});
+
+for (1,2) {
+  $schema->resultset('CD')->create({ artist => 1, year => 1977, title => "fuzzy_$_" });
+}
+
+my $rs = $schema->resultset('CD');
+
+is_deeply
+  $rs->search({}, {
+    columns => {
+      year                          => 'me.year',
+      'single_track.cd.artist.name' => 'artist.name',
+    },
+    join => { single_track => { cd => 'artist' } },
+    order_by => [qw/me.cdid artist.artistid/],
+  })->all_hri,
+  [
+    {
+      single_track => undef,
+      year => 1981
+    },
+    {
+      single_track => undef,
+      year => 1976
+    },
+    {
+      single_track => {
+        cd => {
+          artist => {
+            name => "JMJ"
+          }
+        }
+      },
+      year => 1978
+    },
+    {
+      single_track => undef,
+      year => 1977
+    },
+    {
+      single_track => undef,
+      year => 1977
+    },
+  ],
+  'plain 1:1 descending chain'
+;
+
+is_deeply
+  $rs->search({}, {
+    columns => {
+      'artist'                                  => 'me.artist',
+      'title'                                   => 'me.title',
+      'year'                                    => 'me.year',
+      'single_track.cd.artist.artistid'         => 'artist.artistid',
+      'single_track.cd.artist.cds.cdid'         => 'cds.cdid',
+      'single_track.cd.artist.cds.tracks.title' => 'tracks.title',
+    },
+    join => { single_track => { cd => { artist => { cds => 'tracks' } } } },
+    order_by => [qw/me.cdid artist.artistid cds.cdid tracks.trackid/],
+  })->all_hri,
+  [
+    {
+      artist => 1,
+      single_track => undef,
+      title => "Magnetic Fields",
+      year => 1981
+    },
+    {
+      artist => 1,
+      single_track => undef,
+      title => "Oxygene",
+      year => 1976
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => {
+              cdid => 1,
+              tracks => {
+                title => "m1"
+              }
+            }
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => {
+              cdid => 1,
+              tracks => {
+                title => "m2"
+              }
+            }
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => {
+              cdid => 1,
+              tracks => {
+                title => "m3"
+              }
+            }
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => {
+              cdid => 1,
+              tracks => {
+                title => "m4"
+              }
+            }
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => {
+              cdid => 2,
+              tracks => {
+                title => "o2"
+              }
+            }
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => {
+              cdid => 2,
+              tracks => {
+                title => "o1"
+              }
+            }
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => {
+              cdid => 3,
+              tracks => {
+                title => "e1"
+              }
+            }
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => {
+              cdid => 3,
+              tracks => {
+                title => "e2"
+              }
+            }
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => {
+              cdid => 3,
+              tracks => {
+                title => "e3"
+              }
+            }
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => {
+              cdid => 4,
+              tracks => undef
+            }
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => {
+              cdid => 5,
+              tracks => undef
+            }
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => undef,
+      title => "fuzzy_1",
+      year => 1977
+    },
+    {
+      artist => 1,
+      single_track => undef,
+      title => "fuzzy_2",
+      year => 1977
+    }
+  ],
+  'non-collapsing 1:1:1:M:M chain',
+;
+
+is_deeply
+  $rs->search({}, {
+    columns => {
+      'artist'                                  => 'me.artist',
+      'title'                                   => 'me.title',
+      'year'                                    => 'me.year',
+      'single_track.cd.artist.artistid'         => 'artist.artistid',
+      'single_track.cd.artist.cds.cdid'         => 'cds.cdid',
+      'single_track.cd.artist.cds.tracks.title' => 'tracks.title',
+    },
+    join => { single_track => { cd => { artist => { cds => 'tracks' } } } },
+    order_by => [qw/me.cdid artist.artistid cds.cdid tracks.trackid/],
+    collapse => {}, #hashref to keep older DBIC versions happy (doesn't actually work)
+  })->all_hri,
+  [
+    {
+      artist => 1,
+      single_track => undef,
+      title => "Magnetic Fields",
+      year => 1981
+    },
+    {
+      artist => 1,
+      single_track => undef,
+      title => "Oxygene",
+      year => 1976
+    },
+    {
+      artist => 1,
+      single_track => {
+        cd => {
+          artist => {
+            artistid => 1,
+            cds => [
+              {
+                cdid => 1,
+                tracks => [
+                  {
+                    title => "m1"
+                  },
+                  {
+                    title => "m2"
+                  },
+                  {
+                    title => "m3"
+                  },
+                  {
+                    title => "m4"
+                  }
+                ]
+              },
+              {
+                cdid => 2,
+                tracks => [
+                  {
+                    title => "o2"
+                  },
+                  {
+                    title => "o1"
+                  }
+                ]
+              },
+              {
+                cdid => 3,
+                tracks => [
+                  {
+                    title => "e1"
+                  },
+                  {
+                    title => "e2"
+                  },
+                  {
+                    title => "e3"
+                  }
+                ]
+              },
+              {
+                cdid => 4,
+                tracks => []
+              },
+              {
+                cdid => 5,
+                tracks => []
+              }
+            ]
+          }
+        }
+      },
+      title => "Equinoxe",
+      year => 1978
+    },
+    {
+      artist => 1,
+      single_track => undef,
+      title => "fuzzy_1",
+      year => 1977
+    },
+    {
+      artist => 1,
+      single_track => undef,
+      title => "fuzzy_2",
+      year => 1977
+    }
+  ],
+  'collapsing 1:1:1:M:M chain',
+;
+
+done_testing;
diff --git a/t/prefetch/manual.t b/t/prefetch/manual.t
index 6914cae..646b2d3 100644
--- a/t/prefetch/manual.t
+++ b/t/prefetch/manual.t
@@ -300,9 +300,8 @@ for my $use_next (0, 1) {
       like( $_->title, qr/^e\d/, "correct title" )
         for $cd->tracks;
       ok( defined $cd->single_track, 'single track prefetched on 1987 cd' );
-      # FIXME - crap! skipping prefetch also doesn't work, next commit
-      #is( $cd->single_track->cd->artist->id, 1, 'Single_track->cd->artist prefetched on 1978 cd' );
-      #is( scalar $cd->single_track->cd->artist->cds, 6, '6 cds prefetched on artist' );
+      is( $cd->single_track->cd->artist->id, 1, 'Single_track->cd->artist prefetched on 1978 cd' );
+      is( scalar $cd->single_track->cd->artist->cds, 6, '6 cds prefetched on artist' );
     }
   }
 }
diff --git a/t/resultset/inflate_result_api.t b/t/resultset/inflate_result_api.t
index d97e475..558a662 100644
--- a/t/resultset/inflate_result_api.t
+++ b/t/resultset/inflate_result_api.t
@@ -2,9 +2,12 @@ use strict;
 use warnings;
 
 use Test::More;
+use Test::Warn;
 use lib qw(t/lib);
 use DBICTest;
 
+my $new_collapser_version = DBIx::Class::ResultSet->can('_construct_objects');
+
 my $schema = DBICTest->init_schema(no_populate => 1);
 
 $schema->resultset('Artist')->create({ name => 'JMJ', cds => [{
@@ -42,12 +45,26 @@ $schema->resultset('CD')->create({
   },
 });
 
+$schema->resultset('CD')->create({ artist => 1, year => 1977, title => "fuzzy_1" });
+
 {
   package DBICTest::_IRCapture;
   sub inflate_result { [@_[2,3]] };
 }
 
-is_deeply(
+{
+  package DBICTest::_IRCaptureAround;
+  use base 'DBIx::Class::Row';
+  sub inflate_result { [@_[2,3]] };
+}
+
+warnings_exist
+  { $schema->resultset ('CD')->search ({}, { result_class => 'DBICTest::_IRCapture', prefetch => 'tracks' } )->all }
+  qr/\QResultClass DBICTest::_IRCapture does not inherit from DBIx::Class::Row and therefore its inflate_result() will receive the full prefetched data tree, without any branch definedness checks/,
+  'Legacy inflate_result() API warned',
+if $new_collapser_version;
+
+cmp_structures(
   ([$schema->resultset ('CD')->search ({}, {
     result_class => 'DBICTest::_IRCapture',
     prefetch => { single_track => { cd => 'artist' } },
@@ -96,11 +113,63 @@ is_deeply(
         ] }
       ] }
     ],
+    [
+      { cdid => 4, single_track => undef, artist => 1, genreid => undef, year => 1977, title => "fuzzy_1" },
+      { single_track => [
+        { trackid => undef, title => undef, position => undef, cd => undef, last_updated_at => undef, last_updated_on => undef },
+        {  cd => [
+          { cdid => undef, single_track => undef, artist => undef, genreid => undef, year => undef, title => undef },
+          {
+            artist => [
+              { artistid => undef, name => undef, charfield => undef, rank => undef }
+            ]
+          }
+        ] }
+      ] }
+    ],
   ],
-  'Simple 1:1 descend with classic prefetch ok'
+  'Simple 1:1 descend with classic prefetch legacy'
 );
 
-is_deeply(
+cmp_structures(
+  ([$schema->resultset ('CD')->search ({}, {
+    result_class => 'DBICTest::_IRCaptureAround',
+    prefetch => { single_track => { cd => 'artist' } },
+    order_by => 'me.cdid',
+  })->all]),
+  [
+    [
+      { cdid => 1, single_track => undef, artist => 1, genreid => 1, year => 1981, title => "Magnetic Fields" },
+      { single_track => [] }
+    ],
+    [
+      { cdid => 2, single_track => undef, artist => 1, genreid => undef, year => 1976, title => "Oxygene" },
+      { single_track => [] }
+    ],
+    [
+      { cdid => 3, single_track => 6, artist => 1, genreid => 1, year => 1978, title => "Equinoxe" },
+      { single_track => [
+        { trackid => 6, title => 'o1', position => 1, cd => 2, last_updated_at => undef, last_updated_on => undef },
+        {  cd => [
+          { cdid => 2, single_track => undef, artist => 1, genreid => undef, year => 1976, title => "Oxygene" },
+          {
+            artist => [
+              { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 }
+            ]
+          }
+        ] }
+      ] }
+    ],
+    [
+      { cdid => 4, single_track => undef, artist => 1, genreid => undef, year => 1977, title => "fuzzy_1" },
+      { single_track => [ ] }
+    ],
+  ],
+  'Simple 1:1 descend with classic prefetch pruning'
+);
+
+
+cmp_structures(
   [$schema->resultset ('CD')->search ({}, {
     result_class => 'DBICTest::_IRCapture',
     join => { single_track => { cd => 'artist' } },
@@ -156,11 +225,69 @@ is_deeply(
         ] }
       ] }
     ],
+    [
+      { artist => 1, genreid => undef, year => 1977, title => "fuzzy_1" },
+      { single_track => [
+        undef,
+        {  cd => [
+          undef,
+          {
+            artist => [
+              { artistid => undef }
+            ]
+          }
+        ] }
+      ] }
+    ],
   ],
-  'Simple 1:1 descend with missing selectors ok'
+  'Simple 1:1 descend with missing selectors legacy'
 );
 
-is_deeply(
+cmp_structures(
+  [$schema->resultset ('CD')->search ({}, {
+    result_class => 'DBICTest::_IRCaptureAround',
+    join => { single_track => { cd => 'artist' } },
+    columns => [
+      { 'year'                                    => 'me.year' },
+      { 'genreid'                                 => 'me.genreid' },
+      { 'single_track.cd.artist.artistid'         => 'artist.artistid' },
+      { 'title'                                   => 'me.title' },
+      { 'artist'                                  => 'me.artist' },
+    ],
+    order_by => 'me.cdid',
+  })->all],
+  [
+    [
+      { artist => 1, genreid => 1, year => 1981, title => "Magnetic Fields" },
+      { single_track => [] }
+    ],
+    [
+      { artist => 1, genreid => undef, year => 1976, title => "Oxygene" },
+      { single_track => [ ] }
+    ],
+    [
+      { artist => 1, genreid => 1, year => 1978, title => "Equinoxe" },
+      { single_track => [
+        undef,
+        {  cd => [
+          undef,
+          {
+            artist => [
+              { artistid => 1 }
+            ]
+          }
+        ] }
+      ] }
+    ],
+    [
+      { artist => 1, genreid => undef, year => 1977, title => "fuzzy_1" },
+      { single_track => [] }
+    ],
+  ],
+  'Simple 1:1 descend with missing selectors pruning'
+);
+
+cmp_structures(
   ([$schema->resultset ('CD')->search ({}, {
     result_class => 'DBICTest::_IRCapture',
     prefetch => [ { single_track => { cd => { artist => { cds => 'tracks' } } } } ],
@@ -218,6 +345,95 @@ is_deeply(
               { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
               { cds => [
                 [
+                  { cdid => 4, single_track => undef, artist => 1, genreid => undef, year => 1977, title => "fuzzy_1" },
+                  { tracks => [
+                    [ { trackid => undef, title => undef, position => undef, cd => undef, last_updated_at => undef, last_updated_on => undef } ],
+                  ] },
+                ],
+                [
+                  { cdid => 1, single_track => undef, artist => 1, genreid => 1, year => 1981, title => "Magnetic Fields" },
+                  { tracks => [
+                    [ { trackid => 1, title => 'm1', position => 1, cd => 1, last_updated_at => undef, last_updated_on => undef } ],
+                    [ { trackid => 2, title => 'm2', position => 2, cd => 1, last_updated_at => undef, last_updated_on => undef } ],
+                    [ { trackid => 3, title => 'm3', position => 3, cd => 1, last_updated_at => undef, last_updated_on => undef } ],
+                    [ { trackid => 4, title => 'm4', position => 4, cd => 1, last_updated_at => undef, last_updated_on => undef } ],
+                  ]},
+                ],
+                [
+                  { cdid => 2, single_track => undef, artist => 1, genreid => undef, year => 1976, title => "Oxygene" },
+                  { tracks => [
+                    [ { trackid => 5, title => 'o2', position => 2, cd => 2, last_updated_at => undef, last_updated_on => undef } ],
+                    [ { trackid => 6, title => 'o1', position => 1, cd => 2, last_updated_at => undef, last_updated_on => undef } ],
+                  ]},
+                ],
+                [
+                  { cdid => 3, single_track => 6, artist => 1, genreid => 1, year => 1978, title => "Equinoxe" },
+                  { tracks => [
+                    [ { trackid => 7, title => 'e1', position => 1, cd => 3, last_updated_at => undef, last_updated_on => undef } ],
+                    [ { trackid => 8, title => 'e2', position => 2, cd => 3, last_updated_at => undef, last_updated_on => undef } ],
+                    [ { trackid => 9, title => 'e3', position => 3, cd => 3, last_updated_at => undef, last_updated_on => undef } ],
+                  ]},
+                ],
+              ]},
+            ]
+          }
+        ] }
+      ] }
+    ],
+    [
+      { cdid => 4, single_track => undef, artist => 1, genreid => undef, year => 1977, title => "fuzzy_1" },
+      { single_track => [
+        { trackid => undef, title => undef, position => undef, cd => undef, last_updated_at => undef, last_updated_on => undef },
+        {  cd => [
+          { cdid => undef, single_track => undef, artist => undef, genreid => undef, year => undef, title => undef },
+          {
+            artist => [
+              { artistid => undef, name => undef, charfield => undef, rank => undef },
+              { cds => [ [
+                { cdid => undef, single_track => undef, artist => undef, genreid => undef, year => undef, title => undef },
+                { tracks => [ [
+                  { trackid => undef, title => undef, position => undef, cd => undef, last_updated_at => undef, last_updated_on => undef },
+                ] ] },
+              ]]},
+            ]
+          }
+        ] }
+      ] }
+    ],
+  ],
+  'Collapsing 1:1 ending in chained has_many with classic prefetch legacy'
+);
+
+cmp_structures(
+  ([$schema->resultset ('CD')->search ({}, {
+    result_class => 'DBICTest::_IRCaptureAround',
+    prefetch => [ { single_track => { cd => { artist => { cds => 'tracks' } } } } ],
+    order_by => [qw/me.cdid tracks.trackid/],
+  })->all]),
+  [
+    [
+      { cdid => 1, single_track => undef, artist => 1, genreid => 1, year => 1981, title => "Magnetic Fields" },
+      { single_track => [ ] },
+    ],
+    [
+      { cdid => 2, single_track => undef, artist => 1, genreid => undef, year => 1976, title => "Oxygene" },
+      { single_track => [ ] },
+    ],
+    [
+      { cdid => 3, single_track => 6, artist => 1, genreid => 1, year => 1978, title => "Equinoxe" },
+      { single_track => [
+        { trackid => 6, title => 'o1', position => 1, cd => 2, last_updated_at => undef, last_updated_on => undef },
+        {  cd => [
+          { cdid => 2, single_track => undef, artist => 1, genreid => undef, year => 1976, title => "Oxygene" },
+          {
+            artist => [
+              { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+              { cds => [
+                [
+                  { cdid => 4, single_track => undef, artist => 1, genreid => undef, year => 1977, title => "fuzzy_1" },
+                  { tracks => [ ] },
+                ],
+                [
                   { cdid => 1, single_track => undef, artist => 1, genreid => 1, year => 1981, title => "Magnetic Fields" },
                   { tracks => [
                     [ { trackid => 1, title => 'm1', position => 1, cd => 1, last_updated_at => undef, last_updated_on => undef } ],
@@ -247,11 +463,15 @@ is_deeply(
         ] }
       ] }
     ],
+    [
+      { cdid => 4, single_track => undef, artist => 1, genreid => undef, year => 1977, title => "fuzzy_1" },
+      { single_track => [ ] },
+    ],
   ],
-  'Collapsing 1:1 ending in chained has_many with classic prefetch ok'
+  'Collapsing 1:1 ending in chained has_many with classic prefetch pruning'
 );
 
-is_deeply (
+cmp_structures (
   ([$schema->resultset ('Artist')->search ({}, {
     result_class => 'DBICTest::_IRCapture',
     join => { cds => 'tracks' },
@@ -343,8 +563,130 @@ is_deeply (
         ]},
       ]},
     ],
+    [
+      { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+      { cds => [
+        { cdid => 4, single_track => undef, artist => 1, genreid => undef, year => 1977, title => "fuzzy_1" },
+        { tracks => [
+          { trackid => undef, title => undef, position => undef, cd => undef, last_updated_at => undef, last_updated_on => undef },
+        ]},
+      ]},
+    ],
   ],
-  'Non-Collapsing chained has_many ok'
+  'Non-Collapsing chained has_many legacy'
 );
 
+cmp_structures(
+  ([$schema->resultset ('Artist')->search ({}, {
+    result_class => 'DBICTest::_IRCaptureAround',
+    join => { cds => 'tracks' },
+    '+columns' => [
+      (map { "cds.$_" } $schema->source('CD')->columns),
+      (map { +{ "cds.tracks.$_" => "tracks.$_" } } $schema->source('Track')->columns),
+    ],
+    order_by => [qw/cds.cdid tracks.trackid/],
+  })->all]),
+  [
+    [
+      { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+      { cds => [
+        { cdid => 1, single_track => undef, artist => 1, genreid => 1, year => 1981, title => "Magnetic Fields" },
+        { tracks => [
+          { trackid => 1, title => 'm1', position => 1, cd => 1, last_updated_at => undef, last_updated_on => undef },
+        ]},
+      ]},
+    ],
+    [
+      { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+      { cds => [
+        { cdid => 1, single_track => undef, artist => 1, genreid => 1, year => 1981, title => "Magnetic Fields" },
+        { tracks => [
+          { trackid => 2, title => 'm2', position => 2, cd => 1, last_updated_at => undef, last_updated_on => undef },
+        ]},
+      ]},
+    ],
+    [
+      { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+      { cds => [
+        { cdid => 1, single_track => undef, artist => 1, genreid => 1, year => 1981, title => "Magnetic Fields" },
+        { tracks => [
+          { trackid => 3, title => 'm3', position => 3, cd => 1, last_updated_at => undef, last_updated_on => undef },
+        ]},
+      ]},
+    ],
+    [
+      { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+      { cds => [
+        { cdid => 1, single_track => undef, artist => 1, genreid => 1, year => 1981, title => "Magnetic Fields" },
+        { tracks => [
+          { trackid => 4, title => 'm4', position => 4, cd => 1, last_updated_at => undef, last_updated_on => undef },
+        ]},
+      ]},
+    ],
+    [
+      { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+      { cds => [
+        { cdid => 2, single_track => undef, artist => 1, genreid => undef, year => 1976, title => "Oxygene" },
+        { tracks => [
+          { trackid => 5, title => 'o2', position => 2, cd => 2, last_updated_at => undef, last_updated_on => undef },
+        ]},
+      ]},
+    ],
+    [
+      { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+      { cds => [
+        { cdid => 2, single_track => undef, artist => 1, genreid => undef, year => 1976, title => "Oxygene" },
+        { tracks => [
+          { trackid => 6, title => 'o1', position => 1, cd => 2, last_updated_at => undef, last_updated_on => undef },
+        ]},
+      ]},
+    ],
+    [
+      { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+      { cds => [
+        { cdid => 3, single_track => 6, artist => 1, genreid => 1, year => 1978, title => "Equinoxe" },
+        { tracks => [
+          { trackid => 7, title => 'e1', position => 1, cd => 3, last_updated_at => undef, last_updated_on => undef },
+        ]},
+      ]},
+    ],
+    [
+      { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+      { cds => [
+        { cdid => 3, single_track => 6, artist => 1, genreid => 1, year => 1978, title => "Equinoxe" },
+        { tracks => [
+          { trackid => 8, title => 'e2', position => 2, cd => 3, last_updated_at => undef, last_updated_on => undef },
+        ]},
+      ]},
+    ],
+    [
+      { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+      { cds => [
+        { cdid => 3, single_track => 6, artist => 1, genreid => 1, year => 1978, title => "Equinoxe" },
+        { tracks => [
+          { trackid => 9, title => 'e3', position => 3, cd => 3, last_updated_at => undef, last_updated_on => undef },
+        ]},
+      ]},
+    ],
+    [
+      { artistid => 1, name => 'JMJ', charfield => undef, rank => 13 },
+      { cds => [
+        { cdid => 4, single_track => undef, artist => 1, genreid => undef, year => 1977, title => "fuzzy_1" },
+        { tracks => [ ] },
+      ]},
+    ],
+  ],
+  'Non-Collapsing chained has_many pruning'
+);
+
+sub cmp_structures {
+  my ($left, $right, $msg) = @_;
+
+  local $TODO = "Pruning test won't work on pre-rewrite DBIC"
+    if ($msg||'') =~ /pruning$/ and ! $new_collapser_version;
+
+  local $Test::Builder::Level = $Test::Builder::Level + 1;
+  is_deeply($left, $right, $msg||());
+}
+
 done_testing;
diff --git a/t/resultset/rowparser_internals.t b/t/resultset/rowparser_internals.t
index a89897b..baa9da6 100644
--- a/t/resultset/rowparser_internals.t
+++ b/t/resultset/rowparser_internals.t
@@ -15,7 +15,10 @@ use Data::Dumper;
 $Data::Dumper::Sortkeys = 1;
 
 my $schema = DBICTest->init_schema(no_deploy => 1);
-my $infmap = [qw/single_track.cd.artist.name year/];
+my $infmap = [qw/
+  single_track.cd.artist.name
+  year
+/];
 
 is_same_src (
   $schema->source ('CD')->_mk_row_parser({
@@ -69,6 +72,81 @@ is_same_src (
   '1:1 descending non-collapsing parser terminating with chained 1:M:M',
 );
 
+is_same_src (
+  $schema->source ('CD')->_mk_row_parser({
+    prune_null_branches => 1,
+    inflate_map => $infmap,
+  }),
+  '$_ = [
+    { artist => $_->[5], title => $_->[4], year => $_->[2] },
+    {
+      ( (! defined $_->[0] ) && (! defined $_->[1]) && (! defined $_->[3] ) )
+        ? ( single_track => [] )
+        : ( single_track => [
+          undef,
+          {
+            cd => [
+              undef,
+              {
+                artist => [
+                  { artistid => $_->[1] },
+                  {
+                    ( (! defined $_->[0] ) && ( ! defined $_->[3] ) )
+                      ? ( cds => [] )
+                      : ( cds => [
+                        { cdid => $_->[3] },
+                        {
+                          ( ! defined $_->[0] )
+                            ? ( tracks => [] )
+                            : ( tracks => [{ title => $_->[0] }] )
+                        }
+                      ])
+                  }
+                ]
+              }
+            ]
+          }
+        ])
+    }
+  ] for @{$_[0]}',
+  '1:1 descending non-collapsing null-pruning parser terminating with chained 1:M:M',
+);
+
+is_same_src (
+  $schema->source ('CD')->_mk_row_parser({
+    prune_null_branches => 1,
+    hri_style => 1,
+    inflate_map => $infmap,
+  }),
+  '$_ = {
+      artist => $_->[5], title => $_->[4], year => $_->[2],
+
+      ( (! defined $_->[0] ) && (! defined $_->[1]) && (! defined $_->[3] ) )
+        ? ( single_track => undef )
+        : ( single_track => {
+            cd =>
+              {
+                artist => {
+                    artistid => $_->[1],
+                    ( (! defined $_->[0] ) && ( ! defined $_->[3] ) )
+                      ? ( cds => undef )
+                      : ( cds => {
+                          cdid => $_->[3],
+                          ( ! defined $_->[0] )
+                            ? ( tracks => undef )
+                            : ( tracks => { title => $_->[0] } )
+                        }
+                       )
+                  }
+              }
+          }
+        )
+    } for @{$_[0]}',
+  '1:1 descending non-collapsing null-pruning HRI-direct parser terminating with chained 1:M:M',
+);
+
+
+
 is_deeply (
   ($schema->source('CD')->_resolve_collapse({ as => {map { $infmap->[$_] => $_ } 0 .. $#$infmap} })),
   {
@@ -136,12 +214,18 @@ is_same_src (
       $collapse_idx[2]{$cur_row_ids{1}}{$cur_row_ids{4}}{$cur_row_ids{5}}[1]{artist} ||= $collapse_idx[3]{$cur_row_ids{1}}{$cur_row_ids{4}}{$cur_row_ids{5}} ||= [{ artistid => $cur_row_data->[1] }];
 
       # prefetch data of cds (if available)
-      push @{$collapse_idx[3]{$cur_row_ids{1}}{$cur_row_ids{4}}{$cur_row_ids{5}}[1]{cds}}, $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}} ||= [{ cdid => $cur_row_data->[3] }]
-        unless $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}};
+      (! $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}} )
+        and
+      push @{$collapse_idx[3]{$cur_row_ids{1}}{$cur_row_ids{4}}{$cur_row_ids{5}}[1]{cds}}, (
+        $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}} = [{ cdid => $cur_row_data->[3] }]
+      );
 
       # prefetch data of tracks (if available)
-      push @{$collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}}[1]{tracks}}, $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}} ||= [{ title => $cur_row_data->[0] }]
-        unless $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}};
+      (! $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}} )
+        and
+      push @{$collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}}[1]{tracks}}, (
+        $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}} = [{ title => $cur_row_data->[0] }]
+      );
 
       $_[0][$result_pos++] = $collapse_idx[0]{$cur_row_ids{4}}{$cur_row_ids{5}}
         if $is_new_res;
@@ -151,6 +235,70 @@ is_same_src (
   'Same 1:1 descending terminating with chained 1:M:M but with collapse',
 );
 
+is_same_src (
+  $schema->source ('CD')->_mk_row_parser({
+    inflate_map => $infmap,
+    collapse => 1,
+    prune_null_branches => 1,
+    hri_style => 1,
+  }),
+  ' my($rows_pos, $result_pos, $cur_row_data, %cur_row_ids, @collapse_idx, $is_new_res) = (0, 0);
+
+    while ($cur_row_data = (
+      ( $rows_pos >= 0 and $_[0][$rows_pos++] ) or do { $rows_pos = -1; undef } )
+        ||
+      ( $_[1] and $_[1]->() )
+    ) {
+
+      $cur_row_ids{$_} = defined $cur_row_data->[$_] ? $cur_row_data->[$_] : "\0NULL\xFF$rows_pos\xFF$_\0"
+        for (0, 1, 3, 4, 5);
+
+      # a present cref in $_[1] implies lazy prefetch, implies a supplied stash in $_[2]
+      $_[1] and $result_pos and unshift(@{$_[2]}, $cur_row_data) and last
+        if ( $is_new_res = ! $collapse_idx[0]{$cur_row_ids{4}}{$cur_row_ids{5}} );
+
+      # the rowdata itself for root node
+      $collapse_idx[0]{$cur_row_ids{4}}{$cur_row_ids{5}} ||= { artist => $cur_row_data->[5], title => $cur_row_data->[4], year => $cur_row_data->[2] };
+
+      # prefetch data of single_track (placed in root)
+      $collapse_idx[0]{$cur_row_ids{4}}{$cur_row_ids{5}}{single_track} ||= $collapse_idx[1]{$cur_row_ids{1}}{$cur_row_ids{4}}{$cur_row_ids{5}}
+        if defined $cur_row_data->[1];
+      $collapse_idx[0]{$cur_row_ids{4}}{$cur_row_ids{5}}{single_track} ||= undef;
+
+      # prefetch data of cd (placed in single_track)
+      $collapse_idx[1]{$cur_row_ids{1}}{$cur_row_ids{4}}{$cur_row_ids{5}}{cd} ||= $collapse_idx[2]{$cur_row_ids{1}}{$cur_row_ids{4}}{$cur_row_ids{5}};
+
+      # prefetch data of artist ( placed in single_track->cd)
+      $collapse_idx[2]{$cur_row_ids{1}}{$cur_row_ids{4}}{$cur_row_ids{5}}{artist} ||= $collapse_idx[3]{$cur_row_ids{1}}{$cur_row_ids{4}}{$cur_row_ids{5}} ||= { artistid => $cur_row_data->[1] };
+
+      # prefetch data of cds (if available)
+      ( defined $cur_row_data->[3] )
+        and
+      (! $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}} )
+        and
+      push @{$collapse_idx[3]{$cur_row_ids{1}}{$cur_row_ids{4}}{$cur_row_ids{5}}{cds}}, (
+        $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}} = { cdid => $cur_row_data->[3] }
+      );
+      $collapse_idx[3]{$cur_row_ids{1}}{$cur_row_ids{4}}{$cur_row_ids{5}}{cds} ||= [];
+
+      # prefetch data of tracks (if available)
+      ( defined $cur_row_data->[0] )
+        and
+      (! $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}} )
+        and
+      push @{$collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}}{tracks}}, (
+        $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}} = { title => $cur_row_data->[0] }
+      );
+      $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{3}}{$cur_row_ids{4}}{$cur_row_ids{5}}{tracks} ||= [];
+
+      $_[0][$result_pos++] = $collapse_idx[0]{$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, pruning, hri-style',
+);
+
 $infmap = [qw/
   tracks.lyrics.existing_lyric_versions.text
   existing_single_track.cd.artist.artistid
@@ -238,19 +386,31 @@ is_same_src (
       $collapse_idx[1]{$cur_row_ids{1}}[1]{cd} ||= $collapse_idx[2]{$cur_row_ids{1}};
       $collapse_idx[2]{$cur_row_ids{1}}[1]{artist} ||= $collapse_idx[3]{$cur_row_ids{1}} ||= [{ artistid => $cur_row_data->[1] }];
 
-      push @{ $collapse_idx[3]{$cur_row_ids{1}}[1]{cds} }, $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{6}} ||= [{ cdid => $cur_row_data->[6], genreid => $cur_row_data->[9], year => $cur_row_data->[2] }]
-        unless $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{6}};
+      (! $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{6}} )
+        and
+      push @{ $collapse_idx[3]{$cur_row_ids{1}}[1]{cds} }, (
+        $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{6}} = [{ cdid => $cur_row_data->[6], genreid => $cur_row_data->[9], year => $cur_row_data->[2] }]
+      );
 
-      push @{ $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{6}}[1]{tracks} }, $collapse_idx[5]{$cur_row_ids{1}}{$cur_row_ids{6}}{$cur_row_ids{8}} ||= [{ title => $cur_row_data->[8] }]
-        unless $collapse_idx[5]{$cur_row_ids{1}}{$cur_row_ids{6}}{$cur_row_ids{8}};
+      (! $collapse_idx[5]{$cur_row_ids{1}}{$cur_row_ids{6}}{$cur_row_ids{8}} )
+        and
+      push @{ $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{6}}[1]{tracks} }, (
+        $collapse_idx[5]{$cur_row_ids{1}}{$cur_row_ids{6}}{$cur_row_ids{8}} = [{ title => $cur_row_data->[8] }]
+      );
 
-      push @{ $collapse_idx[0]{$cur_row_ids{1}}[1]{tracks} }, $collapse_idx[6]{$cur_row_ids{1}}{$cur_row_ids{5}} ||= [{ title => $cur_row_data->[5] }]
-        unless $collapse_idx[6]{$cur_row_ids{1}}{$cur_row_ids{5}};
+      (! $collapse_idx[6]{$cur_row_ids{1}}{$cur_row_ids{5}} )
+        and
+      push @{ $collapse_idx[0]{$cur_row_ids{1}}[1]{tracks} }, (
+        $collapse_idx[6]{$cur_row_ids{1}}{$cur_row_ids{5}} = [{ title => $cur_row_data->[5] }]
+      );
 
       $collapse_idx[6]{$cur_row_ids{1}}{$cur_row_ids{5}}[1]{lyrics} ||= $collapse_idx[7]{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}};
 
-      push @{ $collapse_idx[7]{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}}[1]{existing_lyric_versions} }, $collapse_idx[8]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}} ||= [{ lyric_id => $cur_row_data->[10], text => $cur_row_data->[0] }]
-        unless $collapse_idx[8]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}};
+      (! $collapse_idx[8]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}} )
+        and
+      push @{ $collapse_idx[7]{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}}[1]{existing_lyric_versions} }, (
+        $collapse_idx[8]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}} = [{ lyric_id => $cur_row_data->[10], text => $cur_row_data->[0] }]
+      );
 
       $_[0][$result_pos++] = $collapse_idx[0]{$cur_row_ids{1}}
         if $is_new_res;
@@ -261,6 +421,83 @@ is_same_src (
   'Multiple has_many on multiple branches torture test',
 );
 
+is_same_src (
+  $schema->source ('CD')->_mk_row_parser({
+    inflate_map => $infmap,
+    collapse => 1,
+    prune_null_branches => 1,
+  }),
+  ' my ($rows_pos, $result_pos, $cur_row_data, %cur_row_ids, @collapse_idx, $is_new_res) = (0,0);
+
+    while ($cur_row_data = (
+      ( $rows_pos >= 0 and $_[0][$rows_pos++] ) or do { $rows_pos = -1; undef } )
+        ||
+      ( $_[1] and $_[1]->() )
+    ) {
+
+      $cur_row_ids{$_} = defined $cur_row_data->[$_] ? $cur_row_data->[$_] : "\0NULL\xFF$rows_pos\xFF$_\0"
+        for (0, 1, 5, 6, 8, 10);
+
+      # a present cref in $_[1] implies lazy prefetch, implies a supplied stash in $_[2]
+      $_[1] and $result_pos and unshift(@{$_[2]}, $cur_row_data) and last
+        if ( $is_new_res = ! $collapse_idx[0]{$cur_row_ids{1}} );
+
+      $collapse_idx[0]{$cur_row_ids{1}} ||= [{ genreid => $cur_row_data->[4], latest_cd => $cur_row_data->[7], year => $cur_row_data->[3] }];
+
+      $collapse_idx[0]{$cur_row_ids{1}}[1]{existing_single_track} ||= $collapse_idx[1]{$cur_row_ids{1}};
+      $collapse_idx[1]{$cur_row_ids{1}}[1]{cd} ||= $collapse_idx[2]{$cur_row_ids{1}};
+      $collapse_idx[2]{$cur_row_ids{1}}[1]{artist} ||= $collapse_idx[3]{$cur_row_ids{1}} ||= [{ artistid => $cur_row_data->[1] }];
+
+      (defined $cur_row_data->[6])
+        and
+      (! $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{6}} )
+        and
+      push @{ $collapse_idx[3]{$cur_row_ids{1}}[1]{cds} }, (
+        $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{6}} = [{ cdid => $cur_row_data->[6], genreid => $cur_row_data->[9], year => $cur_row_data->[2] }]
+      );
+      $collapse_idx[3]{$cur_row_ids{1}}[1]{cds} ||= [];
+
+      (defined $cur_row_data->[8])
+        and
+      (! $collapse_idx[5]{$cur_row_ids{1}}{$cur_row_ids{6}}{$cur_row_ids{8}} )
+        and
+      push @{ $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{6}}[1]{tracks} }, (
+        $collapse_idx[5]{$cur_row_ids{1}}{$cur_row_ids{6}}{$cur_row_ids{8}} = [{ title => $cur_row_data->[8] }]
+      );
+      $collapse_idx[4]{$cur_row_ids{1}}{$cur_row_ids{6}}[1]{tracks} ||= [];
+
+      (defined $cur_row_data->[5])
+        and
+      (! $collapse_idx[6]{$cur_row_ids{1}}{$cur_row_ids{5}} )
+        and
+      push @{ $collapse_idx[0]{$cur_row_ids{1}}[1]{tracks} }, (
+        $collapse_idx[6]{$cur_row_ids{1}}{$cur_row_ids{5}} = [{ title => $cur_row_data->[5] }]
+      );
+      $collapse_idx[0]{$cur_row_ids{1}}[1]{tracks} ||= [];
+
+      (defined $cur_row_data->[10])
+        and
+      $collapse_idx[6]{$cur_row_ids{1}}{$cur_row_ids{5}}[1]{lyrics} ||= $collapse_idx[7]{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}};
+      $collapse_idx[6]{$cur_row_ids{1}}{$cur_row_ids{5}}[1]{lyrics} ||= [];
+
+      (defined $cur_row_data->[0])
+        and
+      (! $collapse_idx[8]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}} )
+        and
+      push @{ $collapse_idx[7]{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}}[1]{existing_lyric_versions} }, (
+        $collapse_idx[8]{$cur_row_ids{0}}{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}} = [{ lyric_id => $cur_row_data->[10], text => $cur_row_data->[0] }]
+      );
+      $collapse_idx[7]{$cur_row_ids{1}}{$cur_row_ids{5}}{$cur_row_ids{10}}[1]{existing_lyric_versions} ||= [];
+
+      $_[0][$result_pos++] = $collapse_idx[0]{$cur_row_ids{1}}
+        if $is_new_res;
+    }
+
+    splice @{$_[0]}, $result_pos;
+  ',
+  'Multiple has_many on multiple branches with branch pruning torture test',
+);
+
 $infmap = [
   'single_track.trackid',                   # (0) definitive link to root from 1:1:1:1:M:M chain
   'year',                                   # (1) non-unique
@@ -346,17 +583,23 @@ is_same_src (
 
       $collapse_idx[2]{$cur_row_ids{0}}[1]{artist} ||= ($collapse_idx[3]{$cur_row_ids{0}} ||= [{ artistid => $$cur_row_data[6] }]);
 
-      push @{$collapse_idx[3]{$cur_row_ids{0}}[1]{cds}},
-          $collapse_idx[4]{$cur_row_ids{0}}{$cur_row_ids{4}} ||= [{ cdid => $$cur_row_data[4], genreid => $$cur_row_data[7], year => $$cur_row_data[5] }]
-        unless $collapse_idx[4]{$cur_row_ids{0}}{$cur_row_ids{4}};
+      (! $collapse_idx[4]{$cur_row_ids{0}}{$cur_row_ids{4}} )
+        and
+      push @{$collapse_idx[3]{$cur_row_ids{0}}[1]{cds}}, (
+          $collapse_idx[4]{$cur_row_ids{0}}{$cur_row_ids{4}} = [{ cdid => $$cur_row_data[4], genreid => $$cur_row_data[7], year => $$cur_row_data[5] }]
+      );
 
-      push @{$collapse_idx[4]{$cur_row_ids{0}}{$cur_row_ids{4}}[1]{tracks}},
-          $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{4}}{$cur_row_ids{8}} ||= [{ title => $$cur_row_data[8] }]
-        unless $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{4}}{$cur_row_ids{8}};
+      (! $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{4}}{$cur_row_ids{8}} )
+        and
+      push @{$collapse_idx[4]{$cur_row_ids{0}}{$cur_row_ids{4}}[1]{tracks}}, (
+          $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{4}}{$cur_row_ids{8}} = [{ title => $$cur_row_data[8] }]
+      );
 
-      push @{$collapse_idx[0]{$cur_row_ids{10}}[1]{tracks}},
-          $collapse_idx[6]{$cur_row_ids{2}}{$cur_row_ids{3}} ||= [{ cd => $$cur_row_data[2], title => $$cur_row_data[3] }]
-        unless $collapse_idx[6]{$cur_row_ids{2}}{$cur_row_ids{3}};
+      (! $collapse_idx[6]{$cur_row_ids{2}}{$cur_row_ids{3}} )
+        and
+      push @{$collapse_idx[0]{$cur_row_ids{10}}[1]{tracks}}, (
+          $collapse_idx[6]{$cur_row_ids{2}}{$cur_row_ids{3}} = [{ cd => $$cur_row_data[2], title => $$cur_row_data[3] }]
+      );
 
       $_[0][$result_pos++] = $collapse_idx[0]{$cur_row_ids{10}}
         if $is_new_res;
@@ -367,6 +610,84 @@ is_same_src (
   'Multiple has_many on multiple branches with underdefined root torture test',
 );
 
+is_same_src (
+  $schema->source ('CD')->_mk_row_parser({
+    inflate_map => $infmap,
+    collapse => 1,
+    prune_null_branches => 1,
+    hri_style => 1,
+  }),
+  ' my($rows_pos, $result_pos, $cur_row_data, %cur_row_ids, @collapse_idx, $is_new_res) = (0, 0);
+
+    while ($cur_row_data = (
+      ( $rows_pos >= 0 and $_[0][$rows_pos++] ) or do { $rows_pos = -1; undef } )
+        ||
+      ( $_[1] and $_[1]->() )
+    ) {
+
+      $cur_row_ids{$_} = defined $$cur_row_data[$_] ? $$cur_row_data[$_] : "\0NULL\xFF$rows_pos\xFF$_\0"
+        for (0, 2, 3, 4, 8);
+
+      # cache expensive set of ops in a non-existent rowid slot
+      $cur_row_ids{10} = (
+        ( ( defined $cur_row_data->[0] ) && (join "\xFF", q{}, $cur_row_data->[0], q{} ))
+          or
+        ( ( defined $cur_row_data->[2] ) && (join "\xFF", q{}, $cur_row_data->[2], q{} ))
+          or
+        "\0$rows_pos\0"
+      );
+
+      # a present cref in $_[1] implies lazy prefetch, implies a supplied stash in $_[2]
+      $_[1] and $result_pos and unshift(@{$_[2]}, $cur_row_data) and last
+        if ( $is_new_res = ! $collapse_idx[0]{$cur_row_ids{10}} );
+
+      $collapse_idx[0]{$cur_row_ids{10}} ||= { year => $$cur_row_data[1] };
+
+      (defined $cur_row_data->[0])
+        and
+      $collapse_idx[0]{$cur_row_ids{10}}{single_track} ||= ($collapse_idx[1]{$cur_row_ids{0}} ||= { trackid => $$cur_row_data[0] });
+      $collapse_idx[0]{$cur_row_ids{10}}{single_track} ||= undef;
+
+      $collapse_idx[1]{$cur_row_ids{0}}{cd} ||= $collapse_idx[2]{$cur_row_ids{0}};
+
+      $collapse_idx[2]{$cur_row_ids{0}}{artist} ||= ($collapse_idx[3]{$cur_row_ids{0}} ||= { artistid => $$cur_row_data[6] });
+
+      (defined $cur_row_data->[4])
+        and
+      (! $collapse_idx[4]{$cur_row_ids{0}}{$cur_row_ids{4}} )
+        and
+      push @{$collapse_idx[3]{$cur_row_ids{0}}{cds}}, (
+          $collapse_idx[4]{$cur_row_ids{0}}{$cur_row_ids{4}} = { cdid => $$cur_row_data[4], genreid => $$cur_row_data[7], year => $$cur_row_data[5] }
+      );
+      $collapse_idx[3]{$cur_row_ids{0}}{cds} ||= [];
+
+      (defined $cur_row_data->[8])
+        and
+      (! $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{4}}{$cur_row_ids{8}} )
+        and
+      push @{$collapse_idx[4]{$cur_row_ids{0}}{$cur_row_ids{4}}{tracks}}, (
+          $collapse_idx[5]{$cur_row_ids{0}}{$cur_row_ids{4}}{$cur_row_ids{8}} = { title => $$cur_row_data[8] }
+      );
+      $collapse_idx[4]{$cur_row_ids{0}}{$cur_row_ids{4}}{tracks} ||= [];
+
+      (defined $cur_row_data->[2])
+        and
+      (! $collapse_idx[6]{$cur_row_ids{2}}{$cur_row_ids{3}} )
+        and
+      push @{$collapse_idx[0]{$cur_row_ids{10}}{tracks}}, (
+          $collapse_idx[6]{$cur_row_ids{2}}{$cur_row_ids{3}} = { cd => $$cur_row_data[2], title => $$cur_row_data[3] }
+      );
+      $collapse_idx[0]{$cur_row_ids{10}}{tracks} ||= [];
+
+      $_[0][$result_pos++] = $collapse_idx[0]{$cur_row_ids{10}}
+        if $is_new_res;
+    }
+
+    splice @{$_[0]}, $result_pos;
+  ',
+  'Multiple has_many on multiple branches with underdefined root, hri style with branch pruning torture test',
+);
+
 done_testing;
 
 my $deparser;
@@ -382,6 +703,9 @@ sub is_same_src {
     $deparser->coderef2text($cref);
   } @_[0,1];
 
+#use Test::Differences;
+#eq_or_diff($got, $expect);
+
   is ($got, $expect, $_[2]||() )
     or note ("Originals source:\n\n$_[0]\n\n$_[1]\n");
 }