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