def ==(other)
  case other
  when Associations::CollectionProxy, AssociationRelation
    self == other.records
  when Relation
    other.to_sql == to_sql
  when Array
    records == other
  end
end

def _create(attributes, &block)
  model.create(attributes, &block)
end

def _create!(attributes, &block)
  model.create!(attributes, &block)
end

def _exec_scope(...) # :nodoc:
  @delegate_to_model = true
  registry = model.scope_registry
  _scoping(nil, registry) { instance_exec(...) || self }
ensure
  @delegate_to_model = false
end

def _increment_attribute(attribute, value = 1)
  bind = predicate_builder.build_bind_attribute(attribute.name, value.abs)
  expr = table.coalesce(Arel::Nodes::UnqualifiedColumn.new(attribute), 0)
  expr = value < 0 ? expr - bind : expr + bind
  expr.expr
end

def _new(attributes, &block)
  model.new(attributes, &block)
end

def _scoping(scope, registry, all_queries = false)
  previous = registry.current_scope(model, true)
  registry.set_current_scope(model, scope)
  if all_queries
    previous_global = registry.global_current_scope(model, true)
    registry.set_global_current_scope(model, scope)
  end
  yield
ensure
  registry.set_current_scope(model, previous)
  if all_queries
    registry.set_global_current_scope(model, previous_global)
  end
end

def _substitute_values(values)
  values.map do |name, value|
    attr = table[name]
    if Arel.arel_node?(value)
      if value.is_a?(Arel::Nodes::SqlLiteral)
        value = Arel::Nodes::Grouping.new(value)
      end
    else
      type = model.type_for_attribute(attr.name)
      value = predicate_builder.build_bind_attribute(attr.name, type.cast(value))
    end
    [attr, value]
  end
end

def alias_tracker(joins = [], aliases = nil) # :nodoc:
  ActiveRecord::Associations::AliasTracker.create(model.connection_pool, table.name, joins, aliases)
end

def already_in_scope?(registry)
  @delegate_to_model && registry.current_scope(model, true)
end

def any?(*args)
  return false if @none
  return super if args.present? || block_given?
  !empty?
end

def bind_attribute(name, value) # :nodoc:
  if reflection = model._reflect_on_association(name)
    name = reflection.foreign_key
    value = value.read_attribute(reflection.association_primary_key) unless value.nil?
  end
  attr = table[name]
  bind = predicate_builder.build_bind_attribute(attr.name, value)
  yield attr, bind
end

def blank?
  records.blank?
end

def cache_key(timestamp_column = "updated_at")
  @cache_keys ||= {}
  @cache_keys[timestamp_column] ||= model.collection_cache_key(self, timestamp_column)
end

def cache_key_with_version
  if version = cache_version
    "#{cache_key}-#{version}"
  else
    cache_key
  end
end

def cache_version(timestamp_column = :updated_at)
  if model.collection_cache_versioning
    @cache_versions ||= {}
    @cache_versions[timestamp_column] ||= compute_cache_version(timestamp_column)
  end
end

def compute_cache_key(timestamp_column = :updated_at) # :nodoc:
  query_signature = ActiveSupport::Digest.hexdigest(to_sql)
  key = "#{model.model_name.cache_key}/query-#{query_signature}"
  if model.collection_cache_versioning
    key
  else
    "#{key}-#{compute_cache_version(timestamp_column)}"
  end
end

def compute_cache_version(timestamp_column) # :nodoc:
  timestamp_column = timestamp_column.to_s
  if loaded?
    size = records.size
    if size > 0
      timestamp = records.map { |record| record.read_attribute(timestamp_column) }.max
    end
  else
    collection = eager_loading? ? apply_join_dependency : self
    with_connection do |c|
      column = c.visitor.compile(table[timestamp_column])
      select_values = "COUNT(*) AS #{model.adapter_class.quote_column_name("size")}, MAX(%s) AS timestamp"
      if collection.has_limit_or_offset?
        query = collection.select("#{column} AS collection_cache_key_timestamp")
        query._select!(table[Arel.star]) if distinct_value && collection.select_values.empty?
        subquery_alias = "subquery_for_cache_key"
        subquery_column = "#{subquery_alias}.collection_cache_key_timestamp"
        arel = query.build_subquery(subquery_alias, select_values % subquery_column)
      else
        query = collection.unscope(:order)
        query.select_values = [select_values % column]
        arel = query.arel
      end
      size, timestamp = c.select_rows(arel, nil).first
      if size
        column_type = model.type_for_attribute(timestamp_column)
        timestamp = column_type.deserialize(timestamp)
      else
        size = 0
      end
    end
  end
  if timestamp
    "#{size}-#{timestamp.utc.to_fs(model.cache_timestamp_format)}"
  else
    "#{size}"
  end
end

def create(attributes = nil, &block)
  if attributes.is_a?(Array)
    attributes.collect { |attr| create(attr, &block) }
  else
    block = current_scope_restoring_block(&block)
    scoping { _create(attributes, &block) }
  end
end

def create!(attributes = nil, &block)
  if attributes.is_a?(Array)
    attributes.collect { |attr| create!(attr, &block) }
  else
    block = current_scope_restoring_block(&block)
    scoping { _create!(attributes, &block) }
  end
end

def create_or_find_by(attributes, &block)
  with_connection do |connection|
    transaction(requires_new: true) { create(attributes, &block) }
  rescue ActiveRecord::RecordNotUnique
    if connection.transaction_open?
      where(attributes).lock.find_by!(attributes)
    else
      find_by!(attributes)
    end
  end
end

def create_or_find_by!(attributes, &block)
  with_connection do |connection|
    transaction(requires_new: true) { create!(attributes, &block) }
  rescue ActiveRecord::RecordNotUnique
    if connection.transaction_open?
      where(attributes).lock.find_by!(attributes)
    else
      find_by!(attributes)
    end
  end
end

def current_scope_restoring_block(&block)
  current_scope = model.current_scope(true)
  -> record do
    model.current_scope = current_scope
    yield record if block_given?
  end
end

def delete(id_or_array)
  return 0 if id_or_array.nil? || (id_or_array.is_a?(Array) && id_or_array.empty?)
  where(model.primary_key => id_or_array).delete_all
end

def delete_all
  return 0 if @none
  invalid_methods = INVALID_METHODS_FOR_DELETE_ALL.select do |method|
    value = @values[method]
    method == :distinct ? value : value&.any?
  end
  if invalid_methods.any?
    raise ActiveRecordError.new("delete_all doesn't support #{invalid_methods.join(', ')}")
  end
  model.with_connection do |c|
    arel = eager_loading? ? apply_join_dependency.arel : build_arel(c)
    arel.source.left = table
    group_values_arel_columns = arel_columns(group_values.uniq)
    having_clause_ast = having_clause.ast unless having_clause.empty?
    key = if model.composite_primary_key?
      primary_key.map { |pk| table[pk] }
    else
      table[primary_key]
    end
    stmt = arel.compile_delete(key, having_clause_ast, group_values_arel_columns)
    c.delete(stmt, "#{model} Delete All").tap { reset }
  end
end

def delete_by(*args)
  where(*args).delete_all
end

def destroy(id)
  multiple_ids = if model.composite_primary_key?
    id.first.is_a?(Array)
  else
    id.is_a?(Array)
  end
  if multiple_ids
    find(id).each(&:destroy)
  else
    find(id).destroy
  end
end

def destroy_all
  records.each(&:destroy).tap { reset }
end

def destroy_by(*args)
  where(*args).destroy_all
end

def eager_loading?
  @should_eager_load ||=
    eager_load_values.any? ||
    includes_values.any? && (joined_includes_values.any? || references_eager_loaded_tables?)
end

def empty?
  return true if @none
  if loaded?
    records.empty?
  else
    !exists?
  end
end

def empty_scope? # :nodoc:
  @values == model.unscoped.values
end

def encode_with(coder)
  coder.represent_seq(nil, records)
end

def exec_main_query(async: false)
  if @none
    if async
      return FutureResult.wrap([])
    else
      return []
    end
  end
  skip_query_cache_if_necessary do
    if where_clause.contradiction?
      [].freeze
    elsif eager_loading?
      model.with_connection do |c|
        apply_join_dependency do |relation, join_dependency|
          if relation.null_relation?
            [].freeze
          else
            relation = join_dependency.apply_column_aliases(relation)
            @_join_dependency = join_dependency
            c.select_all(relation.arel, "SQL", async: async)
          end
        end
      end
    else
      model.with_connection do |c|
        model._query_by_sql(c, arel, async: async)
      end
    end
  end
end

def exec_queries(&block)
  skip_query_cache_if_necessary do
    rows = if scheduled?
      future = @future_result
      @future_result = nil
      future.result
    else
      exec_main_query
    end
    records = instantiate_records(rows, &block)
    preload_associations(records) unless skip_preloading_value
    records.each(&:readonly!) if readonly_value
    records.each { |record| record.strict_loading!(strict_loading_value) } unless strict_loading_value.nil?
    records
  end
end

def explain(*options)
  ExplainProxy.new(self, options)
end

def find_or_create_by(attributes, &block)
  find_by(attributes) || create_or_find_by(attributes, &block)
end

def find_or_create_by!(attributes, &block)
  find_by(attributes) || create_or_find_by!(attributes, &block)
end

def find_or_initialize_by(attributes, &block)
  find_by(attributes) || new(attributes, &block)
end

def first_or_create(attributes = nil, &block) # :nodoc:
  first || create(attributes, &block)
end

def first_or_create!(attributes = nil, &block) # :nodoc:
  first || create!(attributes, &block)
end

def first_or_initialize(attributes = nil, &block) # :nodoc:
  first || new(attributes, &block)
end

def global_scope?(registry)
  registry.global_current_scope(model, true)
end

def has_limit_or_offset? # :nodoc:
  limit_value || offset_value
end

def initialize(model, table: nil, predicate_builder: nil, values: {})
  if table
    predicate_builder ||= model.predicate_builder.with(TableMetadata.new(model, table))
  else
    table = model.arel_table
    predicate_builder ||= model.predicate_builder
  end
  @model  = model
  @table  = table
  @values = values
  @loaded = false
  @predicate_builder = predicate_builder
  @delegate_to_model = false
  @future_result = nil
  @records = nil
  @async = false
  @none = false
end

def initialize_copy(other)
  @values = @values.dup
  reset
end

def insert(attributes, returning: nil, unique_by: nil, record_timestamps: nil)
  insert_all([ attributes ], returning: returning, unique_by: unique_by, record_timestamps: record_timestamps)
end

def insert!(attributes, returning: nil, record_timestamps: nil)
  insert_all!([ attributes ], returning: returning, record_timestamps: record_timestamps)
end

def insert_all(attributes, returning: nil, unique_by: nil, record_timestamps: nil)
  InsertAll.execute(self, attributes, on_duplicate: :skip, returning: returning, unique_by: unique_by, record_timestamps: record_timestamps)
end

def insert_all!(attributes, returning: nil, record_timestamps: nil)
  InsertAll.execute(self, attributes, on_duplicate: :raise, returning: returning, record_timestamps: record_timestamps)
end

def inspect
  subject = loaded? ? records : annotate("loading for inspect")
  entries = subject.take([limit_value, 11].compact.min).map!(&:inspect)
  entries[10] = "..." if entries.size == 11
  "#<#{self.class.name} [#{entries.join(', ')}]>"
end

def instantiate_records(rows, &block)
  return [].freeze if rows.empty?
  if eager_loading?
    records = @_join_dependency.instantiate(rows, strict_loading_value, &block).freeze
    @_join_dependency = nil
    records
  else
    model._load_from_sql(rows, &block).freeze
  end
end

def joined_includes_values
  includes_values & joins_values
end

def limited_count
  limit_value ? count : limit(2).count
end

def load(&block)
  if !loaded? || scheduled?
    @records = exec_queries(&block)
    @loaded = true
  end
  self
end

def load_async
  with_connection do |c|
    return load if !c.async_enabled?
    unless loaded?
      result = exec_main_query(async: !c.current_transaction.joinable?)
      if result.is_a?(Array)
        @records = result
      else
        @future_result = result
      end
      @loaded = true
    end
  end
  self
end

def load_records(records)
  @records = records.freeze
  @loaded = true
end

def many?
  return false if @none
  return super if block_given?
  return records.many? if loaded?
  limited_count > 1
end

def new(attributes = nil, &block)
  if attributes.is_a?(Array)
    attributes.collect { |attr| new(attr, &block) }
  else
    block = current_scope_restoring_block(&block)
    scoping { _new(attributes, &block) }
  end
end

def none?(*args)
  return true if @none
  return super if args.present? || block_given?
  empty?
end

def one?(*args)
  return false if @none
  return super if args.present? || block_given?
  return records.one? if loaded?
  limited_count == 1
end

def preload_associations(records) # :nodoc:
  preload = preload_values
  preload += includes_values unless eager_loading?
  scope = strict_loading_value ? StrictLoadingScope : nil
  preload.each do |associations|
    ActiveRecord::Associations::Preloader.new(records: records, associations: associations, scope: scope).call
  end
end

def pretty_print(pp)
  subject = loaded? ? records : annotate("loading for pp")
  entries = subject.take([limit_value, 11].compact.min)
  entries[10] = "..." if entries.size == 11
  pp.pp(entries)
end

def readonly?
  readonly_value
end

def records # :nodoc:
  load
  @records
end

def references_eager_loaded_tables?
  joined_tables = build_joins([]).flat_map do |join|
    if join.is_a?(Arel::Nodes::StringJoin)
      tables_in_string(join.left)
    else
      join.left.name
    end
  end
  joined_tables << table.name
  # always convert table names to downcase as in Oracle quoted table names are in uppercase
  joined_tables.map!(&:downcase)
  !(references_values.map(&:to_s) - joined_tables).empty?
end

def reload
  reset
  load
end

def reset
  @future_result&.cancel
  @future_result = nil
  @delegate_to_model = false
  @to_sql = @arel = @loaded = @should_eager_load = nil
  @offsets = @take = nil
  @cache_keys = nil
  @cache_versions = nil
  @records = nil
  self
end

def scheduled?
  !!@future_result
end

def scope_for_create
  hash = where_clause.to_h(model.table_name, equality_only: true)
  create_with_value.each { |k, v| hash[k.to_s] = v } unless create_with_value.empty?
  hash
end

def scoping(all_queries: nil, &block)
  registry = model.scope_registry
  if global_scope?(registry) && all_queries == false
    raise ArgumentError, "Scoping is set to apply to all queries and cannot be unset in a nested block."
  elsif already_in_scope?(registry)
    yield
  else
    _scoping(self, registry, all_queries, &block)
  end
end

def size
  if loaded?
    records.length
  else
    count(:all)
  end
end

def skip_query_cache_if_necessary(&block)
  if skip_query_cache_value
    model.uncached(&block)
  else
    yield
  end
end

def tables_in_string(string)
  return [] if string.blank?
  # always convert table names to downcase as in Oracle quoted table names are in uppercase
  # ignore raw_sql_ that is used by Oracle adapter as alias for limit/offset subqueries
  string.scan(/[a-zA-Z_][.\w]+(?=.?\.)/).map!(&:downcase) - ["raw_sql_"]
end

def then(&block) # :nodoc:
  if @future_result
    @future_result.then do
      yield self
    end
  else
    super
  end
end

def to_ary
  records.dup
end

def to_sql
  @to_sql ||= if eager_loading?
    apply_join_dependency do |relation, join_dependency|
      relation = join_dependency.apply_column_aliases(relation)
      relation.to_sql
    end
  else
    model.with_connection do |conn|
      conn.unprepared_statement { conn.to_sql(arel) }
    end
  end
end

def touch_all(*names, time: nil)
  update_all model.touch_attributes_with_time(*names, time: time)
end

def update(id = :all, attributes) # :nodoc:
  if id == :all
    each { |record| record.update(attributes) }
  else
    model.update(id, attributes)
  end
end

def update!(id = :all, attributes) # :nodoc:
  if id == :all
    each { |record| record.update!(attributes) }
  else
    model.update!(id, attributes)
  end
end

def update_all(updates)
  raise ArgumentError, "Empty list of attributes to change" if updates.blank?
  return 0 if @none
  if updates.is_a?(Hash)
    if model.locking_enabled? &&
        !updates.key?(model.locking_column) &&
        !updates.key?(model.locking_column.to_sym)
      attr = table[model.locking_column]
      updates[attr.name] = _increment_attribute(attr)
    end
    values = _substitute_values(updates)
  else
    values = Arel.sql(model.sanitize_sql_for_assignment(updates, table.name))
  end
  model.with_connection do |c|
    arel = eager_loading? ? apply_join_dependency.arel : build_arel(c)
    arel.source.left = table
    group_values_arel_columns = arel_columns(group_values.uniq)
    having_clause_ast = having_clause.ast unless having_clause.empty?
    key = if model.composite_primary_key?
      primary_key.map { |pk| table[pk] }
    else
      table[primary_key]
    end
    stmt = arel.compile_update(values, key, having_clause_ast, group_values_arel_columns)
    c.update(stmt, "#{model} Update All").tap { reset }
  end
end

def update_counters(counters)
  touch = counters.delete(:touch)
  updates = {}
  counters.each do |counter_name, value|
    attr = table[counter_name]
    updates[attr.name] = _increment_attribute(attr, value)
  end
  if touch
    names = touch if touch != true
    names = Array.wrap(names)
    options = names.extract_options!
    touch_updates = model.touch_attributes_with_time(*names, **options)
    updates.merge!(touch_updates) unless touch_updates.empty?
  end
  update_all updates
end

def upsert(attributes, **kwargs)
  upsert_all([ attributes ], **kwargs)
end

def upsert_all(attributes, on_duplicate: :update, update_only: nil, returning: nil, unique_by: nil, record_timestamps: nil)
  InsertAll.execute(self, attributes, on_duplicate: on_duplicate, update_only: update_only, returning: returning, unique_by: unique_by, record_timestamps: record_timestamps)
end

def values
  @values.dup
end

def values_for_queries # :nodoc:
  @values.except(:extending, :skip_query_cache, :strict_loading)
end

def where_values_hash(relation_table_name = model.table_name) # :nodoc:
  where_clause.to_h(relation_table_name)
end