private ComplexEventChunk processInBatchGroupBy(ComplexEventChunk complexEventChunk) {
        Map<String, ComplexEvent> groupedEvents = new LinkedHashMap<String, ComplexEvent>();
        complexEventChunk.reset();

        synchronized (this) {
            while (complexEventChunk.hasNext()) {
                ComplexEvent event = complexEventChunk.next();
                switch (event.getType()) {

                    case CURRENT:
                    case EXPIRED:
                        eventPopulator.populateStateEvent(event);
                        String groupByKey = groupByKeyGenerator.constructEventKey(event);
                        SiddhiAppContext.startGroupByFlow(groupByKey);
                        try {
                            for (AttributeProcessor attributeProcessor : attributeProcessorList) {
                                attributeProcessor.process(event);
                            }
                            if (!(havingConditionExecutor != null && !havingConditionExecutor.execute(event))) {
                                if ((event.getType() == StreamEvent.Type.CURRENT && currentOn) || (event.getType() ==
                                        StreamEvent.Type.EXPIRED && expiredOn)) {
                                    complexEventChunk.remove();
                                    groupedEvents.put(groupByKey, event);
                                }
                            }
                        } finally {
                            SiddhiAppContext.stopGroupByFlow();
                        }
                        break;
                    case TIMER:
                        break;
                    case RESET:
                        for (AttributeProcessor attributeProcessor : attributeProcessorList) {
                            attributeProcessor.process(event);
                        }
                        break;
                }
            }
        }

        if (groupedEvents.size() != 0) {
            complexEventChunk.clear();
            for (Map.Entry<String, ComplexEvent> groupedEventEntry : groupedEvents.entrySet()) {
                complexEventChunk.add(new GroupedComplexEvent(groupedEventEntry.getKey(),
                        groupedEventEntry.getValue()));
            }
            if (isOrderBy) {
                orderEventChunk(complexEventChunk);
            }
            if (offset != SiddhiConstants.UNKNOWN_STATE) {
                offsetEventChunk(complexEventChunk);
            }
            if (limit != SiddhiConstants.UNKNOWN_STATE) {
                limitEventChunk(complexEventChunk);
            }
            complexEventChunk.reset();
            return complexEventChunk;
        }
        return null;
    }
Processor implementation representing selector portion of the Siddhi query.
modules/siddhi-core/src/main/java/io/siddhi/core/query/selector/QuerySelector.java