/*
 * Druid - a distributed column store.
 * Copyright (C) 2012, 2013  Metamarkets Group Inc.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 */

package io.druid.query;

import com.google.common.collect.Lists;
import com.metamx.common.guava.Accumulator;
import com.metamx.common.guava.Sequence;
import com.metamx.common.guava.Sequences;
import com.metamx.common.guava.Yielder;
import com.metamx.common.guava.Yielders;
import com.metamx.common.guava.YieldingAccumulator;
import org.joda.time.DateTime;

import java.util.List;

/**
 */
public class BySegmentQueryRunner<T> implements QueryRunner<T>
{
  private final String segmentIdentifier;
  private final DateTime timestamp;
  private final QueryRunner<T> base;

  public BySegmentQueryRunner(
      String segmentIdentifier,
      DateTime timestamp,
      QueryRunner<T> base
  )
  {
    this.segmentIdentifier = segmentIdentifier;
    this.timestamp = timestamp;
    this.base = base;
  }

  @Override
  @SuppressWarnings("unchecked")
  public Sequence<T> run(final Query<T> query)
  {
    if (query.getContextBySegment(false)) {
      final Sequence<T> baseSequence = base.run(query);
      return new Sequence<T>()
      {
        @Override
        public <OutType> OutType accumulate(OutType initValue, Accumulator<OutType, T> accumulator)
        {
          List<T> results = Sequences.toList(baseSequence, Lists.<T>newArrayList());

          return accumulator.accumulate(
              initValue,
              (T) new Result<BySegmentResultValueClass<T>>(
                  timestamp,
                  new BySegmentResultValueClass<T>(
                      results,
                      segmentIdentifier,
                      query.getIntervals().get(0)
                  )
              )
          );
        }

        @Override
        public <OutType> Yielder<OutType> toYielder(OutType initValue, YieldingAccumulator<OutType, T> accumulator)
        {
          List<T> results = Sequences.toList(baseSequence, Lists.<T>newArrayList());

          final OutType retVal = accumulator.accumulate(
              initValue,
              (T) new Result<BySegmentResultValueClass<T>>(
                  timestamp,
                  new BySegmentResultValueClass<T>(
                      results,
                      segmentIdentifier,
                      query.getIntervals().get(0)
                  )
              )
          );

          return Yielders.done(retVal, null);
        }
      };
    }

    return base.run(query);
  }
}

@@ -1171,16 +1171,41 @@
 eTime;%0A%0A
+import java.util.Arrays;%0A
 import j
@@ -1839,184 +1839,20 @@
 y);%0A
-      return new Sequence%3CT%3E()%0A      %7B%0A        @Override%0A        public %3COutType%3E OutType accumulate(OutType initValue, Accumulator%3COutType, T%3E accumulator)%0A        %7B%0A         
+%0A      final
  Lis
@@ -1919,29 +1919,24 @@
 rayList());%0A
-%0A    
       return
@@ -1940,691 +1940,50 @@
 urn 
-accumulator.accumulate(%0A              initValue,%0A              (T) new Result%3CBySegmentResultValueClass%3CT%3E%3E(%0A                  timestamp,%0A                  new BySegmentResultValueClass%3CT%3E(%0A                      results,%0A                      segmentIdentifier,%0A                      query.getIntervals().get(0)%0A                  )%0A              )%0A          );%0A        %7D%0A%0A        @Override%0A        public %3COutType%3E Yielder%3COutType%3E toYielder(OutType initValue, YieldingAccumulator%3COutType, T%3E accumulator)%0A        %7B%0A          List%3CT%3E results = Sequences.toList(baseSequence, Lists.%3CT%3EnewArrayList());%0A%0A          final OutType retVal = accumulator.accumulate(%0A              initValue,
+Sequences.simple(%0A          Arrays.asList(
 %0A   
@@ -2293,74 +2293,16 @@
    )
-;%0A
 %0A      
-    return Yielders.done(retVal, null);%0A        %7D%0A      %7D
+)
 ;%0A  

fix bysegment-query-runner