Implemented ignore parameter for PostgreSQL connection

Author: mberlanda

lib/bulk_insert/worker.rb

@@ -3,14 +3,16 @@ class Worker
     attr_reader :connection
     attr_accessor :set_size
     attr_accessor :after_save_callback
+    attr_accessor :adapter_name
+    attr_reader :ignore
 
     def initialize(connection, table_name, column_names, set_size=500, ignore=false)
       @connection = connection
       @set_size = set_size
 
-      @adapter_name = @connection.adapter_name
+      @adapter_name = connection.adapter_name
       # INSERT IGNORE only fails inserts with duplicate keys or unallowed nulls not the whole set of inserts
-      @ignore = ignore ? "IGNORE" : nil
+      @ignore = ignore
 
       columns = connection.columns(table_name)
       column_map = columns.inject({}) { |h, c| h.update(c.name => c) }
@@ -66,34 +68,50 @@ def after_save(&block)
 
     def save!
       if pending?
-        sql = "INSERT #{@ignore} INTO #{@table_name} (#{@column_names}) VALUES "
-        @now = Time.now
-
-        rows = []
-        @set.each do |row|
-          values = []
-          @columns.zip(row) do |column, value|
-            value = @now if value == :__timestamp_placeholder
-
-            if ActiveRecord::VERSION::STRING >= "5.0.0"
-              value = @connection.type_cast_from_column(column, value) if column
-              values << @connection.quote(value)
-            else
-              values << @connection.quote(value, column)
-            end
-          end
-          rows << "(#{values.join(',')})"
-        end
-
-        sql << rows.join(",")
-        @connection.execute(sql)
-
+        @connection.execute(compose_insert_query)
         @after_save_callback.() if @after_save_callback
-
         @set.clear
       end
 
       self
     end
+
+    private
+
+    def compose_insert_query
+      sql = insert_sql_statement
+      @now = Time.now
+
+      rows = []
+      @set.each do |row|
+        values = []
+        @columns.zip(row) do |column, value|
+          value = @now if value == :__timestamp_placeholder
+
+          if ActiveRecord::VERSION::STRING >= "5.0.0"
+            value = @connection.type_cast_from_column(column, value) if column
+            values << @connection.quote(value)
+          else
+            values << @connection.quote(value, column)
+          end
+        end
+        rows << "(#{values.join(',')})"
+      end
+
+      sql << rows.join(",")
+      sql << on_conflict_statement
+      sql
+    end
+
+    def insert_sql_statement
+      ignore = nil
+      ignore = 'IGNORE' if (adapter_name == 'MySQL') && @ignore
+      "INSERT #{ignore} INTO #{@table_name} (#{@column_names}) VALUES "
+    end
+
+    def on_conflict_statement
+      return ' ON CONFLICT DO NOTHING' if (adapter_name == 'PostgreSQL') && @ignore
+      ''
+    end
   end
 end

test/bulk_insert/worker_test.rb

@@ -150,5 +150,43 @@ class BulkInsertWorkerTest < ActiveSupport::TestCase
 
     assert_equal "hello", @insert.after_save_callback.()
   end
-end
 
+  test "adapter dependent default methods" do
+    assert_equal @insert.adapter_name, 'SQLite'
+    assert_equal @insert.send(:insert_sql_statement), "INSERT  INTO \"testings\" (\"greeting\",\"age\",\"happy\",\"created_at\",\"updated_at\",\"color\") VALUES "
+    assert_equal @insert.send(:compose_insert_query), "INSERT  INTO \"testings\" (\"greeting\",\"age\",\"happy\",\"created_at\",\"updated_at\",\"color\") VALUES "
+  end
+
+  test "adapter dependent mysql methods" do
+    mysql_worker = BulkInsert::Worker.new(
+      Testing.connection,
+      Testing.table_name,
+      %w(greeting age happy created_at updated_at color),
+      500, # batch size
+      true) # ignore
+    mysql_worker.adapter_name = 'MySQL'
+
+    assert_equal mysql_worker.adapter_name, 'MySQL'
+    assert_equal (mysql_worker.adapter_name == 'MySQL'), true
+    assert_equal mysql_worker.ignore, true
+    assert_equal ((mysql_worker.adapter_name == 'MySQL') & mysql_worker.ignore), true
+
+    mysql_worker.add ["Yo", 15, false, nil, nil]
+
+    assert_equal mysql_worker.send(:compose_insert_query), "INSERT IGNORE INTO \"testings\" (\"greeting\",\"age\",\"happy\",\"created_at\",\"updated_at\",\"color\") VALUES ('Yo',15,'f',NULL,NULL,'chartreuse')"
+  end
+
+  test "adapter dependent postgresql methods" do
+    pgsql_worker = BulkInsert::Worker.new(
+      Testing.connection,
+      Testing.table_name,
+      %w(greeting age happy created_at updated_at color),
+      500, # batch size
+      true) # ignore
+    pgsql_worker.adapter_name = 'PostgreSQL'
+    pgsql_worker.add ["Yo", 15, false, nil, nil]
+
+    assert_equal pgsql_worker.send(:compose_insert_query), "INSERT  INTO \"testings\" (\"greeting\",\"age\",\"happy\",\"created_at\",\"updated_at\",\"color\") VALUES ('Yo',15,'f',NULL,NULL,'chartreuse') ON CONFLICT DO NOTHING"
+
+  end
+end