<commit_before>require "../term"

module ReQL
  class GroupTerm < Term
    infix_inspect "group"

    def check
      expect_args 2, 3
    end
  end

  class Evaluator
    def eval_term(term : GroupTerm)
      target = eval(term.args[0])
      group_func = eval(term.args[1]).as_function
      aggregation_func = term.args.size >= 3 ? eval(term.args[2]).as_function : nil

      stream_table = {} of Datum => GroupStream
      result_channel = Channel({Datum, Datum} | Exception).new

      group_count = 0

      target.each do |value|
        value = value.as_datum
        group = group_func.eval(self, {value}).as_datum
        stream = stream_table[group]?
        unless stream
          stream = stream_table[group] = GroupStream.new
          group_count += 1
          spawn do
            begin
              func = aggregation_func
              if func
                evaluator = Evaluator.new(@manager, @worker)
                evaluator.vars = @vars.dup
                evaluator.now = @now

                result_channel.send({
                  group,
                  func.eval(evaluator, {stream.not_nil!}).as_datum,
                })
              else
                result_channel.send({
                  group,
                  stream.not_nil!.as_datum,
                })
              end
            rescue exception
              result_channel.send(exception)
            end
          end
        end
        stream.not_nil! << value
      end

      stream_table.each_value &.<<(nil)

      result = [] of Hash(String, Datum)

      group_count.times do
        pair = result_channel.receive
        if pair.is_a? Exception
          raise pair
        end

        result << {
          "group"     => pair[0],
          "reduction" => pair[1],
        }
      end

      Datum.new(result)
    end
  end
end<commit_msg>optime r.group() when there is no aggregation<commit_after>require "../term"

module ReQL
  class GroupTerm < Term
    infix_inspect "group"

    def check
      expect_args 2, 3
    end
  end

  class Evaluator
    def eval_term(term : GroupTerm)
      target = eval(term.args[0])
      group_func = eval(term.args[1]).as_function

      result = [] of Hash(String, Datum)

      if term.args.size == 3
        aggregation_func = eval(term.args[2]).as_function

        stream_table = {} of Datum => GroupStream
        result_channel = Channel({Datum, Datum} | Exception).new

        group_count = 0

        target.each do |value|
          value = value.as_datum
          group = group_func.eval(self, {value}).as_datum
          stream = stream_table[group]?
          unless stream
            stream = stream_table[group] = GroupStream.new
            group_count += 1
            spawn do
              begin
                evaluator = Evaluator.new(@manager, @worker)
                evaluator.vars = @vars.dup
                evaluator.now = @now

                result_channel.send({
                  group,
                  aggregation_func.eval(evaluator, {stream.not_nil!}).as_datum,
                })
              rescue exception
                result_channel.send(exception)
              end
            end
          end
          stream.not_nil! << value
        end

        stream_table.each_value &.<<(nil)

        group_count.times do
          pair = result_channel.receive
          if pair.is_a? Exception
            raise pair
          end

          result << {
            "group"     => pair[0],
            "reduction" => pair[1],
          }
        end
      else
        groups = Hash(Datum, Array(Datum)).new { |h, k| h[k] = [] of Datum }

        target.each do |value|
          value = value.as_datum
          group = group_func.eval(self, {value}).as_datum
          groups[group] << value
        end

        groups.each do |(group, array)|
          result << {
            "group"     => group,
            "reduction" => Datum.new(array),
          }
        end
      end

      Datum.new(result)
    end
  end
end
{'fraction_non_alphanumeric': 0.09406800694961528, 'fraction_numerical': 0.004964010920824026, 'mean_word_length': 1.4573170731707317, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 12, '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': '4976242', 'n_tokens_mistral': 1186, 'n_tokens_neox': 1135, 'n_words': 257}