ActiveRecord Insert Test

ar_bulk_insert.rb

require 'active_record'
require 'benchmark'
require 'faker'

ActiveRecord::Base.establish_connection(
  adapter: 'fdbsql', database: 'test', host: 'localhost', pool: 20
)

class User < ActiveRecord::Base
end

ActiveRecord::Schema.define do
  create_table User.table_name do |t|
    t.string :first_name
    t.string :last_name
    t.string :email
    t.boolean :is_admin
    t.integer :points
  end
end unless ActiveRecord::Schema.table_exists?(User.table_name)

def build_insert_stmt(model, with_id, value_as_param)
  cols = User.column_names.sort
  cols = cols.reject { |a| a == "id" } if !with_id
  "INSERT INTO \"#{model.table_name}\" (#{
    cols.map { |a| "\"#{a}\"" }.join(",")
  }) VALUES (#{
    # Send all columns as string and let SQL Layer do conversion
    cols.map { |a| value_as_param ? "?" : "'%{#{a}}'" }.join(",")
  })"
end

def escape_value(v)
  (v.is_a? String) ? v.gsub("'", "''") : v
end

COLUMNS_BY_SYM = User.columns.map { |c| [c.name.to_sym, c] }.to_h
INSERT_ORDER = COLUMNS_BY_SYM.keys.sort
INSERT_NO_ID_ORDER = INSERT_ORDER.reject { |c| c == :id }
SQL_INSERT_ONE_ROW_LITERALS = build_insert_stmt(User, false, false)
SQL_INSERT_ONE_ROW_PARAM = build_insert_stmt(User, false, true)
SQL_INSERT_ONE_ROW_WITH_ID_PARAM = build_insert_stmt(User, true, true)

puts ""
puts "Generating data..."
test_users = []
10000.times do 
  test_users << {
    first_name: Faker::Name.first_name,
    last_name: Faker::Name.last_name,
    email: Faker::Internet.email,
    is_admin: Faker::Number.number(1) == "0",
    points: Faker::Number.number(5),
  }
end
puts "Done, total count: #{test_users.length}"


TEST_RESULTS = []

def run_test(desc)
  puts "Running #{desc}..."
  ActiveRecord::Base.connection.execute("TRUNCATE TABLE #{User.table_name}")
  TEST_RESULTS << {
    desc: desc,
    time: Benchmark.measure { yield() },
    count: User.count
  }
end


run_test "One Thread, single User.create()" do
  test_users.each do |u|
    User.create(u)
  end
end

run_test "One Thread, 100 User.create()" do
  test_users.each_slice(100) do |sub_u|
    ActiveRecord::Base.transaction do
      sub_u.each do |u|
        User.create(u)
      end
    end
  end
end

run_test "One Thread, single non-param INSERT" do
  test_users.each do |u|
    safe_u = u.map { |k,v| [k, escape_value(v)] }.to_h
    User.connection.exec_query(sprintf(SQL_INSERT_ONE_ROW_LITERALS, safe_u))
  end
end

run_test "One Thread, 100 non-param INSERT" do
  test_users.each_slice(100) do |sub_u|
    ActiveRecord::Base.transaction do
      sub_u.each do |u|
        safe_u = u.map { |k,v| [k, escape_value(v)] }.to_h
        User.connection.exec_query(sprintf(SQL_INSERT_ONE_ROW_LITERALS, safe_u))
      end
    end
  end
end

run_test "One Thread, single param INSERT" do
  test_users.each do |u|
    params = INSERT_NO_ID_ORDER.map { |a| [COLUMNS_BY_SYM[a], u[a]] }
    User.connection.exec_query(SQL_INSERT_ONE_ROW_PARAM, nil, params)
  end
end

run_test "One Thread, 100 param INSERT" do
  test_users.each_slice(100) do |sub_u|
    ActiveRecord::Base.transaction do
      sub_u.each do |u|
        params = INSERT_NO_ID_ORDER.map { |a| [COLUMNS_BY_SYM[a], u[a]] }
        User.connection.exec_query(SQL_INSERT_ONE_ROW_PARAM, nil, params)
      end
    end
  end
end

run_test "Ten Threads, 100 param INSERT" do
  chunk_size = test_users.length / 10
  offset = 0
  threads = test_users.each_slice(chunk_size).map do |chunk|
    t = Thread.new(chunk, offset) do |users, id_offset|
      ActiveRecord::Base::connection_pool.with_connection do |conn|
        users.each_slice(100) do |sub_u|
          tmp_id = id_offset
          begin
            conn.transaction do
              sub_u.each do |u|
                tmp_id += 1
                params = INSERT_NO_ID_ORDER.map { |a| [COLUMNS_BY_SYM[a], (a == :id ? tmp_id : u[a])] }
                conn.exec_query(SQL_INSERT_ONE_ROW_PARAM, nil, params)
              end
            end
          rescue ActiveRecord::StatementInvalid => e
            # This *should* unwrap e to get original exception
            # instead of looking at the message
            retry if e.message.include?("not_committed")
            raise e
          end
          id_offset = tmp_id
        end
      end
    end
    offset += chunk_size
    t
  end
  threads.each { |t| t.join }
end

run_test "Ten Threads, 100 param INSERT with ID, delayed checkTime" do
  chunk_size = test_users.length / 10
  offset = 0
  threads = test_users.each_slice(chunk_size).map do |chunk|
    t = Thread.new(chunk, offset) do |users, id_offset|
      ActiveRecord::Base::connection_pool.with_connection do |conn|
        conn.execute("SET constraintCheckTime = 'DELAYED_WITH_RANGE_CACHE'")
        users.each_slice(100) do |sub_u|
          tmp_id = id_offset
          conn.transaction do
            sub_u.each do |u|
              tmp_id += 1
              params = INSERT_ORDER.map { |a| [COLUMNS_BY_SYM[a], (a == :id ? tmp_id : u[a])] }
              conn.exec_query(SQL_INSERT_ONE_ROW_WITH_ID_PARAM, nil, params)
            end
          end
          id_offset = tmp_id
        end
        conn.execute("SET constraintCheckTime TO DEFAULT")
      end
    end
    offset += chunk_size
    t
  end
  threads.each { |t| t.join }
end

len = TEST_RESULTS.map { |r| r[:desc] }.max_by { |desc| desc.length }.length
format = "%-#{len}s | %-11s | %-10s"

puts
puts sprintf(format, "Description", "Loaded Rows", "Elapsed")
puts "-" * (len + 3 + 3 + 11 + 10)
TEST_RESULTS.sort_by { |r| r[:time].real }.each { |r| 
  puts sprintf(format, r[:desc], r[:count], sprintf("%g", r[:time].real))
}

example output

$ ruby ar_bulk_insert.rb
-- table_exists?("users")
   -> 0.0321s
-- create_table("users")
   -> 0.0442s

Generating data...
Done, total count: 10000
Running One Thread, single User.create()...
Running One Thread, 100 User.create()...
Running One Thread, single non-param INSERT...
Running One Thread, 100 non-param INSERT...
Running One Thread, single param INSERT...
Running One Thread, 100 param INSERT...
Running Ten Threads, 100 param INSERT...
Running Ten Threads, 100 param INSERT with ID, delayed checkTime...

Description                                              | Loaded Rows | Elapsed
-----------------------------------------------------------------------------------
Ten Threads, 100 param INSERT with ID, delayed checkTime | 10000       | 1.26987
One Thread, 100 param INSERT                             | 10000       | 4.36524
One Thread, 100 non-param INSERT                         | 10000       | 5.03679
One Thread, 100 User.create()                            | 10000       | 9.11457
Ten Threads, 100 param INSERT                            | 10000       | 9.77568
One Thread, single param INSERT                          | 10000       | 90.2943
One Thread, single non-param INSERT                      | 10000       | 93.588
One Thread, single User.create()                         | 10000       | 105.696