<commit_before>require_relative 'pass_thru'

module ConceptQL
  module Operators
    class Union < PassThru
      register __FILE__

      desc 'Pools sets of incoming results into a single large set of results.'
      allows_many_upstreams
      category "Combine Streams"
      default_query_columns
      validate_at_least_one_upstream
      validate_no_arguments

      def query(db)
        upstreams.map do |expression|
          expression.evaluate(db).from_self.select(*query_cols)
        end.inject do |q, query|
          q.union(query, all: true)
        end
      end<commit_msg>Optimize union operator by avoiding union in certain cases<commit_after>require_relative 'pass_thru'

module ConceptQL
  module Operators
    class Union < PassThru
      register __FILE__

      desc 'Pools sets of incoming results into a single large set of results.'
      allows_many_upstreams
      category "Combine Streams"
      default_query_columns
      validate_at_least_one_upstream
      validate_no_arguments

      def query(db)
        datasets = upstreams.map do |expression|
          expression.evaluate(db)
        end

        optss = datasets
          .map { |ds| ds.opts.compact }
          .each { |opts| opts.delete(:num_dataset_sources) }

        possibly_combinable = optss.all? { |opts| opts.length == 1 && opts[:from].length == 1 }
        if possibly_combinable
          subqueries = optss.map { |opts| opts[:from].first }
          subqueries.map! { |ds| ds.is_a?(Sequel::SQL::AliasedExpression) ? ds.expression : ds }
          subquery_optss = subqueries.map { |ds| ds.opts.compact }
          subquery_optss.each{|opts| opts.delete(:where)}
          combinable = subquery_optss.uniq.length == 1
        end

        if combinable
          first = subqueries.shift
          first
            .or(Sequel.|(*subqueries.map{|ds| ds.opts[:where] || true}))
            .from_self
        else
          datasets.map!{|ds| ds.from_self.select(*query_cols)}
          datasets.inject do |q, query|
            q.union(query, all: true)
          end
        end
      end
{'fraction_non_alphanumeric': 0.08536003814973772, 'fraction_numerical': 0.001430615164520744, 'mean_word_length': 2.3301587301587303, 'pattern_counts': {'":': 0, '<': 5, '<?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': '6322693', 'n_tokens_mistral': 635, 'n_tokens_neox': 612, 'n_words': 150}