<commit_before>require 'mysql'
require 'date'

# Define some convenient shorthands for specifying times
class Numeric
	def minute ; self * 60 ; end
	def hour ; self * 3600 ; end
	def day ; self * 86400 ; end
	def week ; 7 * day ; end
	def month ; 30 * day ; end
	def year ; 365 * day ; end
	def ago ; (Time.new.to_f - self).to_i ; end
end

class Aggregator
	# Rules for handling tables. Add hashes of that include the keys:
	#  :age    # Rule triggers for rows that are at least this old
	#  :table  # Which tables to use this rule on. Use a regexp or symbol :all.
	#  :reduce # Reduce data to one hour averages
	#  :drop   # Don't reduce, just drop older data
	# Reduce data older than one year to eight hour averages, for all tables:
	#  rules << { :table => :all, :age => 1.year, :reduce => 8.hour }
	# Drop data older than two years from ADSL tables:
	#  rules << { :table => /^adsl/, :age => 2.year, :drop => true }
	attr_accessor :rules
	# Whether or not to index tables before optimization.
	# If this is set to true (default), indexes will be created, if they are missing,
	# on dtime and id. This is recommended for acceptable aggregation performance.
	#  index = true
	attr_accessor :index
	# Whether or not to run OPTIMIZE TABLE after aggregation.
	# Defaults to true.
	#  optimize = true
	attr_accessor :optimize
	# List of tables to exclude. Accepts literal strings (exakt match) and
	# regular expressions. By default certain non time-series tables of RTG.
	# such as +interfaces+ and +routers+ are excluded. If you have custom
	# tables that should not be touched by aggregator, add them here.
	#  excludes << "custom_table"  # Don't touch "custom_table"
	#  excludes << /^dlink/        # Don't touch tables whose name starts with "dlink"
	attr_accessor :excludes
	# Limit run time of aggregator to a certain time span.
	# This may be useful if aggregator is begin run from cron at a
	# certain interval, to avoid overlapping processes.
	# Set to nil (default) to no limit the run time.
	# Note that aggregator always completes a table it has started processing, so
	# the specified runtime might get exceeded.
	#  runlimit = nil         # No limit
	#  runlimit = 50.minutes # Almost guaranteed to be done in an hour.
	attr_accessor :runlimit
	# How ofter to reaggregate a table. Defaults to one month.
	#  reaggregate_interval = 1.month
	attr_accessor :reaggregate_interval
	# How to access the RTG database. Accepts a hash containing connection parameters.
	#  database = { :host => "localhost", :user => "aggregator", :password => "<PASSWORD>", :database => "rtg" }
	attr_accessor :database
	# Whether to do any altering operations on the database or not. If set to false, nothing will be done
	# and all altering SQL queries will be echoed to stdout instead. This might fail, if for example the
	# pruned table doesn't exist.
	#  dry_run = false
	attr_accessor :dry_run
	attr_accessor :verbose

	def initialize
		@rules = []
		@index = true
		@optimize = true
		@excludes = [ 'interfaces', 'routers', 'pruned' ]
		@runlimit = nil
		@reaggregate_interval = 1.year
		@database = nil
		@dry_run = false
		@verbose = false
	end

	# List of all currently active rules.
	def rules
		# Rules should always be returned sorted with oldest first.
		@rules.sort! { |a, b| b[:age] <=> a[:age] }
	end

	# Create SQL commands for aggregating a table/id combination between the specified times to the specified interval.
	# Results in an array of DELETE and INSERT commands.
	def aggregate(table, id, start_time, end_time, interval)
		rows = rows(table, id, start_time, end_time)
		return [] if rows.count < 2
		queries = []
		clusters = cluster_rows(rows, interval)
		clusters.each do |cluster|
			if cluster.length > 1
				summary = summary_row(cluster)
				insert = "INSERT INTO #{table} (id, dtime, counter, rate) VALUES (#{id}, FROM_UNIXTIME(#{summary[0]}), #{summary[1]}, #{summary[2]})"
				delete = "DELETE FROM #{table} WHERE id = #{id} AND dtime IN (" + cluster.collect{ |c| "FROM_UNIXTIME(#{c[0]})" }.join(", ") + ")"
				queries << delete << insert
			end
		end
		return queries
	end

	def drop(table, end_time)
		"DELETE FROM #{table} WHERE dtime <= FROM_UNIXTIME(#{end_time})"
	end

	def run
		to_process = tables
		processing_limit_time = Time.new + runlimit
		while to_process.length > 0 && (runlimit.nil? || Time.new < processing_limit_time)
			table = to_process.delete_at(rand(to_process.length))
			last_pruned = get_pruned(table)
			next if !last_pruned.nil? && DateTime.now < last_pruned + reaggregate_interval / 86400.0

			table_rules = rules_for(table)
			next if table_rules.length == 0

			puts "Looking at #{table}" if @verbose

			if @index
				create_indexes(table)
			end

			queries = []
			prev_rule_end_time = nil
			table_rules.each do |rule|
				end_time = rule[:age].ago
				if rule.key? :drop
					queries << drop(table, end_time)
				elsif rule.key? :reduce
					interval = rule[:reduce]
					start_time = prev_rule_end_time || 0
					ids(table).each do |id|
						queries += aggregate(table, id, start_time, end_time, interval)
					end
				end
				prev_rule_end_time = end_time
			end

			deletes = 0
			inserts = 0
			if !@dry_run
				queries.each do |q|
					connection.query(q)
					if q =~ /INSERT/
						inserts += connection.affected_rows
					elsif q =~ /DELETE/
						deletes += connection.affected_rows
					end
				end
			else
				queries.each { |q| puts q } if @verbose
			end

			puts "  Inserts: #{inserts}" if @verbose
			puts "  Deletes: #{deletes}" if @verbose

			if @optimize
				optimize_table(table)
			end

			set_pruned(table)

			puts "  #{to_process.length} tables left to handle." if @verbose
		end
	end

	private

	#
	# Logic stuff
	#

	def cluster_rows(rows, interval)
		clusters = []
		cur_cluster = []
		interval_end = (rows[0][0] / interval + 1).to_i * interval<commit_msg>Interleave changes with selects<commit_after>require 'mysql'
require 'date'

# Define some convenient shorthands for specifying times
class Numeric
	def minute ; self * 60 ; end
	def hour ; self * 3600 ; end
	def day ; self * 86400 ; end
	def week ; 7 * day ; end
	def month ; 30 * day ; end
	def year ; 365 * day ; end
	def ago ; (Time.new.to_f - self).to_i ; end
end

class Aggregator
	# Rules for handling tables. Add hashes of that include the keys:
	#  :age    # Rule triggers for rows that are at least this old
	#  :table  # Which tables to use this rule on. Use a regexp or symbol :all.
	#  :reduce # Reduce data to one hour averages
	#  :drop   # Don't reduce, just drop older data
	# Reduce data older than one year to eight hour averages, for all tables:
	#  rules << { :table => :all, :age => 1.year, :reduce => 8.hour }
	# Drop data older than two years from ADSL tables:
	#  rules << { :table => /^adsl/, :age => 2.year, :drop => true }
	attr_accessor :rules
	# Whether or not to index tables before optimization.
	# If this is set to true (default), indexes will be created, if they are missing,
	# on dtime and id. This is recommended for acceptable aggregation performance.
	#  index = true
	attr_accessor :index
	# Whether or not to run OPTIMIZE TABLE after aggregation.
	# Defaults to true.
	#  optimize = true
	attr_accessor :optimize
	# List of tables to exclude. Accepts literal strings (exakt match) and
	# regular expressions. By default certain non time-series tables of RTG.
	# such as +interfaces+ and +routers+ are excluded. If you have custom
	# tables that should not be touched by aggregator, add them here.
	#  excludes << "custom_table"  # Don't touch "custom_table"
	#  excludes << /^dlink/        # Don't touch tables whose name starts with "dlink"
	attr_accessor :excludes
	# Limit run time of aggregator to a certain time span.
	# This may be useful if aggregator is begin run from cron at a
	# certain interval, to avoid overlapping processes.
	# Set to nil (default) to no limit the run time.
	# Note that aggregator always completes a table it has started processing, so
	# the specified runtime might get exceeded.
	#  runlimit = nil         # No limit
	#  runlimit = 50.minutes # Almost guaranteed to be done in an hour.
	attr_accessor :runlimit
	# How ofter to reaggregate a table. Defaults to one month.
	#  reaggregate_interval = 1.month
	attr_accessor :reaggregate_interval
	# How to access the RTG database. Accepts a hash containing connection parameters.
	#  database = { :host => "localhost", :user => "aggregator", :password => "<PASSWORD>", :database => "rtg" }
	attr_accessor :database
	# Whether to do any altering operations on the database or not. If set to false, nothing will be done
	# and all altering SQL queries will be echoed to stdout instead. This might fail, if for example the
	# pruned table doesn't exist.
	#  dry_run = false
	attr_accessor :dry_run
	attr_accessor :verbose

	def initialize
		@rules = []
		@index = true
		@optimize = true
		@excludes = [ 'interfaces', 'routers', 'pruned' ]
		@runlimit = nil
		@reaggregate_interval = 1.year
		@database = nil
		@dry_run = false
		@verbose = false
	end

	# List of all currently active rules.
	def rules
		# Rules should always be returned sorted with oldest first.
		@rules.sort! { |a, b| b[:age] <=> a[:age] }
	end

	# Create SQL commands for aggregating a table/id combination between the specified times to the specified interval.
	# Results in an array of DELETE and INSERT commands.
	def aggregate(table, id, start_time, end_time, interval)
		rows = rows(table, id, start_time, end_time)
		return [] if rows.count < 2
		queries = []
		clusters = cluster_rows(rows, interval)
		clusters.each do |cluster|
			if cluster.length > 1
				summary = summary_row(cluster)
				insert = "INSERT INTO #{table} (id, dtime, counter, rate) VALUES (#{id}, FROM_UNIXTIME(#{summary[0]}), #{summary[1]}, #{summary[2]})"
				delete = "DELETE FROM #{table} WHERE id = #{id} AND dtime IN (" + cluster.collect{ |c| "FROM_UNIXTIME(#{c[0]})" }.join(", ") + ")"
				queries << delete << insert
			end
		end
		return queries
	end

	def drop(table, end_time)
		"DELETE FROM #{table} WHERE dtime <= FROM_UNIXTIME(#{end_time})"
	end

	def run
		to_process = tables
		processing_limit_time = Time.new + runlimit
		while to_process.length > 0 && (runlimit.nil? || Time.new < processing_limit_time)
			table = to_process.delete_at(rand(to_process.length))
			last_pruned = get_pruned(table)
			next if !last_pruned.nil? && DateTime.now < last_pruned + reaggregate_interval / 86400.0

			table_rules = rules_for(table)
			next if table_rules.length == 0

			puts "Looking at #{table}" if @verbose

			if @index
				create_indexes(table)
			end

			queries = []
			deletes = 0
			inserts = 0
			delete_qs = 0
			insert_qs = 0
			prev_rule_end_time = nil
			table_rules.each do |rule|
				# Do drops and reduces, and collect SQL statements
				end_time = rule[:age].ago
				if rule.key? :drop
					queries << drop(table, end_time)
				elsif rule.key? :reduce
					interval = rule[:reduce]
					start_time = prev_rule_end_time || 0
					ids(table).each do |id|
						queries += aggregate(table, id, start_time, end_time, interval)
					end
				end
				prev_rule_end_time = end_time

				# Execute the collected statements
				if !@dry_run
					queries.each do |q|
						connection.query(q)
						if q =~ /INSERT/
							inserts += connection.affected_rows
							insert_qs += 1
						elsif q =~ /DELETE/
							deletes += connection.affected_rows
							delete_qs += 1
						end
					end
				else
					queries.each { |q| puts q } if @verbose
				end
				queries = []
			end

			puts "  Inserted #{inserts} rows (individually)." if @verbose
			puts "  Deleted #{deletes} rows in #{delete_qs} queries (about #{(deletes / delete_qs).to_i} rows/q)" if @verbose

			if @optimize
				optimize_table(table)
			end

			set_pruned(table)

			puts "  #{to_process.length} tables left to check." if @verbose
		end
	end

	private

	#
	# Logic stuff
	#

	def cluster_rows(rows, interval)
		clusters = []
		cur_cluster = []
		interval_end = (rows[0][0] / interval + 1).to_i * interval
{'fraction_non_alphanumeric': 0.09993337774816789, 'fraction_numerical': 0.007661558960692871, 'mean_word_length': 3.357402031930334, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6147577', 'n_tokens_mistral': 4413, 'n_tokens_neox': 3829, 'n_words': 1425}