java/pig-galago/src/com/yahoo/pig/impl/eval/window/TupleWindowSpec.java

   1 /*
   2  * Copyright (c) 2007 Yahoo! Inc.  All rights reserved.
   3  * See accompanying LICENSE file.
   4  */
   5 package com.yahoo.pig.impl.eval.window;
   6
   7 import java.io.IOException;
   8 import java.util.*;
   9
  10 import com.yahoo.pig.data.*;
  11
  12 public class TupleWindowSpec extends WindowSpec {
  13         int numTuples;
  14     List<Tuple> window;
  15
  16         public TupleWindowSpec(windowType type, int numTuples){
  17                 super(type);
  18                 this.numTuples = numTuples;
  19         window = new LinkedList<Tuple>();
  20         }
  21
  22         @Override
  23         public DataCollector collector(final DataCollector output) {
  24             return new DataCollector() {
  25
  26                 public void add(Tuple t) throws IOException {
  27                         if (t!=null){
  28
  29                                 if (t instanceof TimestampedTuple){
  30                                         if (((TimestampedTuple) t).isHeartbeat) return;
  31                                 }
  32
  33                         window.add(t);
  34                         while (window.size() > numTuples) {
  35                             window.remove(0);
  36                         }
  37
  38
  39                         }
  40
  41                         // emit entire window content to output collector
  42                 for (Iterator<Tuple> it = window.iterator(); it.hasNext(); ) {
  43                     output.add(it.next());
  44                 }
  45
  46                 output.add(null);
  47                 }
  48             };
  49     }
  50 }