Class: Rollup::Aggregator

Inherits:
Object
  • Object
show all
Defined in:
lib/rollup/aggregator.rb

Instance Method Summary collapse

Constructor Details

#initialize(klass) ⇒ Aggregator

Returns a new instance of Aggregator.



3
4
5
 
# File 'lib/rollup/aggregator.rb', line 3

def initialize(klass)
  @klass = klass # or relation
end

Instance Method Details

#determine_dimension_name(group) ⇒ Object 



97
98
99
100
101
102
103
104
105
106
107
108
109
110
 
# File 'lib/rollup/aggregator.rb', line 97

def determine_dimension_name(group)
  # split by ., ->>, and -> and remove whitespace
  value = group.to_s.split(/\s*((\.)|(->>)|(->))\s*/).last

  # removing starting and ending quotes
  # for simplicity, they don't need to be the same
  value = value[1..-2] if value.match(/\A["'`].+["'`]\z/)

  unless value.match(/\A\w+\z/)
    raise "Cannot determine dimension name: #{group}. Use the dimension_names option"
  end

  value
end

#maybe_clear(clear, name, interval) ⇒ Object 



157
158
159
160
161
162
163
164
165
166
 
# File 'lib/rollup/aggregator.rb', line 157

def maybe_clear(clear, name, interval)
  if clear
    Rollup.transaction do
      Rollup.where(name: name, interval: interval).delete_all
      yield
    end
  else
    yield
  end
end

#perform_calculation(relation, &block) ⇒ Object

calculation can mutate relation, but that’s fine



113
114
115
116
117
118
119
 
# File 'lib/rollup/aggregator.rb', line 113

def perform_calculation(relation, &block)
  if block_given?
    yield(relation)
  else
    relation.count
  end
end

#perform_group(name, column:, interval:, time_zone:, current:, last:, clear:) ⇒ Object

Raises:

  • (ArgumentError) 


35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/rollup/aggregator.rb', line 35

def perform_group(name, column:, interval:, time_zone:, current:, last:, clear:)
  raise ArgumentError, "Cannot use last and clear together" if last && clear

  time_zone ||= Rollup.time_zone

  gd_options = {
    current: current
  }

  # make sure Groupdate global options aren't applied
  gd_options[:time_zone] = time_zone
  gd_options[:week_start] = Rollup.week_start if interval.to_s == "week"
  gd_options[:day_start] = 0 if Utils.date_interval?(interval)

  if last
    gd_options[:last] = last
  elsif !clear
    # if no rollups, compute all intervals
    # if rollups, recompute last interval
    max_time = Rollup.where(name: name, interval: interval).maximum(Utils.time_sql(interval))
    if max_time
      # for MySQL on Ubuntu 18.04 (and likely other platforms)
      if max_time.is_a?(String)
        utc = ActiveSupport::TimeZone["Etc/UTC"]
        max_time = Utils.date_interval?(interval) ? max_time.to_date : utc.parse(max_time).in_time_zone(time_zone)
      end

      # aligns perfectly if time zone doesn't change
      # if time zone does change, there are other problems besides this
      gd_options[:range] = max_time..
    end
  end

  # intervals are stored as given
  # we don't normalize intervals (i.e. change 60s -> 1m)
  case interval.to_s
  when "hour", "day", "week", "month", "quarter", "year"
    @klass.group_by_period(interval, column, **gd_options)
  when /\A\d+s\z/
    @klass.group_by_second(column, n: interval.to_i, **gd_options)
  when /\A\d+m\z/
    @klass.group_by_minute(column, n: interval.to_i, **gd_options)
  else
    raise ArgumentError, "Invalid interval: #{interval}"
  end
end

#prepare_result(result, name, dimension_names, interval) ⇒ Object 



121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
 
# File 'lib/rollup/aggregator.rb', line 121

def prepare_result(result, name, dimension_names, interval)
  raise "Expected calculation to return Hash, not #{result.class.name}" unless result.is_a?(Hash)

  time_class = Utils.date_interval?(interval) ? Date : Time
  dimensions_supported = Utils.dimensions_supported?
  expected_key_size = dimension_names.size + 1

  result.map do |key, value|
    dimensions = {}
    if dimensions_supported && dimension_names.any?
      unless key.is_a?(Array) && key.size == expected_key_size
        raise "Expected result key to be Array with size #{expected_key_size}"
      end
      time = key[-1]
      # may be able to support dimensions in SQLite by sorting dimension names
      dimension_names.each_with_index do |dn, i|
        dimensions[dn] = key[i]
      end
    else
      time = key
    end

    raise "Expected time to be #{time_class.name}, not #{time.class.name}" unless time.is_a?(time_class)
    raise "Expected value to be Numeric or nil, not #{value.class.name}" unless value.is_a?(Numeric) || value.nil?

    record = {
      name: name,
      interval: interval,
      time: time,
      value: value
    }
    record[:dimensions] = dimensions if dimensions_supported
    record
  end
end

#rollup(name, column: nil, interval: "day", dimension_names: nil, time_zone: nil, current: true, last: nil, clear: false, &block) ⇒ Object 



7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
 
# File 'lib/rollup/aggregator.rb', line 7

def rollup(name, column: nil, interval: "day", dimension_names: nil, time_zone: nil, current: true, last: nil, clear: false, &block)
  raise "Name can't be blank" if name.blank?

  column ||= @klass.rollup_column || :created_at
  validate_column(column)

  relation = perform_group(name, column: column, interval: interval, time_zone: time_zone, current: current, last: last, clear: clear)
  result = perform_calculation(relation, &block)

  dimension_names = set_dimension_names(dimension_names, relation)
  records = prepare_result(result, name, dimension_names, interval)

  maybe_clear(clear, name, interval) do
    save_records(records) if records.any?
  end
end

#save_records(records) ⇒ Object 



168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
 
# File 'lib/rollup/aggregator.rb', line 168

def save_records(records)
  # order must match unique index
  # consider using index name instead
  conflict_target = [:name, :interval, :time]
  conflict_target << :dimensions if Utils.dimensions_supported?

  if ActiveRecord::VERSION::MAJOR >= 6
    options = Utils.mysql? ? {} : {unique_by: conflict_target}
    Rollup.upsert_all(records, **options)
  else
    update = Utils.mysql? ? [:value] : {columns: [:value], conflict_target: conflict_target}
    Rollup.import(records,
      on_duplicate_key_update: update,
      validate: false
    )
  end
end

#set_dimension_names(dimension_names, relation) ⇒ Object 



82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/rollup/aggregator.rb', line 82

def set_dimension_names(dimension_names, relation)
  groups = relation.group_values[0..-2]

  if dimension_names
    Utils.check_dimensions
    if dimension_names.size != groups.size
      raise ArgumentError, "Expected dimension_names to be size #{groups.size}, not #{dimension_names.size}"
    end
    dimension_names
  else
    Utils.check_dimensions if groups.any?
    groups.map { |group| determine_dimension_name(group) }
  end
end

#validate_column(column) ⇒ Object

basic version of Active Record disallow_raw_sql! symbol = column (safe), Arel node = SQL (safe), other = untrusted no need to quote/resolve column here, as Groupdate handles it TODO remove this method when gem depends on Groupdate 6+



28
29
30
31
32
33
 
# File 'lib/rollup/aggregator.rb', line 28

def validate_column(column)
  # matches table.column and column
  unless column.is_a?(Symbol) || column.is_a?(Arel::Nodes::SqlLiteral) || /\A\w+(\.\w+)?\z/i.match(column.to_s)
    raise "Non-attribute argument: #{column}. Use Arel.sql() for known-safe values"
  end
end