<commit_before>module Sidekiq
  module Superworker
    class Worker
      def self.create(*args, &block)
        class_name = args.shift.to_sym
        nested_hash = DSLParser.parse(block)
        create_class(class_name, args, nested_hash)
      end

      protected

      def self.create_class(class_name, arg_keys, nested_hash)
        klass = Class.new do
          @class_name = class_name
          @arg_keys = arg_keys
          @nested_hash = nested_hash
          @dsl_hash = DSLHash.new

          class << self
            attr_reader :nested_hash

            def is_a_superworker?
              true
            end

            def perform_async(*arg_values)
              superjob_options = {}

              # If an additional argument value is given, it's the superjob's options
              if (arg_values.length == @arg_keys.length + 1) && arg_values.last.is_a?(Hash)
                superjob_options = arg_values.last
              elsif @arg_keys.length != arg_values.length
                raise "Wrong number of arguments for #{name}.perform_async (#{arg_values.length} for #{@arg_keys.length})"
              end

              @args = Hash[@arg_keys.zip(arg_values)]
              @superjob_id = SecureRandom.hex(12)
              subjobs = create_subjobs(arg_values)
              SuperjobProcessor.create(@superjob_id, @class_name, arg_values, subjobs, superjob_options)
            end

            protected

            def create_subjobs(arg_values)
              records = @dsl_hash.nested_hash_to_records(@nested_hash, @args)
              records.collect do |id, record|
                record[:status] = 'initialized'
                record[:superjob_id] = @superjob_id
                record[:superworker_class] = @class_name
                unless record.key?(:arg_values)
                  record[:arg_values] = record[:arg_keys].collect do |arg_key|
                    # Allow for subjob arg_values to be set within the superworker definition; if a symbol is
                    # used in the DSL, use @args[arg_key], and otherwise use arg_key as the value
                    arg_key.is_a?(Symbol) ? @args[arg_key] : arg_key
                  end
                end
                Sidekiq::Superworker::Subjob.create(record)
              end
            end
          end
        end

        Object.const_set(class_name, klass)
      end
    end
  end
end
<commit_msg>Perform Subjob inserts in a single transaction to speed up mass inserts<commit_after>module Sidekiq
  module Superworker
    class Worker
      def self.create(*args, &block)
        class_name = args.shift.to_sym
        nested_hash = DSLParser.parse(block)
        create_class(class_name, args, nested_hash)
      end

      protected

      def self.create_class(class_name, arg_keys, nested_hash)
        klass = Class.new do
          @class_name = class_name
          @arg_keys = arg_keys
          @nested_hash = nested_hash
          @dsl_hash = DSLHash.new

          class << self
            attr_reader :nested_hash

            def is_a_superworker?
              true
            end

            def perform_async(*arg_values)
              superjob_options = {}

              # If an additional argument value is given, it's the superjob's options
              if (arg_values.length == @arg_keys.length + 1) && arg_values.last.is_a?(Hash)
                superjob_options = arg_values.last
              elsif @arg_keys.length != arg_values.length
                raise "Wrong number of arguments for #{name}.perform_async (#{arg_values.length} for #{@arg_keys.length})"
              end

              @args = Hash[@arg_keys.zip(arg_values)]
              @superjob_id = SecureRandom.hex(12)
              subjobs = create_subjobs(arg_values)
              SuperjobProcessor.create(@superjob_id, @class_name, arg_values, subjobs, superjob_options)
            end

            protected

            def create_subjobs(arg_values)
              records = @dsl_hash.nested_hash_to_records(@nested_hash, @args)
              subjobs = records.collect do |id, record|
                record[:status] = 'initialized'
                record[:superjob_id] = @superjob_id
                record[:superworker_class] = @class_name
                unless record.key?(:arg_values)
                  record[:arg_values] = record[:arg_keys].collect do |arg_key|
                    # Allow for subjob arg_values to be set within the superworker definition; if a symbol is
                    # used in the DSL, use @args[arg_key], and otherwise use arg_key as the value
                    arg_key.is_a?(Symbol) ? @args[arg_key] : arg_key
                  end
                end
                record
              end
              # Perform the inserts in a single transaction to improve the performance of large mass inserts
              Sidekiq::Superworker::Subjob.transaction do
                Sidekiq::Superworker::Subjob.create(subjobs)
              end
            end
          end
        end

        Object.const_set(class_name, klass)
      end
    end
  end
end

{'fraction_non_alphanumeric': 0.07460348541217936, 'fraction_numerical': 0.0011748580379870766, 'mean_word_length': 1.7836512261580382, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, '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': '7129170', 'n_tokens_mistral': 1438, 'n_tokens_neox': 1370, 'n_words': 339}