Tighten up select list processing in ::SQLMaker
[dbsrgits/DBIx-Class.git] / lib / DBIx / Class / SQLMaker / LimitDialects.pm
index 5af9487..0e6eb7e 100644 (file)
@@ -3,9 +3,6 @@ package DBIx::Class::SQLMaker::LimitDialects;
 use warnings;
 use strict;
 
-use List::Util 'first';
-use namespace::clean;
-
 # constants are used not only here, but also in comparison tests
 sub __rows_bindtype () {
   +{ sqlt_datatype => 'integer' }
@@ -61,7 +58,7 @@ sub _LimitOffset {
 
 =head2 LimitXY
 
- SELECT ... LIMIT $offset $limit
+ SELECT ... LIMIT $offset, $limit
 
 Supported by B<MySQL> and any L<SQL::Statement> based DBD
 
@@ -100,7 +97,11 @@ sub _RowNumberOver {
 
   # make up an order if none exists
   my $requested_order = (delete $rs_attrs->{order_by}) || $self->_rno_default_order;
+
+  # the order binds (if any) will need to go at the end of the entire inner select
+  local $self->{order_bind};
   my $rno_ord = $self->_order_by ($requested_order);
+  push @{$self->{select_bind}}, @{$self->{order_bind}};
 
   # this is the order supplement magic
   my $mid_sel = $sq_attrs->{selection_outer};
@@ -117,10 +118,10 @@ sub _RowNumberOver {
   }
 
   # and this is order re-alias magic
-  for ($sq_attrs->{order_supplement}, $sq_attrs->{outer_renames}) {
-    for my $col (keys %$_) {
+  for my $map ($sq_attrs->{order_supplement}, $sq_attrs->{outer_renames}) {
+    for my $col (sort { (length $b) <=> (length $a) } keys %{$map||{}} ) {
       my $re_col = quotemeta ($col);
-      $rno_ord =~ s/$re_col/$_->{$col}/;
+      $rno_ord =~ s/$re_col/$map->{$col}/;
     }
   }
 
@@ -153,7 +154,7 @@ sub _rno_default_order {
 
  SELECT SKIP $offset FIRST $limit * FROM ...
 
-Suported by B<Informix>, almost like LimitOffset. According to
+Supported by B<Informix>, almost like LimitOffset. According to
 L<SQL::Abstract::Limit> C<... SKIP $offset LIMIT $limit ...> is also supported.
 
 =cut
@@ -217,7 +218,7 @@ sub _FirstSkip {
 Depending on the resultset attributes one of:
 
  SELECT * FROM (
-  SELECT *, ROWNUM rownum__index FROM (
+  SELECT *, ROWNUM AS rownum__index FROM (
    SELECT ...
   ) WHERE ROWNUM <= ($limit+$offset)
  ) WHERE rownum__index >= ($offset+1)
@@ -225,7 +226,7 @@ Depending on the resultset attributes one of:
 or
 
  SELECT * FROM (
-  SELECT *, ROWNUM rownum__index FROM (
+  SELECT *, ROWNUM AS rownum__index FROM (
     SELECT ...
   )
  ) WHERE rownum__index BETWEEN ($offset+1) AND ($limit+$offset)
@@ -269,20 +270,20 @@ EOS
   # method, and the slower BETWEEN query is used instead
   #
   # FIXME - this is quite expensive, and does not perform caching of any sort
-  # as soon as some of the DQ work becomes viable consider switching this
-  # over
+  # as soon as some of the SQLA-inlining work becomes viable consider adding
+  # some rudimentary caching support
   if (
     $rs_attrs->{order_by}
       and
-    $rs_attrs->{_rsroot_rsrc}->storage->_order_by_is_stable(
-      $rs_attrs->{from}, $rs_attrs->{order_by}
+    $rs_attrs->{result_source}->schema->storage->_order_by_is_stable(
+      @{$rs_attrs}{qw/from order_by where/}
     )
   ) {
     push @{$self->{limit_bind}}, [ $self->__total_bindtype => $offset + $rows ], [ $self->__offset_bindtype => $offset + 1 ];
 
     return <<EOS;
 SELECT $sq_attrs->{selection_outer} FROM (
-  SELECT $sq_attrs->{selection_outer}, ROWNUM $idx_name FROM (
+  SELECT $sq_attrs->{selection_outer}, ROWNUM AS $idx_name FROM (
     SELECT $sq_attrs->{selection_inner} $sq_attrs->{query_leftover}${order_group_having}
   ) $qalias WHERE ROWNUM <= ?
 ) $qalias WHERE $idx_name >= ?
@@ -293,7 +294,7 @@ EOS
 
     return <<EOS;
 SELECT $sq_attrs->{selection_outer} FROM (
-  SELECT $sq_attrs->{selection_outer}, ROWNUM $idx_name FROM (
+  SELECT $sq_attrs->{selection_outer}, ROWNUM AS $idx_name FROM (
     SELECT $sq_attrs->{selection_inner} $sq_attrs->{query_leftover}${order_group_having}
   ) $qalias
 ) $qalias WHERE $idx_name BETWEEN ? AND ?
@@ -312,87 +313,83 @@ sub _prep_for_skimming_limit {
   $sq_attrs->{order_by_requested} = $self->_order_by ($requested_order);
   $sq_attrs->{grpby_having} = $self->_parse_rs_attrs ($rs_attrs);
 
-  # make up an order unless supplied or sanity check what we are given
-  my $inner_order;
-  if ($sq_attrs->{order_by_requested}) {
-    $self->throw_exception (
-      'Unable to safely perform "skimming type" limit with supplied unstable order criteria'
-    ) unless $rs_attrs->{_rsroot_rsrc}->schema->storage->_order_by_is_stable(
-      $rs_attrs->{from},
-      $requested_order
-    );
-
-    $inner_order = $requested_order;
+  # without an offset things are easy
+  if (! $rs_attrs->{offset}) {
+    $sq_attrs->{order_by_inner} = $sq_attrs->{order_by_requested};
   }
   else {
-    $inner_order = [ map
-      { "$rs_attrs->{alias}.$_" }
-      ( @{
-        $rs_attrs->{_rsroot_rsrc}->_identifying_column_set
-          ||
-        $self->throw_exception(sprintf(
-          'Unable to auto-construct stable order criteria for "skimming type" limit '
-        . "dialect based on source '%s'", $rs_attrs->{_rsroot_rsrc}->name) );
-      } )
-    ];
-  }
-
-  # localise as we already have all the bind values we need
-  local $self->{order_bind};
-
-  $sq_attrs->{order_by_inner} = $self->_order_by ($inner_order);
-
-  my @out_chunks;
-  for my $ch ($self->_order_by_chunks ($inner_order)) {
-    $ch = $ch->[0] if ref $ch eq 'ARRAY';
+    $sq_attrs->{quoted_rs_alias} = $self->_quote ($rs_attrs->{alias});
+
+    # localise as we already have all the bind values we need
+    local $self->{order_bind};
+
+    # make up an order unless supplied or sanity check what we are given
+    my $inner_order;
+    if ($sq_attrs->{order_by_requested}) {
+      $self->throw_exception (
+        'Unable to safely perform "skimming type" limit with supplied unstable order criteria'
+      ) unless ($rs_attrs->{result_source}->schema->storage->_order_by_is_stable(
+        $rs_attrs->{from},
+        $requested_order,
+        $rs_attrs->{where},
+      ));
+
+      $inner_order = $requested_order;
+    }
+    else {
+      $inner_order = [ map
+        { "$rs_attrs->{alias}.$_" }
+        ( @{
+          $rs_attrs->{result_source}->_identifying_column_set
+            ||
+          $self->throw_exception(sprintf(
+            'Unable to auto-construct stable order criteria for "skimming type" limit '
+          . "dialect based on source '%s'", $rs_attrs->{result_source}->name) );
+        } )
+      ];
+    }
 
-    $ch =~ s/\s+ ( ASC|DESC ) \s* $//ix;
-    my $dir = uc ($1||'ASC');
+    $sq_attrs->{order_by_inner} = $self->_order_by ($inner_order);
 
-    push @out_chunks, \join (' ', $ch, $dir eq 'ASC' ? 'DESC' : 'ASC' );
-  }
+    my @out_chunks;
+    for my $ch ($self->_order_by_chunks ($inner_order)) {
+      $ch = $ch->[0] if ref $ch eq 'ARRAY';
 
-  $sq_attrs->{quoted_rs_alias} = $self->_quote ($rs_attrs->{alias});
-  $sq_attrs->{order_by_middle} = $self->_order_by (\@out_chunks);
-  $sq_attrs->{selection_middle} = $sq_attrs->{selection_outer};
-
-  # this is the order supplement magic
-  if (my $extra_order_sel = $sq_attrs->{order_supplement}) {
-    for my $extra_col (sort
-      { $extra_order_sel->{$a} cmp $extra_order_sel->{$b} }
-      keys %$extra_order_sel
-    ) {
-      $sq_attrs->{selection_inner} .= sprintf (', %s AS %s',
-        $extra_col,
-        $extra_order_sel->{$extra_col},
-      );
+      ($ch, my $is_desc) = $self->_split_order_chunk($ch);
 
-      $sq_attrs->{selection_middle} .= ', ' . $extra_order_sel->{$extra_col};
+      # !NOTE! outside chunks come in reverse order ( !$is_desc )
+      push @out_chunks, { ($is_desc ? '-asc' : '-desc') => \$ch };
     }
 
-    # Whatever order bindvals there are, they will be realiased and
-    # reselected, and need to show up at end of the initial inner select
-    push @{$self->{select_bind}}, @{$self->{order_bind}};
-
-    # if this is a part of something bigger, we need to add back all
-    # the extra order_by's, as they may be relied upon by the outside
-    # of a prefetch or something
-    if ($rs_attrs->{_is_internal_subuery}) {
-      $sq_attrs->{selection_outer} .= sprintf ", $extra_order_sel->{$_} AS $_"
-        for sort
-          { $extra_order_sel->{$a} cmp $extra_order_sel->{$b} }
-            grep { $_ !~ /[^\w\-]/ }  # ignore functions
-            keys %$extra_order_sel
-      ;
+    $sq_attrs->{order_by_middle} = $self->_order_by (\@out_chunks);
+
+    # this is the order supplement magic
+    $sq_attrs->{selection_middle} = $sq_attrs->{selection_outer};
+    if (my $extra_order_sel = $sq_attrs->{order_supplement}) {
+      for my $extra_col (sort
+        { $extra_order_sel->{$a} cmp $extra_order_sel->{$b} }
+        keys %$extra_order_sel
+      ) {
+        $sq_attrs->{selection_inner} .= sprintf (', %s AS %s',
+          $extra_col,
+          $extra_order_sel->{$extra_col},
+        );
+
+        $sq_attrs->{selection_middle} .= ', ' . $extra_order_sel->{$extra_col};
+      }
+
+      # Whatever order bindvals there are, they will be realiased and
+      # reselected, and need to show up at end of the initial inner select
+      push @{$self->{select_bind}}, @{$self->{order_bind}};
     }
-  }
 
-  # and this is order re-alias magic
-  for my $map ($sq_attrs->{order_supplement}, $sq_attrs->{outer_renames}) {
-    for my $col (sort { $map->{$a} cmp $map->{$b} } keys %{$map||{}}) {
-      my $re_col = quotemeta ($col);
-      $_ =~ s/$re_col/$map->{$col}/
-        for ($sq_attrs->{order_by_middle}, $sq_attrs->{order_by_requested});
+    # and this is order re-alias magic
+    for my $map ($sq_attrs->{order_supplement}, $sq_attrs->{outer_renames}) {
+      for my $col (sort { (length $b) <=> (length $a) } keys %{$map||{}}) {
+        my $re_col = quotemeta ($col);
+        $_ =~ s/$re_col/$map->{$col}/
+          for ($sq_attrs->{order_by_middle}, $sq_attrs->{order_by_requested});
+      }
     }
   }
 
@@ -425,7 +422,7 @@ sub _Top {
 
   $sql = sprintf ('SELECT TOP %u %s %s %s %s',
     $rows + ($offset||0),
-    $lim->{selection_inner},
+    $offset ? $lim->{selection_inner} : $lim->{selection_original},
     $lim->{query_leftover},
     $lim->{grpby_having},
     $lim->{order_by_inner},
@@ -439,8 +436,7 @@ sub _Top {
     $lim->{order_by_middle},
   ) if $offset;
 
-  $sql = sprintf ('SELECT TOP %u %s FROM ( %s ) %s %s',
-    $rows,
+  $sql = sprintf ('SELECT %s FROM ( %s ) %s %s',
     $lim->{selection_outer},
     $sql,
     $lim->{quoted_rs_alias},
@@ -480,7 +476,7 @@ sub _FetchFirst {
   my $lim = $self->_prep_for_skimming_limit($sql, $rs_attrs);
 
   $sql = sprintf ('SELECT %s %s %s %s FETCH FIRST %u ROWS ONLY',
-    $lim->{selection_inner},
+    $offset ? $lim->{selection_inner} : $lim->{selection_original},
     $lim->{query_leftover},
     $lim->{grpby_having},
     $lim->{order_by_inner},
@@ -496,12 +492,11 @@ sub _FetchFirst {
   ) if $offset;
 
 
-  $sql = sprintf ('SELECT %s FROM ( %s ) %s %s FETCH FIRST %u ROWS ONLY',
+  $sql = sprintf ('SELECT %s FROM ( %s ) %s %s',
     $lim->{selection_outer},
     $sql,
     $lim->{quoted_rs_alias},
     $lim->{order_by_requested},
-    $rows,
   ) if $offset and (
     $lim->{order_by_requested} or $lim->{selection_middle} ne $lim->{selection_outer}
   );
@@ -509,32 +504,6 @@ sub _FetchFirst {
   return $sql;
 }
 
-=head2 RowCountOrGenericSubQ
-
-This is not exactly a limit dialect, but more of a proxy for B<Sybase ASE>.
-If no $offset is supplied the limit is simply performed as:
-
- SET ROWCOUNT $limit
- SELECT ...
- SET ROWCOUNT 0
-
-Otherwise we fall back to L</GenericSubQ>
-
-=cut
-
-sub _RowCountOrGenericSubQ {
-  my $self = shift;
-  my ($sql, $rs_attrs, $rows, $offset) = @_;
-
-  return $self->_GenericSubQ(@_) if $offset;
-
-  return sprintf <<"EOF", $rows, $sql;
-SET ROWCOUNT %d
-%s
-SET ROWCOUNT 0
-EOF
-}
-
 =head2 GenericSubQ
 
  SELECT * FROM (
@@ -548,8 +517,11 @@ This is the most evil limit "dialect" (more of a hack) for I<really> stupid
 databases. It works by ordering the set by some unique column, and calculating
 the amount of rows that have a less-er value (thus emulating a L</RowNum>-like
 index). Of course this implies the set can only be ordered by a single unique
-column. Also note that this technique can be and often is B<excruciatingly
-slow>.
+column.
+
+Also note that this technique can be and often is B<excruciatingly slow>. You
+may have much better luck using L<DBIx::Class::ResultSet/software_limit>
+instead.
 
 Currently used by B<Sybase ASE>, due to lack of any other option.
 
@@ -557,50 +529,115 @@ Currently used by B<Sybase ASE>, due to lack of any other option.
 sub _GenericSubQ {
   my ($self, $sql, $rs_attrs, $rows, $offset) = @_;
 
-  my $root_rsrc = $rs_attrs->{_rsroot_rsrc};
-  my $root_tbl_name = $root_rsrc->name;
+  my $main_rsrc = $rs_attrs->{result_source};
 
-  my ($first_order_by) = do {
-    local $self->{quote_char};
-    map { ref $_ ? $_->[0] : $_ } $self->_order_by_chunks ($rs_attrs->{order_by})
-  } or $self->throw_exception (
-    'Generic Subquery Limit does not work on resultsets without an order. Provide a single, '
-  . 'unique-column order criteria.'
-  );
+  # Explicitly require an order_by
+  # GenSubQ is slow enough as it is, just emulating things
+  # like in other cases is not wise - make the user work
+  # to shoot their DBA in the foot
+  $self->throw_exception (
+    'Generic Subquery Limit does not work on resultsets without an order. Provide a stable, '
+  . 'main-table-based order criteria.'
+  ) unless $rs_attrs->{order_by};
 
-  $first_order_by =~ s/\s+ ( ASC|DESC ) \s* $//ix;
-  my $direction = lc ($1 || 'asc');
+  my $usable_order_colinfo = $main_rsrc->schema->storage->_extract_colinfo_of_stable_main_source_order_by_portion(
+    $rs_attrs
+  );
 
-  my ($first_ord_alias, $first_ord_col) = $first_order_by =~ /^ (?: ([^\.]+) \. )? ([^\.]+) $/x;
+  $self->throw_exception(
+    'Generic Subquery Limit can not work with order criteria based on sources other than the main one'
+  ) if (
+    ! keys %{$usable_order_colinfo||{}}
+      or
+    grep
+      { $_->{-source_alias} ne $rs_attrs->{alias} }
+      (values %$usable_order_colinfo)
+  );
 
-  $self->throw_exception(sprintf
-    "Generic Subquery Limit order criteria can be only based on the root-source '%s'"
-  . " (aliased as '%s')", $root_rsrc->source_name, $rs_attrs->{alias},
-  ) if ($first_ord_alias and $first_ord_alias ne $rs_attrs->{alias});
+###
+###
+### we need to know the directions after we figured out the above - reextract *again*
+### this is eyebleed - trying to get it to work at first
+  my $supplied_order = delete $rs_attrs->{order_by};
 
-  $first_ord_alias ||= $rs_attrs->{alias};
+  my @order_bits = do {
+    local $self->{quote_char};
+    local $self->{order_bind};
+    map { ref $_ ? $_->[0] : $_ } $self->_order_by_chunks ($supplied_order)
+  };
 
-  $self->throw_exception(
-    "Generic Subquery Limit first order criteria '$first_ord_col' must be unique"
-  ) unless $root_rsrc->_identifying_column_set([$first_ord_col]);
+  # truncate to what we'll use
+  $#order_bits = ( (keys %$usable_order_colinfo) - 1 );
 
-  my $sq_attrs = $self->_subqueried_limit_attrs ($sql, $rs_attrs);
+  # @order_bits likely will come back quoted (due to how the prefetch
+  # rewriter operates
+  # Hence supplement the column_info lookup table with quoted versions
+  if ($self->quote_char) {
+    $usable_order_colinfo->{$self->_quote($_)} = $usable_order_colinfo->{$_}
+      for keys %$usable_order_colinfo;
+  }
 
-  my $cmp_op = $direction eq 'desc' ? '>' : '<';
+# calculate the condition
   my $count_tbl_alias = 'rownum__emulation';
+  my $main_alias = $rs_attrs->{alias};
+  my $main_tbl_name = $main_rsrc->name;
 
-  my $order_sql = $self->_order_by (delete $rs_attrs->{order_by});
-  my $group_having_sql = $self->_parse_rs_attrs($rs_attrs);
+  my (@unqualified_names, @qualified_names, @is_desc, @new_order_by);
 
-  my $in_sel = $sq_attrs->{selection_inner};
+  for my $bit (@order_bits) {
 
-  # add the order supplement (if any) as this is what will be used for the outer WHERE
-  $in_sel .= ", $_" for keys %{$sq_attrs->{order_supplement}};
+    ($bit, my $is_desc) = $self->_split_order_chunk($bit);
+
+    push @is_desc, $is_desc;
+    push @unqualified_names, $usable_order_colinfo->{$bit}{-colname};
+    push @qualified_names, $usable_order_colinfo->{$bit}{-fq_colname};
+
+    push @new_order_by, { ($is_desc ? '-desc' : '-asc') => $usable_order_colinfo->{$bit}{-fq_colname} };
+  };
+
+  my (@where_cond, @skip_colpair_stack);
+  for my $i (0 .. $#order_bits) {
+    my $ci = $usable_order_colinfo->{$order_bits[$i]};
+
+    my ($subq_col, $main_col) = map { "$_.$ci->{-colname}" } ($count_tbl_alias, $main_alias);
+    my $cur_cond = { $subq_col => { ($is_desc[$i] ? '>' : '<') => { -ident => $main_col } } };
+
+    push @skip_colpair_stack, [
+      { $main_col => { -ident => $subq_col } },
+    ];
+
+    # we can trust the nullability flag because
+    # we already used it during _id_col_set resolution
+    #
+    if ($ci->{is_nullable}) {
+      push @{$skip_colpair_stack[-1]}, { $main_col => undef, $subq_col=> undef };
+
+      $cur_cond = [
+        {
+          ($is_desc[$i] ? $subq_col : $main_col) => { '!=', undef },
+          ($is_desc[$i] ? $main_col : $subq_col) => undef,
+        },
+        {
+          $subq_col => { '!=', undef },
+          $main_col => { '!=', undef },
+          -and => $cur_cond,
+        },
+      ];
+    }
 
+    push @where_cond, { '-and', => [ @skip_colpair_stack[0..$i-1], $cur_cond ] };
+  }
+
+# reuse the sqlmaker WHERE, this will not be returning binds
+  my $counted_where = do {
+    local $self->{where_bind};
+    $self->where(\@where_cond);
+  };
+
+# construct the rownum condition by hand
   my $rownum_cond;
   if ($offset) {
     $rownum_cond = 'BETWEEN ? AND ?';
-
     push @{$self->{limit_bind}},
       [ $self->__offset_bindtype => $offset ],
       [ $self->__total_bindtype => $offset + $rows - 1]
@@ -608,25 +645,51 @@ sub _GenericSubQ {
   }
   else {
     $rownum_cond = '< ?';
-
     push @{$self->{limit_bind}},
       [ $self->__rows_bindtype => $rows ]
     ;
   }
 
+# and what we will order by inside
+  my $inner_order_sql = do {
+    local $self->{order_bind};
+
+    my $s = $self->_order_by (\@new_order_by);
+
+    $self->throw_exception('Inner gensubq order may not contain binds... something went wrong')
+      if @{$self->{order_bind}};
+
+    $s;
+  };
+
+### resume originally scheduled programming
+###
+###
+
+  # we need to supply the order for the supplements to be properly calculated
+  my $sq_attrs = $self->_subqueried_limit_attrs (
+    $sql, { %$rs_attrs, order_by => \@new_order_by }
+  );
+
+  my $in_sel = $sq_attrs->{selection_inner};
+
+  # add the order supplement (if any) as this is what will be used for the outer WHERE
+  $in_sel .= ", $_" for sort keys %{$sq_attrs->{order_supplement}};
+
+  my $group_having_sql = $self->_parse_rs_attrs($rs_attrs);
+
+
   return sprintf ("
 SELECT $sq_attrs->{selection_outer}
   FROM (
     SELECT $in_sel $sq_attrs->{query_leftover}${group_having_sql}
   ) %s
-WHERE ( SELECT COUNT(*) FROM %s %s WHERE %s $cmp_op %s ) $rownum_cond
-$order_sql
+WHERE ( SELECT COUNT(*) FROM %s %s $counted_where ) $rownum_cond
+$inner_order_sql
   ", map { $self->_quote ($_) } (
     $rs_attrs->{alias},
-    $root_tbl_name,
+    $main_tbl_name,
     $count_tbl_alias,
-    "$count_tbl_alias.$first_ord_col",
-    "$first_ord_alias.$first_ord_col",
   ));
 }
 
@@ -635,7 +698,7 @@ $order_sql
 #
 # Generates inner/outer select lists for various limit dialects
 # which result in one or more subqueries (e.g. RNO, Top, RowNum)
-# Any non-root-table columns need to have their table qualifier
+# Any non-main-table columns need to have their table qualifier
 # turned into a column alias (otherwise names in subqueries clash
 # and/or lose their source table)
 #
@@ -667,23 +730,29 @@ sub _subqueried_limit_attrs {
 
   my ($re_sep, $re_alias) = map { quotemeta $_ } ( $self->{name_sep}, $rs_attrs->{alias} );
 
-  # insulate from the multiple _recurse_fields calls below
-  local $self->{select_bind};
-
   # correlate select and as, build selection index
   my (@sel, $in_sel_index);
   for my $i (0 .. $#{$rs_attrs->{select}}) {
 
     my $s = $rs_attrs->{select}[$i];
-    my $sql_sel = $self->_recurse_fields ($s);
     my $sql_alias = (ref $s) eq 'HASH' ? $s->{-as} : undef;
 
+    my ($sql_sel) = length ref $s
+      # we throw away the @bind here deliberately
+      ? $self->_recurse_fields( $s )
+      : $self->_quote( $s )
+    ;
+
     push @sel, {
+      arg => $s,
       sql => $sql_sel,
-      unquoted_sql => do {
-        local $self->{quote_char};
-        $self->_recurse_fields ($s);
-      },
+      unquoted_sql => ( length ref $s
+        ? do {
+          local $self->{quote_char};
+          ($self->_recurse_fields ($s))[0]; # ignore binds again
+        }
+        : $s
+      ),
       as =>
         $sql_alias
           ||
@@ -693,7 +762,9 @@ sub _subqueried_limit_attrs {
       ,
     };
 
-    $in_sel_index->{$sql_sel}++;
+    # anything with a placeholder in it needs re-selection
+    $in_sel_index->{$sql_sel}++ unless $sql_sel =~ / (?: ^ | \W ) \? (?: \W | $ ) /x;
+
     $in_sel_index->{$self->_quote ($sql_alias)}++ if $sql_alias;
 
     # record unqualified versions too, so we do not have
@@ -709,9 +780,14 @@ sub _subqueried_limit_attrs {
   # unless we are dealing with the current source alias
   # (which will transcend the subqueries as it is necessary
   # for possible further chaining)
+  # same for anything we do not recognize
   my ($sel, $renamed);
   for my $node (@sel) {
+    push @{$sel->{original}}, $node->{sql};
+
     if (
+      ! $in_sel_index->{$node->{sql}}
+        or
       $node->{as} =~ / (?<! ^ $re_alias ) \. /x
         or
       $node->{unquoted_sql} =~ / (?<! ^ $re_alias ) $re_sep /x
@@ -724,7 +800,7 @@ sub _subqueried_limit_attrs {
     }
     else {
       push @{$sel->{inner}}, $node->{sql};
-      push @{$sel->{outer}}, $self->_quote ($node->{as});
+      push @{$sel->{outer}}, $self->_quote (ref $node->{arg} ? $node->{as} : $node->{arg});
     }
   }
 
@@ -733,12 +809,12 @@ sub _subqueried_limit_attrs {
   for my $chunk ($self->_order_by_chunks ($rs_attrs->{order_by})) {
     # order with bind
     $chunk = $chunk->[0] if (ref $chunk) eq 'ARRAY';
-    $chunk =~ s/\s+ (?: ASC|DESC ) \s* $//ix;
+    ($chunk) = $self->_split_order_chunk($chunk);
 
     next if $in_sel_index->{$chunk};
 
     $extra_order_sel->{$chunk} ||= $self->_quote (
-      'ORDER__BY__' . scalar keys %{$extra_order_sel||{}}
+      'ORDER__BY__' . sprintf '%03d', scalar keys %{$extra_order_sel||{}}
     );
   }
 
@@ -756,14 +832,17 @@ sub _unqualify_colname {
   return $fqcn;
 }
 
-1;
-
-=head1 AUTHORS
+=head1 FURTHER QUESTIONS?
 
-See L<DBIx::Class/CONTRIBUTORS>.
+Check the list of L<additional DBIC resources|DBIx::Class/GETTING HELP/SUPPORT>.
 
-=head1 LICENSE
+=head1 COPYRIGHT AND LICENSE
 
-You may distribute this code under the same terms as Perl itself.
+This module is free software L<copyright|DBIx::Class/COPYRIGHT AND LICENSE>
+by the L<DBIx::Class (DBIC) authors|DBIx::Class/AUTHORS>. You can
+redistribute it and/or modify it under the same terms as the
+L<DBIx::Class library|DBIx::Class/COPYRIGHT AND LICENSE>.
 
 =cut
+
+1;