317 行
14 KiB
Ruby
317 行
14 KiB
Ruby
# Common entrypoint for six version resources; legacy endpoints keep their contracts.
|
|
class CountsTableQuery
|
|
MAX_COLUMNS = 100
|
|
class TooManyColumns < ArgumentError; end
|
|
|
|
KEYS = ['resource', 'definition', 'mode', 'at', 'execution_at', 'from', 'through',
|
|
'timezone', 'rows', 'cols', 'conditions', 'drill', 'tags', 'tag_refs', 'match',
|
|
'tag', 'kind', 'resource_id', 'user_id', 'page', 'limit', 'sort', 'locale', 'measures', 'column_order'].freeze
|
|
attr_reader :population, :axes, :page, :limit, :mode
|
|
|
|
def initialize(params)
|
|
QueryState.validate!(params, KEYS + ['measures'])
|
|
raise ArgumentError unless params.values.all? { _1.is_a?(String) }
|
|
|
|
params = CountsDefaults.for(params).slice(*KEYS).merge(params)
|
|
@params = params
|
|
@population = CountsPopulation.new(params)
|
|
@definition = population.definition
|
|
@connection = @definition.model.connection
|
|
@axes = CountsAxes.new(population)
|
|
@page = Integer(params.fetch('page', '1'), 10)
|
|
@limit = Integer(params.fetch('limit', '20'), 10)
|
|
@mode = params.fetch('mode', 'summary')
|
|
raise ArgumentError unless page.between?(1, 1_000_000) && limit.between?(1, 100)
|
|
raise ArgumentError unless ['summary', 'details'].include?(mode)
|
|
|
|
@sort = QueryState.list(params.fetch('sort', '')).map { _1.split(':', -1) }
|
|
allowed = mode == 'summary' ? axes.all.each_index.map { "a#{ _1 }" } + ['count'] :
|
|
@definition.fields.keys + ['is_latest'] +
|
|
(@definition.resource == 'posts' ? ['tags', 'parents'] : []) +
|
|
(@definition.resource == 'tags' ? ['parents', 'deprecated'] : [])
|
|
raise ArgumentError if @sort.map(&:first).uniq.length != @sort.length
|
|
@sort.each do |key, direction, *extra|
|
|
measure = mode == 'summary' && key&.match?(/\Am[0-9]{1,2}\z/)
|
|
measure_values(key) if measure
|
|
if extra.any? || (!allowed.include?(key) && !measure) || !['asc', 'desc'].include?(direction)
|
|
raise ArgumentError
|
|
end
|
|
end
|
|
end
|
|
|
|
def call
|
|
ApplicationRecord.transaction do
|
|
metadata.merge(mode == 'summary' ? summary : details)
|
|
end
|
|
end
|
|
|
|
def metadata
|
|
{ population_count_scope: 'all_recorded_versions', population_count: HistoryAccess.population_count(@definition.model),
|
|
page:, limit:, execution_at: population.at.iso8601(6), at: @params['at'].presence,
|
|
definition: population.history ? 'history' : 'latest',
|
|
timezone: population.timezone.identifier, rows: axes.rows, cols: axes.columns,
|
|
fields: @definition.metadata,
|
|
count: population.scope.except(:select).optimizer_hints('MAX_EXECUTION_TIME(10000)').count }
|
|
end
|
|
|
|
def summary
|
|
expressions = axes.expressions
|
|
selected = expressions.each_with_index.map do |expression, index|
|
|
"#{ expression } AS a#{ index }"
|
|
end
|
|
source = population.scope
|
|
source = source.joins(axes.joins) unless axes.joins.empty?
|
|
sql = if expressions.empty?
|
|
source.except(:select).select('COUNT(DISTINCT v.id) AS count').to_sql
|
|
else
|
|
source.except(:select).group(*expressions.map { Arel.sql(_1) })
|
|
.select(Arel.sql((selected + ['COUNT(DISTINCT v.id) AS count']).join(', ')))
|
|
.to_sql
|
|
end
|
|
prefix = "WITH cells AS (#{ sql })"
|
|
column_values = limited_columns(prefix)
|
|
row_keys = axes.rows.each_index.map { "a#{ _1 }" }
|
|
# Page whole row tuples so a pivot row is never split across display pages.
|
|
row_selection = row_keys.empty? ? '1 AS singleton' : row_keys.join(', ')
|
|
row_group = row_keys.empty? ? 'GROUP BY singleton' : "GROUP BY #{ row_keys.join(', ') }"
|
|
measure_joins = []
|
|
extras = []
|
|
row_sort = @sort.each_with_index.filter_map do |(key, direction), index|
|
|
if key == 'count' || key.match?(/\Am[0-9]+\z/)
|
|
values = key == 'count' ? [] : measure_values(key)
|
|
relation = measure_relation(values)
|
|
matches = row_keys.map { "grouped.#{ _1 } <=> sort#{ index }.#{ _1 }" }
|
|
condition = matches.empty? ? 'TRUE' : matches.join(' AND ')
|
|
measure_joins << "LEFT JOIN (#{ relation }) sort#{ index } ON #{ condition }"
|
|
extras << "sort#{ index }.count AS sort#{ index }"
|
|
"sort#{ index } IS NULL ASC, sort#{ index } #{ direction.upcase }"
|
|
elsif row_keys.include?(key)
|
|
"#{ key } IS NULL ASC, #{ key } #{ direction.upcase }"
|
|
end
|
|
end
|
|
order_rows = row_sort + row_keys.map { "#{ _1 } IS NULL, #{ _1 }" }
|
|
order_rows = ['singleton ASC'] if order_rows.empty?
|
|
grouped = "SELECT #{ row_selection } FROM cells #{ row_group }"
|
|
selection = (['grouped.*'] + extras).join(', ')
|
|
group_sql = "SELECT #{ selection } FROM (#{ grouped }) grouped #{ measure_joins.join(' ') }"
|
|
total = @connection.select_value(<<~SQL).to_i
|
|
#{ prefix }
|
|
SELECT /*+ MAX_EXECUTION_TIME(10000) */ COUNT(*) FROM (#{ group_sql }) row_groups
|
|
SQL
|
|
join = row_keys.map { "cells.#{ _1 } <=> page_rows.#{ _1 }" }.join(' AND ')
|
|
join = 'TRUE' if row_keys.empty?
|
|
order_cols = @sort.filter_map do |key, direction|
|
|
next unless key.match?(/\Aa[0-9]+\z/) && !row_keys.include?(key)
|
|
|
|
"cells.#{ key } #{ direction.upcase }"
|
|
end
|
|
order = ['page_rows.position'] + order_cols +
|
|
axes.all.each_index.map { "cells.a#{ _1 }" }
|
|
records = @connection.select_all(<<~SQL)
|
|
#{ prefix }, row_groups AS (#{ group_sql }),
|
|
numbered AS (SELECT row_groups.*,
|
|
ROW_NUMBER() OVER (ORDER BY #{ order_rows.join(', ') }) AS position FROM row_groups),
|
|
page_rows AS (SELECT * FROM numbered
|
|
WHERE position > #{ (page - 1) * limit } AND position <= #{ page * limit })
|
|
SELECT /*+ MAX_EXECUTION_TIME(10000) */ cells.* FROM cells JOIN page_rows ON #{ join }
|
|
ORDER BY #{ order.join(', ') }
|
|
SQL
|
|
next_row = @connection.select_one(<<~SQL)
|
|
#{ prefix }, row_groups AS (#{ group_sql }),
|
|
numbered AS (SELECT row_groups.*,
|
|
ROW_NUMBER() OVER (ORDER BY #{ order_rows.join(', ') }) AS position FROM row_groups)
|
|
SELECT /*+ MAX_EXECUTION_TIME(10000) */ * FROM numbered
|
|
WHERE position = #{ page * limit + 1 }
|
|
SQL
|
|
row_values = records.map { |record| row_keys.map { record[_1] } }.uniq
|
|
row_depths = CountsPivotLayout.depths(row_keys, @sort)
|
|
column_keys = axes.columns.each_index.map { "a#{ axes.rows.length + _1 }" }
|
|
column_sort = @sort
|
|
if @params.key?('column_order') && column_keys.any? &&
|
|
!(@sort.first&.first == 'count' || @sort.first&.first&.match?(/\Am[0-9]+\z/))
|
|
column_sort = [[column_keys.first, 'asc']] + @sort
|
|
end
|
|
column_depths = CountsPivotLayout.depths(column_keys, column_sort, measures: false)
|
|
row_headers = CountsPivotLayout.headers(row_values, row_depths,
|
|
next_values: next_row && row_keys.map { next_row[_1] }, grand: axes.rows.any?)
|
|
if axes.rows.empty? && axes.columns.any?
|
|
row_headers = [{ values: [], kind: 'grand_total' }]
|
|
end
|
|
column_headers = CountsPivotLayout.headers(column_values, column_depths,
|
|
grand: axes.columns.any?)
|
|
if axes.columns.empty? && axes.rows.any?
|
|
column_headers = [{ values: [], kind: 'grand_total' }]
|
|
end
|
|
totals = CountsPivotTotals.new(population, axes, row_headers, column_headers).call
|
|
{ group_count: total, column_values:, row_headers:, column_headers:, pivot_totals: totals,
|
|
cells: records.map do |row|
|
|
{ values: axes.all.each_index.map { row["a#{ _1 }"] }, count: row['count'].to_i }
|
|
end, axis_labels: axis_labels(records, column_values) }
|
|
end
|
|
|
|
def details
|
|
query = population.scope.select(Arel.sql(
|
|
"(#{ population.latest_sql }) AS counts_latest"))
|
|
order = sort_sql { detail_expression(_1) }
|
|
query = query.order(Arel.sql((order + ['v.id DESC']).join(', ')))
|
|
.limit(limit).offset((page - 1) * limit)
|
|
rows = query.to_a
|
|
if ['posts', 'tags'].include?(@definition.resource)
|
|
point = population.history ? population.at :
|
|
population.timestamp(@params['at']) || population.at
|
|
values = CountsVersionRepr.many(rows, resource: @definition.resource,
|
|
language_code: locale.language_code,
|
|
at: point)
|
|
values.each_with_index { |value, index| value[:event_type] = rows[index].event_type }
|
|
else
|
|
users = User.where(id: rows.map(&:created_by_user_id)).pluck(:id, :name).to_h
|
|
values = rows.map do |row|
|
|
attributes = @definition.fields.keys.to_h do |key|
|
|
[key, row.public_send(@definition.column(key))]
|
|
end
|
|
attributes.merge('id' => row.id, 'resource_id' => row[@definition.keys.first],
|
|
'is_latest' => row.attributes['counts_latest'].to_i == 1,
|
|
'created_by_user' => row.created_by_user_id &&
|
|
{ id: row.created_by_user_id, name: users[row.created_by_user_id] })
|
|
end
|
|
end
|
|
{ versions: values }
|
|
end
|
|
|
|
def each_page
|
|
total = nil
|
|
loop do
|
|
result = call
|
|
total ||= mode == 'summary' ? result[:group_count] : result[:count]
|
|
yield result
|
|
break if page * limit >= total
|
|
|
|
@page += 1
|
|
end
|
|
end
|
|
|
|
private
|
|
|
|
def locale = @locale ||= Locale.find_by!(code: @params.fetch('locale', 'ja'))
|
|
|
|
def limited_columns(prefix)
|
|
keys = axes.columns.each_index.map { "a#{ axes.rows.length + _1 }" }
|
|
return [[]] if keys.empty?
|
|
|
|
configured = @sort.select { |key, _direction| keys.include?(key) }
|
|
directions = (configured + keys.map { [_1, 'asc'] }).uniq { _1.first }
|
|
order = directions.map { |key, direction|
|
|
"#{ key } IS NULL ASC, #{ key } #{ direction.upcase }"
|
|
}.join(', ')
|
|
custom = QueryState.records(@params, 'column_order')
|
|
if custom.any?
|
|
raise ArgumentError unless custom.map { _1['value'] }.uniq.length == custom.length
|
|
|
|
cases = custom.each_with_index.map do |record, index|
|
|
"WHEN #{ keys.first } <=> #{ @connection.quote(record['value']) } THEN #{ index }"
|
|
end.join(' ')
|
|
order = "CASE #{ cases } ELSE #{ custom.length } END, #{ order }"
|
|
end
|
|
values = @connection.select_rows(<<~SQL)
|
|
#{ prefix }
|
|
SELECT /*+ MAX_EXECUTION_TIME(10000) */ DISTINCT #{ keys.join(', ') } FROM cells
|
|
ORDER BY #{ order } LIMIT #{ MAX_COLUMNS + 1 }
|
|
SQL
|
|
raise TooManyColumns if values.length > MAX_COLUMNS
|
|
|
|
values
|
|
end
|
|
|
|
def measure_relation(values)
|
|
row_expressions = axes.expressions.first(axes.rows.length)
|
|
source = population.scope.except(:select, :order)
|
|
fields = axes.rows + axes.columns.first(values.length)
|
|
joins = axes.joins(fields.map { _1.fetch('field') })
|
|
source = source.joins(joins) unless joins.empty?
|
|
values.each_with_index do |value, index|
|
|
expression = axes.expressions[axes.rows.length + index]
|
|
source = source.where("(#{ expression }) <=> ?", value)
|
|
end
|
|
selected = row_expressions.each_with_index.map { |expression, index|
|
|
"#{ expression } AS a#{ index }"
|
|
} + ['COUNT(DISTINCT v.id) AS count']
|
|
source = source.group(*row_expressions.map { Arel.sql(_1) }) if row_expressions.any?
|
|
source.select(Arel.sql(selected.join(', '))).to_sql
|
|
end
|
|
|
|
def measure_values(key)
|
|
records = QueryState.records(@params, 'measures').select { _1['key'] == key }
|
|
raise ArgumentError if records.empty?
|
|
raise ArgumentError unless records.map { _1['axis'] }.sort == (0...records.length).to_a
|
|
|
|
value = records.sort_by { _1['axis'] }.map { _1['value'] }
|
|
unless (value.is_a?(Array) && value.length <= axes.columns.length &&
|
|
value.all? { _1.nil? || _1.is_a?(String) || _1.is_a?(Numeric) })
|
|
raise ArgumentError
|
|
end
|
|
value
|
|
end
|
|
|
|
def detail_expression(key)
|
|
return "(#{ population.latest_sql })" if key == 'is_latest'
|
|
aliases = { 'tags' => 'tags_json', 'parents' => @definition.resource == 'posts' ?
|
|
'parent_post_ids' : 'parent_tag_ids', 'deprecated' => 'deprecated_at' }
|
|
field = aliases.fetch(key, key)
|
|
expression = @definition.expression(field)
|
|
return expression unless @definition.fields[field] == :json
|
|
|
|
"CAST(#{ expression } AS CHAR) COLLATE utf8mb4_bin"
|
|
end
|
|
|
|
def sort_sql
|
|
@sort.flat_map do |key, direction|
|
|
expression = yield key
|
|
["(#{ expression }) IS NULL ASC", "#{ expression } #{ direction.upcase }"]
|
|
end
|
|
end
|
|
|
|
def axis_labels(records, column_values)
|
|
result = {}
|
|
axes.all.each_with_index do |axis, index|
|
|
values = records.map { _1["a#{ index }"] }
|
|
if index >= axes.rows.length
|
|
values += column_values.map { _1[index - axes.rows.length] }
|
|
end
|
|
values = values.compact.uniq
|
|
labels = case axis['field']
|
|
when 'user'
|
|
User.where(id: values).pluck(:id, :name).to_h.transform_keys(&:to_s)
|
|
when 'language_code', 'source_language_code'
|
|
Language.where(code: values).pluck(:code, :name).to_h
|
|
when 'tags'
|
|
tag_labels(values)
|
|
else {}
|
|
end
|
|
result["a#{ index }"] = labels
|
|
end
|
|
result
|
|
end
|
|
|
|
def tag_labels(values)
|
|
internal = values.grep(/\Ainternal:/).map { _1.split(':').last.to_i }
|
|
external = values.grep(/\Aexternal:/).map { _1.split(':').last.to_i }
|
|
point = population.timestamp(@params['at']) || population.at
|
|
labels = {}
|
|
sources = [[TagVersion, internal, 'internal'], [NicoTagVersion, external, 'external']]
|
|
sources.each do |model, ids, kind|
|
|
next if ids.empty?
|
|
|
|
table = model.table_name
|
|
model.where(tag_id: ids).where('created_at <= ?', point)
|
|
.where("NOT EXISTS (SELECT 1 FROM #{ table } newer " +
|
|
"WHERE newer.tag_id = #{ table }.tag_id AND newer.created_at <= ? " +
|
|
"AND newer.version_no > #{ table }.version_no)", point).each do |row|
|
|
names = row.respond_to?(:names) ? row.names : nil
|
|
name = names&.dig(locale.language_code, 'name') || row.name
|
|
labels["#{ kind }:#{ row.tag_id }"] = name
|
|
end
|
|
end
|
|
labels
|
|
end
|
|
end
|