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