Coverage Report

 // This file was automatically generated with the command: 
 //     java org.galagosearch.tupleflow.typebuilder.TypeBuilderMojo ...
 package org.galagosearch.core.types;
 
 import org.galagosearch.tupleflow.Utility;
 import org.galagosearch.tupleflow.ArrayInput;
 import org.galagosearch.tupleflow.ArrayOutput;
 import org.galagosearch.tupleflow.Order;   
 import org.galagosearch.tupleflow.OrderedWriter;
 import org.galagosearch.tupleflow.Type; 
 import org.galagosearch.tupleflow.TypeReader;
 import org.galagosearch.tupleflow.Step; 
 import org.galagosearch.tupleflow.IncompatibleProcessorException;
 import org.galagosearch.tupleflow.ReaderSource;
 import java.io.IOException;             
 import java.io.EOFException;
 import java.io.UnsupportedEncodingException;
 import java.util.ArrayList;
 import java.util.Arrays;   
 import java.util.Comparator;
 import java.util.PriorityQueue;
 import java.util.Collection;
 
 public class WordCount implements Type<WordCount> {
     public String word;
     public long count;
     public long documents; 
     
     public WordCount() {}
     public WordCount(String word, long count, long documents) {
         this.word = word;
         this.count = count;
         this.documents = documents;
     }  
     
     public String toString() {
             return String.format("%s,%d,%d",
                                    word, count, documents);
     } 
 
     public Order<WordCount> getOrder(String... spec) {
         if (Arrays.equals(spec, new String[] { "+word" })) {
             return new WordOrder();
         }
         return null;
     } 
       
     public interface Processor extends Step, org.galagosearch.tupleflow.Processor<WordCount> {
         public void process(WordCount object) throws IOException;
         public void close() throws IOException;
     }                        
     public interface Source extends Step {
     }
     public static class WordOrder implements Order<WordCount> {
         public int hash(WordCount object) {
             int h = 0;
             h += Utility.hash(object.word);
             return h;
         } 
         public Comparator<WordCount> greaterThan() {
             return new Comparator<WordCount>() {
                 public int compare(WordCount one, WordCount two) {
                     int result = 0;
                     do {
                         result = + Utility.compare(one.word, two.word);
                         if(result != 0) break;
                     } while (false);
                     return -result;
                 }
             };
         }     
         public Comparator<WordCount> lessThan() {
             return new Comparator<WordCount>() {
                 public int compare(WordCount one, WordCount two) {
                     int result = 0;
                     do {
                         result = + Utility.compare(one.word, two.word);
                         if(result != 0) break;
                     } while (false);
                     return result;
                 }
             };
         }     
         public TypeReader<WordCount> orderedReader(ArrayInput _input) {
             return new ShreddedReader(_input);
         }    
 
         public TypeReader<WordCount> orderedReader(ArrayInput _input, int bufferSize) {
             return new ShreddedReader(_input, bufferSize);
         }    
         public OrderedWriter<WordCount> orderedWriter(ArrayOutput _output) {
             ShreddedWriter w = new ShreddedWriter(_output);
             return new OrderedWriterClass(w); 
         }                                    
         public static class OrderedWriterClass extends OrderedWriter< WordCount > {
             WordCount last = null;
             ShreddedWriter shreddedWriter = null; 
             
             public OrderedWriterClass(ShreddedWriter s) {
                 this.shreddedWriter = s;
             }
             
             public void process(WordCount object) throws IOException {
                boolean processAll = false;
                if (processAll || last == null || 0 != Utility.compare(object.word, last.word)) { processAll = true; shreddedWriter.processWord(object.word); }
                shreddedWriter.processTuple(object.count, object.documents);
                last = object;
             }           
                  
             public void close() throws IOException {
                 shreddedWriter.close();
             }
             
             public Class<WordCount> getInputClass() {
                 return WordCount.class;
             }
         } 
         public ReaderSource<WordCount> orderedCombiner(Collection<TypeReader<WordCount>> readers, boolean closeOnExit) {
             ArrayList<ShreddedReader> shreddedReaders = new ArrayList();
             
             for (TypeReader<WordCount> reader : readers) {
                 shreddedReaders.add((ShreddedReader)reader);
             }
             
             return new ShreddedCombiner(shreddedReaders, closeOnExit);
         }                  
         public WordCount clone(WordCount object) {
             WordCount result = new WordCount();
             if (object == null) return result;
             result.word = object.word; 
             result.count = object.count; 
             result.documents = object.documents; 
             return result;
         }                 
         public Class<WordCount> getOrderedClass() {
             return WordCount.class;
         }                           
         public String[] getOrderSpec() {
             return new String[] {"+word"};
         }
 
         public static String getSpecString() {
             return "+word";
         }
                            
         public interface ShreddedProcessor extends Step {
             public void processWord(String word) throws IOException;
             public void processTuple(long count, long documents) throws IOException;
             public void close() throws IOException;
         }    
         public interface ShreddedSource extends Step {
         }                                              
         
         public static class ShreddedWriter implements ShreddedProcessor {
             ArrayOutput output;
             ShreddedBuffer buffer = new ShreddedBuffer();
             String lastWord;
             boolean lastFlush = false;
             
             public ShreddedWriter(ArrayOutput output) {
                 this.output = output;
             }                        
             
             public void close() throws IOException {
                 flush();
             }
             
             public void processWord(String word) {
                 lastWord = word;
                 buffer.processWord(word);
             }
             public final void processTuple(long count, long documents) throws IOException {
                 if (lastFlush) {
                     if(buffer.words.size() == 0) buffer.processWord(lastWord);
                     lastFlush = false;
                 }
                 buffer.processTuple(count, documents);
                 if (buffer.isFull())
                     flush();
             }
             public final void flushTuples(int pauseIndex) throws IOException {
                 
                 while (buffer.getReadIndex() < pauseIndex) {
                            
                     output.writeLong(buffer.getCount());
                     output.writeLong(buffer.getDocuments());
                     buffer.incrementTuple();
                 }
             }  
             public final void flushWord(int pauseIndex) throws IOException {
                 while (buffer.getReadIndex() < pauseIndex) {
                     int nextPause = buffer.getWordEndIndex();
                     int count = nextPause - buffer.getReadIndex();
                     
                     output.writeString(buffer.getWord());
                     output.writeInt(count);
                     buffer.incrementWord();
                       
                     flushTuples(nextPause);
                     assert nextPause == buffer.getReadIndex();
                 }
             }
             public void flush() throws IOException { 
                 flushWord(buffer.getWriteIndex());
                 buffer.reset(); 
                 lastFlush = true;
             }                           
         }
         public static class ShreddedBuffer {
             ArrayList<String> words = new ArrayList();
             ArrayList<Integer> wordTupleIdx = new ArrayList();
             int wordReadIdx = 0;
                             
             long[] counts;
             long[] documentss;
             int writeTupleIndex = 0;
             int readTupleIndex = 0;
             int batchSize;
 
             public ShreddedBuffer(int batchSize) {
                 this.batchSize = batchSize;
 
                 counts = new long[batchSize];
                 documentss = new long[batchSize];
             }                              
 
             public ShreddedBuffer() {    
                 this(10000);
             }                                                                                                                    
             
             public void processWord(String word) {
                 words.add(word);
                 wordTupleIdx.add(writeTupleIndex);
             }                                      
             public void processTuple(long count, long documents) {
                 assert words.size() > 0;
                 counts[writeTupleIndex] = count;
                 documentss[writeTupleIndex] = documents;
                 writeTupleIndex++;
             }
             public void resetData() {
                 words.clear();
                 wordTupleIdx.clear();
                 writeTupleIndex = 0;
             }                  
                                  
             public void resetRead() {
                 readTupleIndex = 0;
                 wordReadIdx = 0;
             } 
 
             public void reset() {
                 resetData();
                 resetRead();
             } 
             public boolean isFull() {
                 return writeTupleIndex >= batchSize;
             }
 
             public boolean isEmpty() {
                 return writeTupleIndex == 0;
             }                          
 
             public boolean isAtEnd() {
                 return readTupleIndex >= writeTupleIndex;
             }           
             public void incrementWord() {
                 wordReadIdx++;  
             }                                                                                              
 
             public void autoIncrementWord() {
                 while (readTupleIndex >= getWordEndIndex() && readTupleIndex < writeTupleIndex)
                     wordReadIdx++;
             }                 
             public void incrementTuple() {
                 readTupleIndex++;
             }                    
             public int getWordEndIndex() {
                 if ((wordReadIdx+1) >= wordTupleIdx.size())
                     return writeTupleIndex;
                 return wordTupleIdx.get(wordReadIdx+1);
             }
             public int getReadIndex() {
                 return readTupleIndex;
             }   
 
             public int getWriteIndex() {
                 return writeTupleIndex;
             } 
             public String getWord() {
                 assert readTupleIndex < writeTupleIndex;
                 assert wordReadIdx < words.size();
                 
                 return words.get(wordReadIdx);
             }
             public long getCount() {
                 assert readTupleIndex < writeTupleIndex;
                 return counts[readTupleIndex];
             }                                         
             public long getDocuments() {
                 assert readTupleIndex < writeTupleIndex;
                 return documentss[readTupleIndex];
             }                                         
             public void copyTuples(int endIndex, ShreddedProcessor output) throws IOException {
                 while (getReadIndex() < endIndex) {
                    output.processTuple(getCount(), getDocuments());
                    incrementTuple();
                 }
             }                                                                           
             public void copyUntilIndexWord(int endIndex, ShreddedProcessor output) throws IOException {
                 while (getReadIndex() < endIndex) {
                     output.processWord(getWord());
                     assert getWordEndIndex() <= endIndex;
                     copyTuples(getWordEndIndex(), output);
                     incrementWord();
                 }
             }  
             public void copyUntilWord(ShreddedBuffer other, ShreddedProcessor output) throws IOException {
                 while (!isAtEnd()) {
                     if (other != null) {   
                         assert !other.isAtEnd();
                         int c = + Utility.compare(getWord(), other.getWord());
                     
                         if (c > 0) {
                             break;   
                         }
                         
                         output.processWord(getWord());
                                       
                         copyTuples(getWordEndIndex(), output);
                     } else {
                         output.processWord(getWord());
                         copyTuples(getWordEndIndex(), output);
                     }
                     incrementWord();  
                     
                
                 }
             }
             public void copyUntil(ShreddedBuffer other, ShreddedProcessor output) throws IOException {
                 copyUntilWord(other, output);
             }
             
         }                         
         public static class ShreddedCombiner implements ReaderSource<WordCount>, ShreddedSource {   
             public ShreddedProcessor processor;
             Collection<ShreddedReader> readers;       
             boolean closeOnExit = false;
             boolean uninitialized = true;
             PriorityQueue<ShreddedReader> queue = new PriorityQueue<ShreddedReader>();
             
             public ShreddedCombiner(Collection<ShreddedReader> readers, boolean closeOnExit) {
                 this.readers = readers;                                                       
                 this.closeOnExit = closeOnExit;
             }
                                   
             public void setProcessor(Step processor) throws IncompatibleProcessorException {  
                 if (processor instanceof ShreddedProcessor) {
                     this.processor = new DuplicateEliminator((ShreddedProcessor) processor);
                 } else if (processor instanceof WordCount.Processor) {
                     this.processor = new DuplicateEliminator(new TupleUnshredder((WordCount.Processor) processor));
                 } else if (processor instanceof org.galagosearch.tupleflow.Processor) {
                     this.processor = new DuplicateEliminator(new TupleUnshredder((org.galagosearch.tupleflow.Processor<WordCount>) processor));
                 } else {
                     throw new IncompatibleProcessorException(processor.getClass().getName() + " is not supported by " + this.getClass().getName());                                                                       
                 }
             }                                
             
             public Class<WordCount> getOutputClass() {
                 return WordCount.class;
             }
             
             public void initialize() throws IOException {
                 for (ShreddedReader reader : readers) {
                     reader.fill();                                        
                     
                     if (!reader.getBuffer().isAtEnd())
                         queue.add(reader);
                 }   
 
                 uninitialized = false;
             }
 
             public void run() throws IOException {
                 initialize();
                
                 while (queue.size() > 0) {
                     ShreddedReader top = queue.poll();
                     ShreddedReader next = null;
                     ShreddedBuffer nextBuffer = null; 
                     
                     assert !top.getBuffer().isAtEnd();
                                                   
                     if (queue.size() > 0) {
                         next = queue.peek();
                         nextBuffer = next.getBuffer();
                         assert !nextBuffer.isAtEnd();
                     }
                     
                     top.getBuffer().copyUntil(nextBuffer, processor);
                     if (top.getBuffer().isAtEnd())
                         top.fill();                 
                         
                     if (!top.getBuffer().isAtEnd())
                         queue.add(top);
                 }              
                 
                 if (closeOnExit)
                     processor.close();
             }
 
             public WordCount read() throws IOException {
                 if (uninitialized)
                     initialize();
 
                 WordCount result = null;
 
                 while (queue.size() > 0) {
                     ShreddedReader top = queue.poll();
                     result = top.read();
 
                     if (result != null) {
                         if (top.getBuffer().isAtEnd())
                             top.fill();
 
                         queue.offer(top);
                         break;
                     } 
                 }
 
                 return result;
             }
         } 
         public static class ShreddedReader implements Step, Comparable<ShreddedReader>, TypeReader<WordCount>, ShreddedSource {      
             public ShreddedProcessor processor;
             ShreddedBuffer buffer;
             WordCount last = new WordCount();         
             long updateWordCount = -1;
             long tupleCount = 0;
             long bufferStartCount = 0;  
             ArrayInput input;
             
             public ShreddedReader(ArrayInput input) {
                 this.input = input; 
                 this.buffer = new ShreddedBuffer();
             }                               
             
             public ShreddedReader(ArrayInput input, int bufferSize) { 
                 this.input = input;
                 this.buffer = new ShreddedBuffer(bufferSize);
             }
                  
             public final int compareTo(ShreddedReader other) {
                 ShreddedBuffer otherBuffer = other.getBuffer();
                 
                 if (buffer.isAtEnd() && otherBuffer.isAtEnd()) {
                     return 0;                 
                 } else if (buffer.isAtEnd()) {
                     return -1;
                 } else if (otherBuffer.isAtEnd()) {
                     return 1;
                 }
                                    
                 int result = 0;
                 do {
                     result = + Utility.compare(buffer.getWord(), otherBuffer.getWord());
                     if(result != 0) break;
                 } while (false);                                             
                 
                 return result;
             }
             
             public final ShreddedBuffer getBuffer() {
                 return buffer;
             }                
             
             public final WordCount read() throws IOException {
                 if (buffer.isAtEnd()) {
                     fill();             
                 
                     if (buffer.isAtEnd()) {
                         return null;
                     }
                 }
                       
                 assert !buffer.isAtEnd();
                 WordCount result = new WordCount();
                 
                 result.word = buffer.getWord();
                 result.count = buffer.getCount();
                 result.documents = buffer.getDocuments();
                 
                 buffer.incrementTuple();
                 buffer.autoIncrementWord();
                 
                 return result;
             }           
             
             public final void fill() throws IOException {
                 try {   
                     buffer.reset();
                     
                     if (tupleCount != 0) {
                                                       
                         if(updateWordCount - tupleCount > 0) {
                             buffer.words.add(last.word);
                             buffer.wordTupleIdx.add((int) (updateWordCount - tupleCount));
                         }
                         bufferStartCount = tupleCount;
                     }
                     
                     while (!buffer.isFull()) {
                         updateWord();
                         buffer.processTuple(input.readLong(), input.readLong());
                         tupleCount++;
                     }
                 } catch(EOFException e) {}
             }
 
             public final void updateWord() throws IOException {
                 if (updateWordCount > tupleCount)
                     return;
                      
                 last.word = input.readString();
                 updateWordCount = tupleCount + input.readInt();
                                       
                 buffer.processWord(last.word);
             }
 
             public void run() throws IOException {
                 while (true) {
                     fill();
                     
                     if (buffer.isAtEnd())
                         break;
                     
                     buffer.copyUntil(null, processor);
                 }      
                 processor.close();
             }
             
             public void setProcessor(Step processor) throws IncompatibleProcessorException {  
                 if (processor instanceof ShreddedProcessor) {
                     this.processor = new DuplicateEliminator((ShreddedProcessor) processor);
                 } else if (processor instanceof WordCount.Processor) {
                     this.processor = new DuplicateEliminator(new TupleUnshredder((WordCount.Processor) processor));
                 } else if (processor instanceof org.galagosearch.tupleflow.Processor) {
                     this.processor = new DuplicateEliminator(new TupleUnshredder((org.galagosearch.tupleflow.Processor<WordCount>) processor));
                 } else {
                     throw new IncompatibleProcessorException(processor.getClass().getName() + " is not supported by " + this.getClass().getName());                                                                       
                 }
             }                                
             
             public Class<WordCount> getOutputClass() {
                 return WordCount.class;
             }                
         }
         
         public static class DuplicateEliminator implements ShreddedProcessor {
             public ShreddedProcessor processor;
             WordCount last = new WordCount();
             boolean wordProcess = true;
                                            
             public DuplicateEliminator() {}
             public DuplicateEliminator(ShreddedProcessor processor) {
                 this.processor = processor;
             }
             
             public void setShreddedProcessor(ShreddedProcessor processor) {
                 this.processor = processor;
             }
 
             public void processWord(String word) throws IOException {  
                 if (wordProcess || Utility.compare(word, last.word) != 0) {
                     last.word = word;
                     processor.processWord(word);
                     wordProcess = false;
                 }
             }  
             
             public void resetWord() {
                  wordProcess = true;
             }                                                
                                
             public void processTuple(long count, long documents) throws IOException {
                 processor.processTuple(count, documents);
             } 
             
             public void close() throws IOException {
                 processor.close();
             }                    
         }
         public static class TupleUnshredder implements ShreddedProcessor {
             WordCount last = new WordCount();
             public org.galagosearch.tupleflow.Processor<WordCount> processor;                               
             
             public TupleUnshredder(WordCount.Processor processor) {
                 this.processor = processor;
             }         
             
             public TupleUnshredder(org.galagosearch.tupleflow.Processor<WordCount> processor) {
                 this.processor = processor;
             }
             
             public WordCount clone(WordCount object) {
                 WordCount result = new WordCount();
                 if (object == null) return result;
                 result.word = object.word; 
                 result.count = object.count; 
                 result.documents = object.documents; 
                 return result;
             }                 
             
             public void processWord(String word) throws IOException {
                 last.word = word;
             }   
                 
             
             public void processTuple(long count, long documents) throws IOException {
                 last.count = count;
                 last.documents = documents;
                 processor.process(clone(last));
             }               
             
             public void close() throws IOException {
                 processor.close();
             }
         }     
         public static class TupleShredder implements Processor {
             WordCount last = new WordCount();
             public ShreddedProcessor processor;
             
             public TupleShredder(ShreddedProcessor processor) {
                 this.processor = processor;
             }                              
             
             public WordCount clone(WordCount object) {
                 WordCount result = new WordCount();
                 if (object == null) return result;
                 result.word = object.word; 
                 result.count = object.count; 
                 result.documents = object.documents; 
                 return result;
             }                 
             
             public void process(WordCount object) throws IOException {                                                                                                                                                   
                 boolean processAll = false;
                 if(last == null || Utility.compare(last.word, object.word) != 0 || processAll) { processor.processWord(object.word); processAll = true; }
                 processor.processTuple(object.count, object.documents);                                         
             }
                           
             public Class<WordCount> getInputClass() {
                 return WordCount.class;
             }
             
             public void close() throws IOException {
                 processor.close();
             }                     
         }
     } 
 }    

1		// This file was automatically generated with the command:
2		// java org.galagosearch.tupleflow.typebuilder.TypeBuilderMojo ...
3		package org.galagosearch.core.types;
4
5		import org.galagosearch.tupleflow.Utility;
6		import org.galagosearch.tupleflow.ArrayInput;
7		import org.galagosearch.tupleflow.ArrayOutput;
8		import org.galagosearch.tupleflow.Order;
9		import org.galagosearch.tupleflow.OrderedWriter;
10		import org.galagosearch.tupleflow.Type;
11		import org.galagosearch.tupleflow.TypeReader;
12		import org.galagosearch.tupleflow.Step;
13		import org.galagosearch.tupleflow.IncompatibleProcessorException;
14		import org.galagosearch.tupleflow.ReaderSource;
15		import java.io.IOException;
16		import java.io.EOFException;
17		import java.io.UnsupportedEncodingException;
18		import java.util.ArrayList;
19		import java.util.Arrays;
20		import java.util.Comparator;
21		import java.util.PriorityQueue;
22		import java.util.Collection;
23
24		public class WordCount implements Type<WordCount> {
25		public String word;
26		public long count;
27		public long documents;
28
29	0	public WordCount() {}
30	0	public WordCount(String word, long count, long documents) {
31	0	this.word = word;
32	0	this.count = count;
33	0	this.documents = documents;
34	0	}
35
36		public String toString() {
37	0	return String.format("%s,%d,%d",
38		word, count, documents);
39		}
40
41		public Order<WordCount> getOrder(String... spec) {
42	0	if (Arrays.equals(spec, new String[] { "+word" })) {
43	0	return new WordOrder();
44		}
45	0	return null;
46		}
47
48		public interface Processor extends Step, org.galagosearch.tupleflow.Processor<WordCount> {
49		public void process(WordCount object) throws IOException;
50		public void close() throws IOException;
51		}
52		public interface Source extends Step {
53		}
54	0	public static class WordOrder implements Order<WordCount> {
55		public int hash(WordCount object) {
56	0	int h = 0;
57	0	h += Utility.hash(object.word);
58	0	return h;
59		}
60		public Comparator<WordCount> greaterThan() {
61	0	return new Comparator<WordCount>() {
62	0	public int compare(WordCount one, WordCount two) {
63	0	int result = 0;
64		do {
65	0	result = + Utility.compare(one.word, two.word);
66	0	if(result != 0) break;
67		} while (false);
68	0	return -result;
69		}
70		};
71		}
72		public Comparator<WordCount> lessThan() {
73	0	return new Comparator<WordCount>() {
74	0	public int compare(WordCount one, WordCount two) {
75	0	int result = 0;
76		do {
77	0	result = + Utility.compare(one.word, two.word);
78	0	if(result != 0) break;
79		} while (false);
80	0	return result;
81		}
82		};
83		}
84		public TypeReader<WordCount> orderedReader(ArrayInput _input) {
85	0	return new ShreddedReader(_input);
86		}
87
88		public TypeReader<WordCount> orderedReader(ArrayInput _input, int bufferSize) {
89	0	return new ShreddedReader(_input, bufferSize);
90		}
91		public OrderedWriter<WordCount> orderedWriter(ArrayOutput _output) {
92	0	ShreddedWriter w = new ShreddedWriter(_output);
93	0	return new OrderedWriterClass(w);
94		}
95	0	public static class OrderedWriterClass extends OrderedWriter< WordCount > {
96	0	WordCount last = null;
97	0	ShreddedWriter shreddedWriter = null;
98
99	0	public OrderedWriterClass(ShreddedWriter s) {
100	0	this.shreddedWriter = s;
101	0	}
102
103		public void process(WordCount object) throws IOException {
104	0	boolean processAll = false;
105	0	if (processAll \|\| last == null \|\| 0 != Utility.compare(object.word, last.word)) { processAll = true; shreddedWriter.processWord(object.word); }
106	0	shreddedWriter.processTuple(object.count, object.documents);
107	0	last = object;
108	0	}
109
110		public void close() throws IOException {
111	0	shreddedWriter.close();
112	0	}
113
114		public Class<WordCount> getInputClass() {
115	0	return WordCount.class;
116		}
117		}
118		public ReaderSource<WordCount> orderedCombiner(Collection<TypeReader<WordCount>> readers, boolean closeOnExit) {
119	0	ArrayList<ShreddedReader> shreddedReaders = new ArrayList();
120
121	0	for (TypeReader<WordCount> reader : readers) {
122	0	shreddedReaders.add((ShreddedReader)reader);
123		}
124
125	0	return new ShreddedCombiner(shreddedReaders, closeOnExit);
126		}
127		public WordCount clone(WordCount object) {
128	0	WordCount result = new WordCount();
129	0	if (object == null) return result;
130	0	result.word = object.word;
131	0	result.count = object.count;
132	0	result.documents = object.documents;
133	0	return result;
134		}
135		public Class<WordCount> getOrderedClass() {
136	0	return WordCount.class;
137		}
138		public String[] getOrderSpec() {
139	0	return new String[] {"+word"};
140		}
141
142		public static String getSpecString() {
143	0	return "+word";
144		}
145
146		public interface ShreddedProcessor extends Step {
147		public void processWord(String word) throws IOException;
148		public void processTuple(long count, long documents) throws IOException;
149		public void close() throws IOException;
150		}
151		public interface ShreddedSource extends Step {
152		}
153
154	0	public static class ShreddedWriter implements ShreddedProcessor {
155		ArrayOutput output;
156	0	ShreddedBuffer buffer = new ShreddedBuffer();
157		String lastWord;
158	0	boolean lastFlush = false;
159
160	0	public ShreddedWriter(ArrayOutput output) {
161	0	this.output = output;
162	0	}
163
164		public void close() throws IOException {
165	0	flush();
166	0	}
167
168		public void processWord(String word) {
169	0	lastWord = word;
170	0	buffer.processWord(word);
171	0	}
172		public final void processTuple(long count, long documents) throws IOException {
173	0	if (lastFlush) {
174	0	if(buffer.words.size() == 0) buffer.processWord(lastWord);
175	0	lastFlush = false;
176		}
177	0	buffer.processTuple(count, documents);
178	0	if (buffer.isFull())
179	0	flush();
180	0	}
181		public final void flushTuples(int pauseIndex) throws IOException {
182
183	0	while (buffer.getReadIndex() < pauseIndex) {
184
185	0	output.writeLong(buffer.getCount());
186	0	output.writeLong(buffer.getDocuments());
187	0	buffer.incrementTuple();
188		}
189	0	}
190		public final void flushWord(int pauseIndex) throws IOException {
191	0	while (buffer.getReadIndex() < pauseIndex) {
192	0	int nextPause = buffer.getWordEndIndex();
193	0	int count = nextPause - buffer.getReadIndex();
194
195	0	output.writeString(buffer.getWord());
196	0	output.writeInt(count);
197	0	buffer.incrementWord();
198
199	0	flushTuples(nextPause);
200	0	assert nextPause == buffer.getReadIndex();
201	0	}
202	0	}
203		public void flush() throws IOException {
204	0	flushWord(buffer.getWriteIndex());
205	0	buffer.reset();
206	0	lastFlush = true;
207	0	}
208		}
209	0	public static class ShreddedBuffer {
210	0	ArrayList<String> words = new ArrayList();
211	0	ArrayList<Integer> wordTupleIdx = new ArrayList();
212	0	int wordReadIdx = 0;
213
214		long[] counts;
215		long[] documentss;
216	0	int writeTupleIndex = 0;
217	0	int readTupleIndex = 0;
218		int batchSize;
219
220	0	public ShreddedBuffer(int batchSize) {
221	0	this.batchSize = batchSize;
222
223	0	counts = new long[batchSize];
224	0	documentss = new long[batchSize];
225	0	}
226
227		public ShreddedBuffer() {
228	0	this(10000);
229	0	}
230
231		public void processWord(String word) {
232	0	words.add(word);
233	0	wordTupleIdx.add(writeTupleIndex);
234	0	}
235		public void processTuple(long count, long documents) {
236	0	assert words.size() > 0;
237	0	counts[writeTupleIndex] = count;
238	0	documentss[writeTupleIndex] = documents;
239	0	writeTupleIndex++;
240	0	}
241		public void resetData() {
242	0	words.clear();
243	0	wordTupleIdx.clear();
244	0	writeTupleIndex = 0;
245	0	}
246
247		public void resetRead() {
248	0	readTupleIndex = 0;
249	0	wordReadIdx = 0;
250	0	}
251
252		public void reset() {
253	0	resetData();
254	0	resetRead();
255	0	}
256		public boolean isFull() {
257	0	return writeTupleIndex >= batchSize;
258		}
259
260		public boolean isEmpty() {
261	0	return writeTupleIndex == 0;
262		}
263
264		public boolean isAtEnd() {
265	0	return readTupleIndex >= writeTupleIndex;
266		}
267		public void incrementWord() {
268	0	wordReadIdx++;
269	0	}
270
271		public void autoIncrementWord() {
272	0	while (readTupleIndex >= getWordEndIndex() && readTupleIndex < writeTupleIndex)
273	0	wordReadIdx++;
274	0	}
275		public void incrementTuple() {
276	0	readTupleIndex++;
277	0	}
278		public int getWordEndIndex() {
279	0	if ((wordReadIdx+1) >= wordTupleIdx.size())
280	0	return writeTupleIndex;
281	0	return wordTupleIdx.get(wordReadIdx+1);
282		}
283		public int getReadIndex() {
284	0	return readTupleIndex;
285		}
286
287		public int getWriteIndex() {
288	0	return writeTupleIndex;
289		}
290		public String getWord() {
291	0	assert readTupleIndex < writeTupleIndex;
292	0	assert wordReadIdx < words.size();
293
294	0	return words.get(wordReadIdx);
295		}
296		public long getCount() {
297	0	assert readTupleIndex < writeTupleIndex;
298	0	return counts[readTupleIndex];
299		}
300		public long getDocuments() {
301	0	assert readTupleIndex < writeTupleIndex;
302	0	return documentss[readTupleIndex];
303		}
304		public void copyTuples(int endIndex, ShreddedProcessor output) throws IOException {
305	0	while (getReadIndex() < endIndex) {
306	0	output.processTuple(getCount(), getDocuments());
307	0	incrementTuple();
308		}
309	0	}
310		public void copyUntilIndexWord(int endIndex, ShreddedProcessor output) throws IOException {
311	0	while (getReadIndex() < endIndex) {
312	0	output.processWord(getWord());
313	0	assert getWordEndIndex() <= endIndex;
314	0	copyTuples(getWordEndIndex(), output);
315	0	incrementWord();
316		}
317	0	}
318		public void copyUntilWord(ShreddedBuffer other, ShreddedProcessor output) throws IOException {
319	0	while (!isAtEnd()) {
320	0	if (other != null) {
321	0	assert !other.isAtEnd();
322	0	int c = + Utility.compare(getWord(), other.getWord());
323
324	0	if (c > 0) {
325	0	break;
326		}
327
328	0	output.processWord(getWord());
329
330	0	copyTuples(getWordEndIndex(), output);
331	0	} else {
332	0	output.processWord(getWord());
333	0	copyTuples(getWordEndIndex(), output);
334		}
335	0	incrementWord();
336
337
338		}
339	0	}
340		public void copyUntil(ShreddedBuffer other, ShreddedProcessor output) throws IOException {
341	0	copyUntilWord(other, output);
342	0	}
343
344		}
345	0	public static class ShreddedCombiner implements ReaderSource<WordCount>, ShreddedSource {
346		public ShreddedProcessor processor;
347		Collection<ShreddedReader> readers;
348	0	boolean closeOnExit = false;
349	0	boolean uninitialized = true;
350	0	PriorityQueue<ShreddedReader> queue = new PriorityQueue<ShreddedReader>();
351
352	0	public ShreddedCombiner(Collection<ShreddedReader> readers, boolean closeOnExit) {
353	0	this.readers = readers;
354	0	this.closeOnExit = closeOnExit;
355	0	}
356
357		public void setProcessor(Step processor) throws IncompatibleProcessorException {
358	0	if (processor instanceof ShreddedProcessor) {
359	0	this.processor = new DuplicateEliminator((ShreddedProcessor) processor);
360	0	} else if (processor instanceof WordCount.Processor) {
361	0	this.processor = new DuplicateEliminator(new TupleUnshredder((WordCount.Processor) processor));
362	0	} else if (processor instanceof org.galagosearch.tupleflow.Processor) {
363	0	this.processor = new DuplicateEliminator(new TupleUnshredder((org.galagosearch.tupleflow.Processor<WordCount>) processor));
364		} else {
365	0	throw new IncompatibleProcessorException(processor.getClass().getName() + " is not supported by " + this.getClass().getName());
366		}
367	0	}
368
369		public Class<WordCount> getOutputClass() {
370	0	return WordCount.class;
371		}
372
373		public void initialize() throws IOException {
374	0	for (ShreddedReader reader : readers) {
375	0	reader.fill();
376
377	0	if (!reader.getBuffer().isAtEnd())
378	0	queue.add(reader);
379		}
380
381	0	uninitialized = false;
382	0	}
383
384		public void run() throws IOException {
385	0	initialize();
386
387	0	while (queue.size() > 0) {
388	0	ShreddedReader top = queue.poll();
389	0	ShreddedReader next = null;
390	0	ShreddedBuffer nextBuffer = null;
391
392	0	assert !top.getBuffer().isAtEnd();
393
394	0	if (queue.size() > 0) {
395	0	next = queue.peek();
396	0	nextBuffer = next.getBuffer();
397	0	assert !nextBuffer.isAtEnd();
398		}
399
400	0	top.getBuffer().copyUntil(nextBuffer, processor);
401	0	if (top.getBuffer().isAtEnd())
402	0	top.fill();
403
404	0	if (!top.getBuffer().isAtEnd())
405	0	queue.add(top);
406	0	}
407
408	0	if (closeOnExit)
409	0	processor.close();
410	0	}
411
412		public WordCount read() throws IOException {
413	0	if (uninitialized)
414	0	initialize();
415
416	0	WordCount result = null;
417
418	0	while (queue.size() > 0) {
419	0	ShreddedReader top = queue.poll();
420	0	result = top.read();
421
422	0	if (result != null) {
423	0	if (top.getBuffer().isAtEnd())
424	0	top.fill();
425
426	0	queue.offer(top);
427	0	break;
428		}
429	0	}
430
431	0	return result;
432		}
433		}
434	0	public static class ShreddedReader implements Step, Comparable<ShreddedReader>, TypeReader<WordCount>, ShreddedSource {
435		public ShreddedProcessor processor;
436		ShreddedBuffer buffer;
437	0	WordCount last = new WordCount();
438	0	long updateWordCount = -1;
439	0	long tupleCount = 0;
440	0	long bufferStartCount = 0;
441		ArrayInput input;
442
443	0	public ShreddedReader(ArrayInput input) {
444	0	this.input = input;
445	0	this.buffer = new ShreddedBuffer();
446	0	}
447
448	0	public ShreddedReader(ArrayInput input, int bufferSize) {
449	0	this.input = input;
450	0	this.buffer = new ShreddedBuffer(bufferSize);
451	0	}
452
453		public final int compareTo(ShreddedReader other) {
454	0	ShreddedBuffer otherBuffer = other.getBuffer();
455
456	0	if (buffer.isAtEnd() && otherBuffer.isAtEnd()) {
457	0	return 0;
458	0	} else if (buffer.isAtEnd()) {
459	0	return -1;
460	0	} else if (otherBuffer.isAtEnd()) {
461	0	return 1;
462		}
463
464	0	int result = 0;
465		do {
466	0	result = + Utility.compare(buffer.getWord(), otherBuffer.getWord());
467	0	if(result != 0) break;
468		} while (false);
469
470	0	return result;
471		}
472
473		public final ShreddedBuffer getBuffer() {
474	0	return buffer;
475		}
476
477		public final WordCount read() throws IOException {
478	0	if (buffer.isAtEnd()) {
479	0	fill();
480
481	0	if (buffer.isAtEnd()) {
482	0	return null;
483		}
484		}
485
486	0	assert !buffer.isAtEnd();
487	0	WordCount result = new WordCount();
488
489	0	result.word = buffer.getWord();
490	0	result.count = buffer.getCount();
491	0	result.documents = buffer.getDocuments();
492
493	0	buffer.incrementTuple();
494	0	buffer.autoIncrementWord();
495
496	0	return result;
497		}
498
499		public final void fill() throws IOException {
500		try {
501	0	buffer.reset();
502
503	0	if (tupleCount != 0) {
504
505	0	if(updateWordCount - tupleCount > 0) {
506	0	buffer.words.add(last.word);
507	0	buffer.wordTupleIdx.add((int) (updateWordCount - tupleCount));
508		}
509	0	bufferStartCount = tupleCount;
510		}
511
512	0	while (!buffer.isFull()) {
513	0	updateWord();
514	0	buffer.processTuple(input.readLong(), input.readLong());
515	0	tupleCount++;
516		}
517	0	} catch(EOFException e) {}
518	0	}
519
520		public final void updateWord() throws IOException {
521	0	if (updateWordCount > tupleCount)
522	0	return;
523
524	0	last.word = input.readString();
525	0	updateWordCount = tupleCount + input.readInt();
526
527	0	buffer.processWord(last.word);
528	0	}
529
530		public void run() throws IOException {
531		while (true) {
532	0	fill();
533
534	0	if (buffer.isAtEnd())
535	0	break;
536
537	0	buffer.copyUntil(null, processor);
538		}
539	0	processor.close();
540	0	}
541
542		public void setProcessor(Step processor) throws IncompatibleProcessorException {
543	0	if (processor instanceof ShreddedProcessor) {
544	0	this.processor = new DuplicateEliminator((ShreddedProcessor) processor);
545	0	} else if (processor instanceof WordCount.Processor) {
546	0	this.processor = new DuplicateEliminator(new TupleUnshredder((WordCount.Processor) processor));
547	0	} else if (processor instanceof org.galagosearch.tupleflow.Processor) {
548	0	this.processor = new DuplicateEliminator(new TupleUnshredder((org.galagosearch.tupleflow.Processor<WordCount>) processor));
549		} else {
550	0	throw new IncompatibleProcessorException(processor.getClass().getName() + " is not supported by " + this.getClass().getName());
551		}
552	0	}
553
554		public Class<WordCount> getOutputClass() {
555	0	return WordCount.class;
556		}
557		}
558
559		public static class DuplicateEliminator implements ShreddedProcessor {
560		public ShreddedProcessor processor;
561	0	WordCount last = new WordCount();
562	0	boolean wordProcess = true;
563
564	0	public DuplicateEliminator() {}
565	0	public DuplicateEliminator(ShreddedProcessor processor) {
566	0	this.processor = processor;
567	0	}
568
569		public void setShreddedProcessor(ShreddedProcessor processor) {
570	0	this.processor = processor;
571	0	}
572
573		public void processWord(String word) throws IOException {
574	0	if (wordProcess \|\| Utility.compare(word, last.word) != 0) {
575	0	last.word = word;
576	0	processor.processWord(word);
577	0	wordProcess = false;
578		}
579	0	}
580
581		public void resetWord() {
582	0	wordProcess = true;
583	0	}
584
585		public void processTuple(long count, long documents) throws IOException {
586	0	processor.processTuple(count, documents);
587	0	}
588
589		public void close() throws IOException {
590	0	processor.close();
591	0	}
592		}
593		public static class TupleUnshredder implements ShreddedProcessor {
594	0	WordCount last = new WordCount();
595		public org.galagosearch.tupleflow.Processor<WordCount> processor;
596
597	0	public TupleUnshredder(WordCount.Processor processor) {
598	0	this.processor = processor;
599	0	}
600
601	0	public TupleUnshredder(org.galagosearch.tupleflow.Processor<WordCount> processor) {
602	0	this.processor = processor;
603	0	}
604
605		public WordCount clone(WordCount object) {
606	0	WordCount result = new WordCount();
607	0	if (object == null) return result;
608	0	result.word = object.word;
609	0	result.count = object.count;
610	0	result.documents = object.documents;
611	0	return result;
612		}
613
614		public void processWord(String word) throws IOException {
615	0	last.word = word;
616	0	}
617
618
619		public void processTuple(long count, long documents) throws IOException {
620	0	last.count = count;
621	0	last.documents = documents;
622	0	processor.process(clone(last));
623	0	}
624
625		public void close() throws IOException {
626	0	processor.close();
627	0	}
628		}
629	0	public static class TupleShredder implements Processor {
630	0	WordCount last = new WordCount();
631		public ShreddedProcessor processor;
632
633	0	public TupleShredder(ShreddedProcessor processor) {
634	0	this.processor = processor;
635	0	}
636
637		public WordCount clone(WordCount object) {
638	0	WordCount result = new WordCount();
639	0	if (object == null) return result;
640	0	result.word = object.word;
641	0	result.count = object.count;
642	0	result.documents = object.documents;
643	0	return result;
644		}
645
646		public void process(WordCount object) throws IOException {
647	0	boolean processAll = false;
648	0	if(last == null \|\| Utility.compare(last.word, object.word) != 0 \|\| processAll) { processor.processWord(object.word); processAll = true; }
649	0	processor.processTuple(object.count, object.documents);
650	0	}
651
652		public Class<WordCount> getInputClass() {
653	0	return WordCount.class;
654		}
655
656		public void close() throws IOException {
657	0	processor.close();
658	0	}
659		}
660		}
661		}