<commit_before>module BulkInsert
  class Worker
    attr_reader :connection
    attr_accessor :set_size
    attr_accessor :before_save_callback
    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
      # INSERT IGNORE only fails inserts with duplicate keys or unallowed nulls not the whole set of inserts
      @ignore = ignore

      columns = connection.columns(table_name)
      column_map = columns.inject({}) { |h, c| h.update(c.name => c) }

      @columns = column_names.map { |name| column_map[name.to_s] }
      @table_name = connection.quote_table_name(table_name)
      @column_names = column_names.map { |name| connection.quote_column_name(name) }.join(",")

      @before_save_callback = nil
      @after_save_callback = nil

      @set = []
    end

    def pending?
      @set.any?
    end

    def pending_count
      @set.count
    end

    def add(values)
      save! if @set.length >= set_size

      values = values.with_indifferent_access if values.is_a?(Hash)
      mapped = @columns.map.with_index do |column, index|
          value_exists = values.is_a?(Hash) ? values.key?(column.name) : (index < values.length)
          if !value_exists
            if column.default.present?
              column.default
            elsif column.name == "created_at" || column.name == "updated_at"
              :__timestamp_placeholder
            else
              nil
            end
          else
            values.is_a?(Hash) ? values[column.name] : values[index]
          end
        end

      @set.push(mapped)
      self
    end

    def add_all(rows)
      rows.each { |row| add(row) }
      self
    end

    def before_save(&block)
      @before_save_callback = block
    end

    def after_save(&block)
      @after_save_callback = block
    end

    def save!
      if pending?
        @before_save_callback.(@set) if @before_save_callback
        compose_insert_query.tap { |query| @connection.execute(query) if query }
        @after_save_callback.() if @after_save_callback
        @set.clear
      end

      self
    end

    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

      if !rows.empty?
        sql << rows.join(",")
        sql << on_conflict_statement
        sql
      else
        false
      end
    end

    def insert_sql_statement
      insert_ignore = if ignore
        if adapter_name == "MySQL"
          'IGNORE'
        elsif adapter_name.match(/sqlite.*/i)
          'OR IGNORE'
        else
          '' # Not supported
        end
      end

      "INSERT #{insert_ignore} INTO #{@table_name} (#{@column_names}) VALUES "
    end

    def on_conflict_statement
      (adapter_name == 'PostgreSQL' && ignore ) ? ' ON CONFLICT DO NOTHING' : ''
    end
  end
end
<commit_msg>Add INSERT IGNORE support for Mysql2 adapter<commit_after>module BulkInsert
  class Worker
    attr_reader :connection
    attr_accessor :set_size
    attr_accessor :before_save_callback
    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
      # INSERT IGNORE only fails inserts with duplicate keys or unallowed nulls not the whole set of inserts
      @ignore = ignore

      columns = connection.columns(table_name)
      column_map = columns.inject({}) { |h, c| h.update(c.name => c) }

      @columns = column_names.map { |name| column_map[name.to_s] }
      @table_name = connection.quote_table_name(table_name)
      @column_names = column_names.map { |name| connection.quote_column_name(name) }.join(",")

      @before_save_callback = nil
      @after_save_callback = nil

      @set = []
    end

    def pending?
      @set.any?
    end

    def pending_count
      @set.count
    end

    def add(values)
      save! if @set.length >= set_size

      values = values.with_indifferent_access if values.is_a?(Hash)
      mapped = @columns.map.with_index do |column, index|
          value_exists = values.is_a?(Hash) ? values.key?(column.name) : (index < values.length)
          if !value_exists
            if column.default.present?
              column.default
            elsif column.name == "created_at" || column.name == "updated_at"
              :__timestamp_placeholder
            else
              nil
            end
          else
            values.is_a?(Hash) ? values[column.name] : values[index]
          end
        end

      @set.push(mapped)
      self
    end

    def add_all(rows)
      rows.each { |row| add(row) }
      self
    end

    def before_save(&block)
      @before_save_callback = block
    end

    def after_save(&block)
      @after_save_callback = block
    end

    def save!
      if pending?
        @before_save_callback.(@set) if @before_save_callback
        compose_insert_query.tap { |query| @connection.execute(query) if query }
        @after_save_callback.() if @after_save_callback
        @set.clear
      end

      self
    end

    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

      if !rows.empty?
        sql << rows.join(",")
        sql << on_conflict_statement
        sql
      else
        false
      end
    end

    def insert_sql_statement
      insert_ignore = if ignore
        if adapter_name =~ /^mysql/i
          'IGNORE'
        elsif adapter_name.match(/sqlite.*/i)
          'OR IGNORE'
        else
          '' # Not supported
        end
      end

      "INSERT #{insert_ignore} INTO #{@table_name} (#{@column_names}) VALUES "
    end

    def on_conflict_statement
      (adapter_name == 'PostgreSQL' && ignore ) ? ' ON CONFLICT DO NOTHING' : ''
    end
  end
end

{'fraction_non_alphanumeric': 0.09582799373487114, 'fraction_numerical': 0.0018510608002278229, 'mean_word_length': 2.1148558758314855, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3275888', 'n_tokens_mistral': 2165, 'n_tokens_neox': 2104, 'n_words': 498}