class ActiveRecord::Base

def self.synchronize(instances, keys = [primary_key])
  return if instances.empty?
  conditions = {}
  key_values = keys.map { |key| instances.map(&key.to_sym) }
  keys.zip(key_values).each { |key, values| conditions[key] = values }
  order = keys.map { |key| "#{key} ASC" }.join(",")
  klass = instances.first.class
  fresh_instances = klass.unscoped.where(conditions).order(order)
  instances.each do |instance|
    matched_instance = fresh_instances.detect do |fresh_instance|
      keys.all? { |key| fresh_instance.send(key) == instance.send(key) }
    end
    next unless matched_instance
    instance.send :clear_aggregation_cache
    instance.send :clear_association_cache
    instance.instance_variable_set :@attributes, matched_instance.instance_variable_get(:@attributes)
    if instance.respond_to?(:clear_changes_information)
      instance.clear_changes_information                      # Rails 4.2 and higher
    else
      instance.instance_variable_set :@attributes_cache, {}   # Rails 4.0, 4.1
      instance.changed_attributes.clear                       # Rails 3.2
      instance.previous_changes.clear
    end
    # Since the instance now accurately reflects the record in
    # the database, ensure that instance.persisted? is true.
    instance.instance_variable_set '@new_record', false
    instance.instance_variable_set '@destroyed', false
  end
end

def add_special_rails_stamps( column_names, array_of_attributes, options )
  timestamp_columns = {}
  timestamps        = {}
  if respond_to?(:all_timestamp_attributes_in_model, true) # Rails 5.1 and higher
    timestamp_columns[:create] = timestamp_attributes_for_create_in_model
    timestamp_columns[:update] = timestamp_attributes_for_update_in_model
  else
    instance = new
    timestamp_columns[:create] = instance.send(:timestamp_attributes_for_create_in_model)
    timestamp_columns[:update] = instance.send(:timestamp_attributes_for_update_in_model)
  end
  # use tz as set in ActiveRecord::Base
  timestamp = ActiveRecord::Base.default_timezone == :utc ? Time.now.utc : Time.now
  [:create, :update].each do |action|
    timestamp_columns[action].each do |column|
      column = column.to_s
      timestamps[column] = timestamp
      index = column_names.index(column) || column_names.index(column.to_sym)
      if index
        # replace every instance of the array of attributes with our value
        array_of_attributes.each { |arr| arr[index] = timestamp if arr[index].nil? || action == :update }
      else
        column_names << column
        array_of_attributes.each { |arr| arr << timestamp }
      end
      if supports_on_duplicate_key_update? && action == :update
        connection.add_column_for_on_duplicate_key_update(column, options)
      end
    end
  end
  timestamps
end

def establish_connection_with_activerecord_import(*args)
  conn = establish_connection_without_activerecord_import(*args)
  ActiveRecord::Import.load_from_connection_pool connection_pool
  conn
end

def find_associated_objects_for_import(associated_objects_by_class, model)
  associated_objects_by_class[model.class.name] ||= {}
  return associated_objects_by_class unless model.id
  association_reflections =
    model.class.reflect_on_all_associations(:has_one) +
    model.class.reflect_on_all_associations(:has_many)
  association_reflections.each do |association_reflection|
    associated_objects_by_class[model.class.name][association_reflection.name] ||= []
    association = model.association(association_reflection.name)
    association.loaded!
    # Wrap target in an array if not already
    association = Array(association.target)
    changed_objects = association.select { |a| a.new_record? || a.changed? }
    changed_objects.each do |child|
      child.public_send("#{association_reflection.foreign_key}=", model.id)
      # For polymorphic associations
      association_reflection.type.try do |type|
        child.public_send("#{type}=", model.class.base_class.name)
      end
    end
    associated_objects_by_class[model.class.name][association_reflection.name].concat changed_objects
  end
  associated_objects_by_class
end

def import(*args)
  if args.first.is_a?( Array ) && args.first.first.is_a?(ActiveRecord::Base)
    options = {}
    options.merge!( args.pop ) if args.last.is_a?(Hash)
    models = args.first
    import_helper(models, options)
  else
    import_helper(*args)
  end
end

def import!(*args)
  options = args.last.is_a?( Hash ) ? args.pop : {}
  options[:validate] = true
  options[:raise_error] = true
  import(*args, options)
end

def import_associations(models, options)
  # now, for all the dirty associations, collect them into a new set of models, then recurse.
  # notes:
  #    does not handle associations that reference themselves
  #    should probably take a hash to associations to follow.
  return if models.nil?
  associated_objects_by_class = {}
  models.each { |model| find_associated_objects_for_import(associated_objects_by_class, model) }
  # :on_duplicate_key_update not supported for associations
  options.delete(:on_duplicate_key_update)
  associated_objects_by_class.each_value do |associations|
    associations.each_value do |associated_records|
      associated_records.first.class.import(associated_records, options) unless associated_records.empty?
    end
  end
end

def import_helper( *args )
  options = { validate: true, timestamps: true }
  options.merge!( args.pop ) if args.last.is_a? Hash
  # making sure that current model's primary key is used
  options[:primary_key] = primary_key
  # Don't modify incoming arguments
  on_duplicate_key_update = options[:on_duplicate_key_update]
  if on_duplicate_key_update && on_duplicate_key_update.duplicable?
    options[:on_duplicate_key_update] = if on_duplicate_key_update.is_a?(Hash)
      on_duplicate_key_update.each { |k, v| on_duplicate_key_update[k] = v.dup if v.duplicable? }
    else
      on_duplicate_key_update.dup
    end
  end
  is_validating = options[:validate]
  is_validating = true unless options[:validate_with_context].nil?
  # assume array of model objects
  if args.last.is_a?( Array ) && args.last.first.is_a?(ActiveRecord::Base)
    if args.length == 2
      models = args.last
      column_names = args.first.dup
    else
      models = args.first
      column_names = self.column_names.dup
    end
    if models.first.id.nil? && column_names.include?(primary_key) && columns_hash[primary_key].type == :uuid
      column_names.delete(primary_key)
    end
    default_values = column_defaults
    stored_attrs = respond_to?(:stored_attributes) ? stored_attributes : {}
    serialized_attrs = if defined?(ActiveRecord::Type::Serialized)
      attrs = column_names.select { |c| type_for_attribute(c.to_s).class == ActiveRecord::Type::Serialized }
      Hash[attrs.map { |a| [a, nil] }]
    else
      serialized_attributes
    end
    array_of_attributes = models.map do |model|
      if supports_setting_primary_key_of_imported_objects?
        load_association_ids(model)
      end
      column_names.map do |name|
        if stored_attrs.key?(name.to_sym) ||
           serialized_attrs.key?(name) ||
           default_values.key?(name.to_s)
          model.read_attribute(name.to_s)
        else
          model.read_attribute_before_type_cast(name.to_s)
        end
      end
    end
    # supports array of hash objects
  elsif args.last.is_a?( Array ) && args.last.first.is_a?(Hash)
    if args.length == 2
      array_of_hashes = args.last
      column_names = args.first.dup
      allow_extra_hash_keys = true
    else
      array_of_hashes = args.first
      column_names = array_of_hashes.first.keys
      allow_extra_hash_keys = false
    end
    array_of_attributes = array_of_hashes.map do |h|
      error_message = validate_hash_import(h, column_names, allow_extra_hash_keys)
      raise ArgumentError, error_message if error_message
      column_names.map do |key|
        h[key]
      end
    end
    # supports empty array
  elsif args.last.is_a?( Array ) && args.last.empty?
    return ActiveRecord::Import::Result.new([], 0, [])
    # supports 2-element array and array
  elsif args.size == 2 && args.first.is_a?( Array ) && args.last.is_a?( Array )
    unless args.last.first.is_a?(Array)
      raise ArgumentError, "Last argument should be a two dimensional array '[[]]'. First element in array was a #{args.last.first.class}"
    end
    column_names, array_of_attributes = args
    # dup the passed args so we don't modify unintentionally
    column_names = column_names.dup
    array_of_attributes = array_of_attributes.map(&:dup)
  else
    raise ArgumentError, "Invalid arguments!"
  end
  # Force the primary key col into the insert if it's not
  # on the list and we are using a sequence and stuff a nil
  # value for it into each row so the sequencer will fire later
  symbolized_column_names = Array(column_names).map(&:to_sym)
  symbolized_primary_key = Array(primary_key).map(&:to_sym)
  if !symbolized_primary_key.to_set.subset?(symbolized_column_names.to_set) && connection.prefetch_primary_key? && sequence_name
    column_count = column_names.size
    column_names.concat(Array(primary_key)).uniq!
    columns_added = column_names.size - column_count
    new_fields = Array.new(columns_added)
    array_of_attributes.each { |a| a.concat(new_fields) }
  end
  timestamps = {}
  # record timestamps unless disabled in ActiveRecord::Base
  if record_timestamps && options.delete( :timestamps )
    timestamps = add_special_rails_stamps column_names, array_of_attributes, options
  end
  return_obj = if is_validating
    if models
      import_with_validations( column_names, array_of_attributes, options ) do |validator, failed|
        models.each_with_index do |model, i|
          next if validator.valid_model? model
          raise(ActiveRecord::RecordInvalid, model) if options[:raise_error]
          array_of_attributes[i] = nil
          failed << model
        end
      end
    else
      import_with_validations( column_names, array_of_attributes, options )
    end
  else
    import_without_validations_or_callbacks( column_names, array_of_attributes, options )
  end
  if options[:synchronize]
    sync_keys = options[:synchronize_keys] || [primary_key]
    synchronize( options[:synchronize], sync_keys)
  end
  return_obj.num_inserts = 0 if return_obj.num_inserts.nil?
  # if we have ids, then set the id on the models and mark the models as clean.
  if models && supports_setting_primary_key_of_imported_objects?
    set_attributes_and_mark_clean(models, return_obj, timestamps, options)
    # if there are auto-save associations on the models we imported that are new, import them as well
    import_associations(models, options.dup) if options[:recursive]
  end
  return_obj
end

def import_with_validations( column_names, array_of_attributes, options = {} )
  failed_instances = []
  validator = ActiveRecord::Import::Validator.new(options)
  if block_given?
    yield validator, failed_instances
  else
    # create instances for each of our column/value sets
    arr = validations_array_for_column_names_and_attributes( column_names, array_of_attributes )
    # keep track of the instance and the position it is currently at. if this fails
    # validation we'll use the index to remove it from the array_of_attributes
    arr.each_with_index do |hsh, i|
      model = new
      hsh.each_pair { |k, v| model[k] = v }
      next if validator.valid_model? model
      raise(ActiveRecord::RecordInvalid, model) if options[:raise_error]
      array_of_attributes[i] = nil
      failure = model.dup
      failure.errors.send(:initialize_dup, model.errors)
      failed_instances << failure
    end
  end
  array_of_attributes.compact!
  result = if array_of_attributes.empty? || options[:all_or_none] && failed_instances.any?
    ActiveRecord::Import::Result.new([], 0, [], [])
  else
    import_without_validations_or_callbacks( column_names, array_of_attributes, options )
  end
  ActiveRecord::Import::Result.new(failed_instances, result.num_inserts, result.ids, result.results)
end

def import_without_validations_or_callbacks( column_names, array_of_attributes, options = {} )
  column_names = column_names.map(&:to_sym)
  scope_columns, scope_values = scope_attributes.to_a.transpose
  unless scope_columns.blank?
    scope_columns.zip(scope_values).each do |name, value|
      name_as_sym = name.to_sym
      next if column_names.include?(name_as_sym)
      is_sti = (name_as_sym == inheritance_column.to_sym && self < base_class)
      value = Array(value).first if is_sti
      column_names << name_as_sym
      array_of_attributes.each { |attrs| attrs << value }
    end
  end
  columns = column_names.each_with_index.map do |name, i|
    column = columns_hash[name.to_s]
    raise ActiveRecord::Import::MissingColumnError.new(name.to_s, i) if column.nil?
    column
  end
  columns_sql = "(#{column_names.map { |name| connection.quote_column_name(name) }.join(',')})"
  pre_sql_statements = connection.pre_sql_statements( options )
  insert_sql = ['INSERT', pre_sql_statements, "INTO #{quoted_table_name} #{columns_sql} VALUES "]
  insert_sql = insert_sql.flatten.join(' ')
  values_sql = values_sql_for_columns_and_attributes(columns, array_of_attributes)
  number_inserted = 0
  ids = []
  results = []
  if supports_import?
    # generate the sql
    post_sql_statements = connection.post_sql_statements( quoted_table_name, options )
    batch_size = options[:batch_size] || values_sql.size
    values_sql.each_slice(batch_size) do |batch_values|
      # perform the inserts
      result = connection.insert_many( [insert_sql, post_sql_statements].flatten,
        batch_values,
        options,
        "#{model_name} Create Many Without Validations Or Callbacks" )
      number_inserted += result.num_inserts
      ids += result.ids
      results += result.results
    end
  else
    transaction(requires_new: true) do
      values_sql.each do |values|
        ids << connection.insert(insert_sql + values)
        number_inserted += 1
      end
    end
  end
  ActiveRecord::Import::Result.new([], number_inserted, ids, results)
end

def load_association_ids(model)
  association_reflections = model.class.reflect_on_all_associations(:belongs_to)
  association_reflections.each do |association_reflection|
    next if association_reflection.options[:polymorphic]
    association = model.association(association_reflection.name)
    association = association.target
    if association
      association_primary_key = association_reflection.association_primary_key
      model.public_send("#{association_reflection.foreign_key}=", association.send(association_primary_key))
    end
  end
end

def set_attributes_and_mark_clean(models, import_result, timestamps, options)
  return if models.nil?
  models -= import_result.failed_instances
  # if ids were returned for all models we know all were updated
  if models.size == import_result.ids.size
    import_result.ids.each_with_index do |id, index|
      model = models[index]
      model.id = id
      timestamps.each do |attr, value|
        model.send(attr + "=", value)
      end
    end
  end
  if models.size == import_result.results.size
    columns = Array(options[:returning])
    single_column = "#{columns.first}=" if columns.size == 1
    import_result.results.each_with_index do |result, index|
      model = models[index]
      if single_column
        model.send(single_column, result)
      else
        columns.each_with_index do |column, col_index|
          model.send("#{column}=", result[col_index])
        end
      end
    end
  end
  models.each do |model|
    if model.respond_to?(:clear_changes_information) # Rails 4.0 and higher
      model.clear_changes_information
    else # Rails 3.2
      model.instance_variable_get(:@changed_attributes).clear
    end
    model.instance_variable_set(:@new_record, false)
  end
end

def supports_import?(*args)
  connection.respond_to?(:supports_import?) && connection.supports_import?(*args)
end

def supports_on_duplicate_key_update?
  connection.respond_to?(:supports_on_duplicate_key_update?) && connection.supports_on_duplicate_key_update?
end

def supports_setting_primary_key_of_imported_objects?
  connection.respond_to?(:supports_setting_primary_key_of_imported_objects?) && connection.supports_setting_primary_key_of_imported_objects?
end

def synchronize(instances, key = [ActiveRecord::Base.primary_key])
  self.class.synchronize(instances, key)
end

def validate_hash_import(hash, required_keys, allow_extra_keys) # :nodoc:
  extra_keys = allow_extra_keys ? [] : hash.keys - required_keys
  missing_keys = required_keys - hash.keys
  return nil if extra_keys.empty? && missing_keys.empty?
  if allow_extra_keys
    <<-EOS
 key mismatch.
 importing an array of hashes with provided columns_names, each hash must contain keys for all column_names.
ired keys: #{column_names}
ing keys: #{missing_keys}
: #{hash}
    EOS
  else
    <<-EOS
 key mismatch.
 importing an array of hashes, all hashes must have the same keys.
ou have records that are missing some values, we recommend you either set default values
the missing keys or group these records into batches by key set before importing.
ired keys: #{column_names}
a keys: #{extra_keys}
ing keys: #{missing_keys}
: #{hash}
    EOS
  end
end

def validations_array_for_column_names_and_attributes( column_names, array_of_attributes ) # :nodoc:
  array_of_attributes.map { |values| Hash[column_names.zip(values)] }
end

def values_sql_for_columns_and_attributes(columns, array_of_attributes) # :nodoc:
  # connection gets called a *lot* in this high intensity loop.
  # Reuse the same one w/in the loop, otherwise it would keep being re-retreived (= lots of time for large imports)
  connection_memo = connection
  array_of_attributes.map do |arr|
    my_values = arr.each_with_index.map do |val, j|
      column = columns[j]
      # be sure to query sequence_name *last*, only if cheaper tests fail, because it's costly
      if val.nil? && column.name == primary_key && !sequence_name.blank?
        connection_memo.next_value_for_sequence(sequence_name)
      elsif val.respond_to?(:to_sql)
        "(#{val.to_sql})"
      elsif column
        if respond_to?(:type_caster)                                         # Rails 5.0 and higher
          type = type_for_attribute(column.name)
          val = type.type == :boolean ? type.cast(val) : type.serialize(val)
          connection_memo.quote(val)
        elsif column.respond_to?(:type_cast_from_user)                       # Rails 4.2
          connection_memo.quote(column.type_cast_from_user(val), column)
        else                                                                 # Rails 3.2, 4.0 and 4.1
          if serialized_attributes.include?(column.name)
            val = serialized_attributes[column.name].dump(val)
          end
          # Fixes #443 to support binary (i.e. bytea) columns on PG
          val = column.type_cast(val) unless column.type.to_sym == :binary
          connection_memo.quote(val, column)
        end
      end
    end
    "(#{my_values.join(',')})"
  end
end