<commit_before>require 'bloom-filter'

module Jetpants
  class Shard
    # Runs queries against a slave in the pool to verify sharding key values
    def validate_shard_data
      tables = Table.from_config 'sharded_tables'
      table_statuses = {}
      tables.limited_concurrent_map(8) { |table|
        table.sharding_keys.each do |col|
          range_sql = table.sql_range_check col, @min_id, @max_id

          # use a standby slave, since this query will be very heavy and these shards are live
          db = standby_slaves.last
          result = db.query_return_array range_sql

          if result.first.values.first > 0
            table_statuses[table] = :invalid
          else
            table_statuses[table] = :valid
          end
        end
      }

      table_statuses
    end

    def self.check_duplicate_keys(shards, table, key)
      dbs = []
      shards.each do |shard|
        raise "Invalid shard #{shard}!" unless shard.is_a? Shard
        raise "Attempting to validate table not con" unless shard.has_table? table.name
      end
      raise "Currently only possible to compare 2 shards!" unless shards.count == 2
      raise "Invalid index '#{key}' for table '#{table}'!" if table.indexes[key].nil?
      raise "Only currently implemented for single-column indexes" unless table.indexes[key][:columns].count == 1

      source_shard = shards.first
      source_db = source_shard.standby_slaves.last
      comparison_shard = shards.last
      comparison_db = comparison_shard.standby_slaves.last
      column = table.indexes[key][:columns].first
      dbs = [ source_db, comparison_db ]

      dbs.concurrent_each do |db|
        db.pause_replication
        db.stop_query_killer
        db.disable_monitoring
      end

      chunk_size = 5000

      min_val = source_db.query_return_first_value("SELECT min(#{column}) FROM #{table}")
      max_val = source_db.query_return_first_value("SELECT min(#{column}) FROM #{table}")

      # maximum possible entries and desired error rate
      max_size = (max_val.to_i - min_val.to_i) / Jetpants.shards.count
      filter = BloomFilter.new size: max_size, error_rate: 0.001
      curr_val = min_val

      puts "Generating filter from #{source_shard}"
      while curr_val < max_val do
        vals = source_db.query_return_array("SELECT #{column} FROM #{table} WHERE #{column} > #{val} LIMIT #{chunk_size}").map{ |row| row.values.first }
        vals.each{ |val| filter.insert val }
        curr_val = vals.last
      end

      min_val = comparison_db.query_return_first_value("SELECT min(#{column}) FROM #{table}")
      max_val = comparison_db.query_return_first_value("SELECT min(#{column}) FROM #{table}")
      possible_dupes = []
      curr_val = min_val

      puts "Searching for duplicates in #{comparison_shard}"<commit_msg>Use max() in sql<commit_after>require 'bloom-filter'

module Jetpants
  class Shard
    # Runs queries against a slave in the pool to verify sharding key values
    def validate_shard_data
      tables = Table.from_config 'sharded_tables'
      table_statuses = {}
      tables.limited_concurrent_map(8) { |table|
        table.sharding_keys.each do |col|
          range_sql = table.sql_range_check col, @min_id, @max_id

          # use a standby slave, since this query will be very heavy and these shards are live
          db = standby_slaves.last
          result = db.query_return_array range_sql

          if result.first.values.first > 0
            table_statuses[table] = :invalid
          else
            table_statuses[table] = :valid
          end
        end
      }

      table_statuses
    end

    def self.check_duplicate_keys(shards, table, key)
      dbs = []
      shards.each do |shard|
        raise "Invalid shard #{shard}!" unless shard.is_a? Shard
        raise "Attempting to validate table not con" unless shard.has_table? table.name
      end
      raise "Currently only possible to compare 2 shards!" unless shards.count == 2
      raise "Invalid index '#{key}' for table '#{table}'!" if table.indexes[key].nil?
      raise "Only currently implemented for single-column indexes" unless table.indexes[key][:columns].count == 1

      source_shard = shards.first
      source_db = source_shard.standby_slaves.last
      comparison_shard = shards.last
      comparison_db = comparison_shard.standby_slaves.last
      column = table.indexes[key][:columns].first
      dbs = [ source_db, comparison_db ]

      dbs.concurrent_each do |db|
        db.pause_replication
        db.stop_query_killer
        db.disable_monitoring
      end

      chunk_size = 5000

      min_val = source_db.query_return_first_value("SELECT min(#{column}) FROM #{table}")
      max_val = source_db.query_return_first_value("SELECT max(#{column}) FROM #{table}")

      # maximum possible entries and desired error rate
      max_size = (max_val.to_i - min_val.to_i) / Jetpants.shards.count
      filter = BloomFilter.new size: max_size, error_rate: 0.001
      curr_val = min_val

      puts "Generating filter from #{source_shard}"
      while curr_val < max_val do
        vals = source_db.query_return_array("SELECT #{column} FROM #{table} WHERE #{column} > #{val} LIMIT #{chunk_size}").map{ |row| row.values.first }
        vals.each{ |val| filter.insert val }
        curr_val = vals.last
      end

      min_val = comparison_db.query_return_first_value("SELECT min(#{column}) FROM #{table}")
      max_val = comparison_db.query_return_first_value("SELECT max(#{column}) FROM #{table}")
      possible_dupes = []
      curr_val = min_val

      puts "Searching for duplicates in #{comparison_shard}"
{'fraction_non_alphanumeric': 0.09379996446971042, 'fraction_numerical': 0.004618937644341801, 'mean_word_length': 2.856164383561644, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '4020131', 'n_tokens_mistral': 1753, 'n_tokens_neox': 1706, 'n_words': 446}