@Override
    public StreamEvent query(StateEvent matchingEvent, CompiledCondition compiledCondition,
                             CompiledSelection compiledSelection, Attribute[] outputAttributes)
            throws ConnectionUnavailableException {
        findMatchingEvent = matchingEvent;
        updateStoreTableSize();

        // handle condition type convs
        ComplexEventChunk<StreamEvent> streamEventComplexEventChunk = new ComplexEventChunk<>();
        RecordStoreCompiledCondition recordStoreCompiledCondition;
        RecordStoreCompiledSelection recordStoreCompiledSelection;
        CompiledConditionWithCache compiledConditionWithCache = null;
        CompiledSelectionWithCache compiledSelectionWithCache = null;
        StreamEvent cacheResults;

        if (cacheEnabled) {
            RecordStoreCompiledCondition compiledConditionTemp = (RecordStoreCompiledCondition) compiledCondition;
            compiledConditionWithCache = (CompiledConditionWithCache)
                    compiledConditionTemp.getCompiledCondition();
            recordStoreCompiledCondition = new RecordStoreCompiledCondition(
                    compiledConditionTemp.variableExpressionExecutorMap,
                    compiledConditionWithCache.getStoreCompileCondition(),
                    compiledConditionTemp.getSiddhiQueryContext());

            compiledSelectionWithCache = (CompiledSelectionWithCache) compiledSelection;
            recordStoreCompiledSelection = compiledSelectionWithCache.recordStoreCompiledSelection;
        } else {
            recordStoreCompiledSelection = ((RecordStoreCompiledSelection) compiledSelection);
            recordStoreCompiledCondition = ((RecordStoreCompiledCondition) compiledCondition);
        }

        Map<String, Object> parameterMap = new HashMap<>();
        for (Map.Entry<String, ExpressionExecutor> entry :
                recordStoreCompiledCondition.variableExpressionExecutorMap.entrySet()) {
            parameterMap.put(entry.getKey(), entry.getValue().execute(matchingEvent));
        }
        for (Map.Entry<String, ExpressionExecutor> entry :
                recordStoreCompiledSelection.variableExpressionExecutorMap.entrySet()) {
            parameterMap.put(entry.getKey(), entry.getValue().execute(matchingEvent));
        }

        Iterator<Object[]> records;
        if (cacheEnabled) {
            readWriteLock.writeLock().lock();
            try {
                // when store is smaller than max cache size
                if (storeTableSize <= maxCacheSize && !queryStoreWithoutCheckingCache.get()) {
                    // return results from cache
                    cacheResults = cacheTable.find(compiledConditionWithCache.getCacheCompileCondition(),
                            matchingEvent);
                    if (log.isDebugEnabled()) {
                        log.debug(siddhiAppContext.getName() + "-" +
                                recordStoreCompiledCondition.getSiddhiQueryContext().getName() +
                                ": store table size is smaller than max cache. Sending results from cache");
                    }
                    if (cacheResults == null) {
                        return null;
                    }
                    return executeSelectorOnCacheResults(compiledSelectionWithCache, cacheResults,
                            matchingEvent.getStreamEvent(0));
                } else { // when store is bigger than max cache size
                    if (log.isDebugEnabled() && !queryStoreWithoutCheckingCache.get()) {
                        log.debug(siddhiAppContext.getName() + "-" +
                                recordStoreCompiledCondition.getSiddhiQueryContext().getName() +
                                ": store table size is bigger than cache.");
                    }
                    if (compiledConditionWithCache.isRouteToCache() && !queryStoreWithoutCheckingCache.get()) {
                        if (log.isDebugEnabled()) {
                            log.debug(siddhiAppContext.getName() + "-" +
                                    recordStoreCompiledCondition.getSiddhiQueryContext().
                                            getName() + ": cache constraints satisfied. Checking cache");
                        }
                        // if query conrains all primary keys and has == only for them
                        cacheResults = cacheTable.find(compiledConditionWithCache.getCacheCompileCondition(),
                                matchingEvent);
                        if (cacheResults != null) {
                            if (log.isDebugEnabled()) {
                                log.debug(siddhiAppContext.getName() + "-" + recordStoreCompiledCondition.
                                        getSiddhiQueryContext().getName() + ": cache hit. Sending results from cache");
                            }
                            return executeSelectorOnCacheResults(compiledSelectionWithCache, cacheResults,
                                    matchingEvent.getStreamEvent(0));
                        }

                        if (log.isDebugEnabled()) {
                            log.debug(siddhiAppContext.getName() + "-" +
                                    recordStoreCompiledCondition.getSiddhiQueryContext().
                                            getName() + ": cache miss. Loading from store");
                        }
                        // read all fields of missed entry from store

                        Iterator<Object[]> recordsFromSelectAll;
                        if (recordTableHandler != null) {
                            recordsFromSelectAll = recordTableHandler.query(matchingEvent.getTimestamp(), parameterMap,
                                    recordStoreCompiledCondition.getCompiledCondition(), compiledSelectionForSelectAll,
                                    outputAttributes);
                        } else {
                            recordsFromSelectAll = query(parameterMap,
                                    recordStoreCompiledCondition.getCompiledCondition(),
                                    compiledSelectionForSelectAll, outputAttributes);
                        }
                        if (recordsFromSelectAll == null || !recordsFromSelectAll.hasNext()) {
                            if (log.isDebugEnabled()) {
                                log.debug(siddhiAppContext.getName() + "-" + recordStoreCompiledCondition.
                                        getSiddhiQueryContext().getName() + ": store also miss. sending null");
                            }
                            return null;
                        }
                        Object[] recordSelectAll = recordsFromSelectAll.next();
                        StreamEvent streamEvent = storeEventPool.newInstance();
                        streamEvent.setOutputData(new Object[outputAttributes.length]);
                        System.arraycopy(recordSelectAll, 0, streamEvent.getOutputData(), 0, recordSelectAll.length);

                        if (cacheTable.size() == maxCacheSize) {
                            ((CacheTable) cacheTable).deleteOneEntryUsingCachePolicy();
                        }
                        ((CacheTable) cacheTable).addStreamEventUptoMaxSize(streamEvent);
                        if (log.isDebugEnabled()) {
                            log.debug(siddhiAppContext.getName() + "-" +
                                    recordStoreCompiledCondition.getSiddhiQueryContext().getName() +
                                    ": sending results from cache after loading from store");
                        }
                        cacheResults = cacheTable.find(compiledConditionWithCache.getCacheCompileCondition(),
                                matchingEvent);
                        return executeSelectorOnCacheResults(compiledSelectionWithCache, cacheResults,
                                matchingEvent.getStreamEvent(0));
                    }
                }
            } finally {
                readWriteLock.writeLock().unlock();
            }
        }
        if (log.isDebugEnabled() && !queryStoreWithoutCheckingCache.get()) {
            log.debug(siddhiAppContext.getName() + "-" + recordStoreCompiledCondition.getSiddhiQueryContext().
                    getName() + ": sending results from store");
        }
        // query conditions are not satisfied check from store/ cache not enabled
        if (recordTableHandler != null) {
            records = recordTableHandler.query(matchingEvent.getTimestamp(), parameterMap,
                    recordStoreCompiledCondition.getCompiledCondition(),
                    recordStoreCompiledSelection.compiledSelection, outputAttributes);
        } else {
            records = query(parameterMap, recordStoreCompiledCondition.getCompiledCondition(),
                    recordStoreCompiledSelection.compiledSelection, outputAttributes);
        }
        addStreamEventToChunk(outputAttributes, streamEventComplexEventChunk, records);
        return streamEventComplexEventChunk.getFirst();
    }
An abstract implementation of table. Abstract implementation will handle {@link ComplexEventChunk} so that
developer can directly work with event data.
modules/siddhi-core/src/main/java/io/siddhi/core/table/record/AbstractQueryableRecordTable.java