# 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'].freeze attr_reader :population, :axes, :page, :limit, :mode def initialize(params) raise ArgumentError if (params.keys - KEYS).any? raise ArgumentError unless params.values.all? { _1.is_a?(String) } @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 = params.fetch('sort', '').split(',').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', 'original_created_at'] : []) + (@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?(/\Ameasure_[0-9a-f]+\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 { 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(', ') }" row_sort = @sort.each_with_index.filter_map do |(key, direction), index| if key == 'count' || key.start_with?('measure_') condition = if key == 'count' 'TRUE' else measure_values(key).each_with_index.map do |value, offset| "a#{ axes.rows.length + offset } <=> #{ @connection.quote(value) }" end.join(' AND ').presence || 'TRUE' end ["MAX(CASE WHEN #{ condition } THEN count END) AS sort#{ index }", "sort#{ index } IS NULL ASC, sort#{ index } #{ direction.upcase }"] elsif row_keys.include?(key) [nil, "#{ key } IS NULL ASC, #{ key } #{ direction.upcase }"] end end extras = row_sort.filter_map(&:first) order_rows = row_sort.map(&:last) + row_keys.map { "#{ _1 } IS NULL, #{ _1 }" } order_rows = ['singleton ASC'] if order_rows.empty? group_sql = "SELECT #{ ([row_selection] + extras).join(', ') } FROM cells #{ row_group }" 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 { group_count: total, column_values:, 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? order = keys.map { "#{ _1 } IS NULL, #{ _1 }" }.join(', ') 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_values(key) value = JSON.parse([key.delete_prefix('measure_')].pack('H*')) 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', 'original_created_at' => 'original_created_from' } 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