QueryScanner extends GafferScanner implements RegionScanner { protected static List<GafferScannerProcessor> createProcessors( final Scan scan, final Schema schema, final ElementSerialisation serialisation) { final List<GafferScannerProcessor> processors = new ArrayList<>(); final Set<Class<? extends GafferScannerProcessor>> extraProcessors = getExtraProcessors(scan); final View view = getView(scan); if (null != view) { processors.add(new GroupFilterProcessor(view)); if (extraProcessors.remove(ElementDedupeFilterProcessor.class)) { processors.add(new ElementDedupeFilterProcessor(view.hasEntities(), view.hasEdges(), getDirectedType(scan))); } } if (schema.isAggregationEnabled()) { processors.add(new StoreAggregationProcessor(serialisation, schema)); } processors.add(new ValidationProcessor(schema)); if (null != view) { processors.add(new PreAggregationFilterProcessor(view)); if (schema.isAggregationEnabled()) { processors.add(new QueryAggregationProcessor(serialisation, schema, view)); } processors.add(new PostAggregationFilterProcessor(view)); } if (!extraProcessors.isEmpty()) { throw new RuntimeException("Unrecognised extra processors: " + extraProcessors); } return processors; } QueryScanner(final RegionScanner scanner,
                        final Scan scan,
                        final Schema schema,
                        final ElementSerialisation serialisation); @Override HRegionInfo getRegionInfo(); @Override boolean isFilterDone(); @Override boolean reseek(final byte[] row); @Override long getMaxResultSize(); @Override long getMvccReadPoint(); @Override int getBatch(); @Override boolean nextRaw(final List<Cell> output); @Override boolean nextRaw(final List<Cell> output, final ScannerContext scannerContext);  }
@Test public void shouldConstructProcessorsWhenViewIsNull() throws OperationException, IOException { final Scan scan = mock(Scan.class); given(scan.getAttribute(HBaseStoreConstants.VIEW)).willReturn(null); final List<GafferScannerProcessor> processors = QueryScanner.createProcessors(scan, SCHEMA, serialisation); assertEquals(2, processors.size()); int i = 0; assertTrue(processors.get(i) instanceof StoreAggregationProcessor); assertEquals(SCHEMA, ((StoreAggregationProcessor) processors.get(i)).getSchema()); i++; assertTrue(processors.get(i) instanceof ValidationProcessor); assertEquals(SCHEMA, ((ValidationProcessor) processors.get(i)).getSchema()); }
@Test public void shouldThrowErrorWhenInvalidExtras() throws OperationException, IOException { final Scan scan = mock(Scan.class); given(scan.getAttribute(HBaseStoreConstants.VIEW)).willReturn(null); given(scan.getAttribute(HBaseStoreConstants.EXTRA_PROCESSORS)).willReturn(StringUtil.toCsv(ElementDedupeFilterProcessor.class)); try { QueryScanner.createProcessors(scan, SCHEMA, serialisation); fail("Exception expected"); } catch (final RuntimeException e) { assertNotNull(e.getMessage()); } }