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.instance_variable_set :@association_cache, {}
    instance.send :clear_aggregation_cache if instance.respond_to?(:clear_aggregation_cache, true)
    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 = allocate
    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
  default_timezone = if ActiveRecord.respond_to?(:default_timezone)
    ActiveRecord.default_timezone
  else
    ActiveRecord::Base.default_timezone
  end
  timestamp = 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? }
      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 associated_options(options, association)
  return options unless options.key?(:recursive_on_duplicate_key_update)
  options.merge(
    on_duplicate_key_update: options[:recursive_on_duplicate_key_update][association]
  )
end

def bulk_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 bulk_import!(*args)
  options = args.last.is_a?( Hash ) ? args.pop : {}
  options[:validate] = true
  options[:raise_error] = true
  bulk_import(*args, options)
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|
      Array(association_reflection.inverse_of&.foreign_key || association_reflection.foreign_key).each_with_index do |column, index|
        child.public_send("#{column}=", Array(model.id)[index])
      end
      # For polymorphic associations
      association_name = if model.class.respond_to?(:polymorphic_name)
        model.class.polymorphic_name
      else
        model.class.base_class
      end
      association_reflection.type.try do |type|
        child.public_send("#{type}=", association_name)
      end
    end
    associated_objects_by_class[model.class.name][association_reflection.name].concat changed_objects
  end
  associated_objects_by_class
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 only supported for all fields
  options.delete(:on_duplicate_key_update) unless options[:on_duplicate_key_update] == :all
  # :returning not supported for associations
  options.delete(:returning)
  associated_objects_by_class.each_value do |associations|
    associations.each do |association, associated_records|
      next if associated_records.empty?
      associated_class = associated_records.first.class
      associated_class.bulk_import(associated_records,
                                   associated_options(options, association))
    end
  end
end

def import_helper( *args )
  options = { model: self, validate: true, timestamps: true, track_validation_failures: false }
  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
  options[:locking_column] = locking_column if locking_enabled?
  is_validating = options[:validate_with_context].present? || options[:validate]
  validator = ActiveRecord::Import::Validator.new(self, options)
  # 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 = if connection.respond_to?(:supports_virtual_columns?) && connection.supports_virtual_columns?
        columns.reject(&:virtual?).map(&:name)
      else
        self.column_names.dup
      end
    end
    if models.first.id.nil?
      Array(primary_key).each do |c|
        if column_names.include?(c) && schema_columns_hash[c].type == :uuid
          column_names.delete(c)
        end
      end
    end
    update_attrs = if record_timestamps && options[:timestamps]
      if respond_to?(:timestamp_attributes_for_update, true)
        send(:timestamp_attributes_for_update).map(&:to_sym)
      else
        allocate.send(:timestamp_attributes_for_update_in_model)
      end
    end
    array_of_attributes = []
    models.each do |model|
      if supports_setting_primary_key_of_imported_objects?
        load_association_ids(model)
      end
      if is_validating && !validator.valid_model?(model)
        raise(ActiveRecord::RecordInvalid, model) if options[:raise_error]
        next
      end
      array_of_attributes << column_names.map do |name|
        if model.persisted? &&
           update_attrs && update_attrs.include?(name.to_sym) &&
           !model.send("#{name}_changed?")
          nil
        else
          model.read_attribute(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
  # Don't modify incoming arguments
  on_duplicate_key_update = options[:on_duplicate_key_update]
  if on_duplicate_key_update
    updatable_columns = symbolized_column_names.reject { |c| symbolized_primary_key.include? c }
    options[:on_duplicate_key_update] = if on_duplicate_key_update.is_a?(Hash)
      on_duplicate_key_update.each_with_object({}) do |(k, v), duped_options|
        duped_options[k] = if k == :columns && v == :all
          updatable_columns
        elsif v.duplicable?
          v.dup
        else
          v
        end
      end
    elsif on_duplicate_key_update == :all
      updatable_columns
    elsif on_duplicate_key_update.duplicable?
      on_duplicate_key_update.dup
    else
      on_duplicate_key_update
    end
  end
  timestamps = {}
  # record timestamps unless disabled in ActiveRecord::Base
  if record_timestamps && options[:timestamps]
    timestamps = add_special_rails_stamps column_names, array_of_attributes, options
  end
  return_obj = if is_validating
    import_with_validations( column_names, array_of_attributes, options ) do |failed_instances|
      if models
        models.each_with_index do |m, i|
          next unless m.errors.any?
          failed_instances << (options[:track_validation_failures] ? [i, m] : m)
        end
      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|
          # utilize block initializer syntax to prevent failure when 'mass_assignment_sanitizer = :strict'
          model = new do |m|
            hsh.each_pair { |k, v| m[k] = v }
          end
          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 << (options[:track_validation_failures] ? [i, failure] : failure )
        end
        array_of_attributes.compact!
      end
    end
  else
    import_without_validations_or_callbacks( column_names, array_of_attributes, options )
  end
  if options[:synchronize]
    sync_keys = options[:synchronize_keys] || Array(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
    if options[:recursive]
      options[:on_duplicate_key_update] = on_duplicate_key_update unless on_duplicate_key_update.nil?
      import_associations(models, options.dup.merge(validate: false))
    end
  end
  return_obj
end

def import_with_validations( column_names, array_of_attributes, options = {} )
  failed_instances = []
  yield failed_instances if block_given?
  result = if 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 = {} )
  return ActiveRecord::Import::Result.new([], 0, [], []) if array_of_attributes.empty?
  column_names = column_names.map do |name|
    original_name = attribute_alias?(name) ? attribute_alias(name) : name
    original_name.to_sym
  end
  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) || name_as_sym == inheritance_column.to_sym
      column_names << name_as_sym
      array_of_attributes.each { |attrs| attrs << value }
    end
  end
  if finder_needs_type_condition? && !column_names.include?(inheritance_column.to_sym)
    column_names << inheritance_column.to_sym
    array_of_attributes.each { |attrs| attrs << sti_name }
  end
  columns = column_names.each_with_index.map do |name, i|
    column = schema_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 )
    import_size = values_sql.size
    batch_size = options[:batch_size] || import_size
    run_proc = options[:batch_size].to_i.positive? && options[:batch_progress].respond_to?( :call )
    progress_proc = options[:batch_progress]
    current_batch = 0
    batches = (import_size / batch_size.to_f).ceil
    values_sql.each_slice(batch_size) do |batch_values|
      batch_started_at = Time.now.to_i
      # perform the inserts
      result = connection.insert_many( [insert_sql, post_sql_statements].flatten,
        batch_values,
        options,
        "#{model_name} Create Many" )
      number_inserted += result.num_inserts
      ids += result.ids
      results += result.results
      current_batch += 1
      progress_proc.call(import_size, batches, current_batch, Time.now.to_i - batch_started_at) if run_proc
    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)
  changed_columns = model.changed
  association_reflections = model.class.reflect_on_all_associations(:belongs_to)
  association_reflections.each do |association_reflection|
    next if association_reflection.options[:polymorphic]
    column_names = Array(association_reflection.foreign_key).map(&:to_s)
    column_names.each_with_index do |column_name, column_index|
      next if changed_columns.include?(column_name)
      association = model.association(association_reflection.name)
      association = association.target
      next if association.blank? || model.public_send(column_name).present?
      association_primary_key = Array(association_reflection.association_primary_key.tr("[]:", "").split(", "))[column_index]
      model.public_send("#{column_name}=", association.send(association_primary_key))
    end
  end
end

def schema_columns_hash
  if respond_to?(:ignored_columns) && ignored_columns.any?
    connection.schema_cache.columns_hash(table_name)
  else
    columns_hash
  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) if model.send(attr).nil?
      end
    end
  end
  deserialize_value = lambda do |column, value|
    column = schema_columns_hash[column]
    return value unless column
    if respond_to?(:type_caster)
      type = type_for_attribute(column.name)
      type.deserialize(value)
    elsif column.respond_to?(:type_cast_from_database)
      column.type_cast_from_database(value)
    else
      value
    end
  end
  set_value = lambda do |model, column, value|
    val = deserialize_value.call(column, value)
    if model.attribute_names.include?(column)
      model.send("#{column}=", val)
    else
      attributes = attributes_builder.build_from_database(model.attributes.merge(column => val))
      model.instance_variable_set(:@attributes, attributes)
    end
  end
  columns = Array(options[:returning_columns])
  results = Array(import_result.results)
  if models.size == results.size
    single_column = columns.first if columns.size == 1
    results.each_with_index do |result, index|
      model = models[index]
      if single_column
        set_value.call(model, single_column, result)
      else
        columns.each_with_index do |column, col_index|
          set_value.call(model, column, result[col_index])
        end
      end
    end
  end
  models.each do |model|
    if model.respond_to?(:changes_applied) # Rails 4.1.8 and higher
      model.changes_internally_applied if model.respond_to?(:changes_internally_applied) # legacy behavior for Rails 5.1
      model.changes_applied
    elsif 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: #{required_keys}
ing keys: #{missing_keys}
    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: #{required_keys}
a keys: #{extra_keys}
ing keys: #{missing_keys}
    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? && Array(primary_key).first == column.name && !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.respond_to?(:subtype) && 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 && column.type.to_sym == :binary
          connection_memo.quote(val, column)
        end
      else
        raise ArgumentError, "Number of values (#{arr.length}) exceeds number of columns (#{columns.length})"
      end
    end
    "(#{my_values.join(',')})"
  end
end