Skip duplicate upserts.

lib/pgsync/task.rb

@@ -19,6 +19,15 @@ def quoted_table
       quote_ident_full(table)
     end
 
+    def field_equality(column)
+      null_cond = "(#{quoted_table}.#{quote_ident(column)} IS NULL AND EXCLUDED.#{quote_ident(column)} IS NULL)"
+      if to_types[column] == 'json'
+        "#{quoted_table}.#{quote_ident(column)}::jsonb = EXCLUDED.#{quote_ident(column)}::jsonb OR #{null_cond}"
+      else
+        "#{quoted_table}.#{quote_ident(column)} = EXCLUDED.#{quote_ident(column)} OR #{null_cond}"
+      end
+    end
+
     def perform
       with_notices do
         handle_errors do
@@ -37,6 +46,14 @@ def to_fields
       @to_fields ||= to_columns.map { |c| c[:name] }
     end
 
+    def from_types
+      @from_types ||= from_columns.map { |c| [c[:name], c[:type]] }.to_h
+    end
+
+    def to_types
+      @to_types ||= to_columns.map { |c| [c[:name], c[:type]] }.to_h
+    end
+
     def shared_fields
       @shared_fields ||= to_fields & from_fields
     end
@@ -62,8 +79,6 @@ def notes
         missing_sequences = from_sequences - to_sequences
         notes << "Missing sequences: #{missing_sequences.join(", ")}" if missing_sequences.any?
 
-        from_types = from_columns.map { |c| [c[:name], c[:type]] }.to_h
-        to_types = to_columns.map { |c| [c[:name], c[:type]] }.to_h
         different_types = []
         shared_fields.each do |field|
           if from_types[field] != to_types[field]
@@ -135,13 +150,14 @@ def sync_data
         copy(copy_to_command, dest_table: temp_table, dest_fields: fields)
 
         on_conflict = primary_key.map { |pk| quote_ident(pk) }.join(", ")
+
         action =
           if opts[:preserve]
             "NOTHING"
           else # overwrite or sql clause
             setter = shared_fields.reject { |f| primary_key.include?(f) }.map { |f| "#{quote_ident(f)} = EXCLUDED.#{quote_ident(f)}" }
             if setter.any?
-              "UPDATE SET #{setter.join(", ")}"
+              "UPDATE SET #{setter.join(", ")} WHERE NOT (#{shared_fields.reject { |f| primary_key.include?(f) }.map { |f| field_equality(f) }.join(" AND ")})"
             else
               "NOTHING"
             end

test/sync_test.rb

@@ -19,6 +19,20 @@ def test_overwrite
     assert_result("--overwrite", source, dest, expected)
   end
 
+  def test_overwrite_skips_identical_rows
+    source = 3.times.map { |i| {"id" => i + 1, "title" => "Post #{i + 1}"} }
+    dest = source
+    expected = source
+    assert_result("--overwrite", source, dest, expected)
+    # get ctids (~ table row versions)
+    ctids = conn2.exec("select ctid from posts").to_a
+    # rerun the overwrite
+    assert_works "posts --overwrite", config: true
+    ctids2 = conn2.exec("select ctid from posts").to_a
+    # verify ctids have not changed
+    assert_equal(ctids, ctids2)
+  end
+
   def test_preserve
     source = 3.times.map { |i| {"id" => i + 1, "title" => "Post #{i + 1}"} }
     dest = [{"id" => 1, "title" => "First Post"}, {"id" => 4, "title" => "Post 4"}]