addresses issue #225 - MultiIndex does not expose index structures inp…

modules/batch-indexers/src/main/java/org/terrier/structures/indexing/DiskIndexWriter.java

@@ -52,7 +52,6 @@ public DiskIndexWriter withDirect() {
     @SuppressWarnings("unchecked")
     public IndexOnDisk write(Index source) throws IOException {
         IndexOnDisk target = IndexOnDisk.createNewIndex(path, prefix);
-
         CompressionConfiguration compressionDirectConfig = null;
 
 
@@ -135,6 +134,7 @@ public IndexOnDisk write(Index source) throws IOException {
         AbstractPostingOutputStream invOut = compressionInvertedConfig.getPostingOutputStream(
 				path + ApplicationSetup.FILE_SEPARATOR + prefix + "." + "inverted" + compressionInvertedConfig.getStructureFileExtension());
 
+        int tid = 0;
         while (lexIN.hasNext()) {
 
             // Read in-memory lexicon and postings.
@@ -144,10 +144,17 @@ public IndexOnDisk write(Index source) throws IOException {
             // Write on-disk lexicon and postings.
             BitIndexPointer pointer = invOut.writePostings(postings);
             LexiconEntry le = leFactory.newInstance();
+            //set maxtf to 0, so that it can be updated
+            le.setMaxFrequencyInDocuments(0);
             le.add(term.getValue());
-            le.setTermId(term.getValue().getTermId());
+            // we only need to preserve termids if we are creating a direct index
+            if (direct)
+                le.setTermId(term.getValue().getTermId());
+            else
+                le.setTermId(tid);
             le.setPointer(pointer);
             lexOUT.writeNextEntry(term.getKey(), le);
+            tid++;
         }
 
         IndexUtil.close(lexIN);

modules/core/src/main/java/org/terrier/structures/PostingIndexInputStream.java

@@ -54,5 +54,24 @@ public interface PostingIndexInputStream extends Closeable, Iterator<IterablePos
 	int getEntriesSkipped();
 
 	/** Renders the entire structure to stdout */
-	void print();
+	default void print() {
+		try{
+			int entryIndex = 0;
+			while(this.hasNext())
+			{
+				IterablePosting ip = this.next();
+				entryIndex += this.getEntriesSkipped();
+				System.out.print(entryIndex + " ");
+				while(ip.next() != IterablePosting.EOL)
+				{
+					System.out.print(ip.toString());
+					System.out.print(" ");
+				}
+				System.out.println();
+				entryIndex++;
+			}
+		} catch (Exception e) {
+			throw new RuntimeException(e);
+		}
+	}
 }

modules/core/src/main/java/org/terrier/structures/collections/IteratorUtils.java

@@ -0,0 +1,139 @@
+package org.terrier.structures.collections;
+
+import java.util.function.BiFunction;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+import com.google.common.collect.Iterators;
+import com.google.common.collect.PeekingIterator;
+
+
+public class IteratorUtils {
+
+    public static <T> Iterator<Map.Entry<T,Integer>> addOffset(final Iterator<T> iter, int offset) {
+        return new Iterator<Map.Entry<T,Integer>>() {
+
+            @Override
+            public boolean hasNext() {
+                return iter.hasNext();
+            }
+
+            @Override
+            public Map.Entry<T,Integer> next() {
+                return new MapEntry<T,Integer>(iter.next(), offset);
+            }
+        };
+    }
+
+    public static <X,Y> Iterator<Map.Entry<X,Y>> zip(final Iterator<X> iterX, final Iterator<Y> iterY) {
+        return new Iterator<Map.Entry<X,Y>>() {
+
+            @Override
+            public boolean hasNext() {
+                return iterX.hasNext();
+            }
+
+            @Override
+            public Map.Entry<X,Y> next() {
+                return new MapEntry(iterX.next(), iterY.next());
+            }
+        };
+    }
+
+    @SafeVarargs
+    public static <T> Iterator<T> merge(Comparator<? super T> ordering, Iterator<T>... iterators) {
+        return merge(ordering, iterators, 0, iterators.length);
+    }
+
+    @SafeVarargs
+    public static <T> Iterator<T> merge(Comparator<? super T> ordering, BiFunction<T,T,T> merger, Iterator<T>... iterators) {
+        return merge(ordering, merger, iterators, 0, iterators.length);
+    }
+
+    private static <T> Iterator<T> merge(Comparator<? super T> ordering, Iterator<T>[] iterators, int offset, int length) {
+        if (length == 0) {
+            return Collections.emptyIterator();
+        }
+        if (length == 1) {
+            return iterators[offset];
+        }
+        return new MergedIterator<>(ordering,
+            merge(ordering, iterators, offset, length / 2),
+            merge(ordering, iterators, offset + length / 2, length - (length / 2)));
+    }
+
+    private static <T> Iterator<T> merge(Comparator<? super T> ordering,  BiFunction<T,T,T> merger, Iterator<T>[] iterators, int offset, int length) {
+        if (length == 0) {
+            return Collections.emptyIterator();
+        }
+        if (length == 1) {
+            return iterators[offset];
+        }
+        return new MergingMergedIterator<>(ordering, merger,
+            merge(ordering, merger, iterators, offset, length / 2),
+            merge(ordering, merger, iterators, offset + length / 2, length - (length / 2)));
+    }
+
+    static class MergedIterator<T> implements Iterator<T> {
+        final Comparator<? super T> comparator;
+        final PeekingIterator<T> first;
+        final PeekingIterator<T> second;
+
+        private MergedIterator(Comparator<? super T> comparator, Iterator<T> first, Iterator<T> second) {
+            this.comparator = comparator;
+            this.first = Iterators.peekingIterator(first);
+            this.second = Iterators.peekingIterator(second);
+        }
+
+        @Override
+        public boolean hasNext() {
+            return first.hasNext() || second.hasNext();
+        }
+
+        @Override
+        public T next() {
+            if (!first.hasNext()) {
+                if (!second.hasNext()) {
+                    throw new NoSuchElementException();
+                }
+                return second.next();
+            }
+
+            return !second.hasNext() || (comparator.compare(first.peek(), second.peek()) <= 0) ? first.next() : second.next();
+        }
+    }
+
+    static class MergingMergedIterator<T> extends MergedIterator<T> {
+        BiFunction<T,T,T> merger;
+
+        private MergingMergedIterator(
+            Comparator<? super T> comparator, BiFunction<T,T,T> merger, Iterator<T> first, Iterator<T> second) {
+            super(comparator, first, second);
+            this.merger = merger;
+        }
+
+        @Override
+        public T next() {
+            if (!first.hasNext()) {
+                if (!second.hasNext()) {
+                    throw new NoSuchElementException();
+                }
+                return second.next();
+            }
+            if (! second.hasNext()) {
+                return first.next();
+            }
+            final int cmp = comparator.compare(first.peek(), second.peek());
+            if (cmp == 0)
+                return merger.apply(first.next(), second.next()); //move both interators on
+            else if (cmp > 0)
+                return second.next();
+            return first.next();
+        }
+    }
+
+
+}

modules/realtime/src/main/java/org/terrier/realtime/multi/MultiIndex.java

@@ -30,17 +30,23 @@
 import java.io.IOException;
 import java.io.Flushable;
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 
+import org.apache.commons.collections4.iterators.IteratorChain;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.terrier.realtime.matching.IncrementalSelectiveMatching;
 import org.terrier.querying.IndexRef;
+import org.terrier.structures.collections.IteratorUtils;
+import org.terrier.structures.collections.MapEntry;
 import org.terrier.structures.CollectionStatistics;
 import org.terrier.structures.DocumentIndex;
 import org.terrier.structures.Index;
 import org.terrier.structures.IndexFactory;
 import org.terrier.structures.Lexicon;
+import org.terrier.structures.LexiconEntry;
 import org.terrier.structures.MetaIndex;
 import org.terrier.structures.Pointer;
 import org.terrier.structures.PostingIndex;
@@ -215,9 +221,49 @@ public boolean hasIndexStructure(String structureName) {
 		return false;
 	}
 
-	/** Not implemented. */
+	List<Iterator<?>> getIndexStructureInputStream_Iterators(String structureName) {
+		List<Iterator<?>> iters = new ArrayList<>();
+		for (Index index : selectiveMatchingPolicy.getSelectedIndices(indices)) {
+			Iterator<?> iter = (Iterator<?>) index.getIndexStructureInputStream(structureName);
+			if (iter == null) {
+				return null;
+			}
+			iters.add(iter);
+		}
+		return iters;
+	}
+
+	/** Returns an IteratorChain of the underlying constituent index structures */
 	public Object getIndexStructureInputStream(String structureName) {
-		return null;
+
+		// use special class for an invert index input stream
+		if (structureName.equals("inverted"))
+		{
+			return new MultiInvertedIndexInputStream(
+				(Iterator<Map.Entry<String,LexiconEntry>>) getIndexStructureInputStream("lexicon"), 
+				this);
+		}
+
+		// get iterators for each subindex
+		List<Iterator<?>> iters = getIndexStructureInputStream_Iterators(structureName);
+
+		// now merge the iterators. how they are merged depends on the particular index structure
+
+		// support document-wise structures for now
+		if (structureName.equals("document") || structureName.equals("meta"))
+			return new IteratorChain(iters);
+
+		if (structureName.equals("lexicon"))
+			return IteratorUtils.merge(  
+				// comparator
+				(Map.Entry<String,LexiconEntry> term1, Map.Entry<String,LexiconEntry> term2) -> term1.getKey().compareTo(term2.getKey()), 
+				// merger
+				(Map.Entry<String,LexiconEntry> term1, Map.Entry<String,LexiconEntry> term2) -> new MapEntry(  
+					term1.getKey(), 
+					new MultiLexiconEntry(new LexiconEntry[]{term1.getValue(), term2.getValue()}, 0)),
+				// iterators
+				(Iterator<Map.Entry<String,LexiconEntry>>[]) iters.toArray(new Iterator<?>[iters.size()]));
+		throw new UnsupportedOperationException("I dont know how to merge the input streams of " + structureName);
 	}
 
 	/** {@inheritDoc} */

modules/realtime/src/main/java/org/terrier/realtime/multi/MultiInvertedIndexInputStream.java

@@ -0,0 +1,67 @@
+package org.terrier.realtime.multi;
+import org.terrier.structures.PostingIndexInputStream;
+import org.terrier.structures.Index;
+import org.terrier.structures.LexiconEntry;
+import org.terrier.structures.postings.IterablePosting;
+import org.terrier.structures.Pointer;
+import org.terrier.structures.IndexUtil;
+import java.util.Iterator;
+import java.util.Map;
+import java.io.IOException;
+
+/** This is not a streaming implementation */
+public class MultiInvertedIndexInputStream implements PostingIndexInputStream {
+
+    Iterator<Map.Entry<String,LexiconEntry>> iterLex;
+    Index mindex;
+    LexiconEntry le;
+
+    public MultiInvertedIndexInputStream(Iterator<Map.Entry<String,LexiconEntry>> iterLex, Index mindex) {
+        this.mindex = mindex;
+        this.iterLex = iterLex;
+    }
+
+    public IterablePosting getNextPostings() throws IOException {
+        if (! iterLex.hasNext())
+        {
+            return null;
+        }
+        String t = iterLex.next().getKey();
+        // this is inefficient 
+        le = mindex.getLexicon().getLexiconEntry(t);
+        return mindex.getInvertedIndex().getPostings(le);
+    }
+
+    public boolean hasNext() {
+        return iterLex.hasNext();
+    }
+
+    public IterablePosting next() {
+        try{
+            String t = iterLex.next().getKey();
+            // this is inefficient 
+            le = mindex.getLexicon().getLexiconEntry(t);
+            return mindex.getInvertedIndex().getPostings(le);
+        } catch (IOException ioe) {
+            throw new RuntimeException(ioe);
+        }
+    }
+
+    /** Returns the number of postings in the current IterablePosting object */
+    public int getNumberOfCurrentPostings() {
+        return le.getNumberOfEntries();
+    }
+
+    /** Returns the pointer associated with the current postings being accessed */
+    public Pointer getCurrentPointer() {
+        return le;
+    }
+
+    public int getEntriesSkipped() {
+        return 0;
+    }
+
+    public void close() throws IOException {
+        IndexUtil.close(iterLex);
+    }
+}

modules/realtime/src/main/java/org/terrier/realtime/multi/MultiLexicon.java

@@ -245,8 +245,6 @@ public LexiconEntryIterator() {
 					uniqueTerms.add(t);
 				Collections.sort(uniqueTerms);
 			}
-
-
 		}
 
 		@Override