@SuppressWarnings("unchecked")
    private static <R> QueryResult<R> runRangeQuery(
        final Query<R> query,
        final PositionBound positionBound,
        final QueryConfig config,
        final StateStore store
    ) {
        if (!(store instanceof KeyValueStore)) {
            return QueryResult.forUnknownQueryType(query, store);
        }
        final KeyValueStore<Bytes, byte[]> kvStore = (KeyValueStore<Bytes, byte[]>) store;
        final RangeQuery<Bytes, byte[]> rangeQuery = (RangeQuery<Bytes, byte[]>) query;
        final Optional<Bytes> lowerRange = rangeQuery.getLowerBound();
        final Optional<Bytes> upperRange = rangeQuery.getUpperBound();
        final ResultOrder order = rangeQuery.resultOrder();
        final KeyValueIterator<Bytes, byte[]> iterator;
        try {
            if (!lowerRange.isPresent() && !upperRange.isPresent() && !order.equals(ResultOrder.DESCENDING)) {
                iterator = kvStore.all();
            } else if (!order.equals(ResultOrder.DESCENDING)) {
                iterator = kvStore.range(lowerRange.orElse(null), upperRange.orElse(null));
            } else if (!lowerRange.isPresent() && !upperRange.isPresent()) {
                iterator = kvStore.reverseAll();
            } else {
                iterator = kvStore.reverseRange(lowerRange.orElse(null), upperRange.orElse(null));
            }
            final R result = (R) iterator;
            return QueryResult.forResult(result);
        } catch (final Exception e) {
            final String message = parseStoreException(e, store, query);
            return QueryResult.forFailure(
                FailureReason.STORE_EXCEPTION,
                message
            );
        }
    }
a utility interface to facilitate stores' query dispatch logic,
allowing them to generically store query execution logic as the values
in a map.
streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java