183 行
7.1 KiB
Ruby
183 行
7.1 KiB
Ruby
class CountsAxes
|
|
MAX_AXES = 8
|
|
MASKED_TEXT = '********'.freeze
|
|
MASKED_VALUE = '__counts_private__'.freeze
|
|
PUBLIC_FIELDS = ['resource_id', 'version_no', 'event_type', 'user',
|
|
'created_at'].freeze
|
|
class TooManyAxes < ArgumentError; end
|
|
class PrivateCondition < ArgumentError; end
|
|
|
|
GRAINS = ['year', 'quarter', 'month', 'week', 'day', 'hour'].freeze
|
|
|
|
def self.private_field?(field)
|
|
!PUBLIC_FIELDS.include?(field) && field != 'user.id'
|
|
end
|
|
|
|
def self.visible_ids_sql(definition)
|
|
"SELECT counts_visible.id FROM #{ definition.table } counts_visible WHERE " +
|
|
HistoryAccess.sql(definition.model, 'counts_visible')
|
|
end
|
|
|
|
def self.visibility_sql(definition, field)
|
|
return 'TRUE' if HistoryAccess.admin?
|
|
|
|
source = visible_ids_sql(definition)
|
|
visible = "v.id IN (#{ source })"
|
|
return visible unless field == 'changed_field'
|
|
|
|
"(#{ visible }) AND (previous.id IS NULL OR previous.id IN (#{ source }))"
|
|
end
|
|
|
|
def self.masked_sql(expression, definition, field)
|
|
return expression if HistoryAccess.admin? || !private_field?(field)
|
|
|
|
visible = visibility_sql(definition, field)
|
|
quoted = ApplicationRecord.connection.quote(MASKED_VALUE)
|
|
"CASE WHEN #{ visible } THEN #{ expression } ELSE #{ quoted } END"
|
|
end
|
|
|
|
def self.time_expression(expression, grain, timezone)
|
|
raise ArgumentError unless GRAINS.include?(grain)
|
|
|
|
connection = ApplicationRecord.connection
|
|
# Use the installed TZInfo database, including DST, rather than MySQL timezone tables.
|
|
first = Time.utc(1900)
|
|
last = Time.utc(2101)
|
|
offset = timezone.period_for_utc(first).utc_total_offset
|
|
clauses = timezone.transitions_up_to(last, first).filter_map do |transition|
|
|
next if transition.offset.utc_total_offset == offset
|
|
|
|
previous = offset
|
|
offset = transition.offset.utc_total_offset
|
|
"WHEN #{ expression } < #{ connection.quote(transition.at.to_time) } THEN #{ previous }"
|
|
end
|
|
seconds = clauses.empty? ? offset.to_s : "CASE #{ clauses.join(' ') } ELSE #{ offset } END"
|
|
local = "DATE_ADD(#{ expression }, INTERVAL (#{ seconds }) SECOND)"
|
|
case grain
|
|
when 'year' then "DATE_FORMAT(#{ local }, '%Y')"
|
|
when 'quarter' then "CONCAT(YEAR(#{ local }), '-Q', QUARTER(#{ local }))"
|
|
when 'month' then "DATE_FORMAT(#{ local }, '%Y-%m')"
|
|
when 'week' then "DATE_FORMAT(#{ local }, '%x-W%v')"
|
|
when 'day' then "DATE_FORMAT(#{ local }, '%Y-%m-%d')"
|
|
else "DATE_FORMAT(#{ local }, '%Y-%m-%d %H:00')"
|
|
end
|
|
end
|
|
|
|
attr_reader :rows, :columns, :all
|
|
|
|
def initialize(population)
|
|
@population = population
|
|
@definition = population.definition
|
|
@rows = parse(population.params.fetch('rows', ''))
|
|
@columns = parse(population.params.fetch('cols', ''))
|
|
@all = rows + columns
|
|
raise TooManyAxes if all.length > MAX_AXES
|
|
raise ArgumentError if all.map { _1['field'] }.uniq.length != all.length
|
|
end
|
|
|
|
def expressions
|
|
@expressions ||= all.map do |axis|
|
|
field = axis.fetch('field')
|
|
expression = case field
|
|
when 'tags'
|
|
"CASE WHEN tag_entry.tag_id IS NOT NULL " +
|
|
"THEN CONCAT('internal:', tag_entry.tag_id) " +
|
|
"ELSE CONCAT('external:', tag_entry.external_tag_id) END"
|
|
when 'changed_field' then 'changed.field_key'
|
|
else
|
|
expression = @definition.expression(field)
|
|
@definition.fields[field] == :datetime ? self.class.time_expression(
|
|
expression, axis.fetch('grain', 'month'), @population.timezone) : expression
|
|
end
|
|
if !HistoryAccess.admin? && self.class.private_field?(field)
|
|
expression = self.class.masked_sql(
|
|
"CAST((#{ expression }) AS CHAR)", @definition, field)
|
|
end
|
|
expression
|
|
end
|
|
end
|
|
|
|
def grouped_sql(scope, indexes = all.each_index.to_a)
|
|
keys = indexes.map { "a#{ _1 }" }
|
|
selected = indexes.map { "#{ expressions[_1] } AS a#{ _1 }" }
|
|
identities = ['v.id']
|
|
if rows.any? { _1['field'] == 'tags' }
|
|
identities << self.class.masked_sql(
|
|
"COALESCE(CONCAT('internal:', tag_entry.tag_id), " +
|
|
"CONCAT('external:', tag_entry.external_tag_id))", @definition, 'tags')
|
|
if rows.any? { _1['field'] == 'changed_field' }
|
|
identities << self.class.masked_sql('changed.field_key', @definition, 'changed_field')
|
|
end
|
|
end
|
|
members = identities.each_index.map { "counts_member_#{ _1 }" }
|
|
selected += identities.each_with_index.map { |expression, index|
|
|
"#{ expression } AS #{ members[index] }"
|
|
}
|
|
# Evaluate correlated visibility subqueries before grouping. LIMIT prevents
|
|
# derived-table merging, so ONLY_FULL_GROUP_BY sees ordinary projected columns.
|
|
projection = scope.except(:select, :order).select(Arel.sql(selected.join(', ')))
|
|
.limit(18_446_744_073_709_551_615).to_sql
|
|
fields = keys + ["COUNT(DISTINCT #{ members.join(', ') }) AS count"]
|
|
grouping = keys.empty? ? '' : " GROUP BY #{ keys.join(', ') }"
|
|
"SELECT #{ fields.join(', ') } FROM (#{ projection }) counts_values#{ grouping }"
|
|
end
|
|
|
|
def joins(fields = all.map { _1['field'] })
|
|
result = ''
|
|
if fields.include?('tags')
|
|
document = 'CAST(v.tags_json AS JSON)'
|
|
unless HistoryAccess.admin?
|
|
visible = self.class.visible_ids_sql(@definition)
|
|
result += " LEFT JOIN (#{ visible }) counts_tags_visible " +
|
|
'ON counts_tags_visible.id = v.id '
|
|
document = "CASE WHEN counts_tags_visible.id IS NOT NULL " +
|
|
"THEN #{ document } ELSE JSON_ARRAY(" +
|
|
"JSON_OBJECT('tag_id', 0)) END"
|
|
end
|
|
result += <<~SQL.squish
|
|
JOIN JSON_TABLE(#{ document }, '$[*]' COLUMNS (
|
|
tag_id BIGINT PATH '$.tag_id' NULL ON EMPTY,
|
|
external_tag_id BIGINT PATH '$.external_tag_id' NULL ON EMPTY)) tag_entry
|
|
|
|
SQL
|
|
end
|
|
if fields.include?('changed_field')
|
|
fields = @definition.changes.map do |field|
|
|
"SELECT #{ ApplicationRecord.connection.quote(field) } AS field_key"
|
|
end.join(' UNION ALL ')
|
|
condition = @definition.changes.map do |field|
|
|
"(changed.field_key = '#{ field }' AND #{ @population.changed_sql(field) })"
|
|
end.join(' OR ')
|
|
unless HistoryAccess.admin?
|
|
visible = self.class.visibility_sql(@definition, 'changed_field')
|
|
marker = ApplicationRecord.connection.quote(MASKED_VALUE)
|
|
fields += " UNION ALL SELECT #{ marker } AS field_key"
|
|
condition = "((#{ visible }) AND (#{ condition })) OR " +
|
|
"(NOT (#{ visible }) AND changed.field_key = #{ marker })"
|
|
end
|
|
result += " JOIN (#{ fields }) changed ON (#{ condition })"
|
|
end
|
|
result
|
|
end
|
|
|
|
private
|
|
|
|
def parse(value)
|
|
axes = QueryState.axes(value)
|
|
raise ArgumentError unless axes.is_a?(Array)
|
|
|
|
axes.each do |axis|
|
|
raise ArgumentError unless axis.is_a?(Hash) && (axis.keys - ['field', 'grain']).empty?
|
|
raise ArgumentError unless @definition.axes(@population.history).include?(axis['field'])
|
|
if @definition.fields[axis['field']] == :datetime
|
|
raise ArgumentError unless GRAINS.include?(axis.fetch('grain', 'month'))
|
|
|
|
axis['grain'] ||= 'month'
|
|
elsif axis.key?('grain')
|
|
raise ArgumentError
|
|
end
|
|
end
|
|
axes
|
|
end
|
|
end
|