@Test @Ignore public void testOnDiskStreamingIterator() { Database db = (Database) store; int count = TestData.getScaleCount() * 5; Set<Revision<PrimaryKey, Text, Value>> expected = Sets .newLinkedHashSetWithExpectedSize(count); for (int i = 0; i < count; ++i) { Write write = Write.add(TestData.getSimpleString(), TestData.getTObject(), i); db.accept(write); Revision<PrimaryKey, Text, Value> revision = Revision .createPrimaryRevision(write.getRecord(), write.getKey(), write.getValue(), write.getVersion(), write.getType()); expected.add(revision); Variables.register("expected_" + i, revision); if(i % 100 == 0) { db.triggerSync(); } } db.triggerSync(); Iterator<Revision<PrimaryKey, Text, Value>> it = Database .onDiskStreamingIterator(db.getBackingStore()); Iterator<Revision<PrimaryKey, Text, Value>> it2 = expected.iterator(); int i = 0; while (it.hasNext()) { Revision<PrimaryKey, Text, Value> actual = it.next(); Assert.assertEquals(it2.next(), actual); Variables.register("actual_" + i, actual); ++i; } }
public static Iterator<Revision<PrimaryKey, Text, Value>> onDiskStreamingIterator( final String dbStore) { return new ReadOnlyIterator<Revision<PrimaryKey, Text, Value>>() { private final String directory = FileSystem.makePath(dbStore, SEGMENTS_SUBDIRECTORY); private final Iterator<String> fileIt = FileSystem .fileOnlyIterator(directory); private Iterator<Revision<PrimaryKey, Text, Value>> it = null; { flip(); } @Override public boolean hasNext() { if(it == null) { return false; } else if(!it.hasNext() && fileIt.hasNext()) { flip(); return hasNext(); } else if(!it.hasNext()) { return false; } else { return true; } } @Override public Revision<PrimaryKey, Text, Value> next() { if(hasNext()) { return it.next(); } else { return null; } } private void flip() { if(fileIt.hasNext()) { Path file = Paths.get(fileIt.next()); try { it = Segment.load(file).table().iterator(); } catch (SegmentLoadingException e) { flip(); } } } }; }
Database extends BaseStore implements PermanentStore { public static Iterator<Revision<PrimaryKey, Text, Value>> onDiskStreamingIterator( final String dbStore) { return new ReadOnlyIterator<Revision<PrimaryKey, Text, Value>>() { private final String directory = FileSystem.makePath(dbStore, SEGMENTS_SUBDIRECTORY); private final Iterator<String> fileIt = FileSystem .fileOnlyIterator(directory); private Iterator<Revision<PrimaryKey, Text, Value>> it = null; { flip(); } @Override public boolean hasNext() { if(it == null) { return false; } else if(!it.hasNext() && fileIt.hasNext()) { flip(); return hasNext(); } else if(!it.hasNext()) { return false; } else { return true; } } @Override public Revision<PrimaryKey, Text, Value> next() { if(hasNext()) { return it.next(); } else { return null; } } private void flip() { if(fileIt.hasNext()) { Path file = Paths.get(fileIt.next()); try { it = Segment.load(file).table().iterator(); } catch (SegmentLoadingException e) { flip(); } } } }; } }
Database extends BaseStore implements PermanentStore { public static Iterator<Revision<PrimaryKey, Text, Value>> onDiskStreamingIterator( final String dbStore) { return new ReadOnlyIterator<Revision<PrimaryKey, Text, Value>>() { private final String directory = FileSystem.makePath(dbStore, SEGMENTS_SUBDIRECTORY); private final Iterator<String> fileIt = FileSystem .fileOnlyIterator(directory); private Iterator<Revision<PrimaryKey, Text, Value>> it = null; { flip(); } @Override public boolean hasNext() { if(it == null) { return false; } else if(!it.hasNext() && fileIt.hasNext()) { flip(); return hasNext(); } else if(!it.hasNext()) { return false; } else { return true; } } @Override public Revision<PrimaryKey, Text, Value> next() { if(hasNext()) { return it.next(); } else { return null; } } private void flip() { if(fileIt.hasNext()) { Path file = Paths.get(fileIt.next()); try { it = Segment.load(file).table().iterator(); } catch (SegmentLoadingException e) { flip(); } } } }; } Database();  Database(Path directory);  Database(String directory); }
Database extends BaseStore implements PermanentStore { public static Iterator<Revision<PrimaryKey, Text, Value>> onDiskStreamingIterator( final String dbStore) { return new ReadOnlyIterator<Revision<PrimaryKey, Text, Value>>() { private final String directory = FileSystem.makePath(dbStore, SEGMENTS_SUBDIRECTORY); private final Iterator<String> fileIt = FileSystem .fileOnlyIterator(directory); private Iterator<Revision<PrimaryKey, Text, Value>> it = null; { flip(); } @Override public boolean hasNext() { if(it == null) { return false; } else if(!it.hasNext() && fileIt.hasNext()) { flip(); return hasNext(); } else if(!it.hasNext()) { return false; } else { return true; } } @Override public Revision<PrimaryKey, Text, Value> next() { if(hasNext()) { return it.next(); } else { return null; } } private void flip() { if(fileIt.hasNext()) { Path file = Paths.get(fileIt.next()); try { it = Segment.load(file).table().iterator(); } catch (SegmentLoadingException e) { flip(); } } } }; } Database();  Database(Path directory);  Database(String directory); static Iterator<Revision<PrimaryKey, Text, Value>> onDiskStreamingIterator(
            final String dbStore); @Override void accept(Write write); @Override void accept(Write write, boolean sync); @Override Map<Long, String> audit(long record); @Override Map<Long, String> audit(String key, long record); @Override Map<TObject, Set<Long>> browse(String key); @Override Map<TObject, Set<Long>> browse(String key, long timestamp); @Override Map<Long, Set<TObject>> chronologize(String key, long record,
            long start, long end); @Override boolean contains(long record); @Override Map<Long, Set<TObject>> doExplore(long timestamp, String key,
            Operator operator, TObject... values); @Override Map<Long, Set<TObject>> doExplore(String key, Operator operator,
            TObject... values); String dump(String id); @Override Set<TObject> gather(String key, long record); @Override Set<TObject> gather(String key, long record, long timestamp); @Restricted String getBackingStore(); @ManagedOperation List<String> getDumpList(); @Override Memory memory(); @Override Set<Long> search(String key, String query); @Override Map<String, Set<TObject>> select(long record); @Override Map<String, Set<TObject>> select(long record, long timestamp); @Override Set<TObject> select(String key, long record); @Override Set<TObject> select(String key, long record, long timestamp); @Override void start(); @Override void stop(); @Override void sync(); @GuardedBy("triggerSync(boolean)") void triggerSync(); @Override boolean verify(String key, TObject value, long record); @Override boolean verify(String key, TObject value, long record,
            long timestamp); }
Database extends BaseStore implements PermanentStore { public static Iterator<Revision<PrimaryKey, Text, Value>> onDiskStreamingIterator( final String dbStore) { return new ReadOnlyIterator<Revision<PrimaryKey, Text, Value>>() { private final String directory = FileSystem.makePath(dbStore, SEGMENTS_SUBDIRECTORY); private final Iterator<String> fileIt = FileSystem .fileOnlyIterator(directory); private Iterator<Revision<PrimaryKey, Text, Value>> it = null; { flip(); } @Override public boolean hasNext() { if(it == null) { return false; } else if(!it.hasNext() && fileIt.hasNext()) { flip(); return hasNext(); } else if(!it.hasNext()) { return false; } else { return true; } } @Override public Revision<PrimaryKey, Text, Value> next() { if(hasNext()) { return it.next(); } else { return null; } } private void flip() { if(fileIt.hasNext()) { Path file = Paths.get(fileIt.next()); try { it = Segment.load(file).table().iterator(); } catch (SegmentLoadingException e) { flip(); } } } }; } Database();  Database(Path directory);  Database(String directory); static Iterator<Revision<PrimaryKey, Text, Value>> onDiskStreamingIterator(
            final String dbStore); @Override void accept(Write write); @Override void accept(Write write, boolean sync); @Override Map<Long, String> audit(long record); @Override Map<Long, String> audit(String key, long record); @Override Map<TObject, Set<Long>> browse(String key); @Override Map<TObject, Set<Long>> browse(String key, long timestamp); @Override Map<Long, Set<TObject>> chronologize(String key, long record,
            long start, long end); @Override boolean contains(long record); @Override Map<Long, Set<TObject>> doExplore(long timestamp, String key,
            Operator operator, TObject... values); @Override Map<Long, Set<TObject>> doExplore(String key, Operator operator,
            TObject... values); String dump(String id); @Override Set<TObject> gather(String key, long record); @Override Set<TObject> gather(String key, long record, long timestamp); @Restricted String getBackingStore(); @ManagedOperation List<String> getDumpList(); @Override Memory memory(); @Override Set<Long> search(String key, String query); @Override Map<String, Set<TObject>> select(long record); @Override Map<String, Set<TObject>> select(long record, long timestamp); @Override Set<TObject> select(String key, long record); @Override Set<TObject> select(String key, long record, long timestamp); @Override void start(); @Override void stop(); @Override void sync(); @GuardedBy("triggerSync(boolean)") void triggerSync(); @Override boolean verify(String key, TObject value, long record); @Override boolean verify(String key, TObject value, long record,
            long timestamp);  }