@Test public void testLimitedWriter() { final View<TestEntity> range = ds .fromAfter(NAMES[0], "1").to(NAMES[0], "5") .fromAfter(NAMES[1], "1").to(NAMES[1], "5"); DatasetWriter<TestEntity> writer = range.newWriter(); try { writer.write(newTestEntity("3", "3")); writer.write(newTestEntity("5", "5")); } finally { writer.close(); } }
@Override public DatasetWriter<E> newWriter() { final EntityBatch<E> wrappedWriter = dataset.getDao().newBatch(); if (constraints.isUnbounded()) { wrappedWriter.initialize(); return wrappedWriter; } final StorageKey partitionStratKey = new StorageKey(dataset.getDescriptor().getPartitionStrategy()); AbstractDatasetWriter<E> writer = new AbstractFlushableDatasetWriter<E>() { private Constraints.KeyPredicate keyPredicate = constraints.toKeyPredicate(); @Override public void initialize() { wrappedWriter.initialize(); } @Override public void write(E entity) { StorageKey key = getAccessor().keyFor(entity, constraints.getProvidedValues(), partitionStratKey); if (!keyPredicate.apply(key)) { throw new IllegalArgumentException("View does not contain entity: " + entity); } wrappedWriter.write(entity); } @Override public void flush() { wrappedWriter.flush(); } @Override public void close() { wrappedWriter.close(); } @Override public boolean isOpen() { return wrappedWriter.isOpen(); } }; writer.initialize(); return writer; }
DaoView extends AbstractRefinableView<E> implements InputFormatAccessor<E> { @Override public DatasetWriter<E> newWriter() { final EntityBatch<E> wrappedWriter = dataset.getDao().newBatch(); if (constraints.isUnbounded()) { wrappedWriter.initialize(); return wrappedWriter; } final StorageKey partitionStratKey = new StorageKey(dataset.getDescriptor().getPartitionStrategy()); AbstractDatasetWriter<E> writer = new AbstractFlushableDatasetWriter<E>() { private Constraints.KeyPredicate keyPredicate = constraints.toKeyPredicate(); @Override public void initialize() { wrappedWriter.initialize(); } @Override public void write(E entity) { StorageKey key = getAccessor().keyFor(entity, constraints.getProvidedValues(), partitionStratKey); if (!keyPredicate.apply(key)) { throw new IllegalArgumentException("View does not contain entity: " + entity); } wrappedWriter.write(entity); } @Override public void flush() { wrappedWriter.flush(); } @Override public void close() { wrappedWriter.close(); } @Override public boolean isOpen() { return wrappedWriter.isOpen(); } }; writer.initialize(); return writer; } }
DaoView extends AbstractRefinableView<E> implements InputFormatAccessor<E> { @Override public DatasetWriter<E> newWriter() { final EntityBatch<E> wrappedWriter = dataset.getDao().newBatch(); if (constraints.isUnbounded()) { wrappedWriter.initialize(); return wrappedWriter; } final StorageKey partitionStratKey = new StorageKey(dataset.getDescriptor().getPartitionStrategy()); AbstractDatasetWriter<E> writer = new AbstractFlushableDatasetWriter<E>() { private Constraints.KeyPredicate keyPredicate = constraints.toKeyPredicate(); @Override public void initialize() { wrappedWriter.initialize(); } @Override public void write(E entity) { StorageKey key = getAccessor().keyFor(entity, constraints.getProvidedValues(), partitionStratKey); if (!keyPredicate.apply(key)) { throw new IllegalArgumentException("View does not contain entity: " + entity); } wrappedWriter.write(entity); } @Override public void flush() { wrappedWriter.flush(); } @Override public void close() { wrappedWriter.close(); } @Override public boolean isOpen() { return wrappedWriter.isOpen(); } }; writer.initialize(); return writer; } DaoView(DaoDataset<E> dataset, Class<E> type); private  DaoView(DaoView<E> view, Constraints constraints); private  DaoView(DaoView<?> view, Schema schema, Class<E> type); }
DaoView extends AbstractRefinableView<E> implements InputFormatAccessor<E> { @Override public DatasetWriter<E> newWriter() { final EntityBatch<E> wrappedWriter = dataset.getDao().newBatch(); if (constraints.isUnbounded()) { wrappedWriter.initialize(); return wrappedWriter; } final StorageKey partitionStratKey = new StorageKey(dataset.getDescriptor().getPartitionStrategy()); AbstractDatasetWriter<E> writer = new AbstractFlushableDatasetWriter<E>() { private Constraints.KeyPredicate keyPredicate = constraints.toKeyPredicate(); @Override public void initialize() { wrappedWriter.initialize(); } @Override public void write(E entity) { StorageKey key = getAccessor().keyFor(entity, constraints.getProvidedValues(), partitionStratKey); if (!keyPredicate.apply(key)) { throw new IllegalArgumentException("View does not contain entity: " + entity); } wrappedWriter.write(entity); } @Override public void flush() { wrappedWriter.flush(); } @Override public void close() { wrappedWriter.close(); } @Override public boolean isOpen() { return wrappedWriter.isOpen(); } }; writer.initialize(); return writer; } DaoView(DaoDataset<E> dataset, Class<E> type); private  DaoView(DaoView<E> view, Constraints constraints); private  DaoView(DaoView<?> view, Schema schema, Class<E> type); @Override DatasetReader<E> newReader(); @Override DatasetWriter<E> newWriter(); @Override InputFormat<E, Void> getInputFormat(Configuration conf); }
DaoView extends AbstractRefinableView<E> implements InputFormatAccessor<E> { @Override public DatasetWriter<E> newWriter() { final EntityBatch<E> wrappedWriter = dataset.getDao().newBatch(); if (constraints.isUnbounded()) { wrappedWriter.initialize(); return wrappedWriter; } final StorageKey partitionStratKey = new StorageKey(dataset.getDescriptor().getPartitionStrategy()); AbstractDatasetWriter<E> writer = new AbstractFlushableDatasetWriter<E>() { private Constraints.KeyPredicate keyPredicate = constraints.toKeyPredicate(); @Override public void initialize() { wrappedWriter.initialize(); } @Override public void write(E entity) { StorageKey key = getAccessor().keyFor(entity, constraints.getProvidedValues(), partitionStratKey); if (!keyPredicate.apply(key)) { throw new IllegalArgumentException("View does not contain entity: " + entity); } wrappedWriter.write(entity); } @Override public void flush() { wrappedWriter.flush(); } @Override public void close() { wrappedWriter.close(); } @Override public boolean isOpen() { return wrappedWriter.isOpen(); } }; writer.initialize(); return writer; } DaoView(DaoDataset<E> dataset, Class<E> type); private  DaoView(DaoView<E> view, Constraints constraints); private  DaoView(DaoView<?> view, Schema schema, Class<E> type); @Override DatasetReader<E> newReader(); @Override DatasetWriter<E> newWriter(); @Override InputFormat<E, Void> getInputFormat(Configuration conf);  }