Class: Deimos::ActiveRecordConsume::MassUpdater

Inherits:
Object
  • Object
show all
Defined in:
lib/deimos/active_record_consume/mass_updater.rb

Overview

Responsible for updating the database itself.

Instance Method Summary collapse

Constructor Details

#initialize(klass, key_col_proc: nil, col_proc: nil, replace_associations: true, bulk_import_id_generator: nil) ⇒ MassUpdater

Returns a new instance of MassUpdater.

Parameters:

  • klass (Class < ActiveRecord::Base]) —

    lass [Class < ActiveRecord::Base]

  • key_col_proc (Proc<Class < ActiveRecord::Base>]) (defaults to: nil) —

    ey_col_proc [Proc<Class < ActiveRecord::Base>]

  • col_proc (Proc<Class < ActiveRecord::Base>]) (defaults to: nil) —

    ol_proc [Proc<Class < ActiveRecord::Base>]

  • replace_associations (Boolean) (defaults to: true) 


22
23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/deimos/active_record_consume/mass_updater.rb', line 22

def initialize(klass, key_col_proc: nil, col_proc: nil,
               replace_associations: true, bulk_import_id_generator: nil)
  @klass = klass
  @replace_associations = replace_associations
  @bulk_import_id_generator = bulk_import_id_generator

  @key_cols = {}
  @key_col_proc = key_col_proc

  @columns = {}
  @col_proc = col_proc
end

Instance Method Details

#columns(klass) ⇒ Object

Parameters:

  • klass (Class < ActiveRecord::Base]) —

    lass [Class < ActiveRecord::Base]



36
37
38
 
# File 'lib/deimos/active_record_consume/mass_updater.rb', line 36

def columns(klass)
  @columns[klass] ||= @col_proc&.call(klass) || self.default_cols(klass)
end

#default_cols(klass) ⇒ Object

Parameters:

  • klass (Class < ActiveRecord::Base]) —

    lass [Class < ActiveRecord::Base]



14
15
16
 
# File 'lib/deimos/active_record_consume/mass_updater.rb', line 14

def default_cols(klass)
  klass.column_names - %w(created_at updated_at)
end

#default_keys(klass) ⇒ Object

Parameters:

  • klass (Class < ActiveRecord::Base]) —

    lass [Class < ActiveRecord::Base]



9
10
11
 
# File 'lib/deimos/active_record_consume/mass_updater.rb', line 9

def default_keys(klass)
  [klass.primary_key]
end

#import_associations(record_list) ⇒ Object

Imports associated objects and import them to database table The base table is expected to contain bulk_import_id column for indexing associated objects with id

Parameters:



71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/deimos/active_record_consume/mass_updater.rb', line 71

def import_associations(record_list)
  record_list.fill_primary_keys!

  import_id = @replace_associations ? @bulk_import_id_generator&.call : nil
  record_list.associations.each do |assoc|
    sub_records = record_list.map { |r| r.sub_records(assoc.name, import_id) }.flatten
    next unless sub_records.any?

    sub_record_list = BatchRecordList.new(sub_records)

    save_records_to_database(sub_record_list)
    record_list.delete_old_records(assoc, import_id) if import_id
  end
end

#key_cols(klass) ⇒ Object

Parameters:

  • klass (Class < ActiveRecord::Base]) —

    lass [Class < ActiveRecord::Base]



41
42
43
 
# File 'lib/deimos/active_record_consume/mass_updater.rb', line 41

def key_cols(klass)
  @key_cols[klass] ||= @key_col_proc&.call(klass) || self.default_keys(klass)
end

#mass_update(record_list) ⇒ Array<ActiveRecord::Base>

Parameters:

Returns:

  • (Array<ActiveRecord::Base>) 


88
89
90
91
92
93
94
95
96
97
 
# File 'lib/deimos/active_record_consume/mass_updater.rb', line 88

def mass_update(record_list)
  # The entire batch should be treated as one transaction so that if
  # any message fails, the whole thing is rolled back or retried
  # if there is deadlock
  Deimos::Utils::DeadlockRetry.wrap(Deimos.config.tracer.active_span.get_tag('topic')) do
    save_records_to_database(record_list)
    import_associations(record_list) if record_list.associations.any?
  end
  record_list.records
end

#save_records_to_database(record_list) ⇒ Object

Parameters:



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/deimos/active_record_consume/mass_updater.rb', line 46

def save_records_to_database(record_list)
  columns = self.columns(record_list.klass)
  key_cols = self.key_cols(record_list.klass)
  record_list.records.each(&:validate!)

  options = if @key_cols.empty?
              {} # Can't upsert with no key, just do regular insert
            elsif ActiveRecord::Base.connection.adapter_name.downcase =~ /mysql/
              {
                on_duplicate_key_update: columns
              }
            else
              {
                on_duplicate_key_update: {
                  conflict_target: key_cols,
                  columns: columns
                }
              }
            end
  record_list.klass.import!(columns, record_list.records, options)
end