Deduplicate code in rownum limit emulation

Use the RSRC unique constraint traversal to determine if an order_by is stable

diff --git a/lib/DBIx/Class/SQLMaker/LimitDialects.pm b/lib/DBIx/Class/SQLMaker/LimitDialects.pm
index 6469bb0..6bb7b92 100644 (file)
--- a/lib/DBIx/Class/SQLMaker/LimitDialects.pm
+++ b/lib/DBIx/Class/SQLMaker/LimitDialects.pm
@@ -259,11 +259,16 @@ sub _RowNum {
   # ordered by a unique set of columns, it is not safe to use the faster
   # method, and the slower BETWEEN query is used instead
   #
-  # FIXME - this is quite expensive, and doe snot perform caching of any sort
+  # 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
-  if ( __order_by_is_unique($rs_attrs) ) {
-
+  if (
+    $rs_attrs->{order_by}
+      and
+    $rs_attrs->{_rsroot_rsrc}->storage->_order_by_is_stable(
+      $rs_attrs->{from}, $rs_attrs->{order_by}
+    )
+  ) {
     # if offset is 0 (first page) the we can skip a subquery
     if (! $offset) {
       push @{$self->{limit_bind}}, [ $self->__rows_bindtype => $rows ];
@@ -299,43 +304,6 @@ EOS
   }
 }
 
-# determine if the supplied order_by contains a unique column (set)
-sub __order_by_is_unique {
-  my $rs_attrs = shift;
-  my $rsrc = $rs_attrs->{_rsroot_rsrc};
-  my $order_by = $rs_attrs->{order_by}
-    || return 0;
-
-  my $storage = $rsrc->schema->storage;
-
-  my @order_by_cols = map { $_->[0] } $storage->_extract_order_criteria($order_by)
-    or return 0;
-
-  my $colinfo =
-    $storage->_resolve_column_info($rs_attrs->{from}, \@order_by_cols);
-
-  my $sources = {
-    map {( "$_" => $_ )} map { $_->{-result_source} } values %$colinfo
-  };
-
-  my $supplied_order = {
-    map { $_ => 1 }
-    grep { exists $colinfo->{$_} and ! $colinfo->{$_}{is_nullable} }
-    @order_by_cols
-  };
-
-  return 0 unless keys %$supplied_order;
-
-  for my $uks (
-    map { values %$_ } map { +{ $_->unique_constraints } } values %$sources
-  ) {
-    return 1
-      unless first { ! exists $supplied_order->{$_} } @$uks;
-  }
-
-  return 0;
-}
-
 # used by _Top and _FetchFirst below
 sub _prep_for_skimming_limit {
   my ( $self, $sql, $rs_attrs ) = @_;

diff --git a/lib/DBIx/Class/Storage/DBIHacks.pm b/lib/DBIx/Class/Storage/DBIHacks.pm
index 276cefd..66790f4 100644 (file)
--- a/lib/DBIx/Class/Storage/DBIHacks.pm
+++ b/lib/DBIx/Class/Storage/DBIHacks.pm
@@ -680,4 +680,24 @@ sub _extract_order_criteria {
   }
 }
 
+sub _order_by_is_stable {
+  my ($self, $ident, $order_by) = @_;
+
+  my $colinfo = $self->_resolve_column_info(
+    $ident, [ map { $_->[0] } $self->_extract_order_criteria($order_by) ]
+  );
+
+  return undef unless keys %$colinfo;
+
+  my $cols_per_src;
+  $cols_per_src->{$_->{-source_alias}}{$_->{-colname}} = $_ for values %$colinfo;
+
+  for (values %$cols_per_src) {
+    my $src = (values %$_)[0]->{-result_source};
+    return 1 if $src->_identifying_column_set($_);
+  }
+
+  return undef;
+}
+
 1;

diff --git a/t/sqlmaker/limit_dialects/rownum.t b/t/sqlmaker/limit_dialects/rownum.t
index 14ac5cd..522b4d4 100644 (file)
--- a/t/sqlmaker/limit_dialects/rownum.t
+++ b/t/sqlmaker/limit_dialects/rownum.t
@@ -79,6 +79,37 @@ for my $test_set (
       [ $TOTAL => 4 ],
       [ $OFFSET => 4 ],
     ],
+  },
+ {
+    name => 'Rownum subsel aliasing works correctly with non-unique order_by',
+    rs => $rs->search_rs(undef, {
+      rows => 1,
+      offset => 3,
+      columns => [
+        { id => 'foo.id' },
+        { 'bar.id' => 'bar.id' },
+        { bleh => \'TO_CHAR (foo.womble, "blah")' },
+      ],
+      order_by => 'artist',
+    }),
+    sql => '(
+      SELECT id, bar__id, bleh
+      FROM (
+        SELECT id, bar__id, bleh, ROWNUM rownum__index
+        FROM (
+          SELECT foo.id AS id, bar.id AS bar__id, TO_CHAR(foo.womble, "blah") AS bleh
+            FROM cd me
+          WHERE id = ?
+          ORDER BY artist
+        ) me
+      ) me
+      WHERE rownum__index BETWEEN ? and ?
+    )',
+    binds => [
+      $where_bind,
+      [ $OFFSET => 4 ],
+      [ $TOTAL => 4 ],
+    ],
   }, {
     name => 'Rownum subsel aliasing #2 works correctly',
     rs => $rs->search_rs(undef, {