private ComplexEventChunk<ComplexEvent> processGroupBy(ComplexEventChunk complexEventChunk) {
        complexEventChunk.reset();
        ComplexEventChunk<ComplexEvent> currentComplexEventChunk = new ComplexEventChunk<ComplexEvent>
                ();

        synchronized (this) {
            int limitCount = 0;
            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 ((event.getType() == StreamEvent.Type.CURRENT && currentOn) || (event.getType() ==
                                    StreamEvent.Type.EXPIRED && expiredOn)) {
                                if (!(havingConditionExecutor != null && !havingConditionExecutor.execute(event))) {
                                    complexEventChunk.remove();
                                    if (limit == SiddhiConstants.UNKNOWN_STATE) {
                                        currentComplexEventChunk.add(new GroupedComplexEvent(groupByKey, event));
                                    } else {
                                        if (limitCount < limit) {
                                            currentComplexEventChunk.add(new GroupedComplexEvent(groupByKey, event));
                                            limitCount++;
                                        }
                                    }
                                }
                            }
                        } finally {
                            SiddhiAppContext.stopGroupByFlow();
                        }
                        break;
                    case TIMER:
                        break;
                    case RESET:
                        for (AttributeProcessor attributeProcessor : attributeProcessorList) {
                            attributeProcessor.process(event);
                        }
                        break;
                }

            }
        }
        if (isOrderBy) {
            orderEventChunk(complexEventChunk);
        }
        if (offset != SiddhiConstants.UNKNOWN_STATE) {
            offsetEventChunk(complexEventChunk);
        }
        if (limit != SiddhiConstants.UNKNOWN_STATE) {
            limitEventChunk(complexEventChunk);
        }
        currentComplexEventChunk.reset();
        if (currentComplexEventChunk.hasNext()) {
            return currentComplexEventChunk;
        }
        return null;
    }
Processor implementation representing selector portion of the Siddhi query.
modules/siddhi-core/src/main/java/io/siddhi/core/query/selector/QuerySelector.java