Use a qualified primary key for Model.paged_each and for the paged_operations plugin when using joined datasets

This adds an unambiguous_primary_key private model dataset method,
which returns the qualified primary key for the model if the dataset
is joined, or the unqualified primary key for the model otherwise.

This changes the _force_primary_key_order private model dataset
method to use unambiguous_primary_key, so that it will use a
qualified version of the primary key automatically if the dataset
is joined.  It makes the same change to the _paged_operations_pk
method in the paged_operations plugin.

Fixes #2300

Co-authored-by: Dave Kroondyk <[email protected]>

Author: jeremyevans

CHANGELOG

@@ -1,5 +1,7 @@
 === master
 
+* Use a qualified primary key for Model.paged_each and for the paged_operations plugin when using joined datasets (jeremyevans, davekaro) (#2299, #2300)
+
 * Add Model.qualified_primary_key for returning a qualified version of the model's primary key (jeremyevans)
 
 * Support sql_literal_allow_caching? method on objects that respond to sql_literal_append/sql_literal (jeremyevans)

lib/sequel/model/base.rb

@@ -2278,7 +2278,7 @@ def with_pk!(pk)
       # Return the dataset ordered by the model's primary key.  This should not
       # be used if the model does not have a primary key.
       def _force_primary_key_order
-        cached_dataset(:_pk_order_ds){order(*model.primary_key)}
+        cached_dataset(:_pk_order_ds){order(*unambiguous_primary_key)}
       end
 
       # If the dataset is not already ordered, and the model has a primary key,
@@ -2306,6 +2306,15 @@ def _with_pk_loader
         end
       end
 
+      # The primary key for the dataset's model, qualified if the dataset is joined.
+      def unambiguous_primary_key
+        if joined_dataset?
+          model.qualified_primary_key
+        else
+          model.primary_key
+        end
+      end
+
       def non_sql_option?(key)
         super || key == :model
       end

lib/sequel/plugins/paged_operations.rb

@@ -157,13 +157,16 @@ def _paged_operations_pk(meth)
             raise Error, "the paged_operations plugin is not supported on DB2 when using emulated offsets, set the :offset_strategy Database option to 'limit_offset' or 'offset_fetch'"
           end
 
-          case pk = model.primary_key
+          case pk = unambiguous_primary_key
           when Symbol
             Sequel.identifier(pk)
           when Array
             raise Error, "cannot use #{meth} on a model with a composite primary key"
-          else
+          when nil
             raise Error, "cannot use #{meth} on a model without a primary key"
+          else
+            # Likely SQL::QualifiedIdentifier, if the dataset is joined.
+            pk
           end
         end
 

spec/extensions/paged_operations_spec.rb

@@ -262,3 +262,58 @@ def db.server_version(_=nil); 11020002; end
     ]
   end
 end
+
+describe "paged_operations plugin with joined dataset" do
+  before do
+    @db = Sequel.connect("mock://postgres")
+    @c = Class.new(Sequel::Model(@db[:albums]))
+    @c.plugin :paged_operations
+    @db.sqls
+    @db.fetch = [[{:id=>1002}], [{:id=>2002}]]
+    @db.numrows = [1000, 1000, 2]
+    @ds = @c.dataset.
+      qualify.
+      from(:albums, :artists).
+      where{albums[:id] =~ artists[:album_id]}.
+      with_quote_identifiers(false)
+  end
+
+  it "#paged_delete should use qualified columns when joining" do
+    @ds.paged_delete.must_equal 2002
+    @db.sqls.must_equal [
+      "SELECT albums.id FROM albums, artists WHERE (albums.id = artists.album_id) ORDER BY albums.id LIMIT 1 OFFSET 1000",
+      "DELETE FROM albums USING artists WHERE ((albums.id = artists.album_id) AND (albums.id < 1002))",
+      "SELECT albums.id FROM albums, artists WHERE (albums.id = artists.album_id) ORDER BY albums.id LIMIT 1 OFFSET 1000",
+      "DELETE FROM albums USING artists WHERE ((albums.id = artists.album_id) AND (albums.id < 2002))",
+      "SELECT albums.id FROM albums, artists WHERE (albums.id = artists.album_id) ORDER BY albums.id LIMIT 1 OFFSET 1000",
+      "DELETE FROM albums USING artists WHERE (albums.id = artists.album_id)"
+    ]
+  end
+
+  it "#paged_update should update using multiple queries" do
+    @ds.paged_update({x: Sequel[:artists][:y]}).must_equal 2002
+    @db.sqls.must_equal [
+      "SELECT albums.id FROM albums, artists WHERE (albums.id = artists.album_id) ORDER BY albums.id LIMIT 1 OFFSET 1000",
+      "UPDATE albums SET x = artists.y FROM artists WHERE ((albums.id = artists.album_id) AND (albums.id < 1002))",
+      "SELECT albums.id FROM albums, artists WHERE ((albums.id = artists.album_id) AND (albums.id >= 1002)) ORDER BY albums.id LIMIT 1 OFFSET 1000",
+      "UPDATE albums SET x = artists.y FROM artists WHERE ((albums.id = artists.album_id) AND (albums.id < 2002) AND (albums.id >= 1002))",
+      "SELECT albums.id FROM albums, artists WHERE ((albums.id = artists.album_id) AND (albums.id >= 2002)) ORDER BY albums.id LIMIT 1 OFFSET 1000",
+      "UPDATE albums SET x = artists.y FROM artists WHERE ((albums.id = artists.album_id) AND (albums.id >= 2002))"
+    ]
+  end
+
+  it "#paged_datasets should yield multiple datasets making up dataset" do
+    sqls = []
+    @ds.paged_datasets{|ds| sqls << ds.sql}
+    sqls.must_equal [
+      "SELECT albums.* FROM albums, artists WHERE ((albums.id = artists.album_id) AND (albums.id < 1002))",
+      "SELECT albums.* FROM albums, artists WHERE ((albums.id = artists.album_id) AND (albums.id < 2002) AND (albums.id >= 1002))",
+      "SELECT albums.* FROM albums, artists WHERE ((albums.id = artists.album_id) AND (albums.id >= 2002))"
+    ]
+    @db.sqls.must_equal [
+      "SELECT albums.id FROM albums, artists WHERE (albums.id = artists.album_id) ORDER BY albums.id LIMIT 1 OFFSET 1000",
+      "SELECT albums.id FROM albums, artists WHERE ((albums.id = artists.album_id) AND (albums.id >= 1002)) ORDER BY albums.id LIMIT 1 OFFSET 1000",
+      "SELECT albums.id FROM albums, artists WHERE ((albums.id = artists.album_id) AND (albums.id >= 2002)) ORDER BY albums.id LIMIT 1 OFFSET 1000"
+    ]
+  end
+end

spec/model/dataset_methods_spec.rb

@@ -122,6 +122,11 @@ def destroy
     @c.db.sqls.must_equal ['BEGIN', 'SELECT * FROM items ORDER BY id LIMIT 5 OFFSET 0', 'COMMIT']
   end
 
+  it "#paged_each should order by qualified primary key if the dataset is joined" do
+    @c.join(:x, :a=>:b).paged_each{|r| r.must_be_kind_of(@c)}
+    @c.db.sqls.must_equal ['BEGIN', 'SELECT * FROM items INNER JOIN x ON (x.a = items.b) ORDER BY items.id LIMIT 1000 OFFSET 0', 'COMMIT']
+  end
+
   it "#paged_each should use existing order if there is one" do
     @c.order(:foo).paged_each{|r| r.must_be_kind_of(@c)}
     @c.db.sqls.must_equal ['BEGIN', 'SELECT * FROM items ORDER BY foo LIMIT 1000 OFFSET 0', 'COMMIT']