From c8778ce527575c545a864ccbc3d98e3502fbb2a2 Mon Sep 17 00:00:00 2001
From: Ramin Gharib <ramin.gharib@bakdata.com>
Date: Thu, 15 Sep 2022 08:26:58 +0200
Subject: [PATCH] Add range to topic directive with range data fetcher (#62)

Closes #57 and #59
---
 .../api/client/DefaultMirrorClient.java       |  25 +++-
 .../quick/common/api/client/MirrorClient.java |  19 ++-
 .../api/client/PartitionedMirrorClient.java   |  40 ++++--
 .../common/api/model/mirror/MirrorHost.java   |  10 +-
 .../api/model/mirror/MirrorHostTest.java      |  66 +++++++++
 e2e/functional/range/query-range.gql          |   8 ++
 e2e/functional/range/result-range.json        |  20 +++
 e2e/functional/range/schema.gql               |  19 +++
 e2e/functional/range/tests.bats               |  66 +++++++++
 e2e/functional/range/user-requests.json       |  62 ++++++++
 .../directives/topic/TopicDirective.java      |  36 ++++-
 .../topic/rule/fetcher/DataFetcherRules.java  |   5 +-
 .../rule/fetcher/ListArgumentFetcherRule.java |   4 +-
 .../topic/rule/fetcher/QueryFetcherRule.java  |   2 +-
 ...tchRule.java => QueryListFetcherRule.java} |   2 +-
 .../topic/rule/fetcher/RangeFetcherRule.java  |  83 +++++++++++
 .../topic/rule/validation/KeyInformation.java |  76 +++-------
 .../rule/validation/MultipleArguments.java    |  37 -----
 .../topic/rule/validation/RangeArguments.java |  80 +++++++++++
 .../rule/validation/ValidationRules.java      |   9 +-
 .../rule/validation/ValidationUtility.java    |  77 ++++++++++
 .../gateway/fetcher/DataFetcherClient.java    |   3 +
 .../quick/gateway/fetcher/DeferFetcher.java   |   2 +-
 .../quick/gateway/fetcher/FetcherFactory.java |  22 +--
 .../gateway/fetcher/KeyFieldFetcher.java      |   8 +-
 .../fetcher/MirrorDataFetcherClient.java      |   6 +
 .../gateway/fetcher/MutationFetcher.java      |   3 +-
 .../gateway/fetcher/QueryListFetcher.java     |   9 +-
 .../gateway/fetcher/RangeQueryFetcher.java    |  77 ++++++++++
 .../gateway/ControllerReturnSchemaTest.java   |   7 +-
 .../gateway/ControllerUpdateSchemaTest.java   |   5 +-
 .../quick/gateway/GatewayConfigTest.java      |  13 +-
 .../quick/gateway/GatewayInitializerTest.java |   2 -
 .../gateway/GraphQLQueryExecutionTest.java    |  81 +++++++++--
 .../gateway/GraphQLSchemaGeneratorTest.java   |  90 ++++++++----
 .../quick/gateway/GraphQLSecurityTest.java    |   6 +-
 .../quick/gateway/GraphQLTestUtil.java        |  16 ++-
 .../fetcher/QueryListArgumentFetcherTest.java |  18 +--
 .../gateway/fetcher/QueryListFetcherTest.java |   1 -
 .../fetcher/RangeQueryFetcherTest.java        | 132 ++++++++++++++++++
 .../quick/gateway/security/ApiKeyTest.java    |   2 -
 .../src/test/resources/application-test.yaml  |   5 +-
 .../shouldConvertQueryWithRange.graphql       |  19 +++
 ...shouldConvertQueryWithRangeOnField.graphql |  12 ++
 ...ndInputNameDifferentInNonQueryType.graphql |   1 +
 ...rgAndInputNameDifferentInQueryType.graphql |   1 +
 ...ConvertIfMissingKeyInfoInBasicType.graphql |   1 +
 ...ConvertIfMissingKeyInfoInQueryType.graphql |   1 +
 ...onvertIfMutationDoesNotHaveTwoArgs.graphql |   1 +
 ...tConvertIfRangeToArgumentIsMissing.graphql |  19 +++
 ...IfKeyArgumentInRangeQueryIsMissing.graphql |  18 +++
 ...CovertIfRangeFromArgumentIsMissing.graphql |  19 +++
 ...ldNotCovertIfRangeIsDefinedOnField.graphql |  13 ++
 ...tIfReturnTypeOfRangeQueryIsNotList.graphql |  20 +++
 .../execution/shouldExecuteRange.graphql      |  16 +++
 .../execution/shouldExecuteRangeQuery.graphql |   5 +
 56 files changed, 1167 insertions(+), 233 deletions(-)
 create mode 100644 common/src/test/java/com/bakdata/quick/common/api/model/mirror/MirrorHostTest.java
 create mode 100644 e2e/functional/range/query-range.gql
 create mode 100644 e2e/functional/range/result-range.json
 create mode 100644 e2e/functional/range/schema.gql
 create mode 100644 e2e/functional/range/tests.bats
 create mode 100644 e2e/functional/range/user-requests.json
 rename gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/{QueryListFetchRule.java => QueryListFetcherRule.java} (97%)
 create mode 100644 gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/RangeFetcherRule.java
 delete mode 100644 gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/MultipleArguments.java
 create mode 100644 gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/RangeArguments.java
 create mode 100644 gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/ValidationUtility.java
 create mode 100644 gateway/src/main/java/com/bakdata/quick/gateway/fetcher/RangeQueryFetcher.java
 create mode 100644 gateway/src/test/java/com/bakdata/quick/gateway/fetcher/RangeQueryFetcherTest.java
 create mode 100644 gateway/src/test/resources/schema/conversion/shouldConvertQueryWithRange.graphql
 create mode 100644 gateway/src/test/resources/schema/conversion/shouldConvertQueryWithRangeOnField.graphql
 create mode 100644 gateway/src/test/resources/schema/conversion/shouldNotConvertIfRangeToArgumentIsMissing.graphql
 create mode 100644 gateway/src/test/resources/schema/conversion/shouldNotCovertIfKeyArgumentInRangeQueryIsMissing.graphql
 create mode 100644 gateway/src/test/resources/schema/conversion/shouldNotCovertIfRangeFromArgumentIsMissing.graphql
 create mode 100644 gateway/src/test/resources/schema/conversion/shouldNotCovertIfRangeIsDefinedOnField.graphql
 create mode 100644 gateway/src/test/resources/schema/conversion/shouldNotCovertIfReturnTypeOfRangeQueryIsNotList.graphql
 create mode 100644 gateway/src/test/resources/schema/execution/shouldExecuteRange.graphql
 create mode 100644 gateway/src/test/resources/schema/execution/shouldExecuteRangeQuery.graphql

diff --git a/common/src/main/java/com/bakdata/quick/common/api/client/DefaultMirrorClient.java b/common/src/main/java/com/bakdata/quick/common/api/client/DefaultMirrorClient.java
index 360f24d3..8acd111f 100644
--- a/common/src/main/java/com/bakdata/quick/common/api/client/DefaultMirrorClient.java
+++ b/common/src/main/java/com/bakdata/quick/common/api/client/DefaultMirrorClient.java
@@ -40,27 +40,27 @@ public class DefaultMirrorClient<K, V> implements MirrorClient<K, V> {
     /**
      * Constructor for the client.
      *
-     * @param topicName     name of the topic the mirror is deployed
-     * @param client        http client
-     * @param mirrorConfig  configuration of the mirror host
+     * @param topicName name of the topic the mirror is deployed
+     * @param client http client
+     * @param mirrorConfig configuration of the mirror host
      * @param valueResolver the value's {@link TypeResolver}
      * @param requestManager a manager for sending requests to the mirror and processing responses
      */
     public DefaultMirrorClient(final String topicName, final HttpClient client, final MirrorConfig mirrorConfig,
-                               final TypeResolver<V> valueResolver, final MirrorRequestManager requestManager) {
+        final TypeResolver<V> valueResolver, final MirrorRequestManager requestManager) {
         this(new MirrorHost(topicName, mirrorConfig), client, valueResolver, requestManager);
     }
 
     /**
      * Constructor that can be used when the mirror client is based on an IP or other non-standard host.
      *
-     * @param mirrorHost   host to use
-     * @param client       http client
+     * @param mirrorHost host to use
+     * @param client http client
      * @param typeResolver the value's {@link TypeResolver}
      * @param requestManager a manager for sending requests to the mirror and processing responses
      */
     public DefaultMirrorClient(final MirrorHost mirrorHost, final HttpClient client, final TypeResolver<V> typeResolver,
-                               final MirrorRequestManager requestManager) {
+        final MirrorRequestManager requestManager) {
         this.host = mirrorHost;
         this.parser = new MirrorValueParser<>(typeResolver, client.objectMapper());
         this.mirrorRequestManager = requestManager;
@@ -91,6 +91,17 @@ public List<V> fetchValues(final List<K> keys) {
             Collections.emptyList());
     }
 
+    @Override
+    @Nullable
+    public List<V> fetchRange(final K key, final String from, final String rangeTo) {
+        final ResponseWrapper response = this.mirrorRequestManager.makeRequest(
+            this.host.forRange(key.toString(), from, rangeTo)
+        );
+        return Objects.requireNonNullElse(
+            this.mirrorRequestManager.processResponse(response, this.parser::deserializeList),
+            Collections.emptyList());
+    }
+
     @Override
     public boolean exists(final K key) {
         return this.fetchValue(key) != null;
diff --git a/common/src/main/java/com/bakdata/quick/common/api/client/MirrorClient.java b/common/src/main/java/com/bakdata/quick/common/api/client/MirrorClient.java
index ee76ece4..6691f556 100644
--- a/common/src/main/java/com/bakdata/quick/common/api/client/MirrorClient.java
+++ b/common/src/main/java/com/bakdata/quick/common/api/client/MirrorClient.java
@@ -29,7 +29,7 @@
 public interface MirrorClient<K, V> {
 
     /**
-     * fetches the value of the given key from the mirror topic.
+     * Fetches the value of the given key from the mirror topic.
      *
      * @param key a key to be fetched
      * @return a list of values. If the requested mirror responds with a NOT_FOUND code the function returns null.
@@ -38,14 +38,14 @@ public interface MirrorClient<K, V> {
     V fetchValue(final K key);
 
     /**
-     * fetches all the values of a mirror topic.
+     * Fetches all the values of a mirror topic.
      *
      * @return returns a list of all values in a topic. null.
      */
     List<V> fetchAll();
 
     /**
-     * fetches the values of a list of keys from the mirror topic.
+     * Fetches the values of a list of keys from the mirror topic.
      *
      * @param keys list of keys to be fetched
      * @return a list of values. If the requested mirror responds with a NOT_FOUND code the function returns null.
@@ -54,7 +54,18 @@ public interface MirrorClient<K, V> {
     List<V> fetchValues(final List<K> keys);
 
     /**
-     * checks if a key exists in mirror topic.
+     * Fetches a range of a given key from the mirror topic.
+     *
+     * @param key a key to be fetched
+     * @param from lower bound of the range field
+     * @param to higher bound of the range field
+     * @return a list of values.
+     */
+    @Nullable
+    List<V> fetchRange(final K key, final String from, final String to);
+
+    /**
+     * Checks if a key exists in mirror topic.
      *
      * @return True/False if key exists in mirror topic
      */
diff --git a/common/src/main/java/com/bakdata/quick/common/api/client/PartitionedMirrorClient.java b/common/src/main/java/com/bakdata/quick/common/api/client/PartitionedMirrorClient.java
index d7578307..63cb5c21 100644
--- a/common/src/main/java/com/bakdata/quick/common/api/client/PartitionedMirrorClient.java
+++ b/common/src/main/java/com/bakdata/quick/common/api/client/PartitionedMirrorClient.java
@@ -37,8 +37,8 @@
 import org.apache.kafka.common.serialization.Serde;
 
 /**
- * MirrorClient that has access to information about partition-host mapping.
- * This information enables it to efficiently route requests in the case when there is more than one mirror replica.
+ * MirrorClient that has access to information about partition-host mapping. This information enables it to efficiently
+ * route requests in the case when there is more than one mirror replica.
  *
  * @param <K> key type
  * @param <V> value type
@@ -58,19 +58,18 @@ public class PartitionedMirrorClient<K, V> implements MirrorClient<K, V> {
     private final List<MirrorHost> knownHosts;
 
     /**
-     * Next to its default task of instantiation PartitionHost, it takes responsibility for
-     * creating several business objects and initializing the PartitionRouter with a mapping
-     * retrieved from StreamController.
+     * Next to its default task of instantiation PartitionHost, it takes responsibility for creating several business
+     * objects and initializing the PartitionRouter with a mapping retrieved from StreamController.
      *
-     * @param mirrorHost      mirror host to use
-     * @param client          http client
-     * @param keySerde        the serde for the key
-     * @param valueResolver   the value's {@link TypeResolver}
+     * @param mirrorHost mirror host to use
+     * @param client http client
+     * @param keySerde the serde for the key
+     * @param valueResolver the value's {@link TypeResolver}
      * @param partitionFinder strategy for finding partitions
      */
     public PartitionedMirrorClient(final MirrorHost mirrorHost, final HttpClient client,
-                                   final Serde<K> keySerde, final TypeResolver<V> valueResolver,
-                                   final PartitionFinder partitionFinder) {
+        final Serde<K> keySerde, final TypeResolver<V> valueResolver,
+        final PartitionFinder partitionFinder) {
         this.topicName = mirrorHost.getHost();
         this.streamsStateHost = StreamsStateHost.fromMirrorHost(mirrorHost);
         this.client = client;
@@ -121,14 +120,29 @@ public List<V> fetchValues(final List<K> keys) {
         return keys.stream().map(this::fetchValue).collect(Collectors.toList());
     }
 
+    @Override
+    @Nullable
+    public List<V> fetchRange(final K key, final String from, final String to) {
+        final MirrorHost currentKeyHost = Objects.requireNonNull(this.router.findHost(key),
+            String.format("Could not find the a Mirror host %s for key %s", this.topicName, key));
+        final ResponseWrapper response = this.requestManager
+            .makeRequest(Objects.requireNonNull(currentKeyHost).forRange(key.toString(), from, to));
+        if (response.isUpdateCacheHeaderSet()) {
+            log.debug("The update header has been set for host {} and key {}. Updating router info.",
+                this.topicName,
+                key);
+            this.updateRouterInfo();
+        }
+        return this.requestManager.processResponse(response, this.parser::deserializeList);
+    }
+
     @Override
     public boolean exists(final K key) {
         return this.fetchValue(key) != null;
     }
 
     /**
-     * Responsible for fetching the information about the partition - host mapping from
-     * the mirror.
+     * Responsible for fetching the information about the partition - host mapping from the mirror.
      *
      * @return a mapping between a partition (a number) and a corresponding host
      */
diff --git a/common/src/main/java/com/bakdata/quick/common/api/model/mirror/MirrorHost.java b/common/src/main/java/com/bakdata/quick/common/api/model/mirror/MirrorHost.java
index c91fb25a..7ac8ddb2 100644
--- a/common/src/main/java/com/bakdata/quick/common/api/model/mirror/MirrorHost.java
+++ b/common/src/main/java/com/bakdata/quick/common/api/model/mirror/MirrorHost.java
@@ -32,7 +32,7 @@ public class MirrorHost {
     /**
      * Default constructor.
      *
-     * @param host   the host of the mirror. This can be a service name or an IP.
+     * @param host the host of the mirror. This can be a service name or an IP.
      * @param config mirror config to use. This can set the service prefix and REST path.
      */
     public MirrorHost(final String host, final MirrorConfig config) {
@@ -70,6 +70,14 @@ public String forAll() {
         return url;
     }
 
+    /**
+     * Generates a URL for fetching a range of keys.
+     */
+    public String forRange(final String key, final String from, final String to) {
+        return String.format("http://%s%s/%s/range/%s?from=%s&to=%s", this.config.getPrefix(), this.host,
+            this.config.getPath(), key, from, to);
+    }
+
     /**
      * Generates a URL without any keys.
      */
diff --git a/common/src/test/java/com/bakdata/quick/common/api/model/mirror/MirrorHostTest.java b/common/src/test/java/com/bakdata/quick/common/api/model/mirror/MirrorHostTest.java
new file mode 100644
index 00000000..3a8d4de2
--- /dev/null
+++ b/common/src/test/java/com/bakdata/quick/common/api/model/mirror/MirrorHostTest.java
@@ -0,0 +1,66 @@
+/*
+ *    Copyright 2022 bakdata GmbH
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+package com.bakdata.quick.common.api.model.mirror;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.bakdata.quick.common.config.MirrorConfig;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+class MirrorHostTest {
+
+    private static final String MIRROR_HOST_PREFIX = "quick-mirror";
+    private static final String MIRROR_HOST_PATH = "mirror";
+
+    @Test
+    void shouldConstructCorrectUrlForKeyRequest() {
+        final MirrorHost mirrorHost = new MirrorHost("test-for-key", new MirrorConfig());
+        final String actual = mirrorHost.forKey("give-me-key");
+        final String url = "http://%s-test-for-key/%s/give-me-key";
+        final String expected = String.format(url, MIRROR_HOST_PREFIX, MIRROR_HOST_PATH);
+        assertThat(actual).isEqualTo(expected);
+    }
+
+    @Test
+    void shouldConstructCorrectUrlForKeysRequest() {
+        final MirrorHost mirrorHost = new MirrorHost("test-for-keys", new MirrorConfig());
+        final String actual = mirrorHost.forKeys(List.of("test-1", "test-2", "test-3"));
+        final String url = "http://%s-test-for-keys/%s/keys?ids=test-1,test-2,test-3";
+        final String expected = String.format(url, MIRROR_HOST_PREFIX, MIRROR_HOST_PATH);
+        assertThat(actual).isEqualTo(expected);
+    }
+
+    @Test
+    void shouldConstructCorrectUrlForAllRequest() {
+        final MirrorHost mirrorHost = new MirrorHost("test-for-all", new MirrorConfig());
+        final String actual = mirrorHost.forAll();
+        final String url = "http://%s-test-for-all/%s";
+        final String expected = String.format(url, MIRROR_HOST_PREFIX, MIRROR_HOST_PATH);
+        assertThat(actual).isEqualTo(expected);
+    }
+
+    @Test
+    void shouldConstructCorrectUrlForRangeRequest() {
+        final MirrorHost mirrorHost = new MirrorHost("test-for-rage", new MirrorConfig());
+        final String actual = mirrorHost.forRange("test-key", "range-field-from", "range-field-to");
+        final String url = "http://%s-test-for-rage/%s/range/%s?from=%s&to=%s";
+        final String expected =
+            String.format(url, MIRROR_HOST_PREFIX, MIRROR_HOST_PATH, "test-key", "range-field-from", "range-field-to");
+        assertThat(actual).isEqualTo(expected);
+    }
+}
diff --git a/e2e/functional/range/query-range.gql b/e2e/functional/range/query-range.gql
new file mode 100644
index 00000000..a3fc8852
--- /dev/null
+++ b/e2e/functional/range/query-range.gql
@@ -0,0 +1,8 @@
+query {
+    userRequests(userId: 123, timestampFrom:1, timestampTo:3) {
+        userId
+        serviceId
+        requests
+        success
+    }
+}
diff --git a/e2e/functional/range/result-range.json b/e2e/functional/range/result-range.json
new file mode 100644
index 00000000..9839c826
--- /dev/null
+++ b/e2e/functional/range/result-range.json
@@ -0,0 +1,20 @@
+[
+  {
+    "userId": 123,
+    "serviceId": "abc",
+    "requests": 5,
+    "success": 2
+  },
+  {
+    "userId": 123,
+    "serviceId": "def",
+    "requests": 8,
+    "success": 4
+  },
+  {
+    "userId": 123,
+    "serviceId": "def",
+    "requests": 6,
+    "success": 5
+  }
+]
diff --git a/e2e/functional/range/schema.gql b/e2e/functional/range/schema.gql
new file mode 100644
index 00000000..cafb9bc1
--- /dev/null
+++ b/e2e/functional/range/schema.gql
@@ -0,0 +1,19 @@
+type Query {
+    userRequests(
+        userId: Int
+        timestampFrom: Int
+        timestampTo: Int
+    ): [UserRequests]
+    @topic(name: "user-request-range",
+        keyArgument: "userId",
+        rangeFrom: "timestampFrom",
+        rangeTo: "timestampTo")
+}
+
+type UserRequests {
+    userId: Int
+    serviceId: Int
+    timestamp: Int
+    requests: Int
+    success: Int
+}
diff --git a/e2e/functional/range/tests.bats b/e2e/functional/range/tests.bats
new file mode 100644
index 00000000..d1b0acb0
--- /dev/null
+++ b/e2e/functional/range/tests.bats
@@ -0,0 +1,66 @@
+#!/usr/bin/env ./test/libs/bats/bin/bats
+# shellcheck shell=bats
+
+CONTENT_TYPE="content-type:application/json"
+API_KEY="X-API-Key:${X_API_KEY}"
+TOPIC="user-request-range"
+TYPE="UserRequests"
+GATEWAY="range-gateway-test"
+INGEST_URL="${HOST}/ingest/${TOPIC}"
+GRAPHQL_URL="${HOST}/gateway/${GATEWAY}/graphql"
+GRAPHQL_CLI="gql-cli ${GRAPHQL_URL} -H ${API_KEY}"
+
+
+setup() {
+    if [ "$BATS_TEST_NUMBER" -eq 1 ]; then
+        printf "creating context for %s\n" "$HOST"
+        printf "with API_KEY: %s\n" "${X_API_KEY}"
+        quick context create --host "${HOST}" --key "${X_API_KEY}"
+    fi
+}
+
+@test "should deploy product-gateway" {
+    run quick gateway create ${GATEWAY}
+    echo "$output"
+    sleep 30
+    [ "$status" -eq 0 ]
+    [ "$output" = "Create gateway ${GATEWAY} (this may take a few seconds)" ]
+}
+
+@test "should apply schema to range-gateway" {
+    run quick gateway apply ${GATEWAY} -f schema.gql
+    echo "$output"
+    [ "$status" -eq 0 ]
+    [ "$output" = "Applied schema to gateway ${GATEWAY}" ]
+}
+
+#TODO: Remove skip after implementing the Mirror and query service
+@test "should create user-request-range topic with key integer and value schema" {
+    skip
+    run quick topic create ${TOPIC} --key-type integer --value-type schema --schema "${GATEWAY}.${TYPE}" --range-field timestamp --point
+    echo "$output"
+    [ "$status" -eq 0 ]
+    [ "$output" = "Created new topic ${TOPIC}" ]
+}
+
+@test "should ingest valid data in user-request-range" {
+    skip
+    curl --request POST --url "${INGEST_URL}" --header "${CONTENT_TYPE}" --header "${API_KEY}" --data "@./user-requests.json"
+}
+
+@test "should retrieve range of inserted items" {
+    skip
+    sleep 30
+    result="$(${GRAPHQL_CLI} < query-single.gql | jq -j .userRequests)"
+    expected="$(cat result-single.json)"
+    echo "$result"
+    [ "$result" = "$expected" ]
+}
+
+teardown() {
+    if [[ "${#BATS_TEST_NAMES[@]}" -eq "$BATS_TEST_NUMBER" ]]; then
+        quick gateway delete ${GATEWAY}
+        #TODO: Uncomment the topic deletion after the Range Mirror implementation is ready
+        # quick topic delete ${TOPIC}
+    fi
+}
diff --git a/e2e/functional/range/user-requests.json b/e2e/functional/range/user-requests.json
new file mode 100644
index 00000000..478944d5
--- /dev/null
+++ b/e2e/functional/range/user-requests.json
@@ -0,0 +1,62 @@
+[
+  {
+    "key": 123,
+    "value": {
+      "userId": 123,
+      "serviceId": "abc",
+      "timestamp": 1,
+      "requests": 5,
+      "success": 2
+    }
+  },
+  {
+    "key": 123,
+    "value": {
+      "userId": 123,
+      "serviceId": "def",
+      "timestamp": 2,
+      "requests": 8,
+      "success": 4
+    }
+  },
+  {
+    "key": 123,
+    "value": {
+      "userId": 123,
+      "serviceId": "def",
+      "timestamp": 3,
+      "requests": 6,
+      "success": 5
+    }
+  },
+  {
+    "key": 123,
+    "value": {
+      "userId": 123,
+      "serviceId": "abc",
+      "timestamp": 4,
+      "requests": 6,
+      "success": 5
+    }
+  },
+  {
+    "key": 456,
+    "value": {
+      "userId": 456,
+      "serviceId": "hij",
+      "timestamp": 1,
+      "requests": 10,
+      "success": 8
+    }
+  },
+  {
+    "key": 456,
+    "value": {
+      "userId": 456,
+      "serviceId": "abc",
+      "timestamp": 2,
+      "requests": 3,
+      "success": 3
+    }
+  }
+]
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/TopicDirective.java b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/TopicDirective.java
index 28e7897b..052c52ca 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/TopicDirective.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/TopicDirective.java
@@ -37,7 +37,9 @@
  * directive @topic(
  *     name: String!,
  *     keyArgument: String,
- *     keyField: String
+ *     keyField: String,
+ *     rangeFrom: String,
+ *     rangeTo: String
  * ) on FIELD_DEFINITION
  * }</pre>
  */
@@ -49,6 +51,8 @@ public final class TopicDirective implements QuickDirective {
     private static final String TOPIC_NAME_ARG_NAME = "name";
     private static final String KEY_ARGUMENT_ARG_NAME = "keyArgument";
     private static final String KEY_FIELD_ARG_NAME = "keyField";
+    private static final String RANGE_FROM_ARG_NAME = "rangeFrom";
+    private static final String RANGE_TO_ARG_NAME = "rangeTo";
 
     static {
         DEFINITION = DirectiveDefinition.newDirectiveDefinition()
@@ -68,6 +72,16 @@ public final class TopicDirective implements QuickDirective {
                     .name(KEY_FIELD_ARG_NAME)
                     .type(STRING)
                     .build())
+            .inputValueDefinition(
+                InputValueDefinition.newInputValueDefinition()
+                    .name(RANGE_FROM_ARG_NAME)
+                    .type(STRING)
+                    .build())
+            .inputValueDefinition(
+                InputValueDefinition.newInputValueDefinition()
+                    .name(RANGE_TO_ARG_NAME)
+                    .type(STRING)
+                    .build())
             .directiveLocation(
                 DirectiveLocation.newDirectiveLocation()
                     .name(FIELD_DEFINITION.name())
@@ -80,13 +94,19 @@ public final class TopicDirective implements QuickDirective {
     private final String keyArgument;
     @Nullable
     private final String keyField;
+    @Nullable
+    private final String rangeFrom;
+    @Nullable
+    private final String rangeTo;
 
     private TopicDirective(final String topicName, @Nullable final String keyArgument,
-        @Nullable final String keyField) {
+        @Nullable final String keyField, @Nullable final String rangeFrom, @Nullable final String rangeTo) {
         Objects.requireNonNull(topicName);
         this.topicName = topicName;
         this.keyArgument = keyArgument;
         this.keyField = keyField;
+        this.rangeFrom = rangeFrom;
+        this.rangeTo = rangeTo;
     }
 
     /**
@@ -97,7 +117,9 @@ public static TopicDirective fromArguments(final Collection<? extends GraphQLArg
             "Topic name in topic directive must be set");
         final String keyArgument = QuickDirective.extractArgument(arguments, KEY_ARGUMENT_ARG_NAME);
         final String keyField = QuickDirective.extractArgument(arguments, KEY_FIELD_ARG_NAME);
-        return new TopicDirective(topicName, keyArgument, keyField);
+        final String rangeFrom = QuickDirective.extractArgument(arguments, RANGE_FROM_ARG_NAME);
+        final String rangeTo = QuickDirective.extractArgument(arguments, RANGE_TO_ARG_NAME);
+        return new TopicDirective(topicName, keyArgument, keyField, rangeFrom, rangeTo);
     }
 
     public boolean hasKeyArgument() {
@@ -107,4 +129,12 @@ public boolean hasKeyArgument() {
     public boolean hasKeyField() {
         return this.keyField != null;
     }
+
+    public boolean hasRangeFrom() {
+        return this.rangeFrom != null;
+    }
+
+    public boolean hasRangeTo() {
+        return this.rangeTo != null;
+    }
 }
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/DataFetcherRules.java b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/DataFetcherRules.java
index bbee4165..a9b8f0ce 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/DataFetcherRules.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/DataFetcherRules.java
@@ -31,11 +31,12 @@ public class DataFetcherRules implements TopicDirectiveRules {
         rules = List.of(
             new SubscriptionRule(),
             new QueryFetcherRule(),
-            new QueryListFetchRule(),
+            new QueryListFetcherRule(),
             new ListArgumentFetcherRule(),
             new ModificationRule(),
             new ModificationListRule(),
-            new MutationRule()
+            new MutationRule(),
+            new RangeFetcherRule()
         );
     }
 
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/ListArgumentFetcherRule.java b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/ListArgumentFetcherRule.java
index d73b3b1f..54c80e9d 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/ListArgumentFetcherRule.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/ListArgumentFetcherRule.java
@@ -64,7 +64,9 @@ public List<DataFetcherSpecification> extractDataFetchers(final TopicDirectiveCo
 
     @Override
     public boolean isValid(final TopicDirectiveContext context) {
-        return context.getTopicDirective().getKeyArgument() != null
+        return context.getTopicDirective().hasKeyArgument()
+            && !context.getTopicDirective().hasRangeFrom()
+            && !context.getTopicDirective().hasRangeTo()
             && context.isListType()
             && !context.getParentContainerName().equals(GraphQLUtils.SUBSCRIPTION_TYPE)
             && GraphQLTypeUtil.isList(context.getEnvironment().getElement().getType());
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/QueryFetcherRule.java b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/QueryFetcherRule.java
index eb74ddc5..ff82d2e7 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/QueryFetcherRule.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/QueryFetcherRule.java
@@ -65,7 +65,7 @@ public List<DataFetcherSpecification> extractDataFetchers(final TopicDirectiveCo
 
     @Override
     public boolean isValid(final TopicDirectiveContext context) {
-        return context.getTopicDirective().getKeyArgument() != null
+        return context.getTopicDirective().hasKeyArgument()
             && !context.isListType()
             && !context.getParentContainerName().equals(GraphQLUtils.SUBSCRIPTION_TYPE);
     }
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/QueryListFetchRule.java b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/QueryListFetcherRule.java
similarity index 97%
rename from gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/QueryListFetchRule.java
rename to gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/QueryListFetcherRule.java
index 7dce7d1c..65ba1aac 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/QueryListFetchRule.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/QueryListFetcherRule.java
@@ -42,7 +42,7 @@
  *
  * @see com.bakdata.quick.gateway.fetcher.QueryListFetcher
  */
-public class QueryListFetchRule implements DataFetcherRule {
+public class QueryListFetcherRule implements DataFetcherRule {
 
     @Override
     public List<DataFetcherSpecification> extractDataFetchers(final TopicDirectiveContext context) {
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/RangeFetcherRule.java b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/RangeFetcherRule.java
new file mode 100644
index 00000000..132f38b1
--- /dev/null
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/fetcher/RangeFetcherRule.java
@@ -0,0 +1,83 @@
+/*
+ *    Copyright 2022 bakdata GmbH
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+package com.bakdata.quick.gateway.directives.topic.rule.fetcher;
+
+import com.bakdata.quick.common.graphql.GraphQLUtils;
+import com.bakdata.quick.gateway.DataFetcherSpecification;
+import com.bakdata.quick.gateway.directives.topic.TopicDirectiveContext;
+import graphql.schema.DataFetcher;
+import graphql.schema.FieldCoordinates;
+import graphql.schema.GraphQLTypeUtil;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Rule for range query fetcher.
+ *
+ * <p>
+ * <h2>Example:</h2>
+ * <pre>{@code
+ * type Query {
+ *     userRequests(
+ *         userId: Int
+ *         timestampFrom: Int
+ *         timestampTo: Int
+ *     ): [UserRequests] @topic(name: "user-request-range",
+ *                              keyArgument: "userId",
+ *                              rangeFrom: "timestampFrom",
+ *                              rangeTo: "timestampTo")
+ * }
+ *
+ * type UserRequests {
+ *     userId: Int
+ *     serviceId: Int
+ *     timestamp: Int
+ *     requests: Int
+ *     success: Int
+ * }
+ * }</pre>
+ *
+ * @see com.bakdata.quick.gateway.fetcher.RangeQueryFetcher
+ */
+public class RangeFetcherRule implements DataFetcherRule {
+    @Override
+    public List<DataFetcherSpecification> extractDataFetchers(final TopicDirectiveContext context) {
+        Objects.requireNonNull(context.getTopicDirective().getKeyArgument());
+        Objects.requireNonNull(context.getTopicDirective().getRangeFrom());
+        Objects.requireNonNull(context.getTopicDirective().getRangeTo());
+        final DataFetcher<?> dataFetcher = context.getFetcherFactory().rangeFetcher(
+            context.getTopicDirective().getTopicName(),
+            context.getTopicDirective().getKeyArgument(),
+            context.getTopicDirective().getRangeFrom(),
+            context.getTopicDirective().getRangeTo(),
+            context.isNullable()
+        );
+        final FieldCoordinates coordinates = this.currentCoordinates(context);
+        return List.of(DataFetcherSpecification.of(coordinates, dataFetcher));
+    }
+
+    @Override
+    public boolean isValid(final TopicDirectiveContext context) {
+        return context.getTopicDirective().hasKeyArgument()
+            && context.getTopicDirective().hasRangeFrom()
+            && context.getTopicDirective().hasRangeTo()
+            && !context.getParentContainerName().equals(GraphQLUtils.SUBSCRIPTION_TYPE)
+            && context.getParentContainerName().equals(GraphQLUtils.QUERY_TYPE)
+            && GraphQLTypeUtil.isList(context.getEnvironment().getElement().getType());
+    }
+}
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/KeyInformation.java b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/KeyInformation.java
index 10086444..784da27c 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/KeyInformation.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/KeyInformation.java
@@ -16,6 +16,8 @@
 
 package com.bakdata.quick.gateway.directives.topic.rule.validation;
 
+import static com.bakdata.quick.gateway.directives.topic.rule.validation.ValidationUtility.makeCheckForKeyArgument;
+
 import com.bakdata.quick.common.graphql.GraphQLUtils;
 import com.bakdata.quick.gateway.directives.topic.TopicDirectiveContext;
 import graphql.language.InputValueDefinition;
@@ -28,10 +30,10 @@
  * Validation for {@link com.bakdata.quick.gateway.directives.topic.TopicDirective}
  *
  * <p>
- * When a user declares a non-mutation or a non-subscription type with a return type that is not a list,
- * they have to provide key information - either keyField or keyArgument.
- * The following example presents the correct way of providing key information (key argument is present
- * when the return type is not a list, and it is the same as the input name (id):
+ * When a user declares a non-mutation or a non-subscription type with a return type that is not a list, they have to
+ * provide key information - either keyField or keyArgument. The following example presents the correct way of providing
+ * key information (key argument is present when the return type is not a list, and it is the same as the input name
+ * (id):
  * <pre>{@code
  * type Product {
  *     id: ID!
@@ -40,10 +42,10 @@
  * type Query {
  *     getProduct(id: ID): Product @topic(name: "product-topic", keyArgument: "id")
  * }
-}</pre>
- * On the other hand, the example below depicts a code example in which the rules for providing
- * key information are violated (the directive in the product field does not have the key argument at all,
- * and the one in the url field has an incorrect value of the key argument - it should be productId):
+ * }</pre>
+ * On the other hand, the example below depicts a code example in which the rules for providing key information are
+ * violated (the directive in the product field does not have the key argument at all, and the one in the url field has
+ * an incorrect value of the key argument - it should be productId):
  * <pre>{@code
  * type Query {
  *     getProduct(productId: ID): ProductInfo
@@ -58,62 +60,20 @@ public class KeyInformation implements ValidationRule {
 
     @Override
     public Optional<String> validate(final TopicDirectiveContext context) {
-        if (this.checkIfBasicContextPropertiesAreInvalid(context)) {
+        if (checkIfBasicContextPropertiesAreInvalid(context)) {
             return Optional.of("When the return type is not a list for a non-mutation and non-subscription type,"
-                    + " key information (keyArgument or keyField) is needed.");
+                + " key information (keyArgument or keyField) is needed.");
         }
         // additional check for key arguments
-        final Optional<String> additionalKeyArgCheckResult = makeCheckForKeyArgument(context);
-        if (additionalKeyArgCheckResult.isPresent()) {
-            return additionalKeyArgCheckResult;
-        }
+        return makeCheckForKeyArgument(context);
         // TODO: additional check for key field
-        return Optional.empty();
-    }
-
-    private Optional<String> makeCheckForKeyArgument(final TopicDirectiveContext context) {
-        if (context.getTopicDirective().getKeyArgument() != null) {
-            final boolean inputNameAndKeyArgsMatch;
-            if (context.getParentContainerName().equals(GraphQLUtils.QUERY_TYPE)) {
-                inputNameAndKeyArgsMatch = this.checkIfInputNameAndKeyArgMatchInQueryType(context);
-            } else {
-                inputNameAndKeyArgsMatch = this.checkIfInputNameAndKeyArgMatchInNonQueryType(context);
-            }
-            if (!inputNameAndKeyArgsMatch) {
-                return Optional.of("Key argument has to be identical to the input name.");
-            }
-        }
-        return Optional.empty();
     }
 
-    private boolean checkIfBasicContextPropertiesAreInvalid(final TopicDirectiveContext context) {
+    private static boolean checkIfBasicContextPropertiesAreInvalid(final TopicDirectiveContext context) {
         return !context.getParentContainerName().equals(GraphQLUtils.MUTATION_TYPE)
-                && !context.getParentContainerName().equals(GraphQLUtils.SUBSCRIPTION_TYPE)
-                && !context.isListType()
-                && !context.getTopicDirective().hasKeyArgument()
-                && !context.getTopicDirective().hasKeyField();
-    }
-
-    private boolean checkIfInputNameAndKeyArgMatchInQueryType(final TopicDirectiveContext context) {
-        final String keyArg = context.getTopicDirective().getKeyArgument();
-        return context.getEnvironment().getElement().getDefinition().getInputValueDefinitions()
-                .stream()
-                .map(InputValueDefinition::getName)
-                .anyMatch(name -> Objects.equals(name, keyArg));
-    }
-
-    private boolean checkIfInputNameAndKeyArgMatchInNonQueryType(final TopicDirectiveContext context) {
-        final Optional<TypeDefinition> queryTypeDef = context.getEnvironment().getRegistry().getType("Query");
-        if (queryTypeDef.isEmpty()) {
-            throw new IllegalStateException("Something went wrong - The query type is mandatory.");
-        } else {
-            final String keyArg = context.getTopicDirective().getKeyArgument();
-            // We can cast here because we know that Query must be of the type ObjectTypeDefinition
-            return ((ObjectTypeDefinition) queryTypeDef.get()).getFieldDefinitions()
-                    .stream()
-                    .flatMap(fieldDef -> fieldDef.getInputValueDefinitions().stream())
-                    .map(InputValueDefinition::getName)
-                    .anyMatch(name -> Objects.equals(name, keyArg));
-        }
+            && !context.getParentContainerName().equals(GraphQLUtils.SUBSCRIPTION_TYPE)
+            && !context.isListType()
+            && !context.getTopicDirective().hasKeyArgument()
+            && !context.getTopicDirective().hasKeyField();
     }
 }
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/MultipleArguments.java b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/MultipleArguments.java
deleted file mode 100644
index 516c9de3..00000000
--- a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/MultipleArguments.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- *    Copyright 2022 bakdata GmbH
- *
- *    Licensed under the Apache License, Version 2.0 (the "License");
- *    you may not use this file except in compliance with the License.
- *    You may obtain a copy of the License at
- *
- *        http://www.apache.org/licenses/LICENSE-2.0
- *
- *    Unless required by applicable law or agreed to in writing, software
- *    distributed under the License is distributed on an "AS IS" BASIS,
- *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- *    See the License for the specific language governing permissions and
- *    limitations under the License.
- */
-
-package com.bakdata.quick.gateway.directives.topic.rule.validation;
-
-import com.bakdata.quick.gateway.directives.topic.TopicDirectiveContext;
-import java.util.Optional;
-
-/**
- * Validation for {@link com.bakdata.quick.gateway.directives.topic.TopicDirective}
- *
- * <p>
- * The query type only supports single arguments. Therefore, if the size is bigger than one this should be considered as
- * not supported.
- */
-public class MultipleArguments implements ValidationRule {
-    @Override
-    public Optional<String> validate(final TopicDirectiveContext context) {
-        if (context.getEnvironment().getElement().getArguments().size() > 1) {
-            return Optional.of("Only single arguments are supported");
-        }
-        return Optional.empty();
-    }
-}
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/RangeArguments.java b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/RangeArguments.java
new file mode 100644
index 00000000..d6046c49
--- /dev/null
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/RangeArguments.java
@@ -0,0 +1,80 @@
+/*
+ *    Copyright 2022 bakdata GmbH
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+package com.bakdata.quick.gateway.directives.topic.rule.validation;
+
+import com.bakdata.quick.common.graphql.GraphQLUtils;
+import com.bakdata.quick.gateway.directives.topic.TopicDirectiveContext;
+import java.util.Optional;
+
+/**
+ * Validation for range queries.
+ *
+ * <p>
+ * These rules should apply:
+ * <ol>
+ * <li> The Parent container should be a Query and not Mutation/Subscription
+ * <li> Both rangeFrom and rangeTo fields should exist in the topic directive
+ * <li> A valid keyArgument should exist in the topic directive
+ * <li> The return type of the query should be list
+ * </ol>
+ *
+ * <p>
+ * <h2>Valid schema:</h2>
+ * <pre>{@code
+ * type Query {
+ *     userRequests(
+ *         userId: Int
+ *         timestampFrom: Int
+ *         timestampTo: Int
+ *     ): [UserRequests] @topic(name: "user-request-range",
+ *                              keyArgument: "userId",
+ *                              rangeFrom: "timestampFrom",
+ *                              rangeTo: "timestampTo")
+ * }
+ *
+ * type UserRequests {
+ *     userId: Int
+ *     serviceId: Int
+ *     timestamp: Int
+ *     requests: Int
+ *     success: Int
+ * }
+ * }</pre>
+ */
+public class RangeArguments implements ValidationRule {
+    @Override
+    public Optional<String> validate(final TopicDirectiveContext context) {
+        if (hasRangeFromAndRangeTo(context)) {
+            if (!context.getParentContainerName().equals(GraphQLUtils.QUERY_TYPE)) {
+                return Optional.of("Range queries are only supported on Query types.");
+            } else if (!context.getTopicDirective().hasKeyArgument()) {
+                return Optional.of("You must define a keyArgument.");
+            } else if (!context.isListType()) {
+                return Optional.of("The return type of range queries should be a list.");
+            }
+            return ValidationUtility.makeCheckForKeyArgument(context);
+        } else if (!context.getTopicDirective().hasRangeFrom() && !context.getTopicDirective().hasRangeTo()) {
+            return Optional.empty();
+        }
+        return Optional.of("Both rangeFrom and rangeTo arguments should be set.");
+    }
+
+    private static boolean hasRangeFromAndRangeTo(final TopicDirectiveContext context) {
+        return context.getTopicDirective().hasRangeFrom()
+            && context.getTopicDirective().hasRangeTo();
+    }
+}
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/ValidationRules.java b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/ValidationRules.java
index a3b3fed3..97a80e25 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/ValidationRules.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/ValidationRules.java
@@ -30,10 +30,11 @@ public class ValidationRules implements TopicDirectiveRules {
 
     static {
         VALIDATION_RULES = List.of(
-                new SubscriptionList(),
-                new ExclusiveArguments(),
-                new KeyInformation(),
-                new MutationRequiresTwoArguments()
+            new SubscriptionList(),
+            new ExclusiveArguments(),
+            new KeyInformation(),
+            new MutationRequiresTwoArguments(),
+            new RangeArguments()
         );
     }
 
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/ValidationUtility.java b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/ValidationUtility.java
new file mode 100644
index 00000000..a67610c4
--- /dev/null
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/directives/topic/rule/validation/ValidationUtility.java
@@ -0,0 +1,77 @@
+/*
+ *    Copyright 2022 bakdata GmbH
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+package com.bakdata.quick.gateway.directives.topic.rule.validation;
+
+import com.bakdata.quick.common.graphql.GraphQLUtils;
+import com.bakdata.quick.gateway.directives.topic.TopicDirectiveContext;
+import graphql.language.InputValueDefinition;
+import graphql.language.ObjectTypeDefinition;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * Utilities to validate the topic directive.
+ */
+public final class ValidationUtility {
+
+    private ValidationUtility() {
+    }
+
+    /**
+     * Checks if the keyArgument of a topic directive is valid or not.
+     *
+     * @param context Topic directive context
+     * @return Optional empty if the keyArgument is valid
+     */
+    public static Optional<String> makeCheckForKeyArgument(final TopicDirectiveContext context) {
+        if (context.getTopicDirective().getKeyArgument() != null) {
+            final boolean inputNameAndKeyArgsMatch;
+            if (context.getParentContainerName().equals(GraphQLUtils.QUERY_TYPE)) {
+                inputNameAndKeyArgsMatch = checkIfInputNameAndKeyArgMatchInQueryType(context);
+            } else {
+                inputNameAndKeyArgsMatch = checkIfInputNameAndKeyArgMatchInNonQueryType(context);
+            }
+            if (!inputNameAndKeyArgsMatch) {
+                return Optional.of("Key argument has to be identical to the input name.");
+            }
+        }
+        return Optional.empty();
+    }
+
+    private static boolean checkIfInputNameAndKeyArgMatchInQueryType(final TopicDirectiveContext context) {
+        final String keyArg = context.getTopicDirective().getKeyArgument();
+        return context.getEnvironment().getElement().getDefinition().getInputValueDefinitions()
+            .stream()
+            .map(InputValueDefinition::getName)
+            .anyMatch(name -> Objects.equals(name, keyArg));
+    }
+
+    private static boolean checkIfInputNameAndKeyArgMatchInNonQueryType(final TopicDirectiveContext context) {
+        final Optional<?> queryTypeDef = context.getEnvironment().getRegistry().getType("Query");
+        if (queryTypeDef.isEmpty()) {
+            throw new IllegalStateException("Something went wrong - The query type is mandatory.");
+        } else {
+            final String keyArg = context.getTopicDirective().getKeyArgument();
+            // We can cast here because we know that Query must be of the type ObjectTypeDefinition
+            return ((ObjectTypeDefinition) queryTypeDef.get()).getFieldDefinitions()
+                .stream()
+                .flatMap(fieldDef -> fieldDef.getInputValueDefinitions().stream())
+                .map(InputValueDefinition::getName)
+                .anyMatch(name -> Objects.equals(name, keyArg));
+        }
+    }
+}
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/DataFetcherClient.java b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/DataFetcherClient.java
index d189f0a0..7d082069 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/DataFetcherClient.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/DataFetcherClient.java
@@ -67,4 +67,7 @@ public interface DataFetcherClient<V> {
      */
     @Nullable
     List<V> fetchList();
+
+    @Nullable
+    List<V> fetchRange(final String id, final String from, final String to);
 }
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/DeferFetcher.java b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/DeferFetcher.java
index 0e8acb2a..b158d9dd 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/DeferFetcher.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/DeferFetcher.java
@@ -57,7 +57,7 @@ public class DeferFetcher implements DataFetcher<DataFetcherResult<Object>> {
     private static final Object PLACEHOLDER = new Object();
 
     /**
-     * Retrieves an argument from the user query.
+     * Retrieves the value of an argument from the user query.
      *
      * <p>
      * With this, arguments passed through with this class are handled correctly.
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/FetcherFactory.java b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/FetcherFactory.java
index 9dfb7b8f..105b9146 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/FetcherFactory.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/FetcherFactory.java
@@ -55,7 +55,7 @@ public class FetcherFactory {
      */
     @VisibleForTesting
     public FetcherFactory(final KafkaConfig kafkaConfig, final ObjectMapper objectMapper,
-                          final ClientSupplier clientSupplier, final TopicTypeService topicTypeService) {
+        final ClientSupplier clientSupplier, final TopicTypeService topicTypeService) {
         this.kafkaConfig = kafkaConfig;
         this.objectMapper = objectMapper;
         this.clientSupplier = clientSupplier;
@@ -69,14 +69,14 @@ public FetcherFactory(final KafkaConfig kafkaConfig, final ObjectMapper objectMa
      */
     @Inject
     public FetcherFactory(final KafkaConfig kafkaConfig,
-                          final HttpClient client, final MirrorConfig mirrorConfig,
-                          final TopicTypeService topicTypeService) {
+        final HttpClient client, final MirrorConfig mirrorConfig,
+        final TopicTypeService topicTypeService) {
         this(kafkaConfig, client.objectMapper(),
             new DefaultClientSupplier(client, topicTypeService, mirrorConfig), topicTypeService);
     }
 
     public <K, V> DataFetcher<Publisher<V>> subscriptionFetcher(final String topic, final String operationName,
-                                                                @Nullable final String argument) {
+        @Nullable final String argument) {
         final Lazy<QuickTopicData<K, V>> topicData = this.getTopicData(topic);
         return new SubscriptionFetcher<>(this.kafkaConfig, topicData, operationName, argument);
     }
@@ -87,24 +87,30 @@ public <V> DataFetcher<V> queryFetcher(final String topic, final String argument
     }
 
     public <V> DataFetcher<List<V>> queryListFetcher(final String topic, final boolean isNullable,
-                                                     final boolean hasNullableElements) {
+        final boolean hasNullableElements) {
         final DataFetcherClient<V> client = this.clientSupplier.createClient(topic);
         return new QueryListFetcher<>(client, isNullable, hasNullableElements);
     }
 
     public <V> DataFetcher<List<V>> listArgumentFetcher(final String topic, final String argument,
-                                                        final boolean isNullable, final boolean hasNullableElements) {
+        final boolean isNullable, final boolean hasNullableElements) {
         final DataFetcherClient<V> client = this.clientSupplier.createClient(topic);
         return new ListArgumentFetcher<>(argument, client, isNullable, hasNullableElements);
     }
 
+    public <V> DataFetcher<List<V>> rangeFetcher(final String topic, final String argument, final String rangeFrom,
+        final String rangeTo, final boolean isNullable) {
+        final DataFetcherClient<V> client = this.clientSupplier.createClient(topic);
+        return new RangeQueryFetcher<>(argument, client, rangeFrom, rangeTo, isNullable);
+    }
+
     /**
      * Creates a MutationFetcher object.
      *
      * @see MutationFetcher
      */
     public <K, V> DataFetcher<V> mutationFetcher(final String topic, final String keyArgumentName,
-                                                 final String valueArgumentName) {
+        final String valueArgumentName) {
         final Lazy<QuickTopicData<K, V>> data = this.getTopicDataWithTopicTypeService(topic);
         return new MutationFetcher<>(topic,
             keyArgumentName,
@@ -124,7 +130,7 @@ public DataFetcher<Object> keyFieldFetcher(final String topic, final String keyF
     }
 
     public <K, V> SubscriptionProvider<K, V> subscriptionProvider(final String topic, final String operationName,
-                                                                  @Nullable final String argument) {
+        @Nullable final String argument) {
         return new KafkaSubscriptionProvider<>(this.kafkaConfig, this.getTopicData(topic), operationName,
             argument);
     }
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/KeyFieldFetcher.java b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/KeyFieldFetcher.java
index e864dca3..5987d485 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/KeyFieldFetcher.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/KeyFieldFetcher.java
@@ -42,7 +42,7 @@
  *
  * <p>
  * Consider the following schema:
- * <pre>
+ * <pre>{@code
  *   type Query {
  *       findPurchase(purchaseId: ID): Purchase @topic(name: "purchase", keyArgument: "purchaseId")
  *   }
@@ -56,7 +56,7 @@
  *   type Product {
  *       ...
  *   }
- * </pre>
+ * }</pre>
  *
  * <p>
  * First, the data fetcher connected with findPurchase is called and returns a Purchase object with a missing product
@@ -73,8 +73,8 @@ public class KeyFieldFetcher<T> implements DataFetcher<Object> {
      * Constructor.
      *
      * @param objectMapper json handler
-     * @param argument     name of the argument to extract key from
-     * @param client       underlying HTTP mirror client
+     * @param argument name of the argument to extract key from
+     * @param client underlying HTTP mirror client
      */
     public KeyFieldFetcher(final ObjectMapper objectMapper, final String argument, final DataFetcherClient<T> client) {
         this.objectMapper = objectMapper;
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/MirrorDataFetcherClient.java b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/MirrorDataFetcherClient.java
index 7149d578..1b594539 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/MirrorDataFetcherClient.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/MirrorDataFetcherClient.java
@@ -77,6 +77,12 @@ public List<V> fetchList() {
         return this.mirrorClient.get().fetchAll();
     }
 
+    @Override
+    @Nullable
+    public List<V> fetchRange(final String id, final String from, final String to) {
+        return this.mirrorClient.get().fetchRange(id, from, to);
+    }
+
     private PartitionedMirrorClient<String, V> createMirrorClient(final String host,
                                                                   final MirrorConfig mirrorConfig,
                                                                   final HttpClient client,
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/MutationFetcher.java b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/MutationFetcher.java
index 54cba2fe..9a3218a3 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/MutationFetcher.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/MutationFetcher.java
@@ -20,6 +20,7 @@
 import com.bakdata.quick.common.type.QuickTopicData;
 import com.bakdata.quick.common.util.Lazy;
 import com.bakdata.quick.gateway.ingest.KafkaIngestService;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import graphql.GraphQLException;
@@ -71,7 +72,7 @@ public MutationFetcher(final String topic,
 
     @Override
     @Nullable
-    public V get(final DataFetchingEnvironment environment) throws Exception {
+    public V get(final DataFetchingEnvironment environment) throws JsonProcessingException {
         log.debug("Incoming request: Ingest payload for topic {}", this.topic);
 
         final Optional<?> keyInputArgument = DeferFetcher.getArgument(this.keyInputArgumentName, environment);
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/QueryListFetcher.java b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/QueryListFetcher.java
index 813edd48..d3ecf21a 100644
--- a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/QueryListFetcher.java
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/QueryListFetcher.java
@@ -29,10 +29,11 @@
  *
  * <p>
  * Consider the following schema:
- * <pre>
+ * <pre>{@code
  *   type Query {
  *       allPurchases: [Purchase]
  *   }
+ * }
  * </pre>
  *
  * <p>
@@ -47,8 +48,8 @@ public class QueryListFetcher<T> implements DataFetcher<List<T>> {
     /**
      * Standard constructor.
      *
-     * @param dataFetcherClient   mirror http client
-     * @param isNullable          true if list itself can be null
+     * @param dataFetcherClient mirror http client
+     * @param isNullable true if list itself can be null
      * @param hasNullableElements true if list elements can be null
      */
     public QueryListFetcher(final DataFetcherClient<T> dataFetcherClient, final boolean isNullable,
@@ -77,6 +78,4 @@ public List<T> get(final DataFetchingEnvironment environment) {
 
         return values;
     }
-
-
 }
diff --git a/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/RangeQueryFetcher.java b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/RangeQueryFetcher.java
new file mode 100644
index 00000000..e6b430a8
--- /dev/null
+++ b/gateway/src/main/java/com/bakdata/quick/gateway/fetcher/RangeQueryFetcher.java
@@ -0,0 +1,77 @@
+/*
+ *    Copyright 2022 bakdata GmbH
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+package com.bakdata.quick.gateway.fetcher;
+
+import edu.umd.cs.findbugs.annotations.Nullable;
+import graphql.schema.DataFetcher;
+import graphql.schema.DataFetchingEnvironment;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Data Fetcher that takes the query's key, rangeFrom, and rangeTo arguments and fetches values from the mirror's range
+ * index.
+ */
+public class RangeQueryFetcher<T> implements DataFetcher<List<T>> {
+    private final String argument;
+    private final String rangeFrom;
+    private final String rangeTo;
+    private final DataFetcherClient<T> dataFetcherClient;
+    private final boolean isNullable;
+
+    /**
+     * Standard constructor.
+     *
+     * @param argument name of the argument to extract key from
+     * @param dataFetcherClient underlying HTTP mirror client
+     * @param rangeFrom name of the range from field
+     * @param rangeTo name of the range to field
+     * @param isNullable true if list itself can be null
+     */
+    public RangeQueryFetcher(final String argument,
+        final DataFetcherClient<T> dataFetcherClient,
+        final String rangeFrom, final String rangeTo, final boolean isNullable) {
+        this.dataFetcherClient = dataFetcherClient;
+        this.argument = argument;
+        this.rangeFrom = rangeFrom;
+        this.rangeTo = rangeTo;
+        this.isNullable = isNullable;
+    }
+
+    @Override
+    @Nullable
+    public List<T> get(final DataFetchingEnvironment environment) {
+        final Object argumentValue = DeferFetcher.getArgument(this.argument, environment)
+            .orElseThrow(() -> new RuntimeException("Could not find argument " + this.argument));
+        final Object rangeFromValue = DeferFetcher.getArgument(this.rangeFrom, environment)
+            .orElseThrow(() -> new RuntimeException("Could not find argument " + this.rangeFrom));
+        final Object rangeToValue = DeferFetcher.getArgument(this.rangeTo, environment)
+            .orElseThrow(() -> new RuntimeException("Could not find argument " + this.rangeTo));
+
+        final List<T> results = this.dataFetcherClient.fetchRange(argumentValue.toString(), rangeFromValue.toString(),
+            rangeToValue.toString());
+
+        // got null but schema doesn't allow null
+        // semantically, there is no difference between null and an empty list for us in this case
+        // we therefore continue gracefully by simply returning a list and not throwing an exception
+        if (results == null && !this.isNullable) {
+            return Collections.emptyList();
+        }
+
+        return results;
+    }
+}
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/ControllerReturnSchemaTest.java b/gateway/src/test/java/com/bakdata/quick/gateway/ControllerReturnSchemaTest.java
index 8c21ff62..3a7b0e94 100644
--- a/gateway/src/test/java/com/bakdata/quick/gateway/ControllerReturnSchemaTest.java
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/ControllerReturnSchemaTest.java
@@ -26,14 +26,13 @@
 import com.bakdata.quick.common.api.model.TopicWriteType;
 import com.bakdata.quick.common.api.model.gateway.SchemaData;
 import com.bakdata.quick.common.type.QuickTopicType;
-import io.micronaut.context.annotation.Property;
 import io.micronaut.core.type.Argument;
 import io.micronaut.http.HttpMethod;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.client.BlockingHttpClient;
-import io.micronaut.rxjava2.http.client.RxHttpClient;
 import io.micronaut.http.client.annotation.Client;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
+import io.micronaut.rxjava2.http.client.RxHttpClient;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
 import java.io.IOException;
@@ -43,8 +42,7 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-@Property(name = "quick.kafka.bootstrap-server", value = "dummy:1234")
-@Property(name = "quick.kafka.schema-registry-url", value = "http://dummy")
+
 @MicronautTest
 class ControllerReturnSchemaTest {
 
@@ -118,5 +116,4 @@ void shouldThrowErrorIfTypeDoesNotExist() {
                     .startsWith("Type nope does not exist")
             );
     }
-
 }
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/ControllerUpdateSchemaTest.java b/gateway/src/test/java/com/bakdata/quick/gateway/ControllerUpdateSchemaTest.java
index d097fd9f..ddf7d298 100644
--- a/gateway/src/test/java/com/bakdata/quick/gateway/ControllerUpdateSchemaTest.java
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/ControllerUpdateSchemaTest.java
@@ -24,21 +24,18 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.micronaut.context.ApplicationContext;
-import io.micronaut.context.annotation.Property;
 import io.micronaut.http.HttpMethod;
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.client.BlockingHttpClient;
-import io.micronaut.rxjava2.http.client.RxHttpClient;
 import io.micronaut.http.client.annotation.Client;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
+import io.micronaut.rxjava2.http.client.RxHttpClient;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
 import java.util.Optional;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Test;
 
-@Property(name = "quick.kafka.bootstrap-server", value = "dummy:1234")
-@Property(name = "quick.kafka.schema-registry-url", value = "http://dummy")
 @MicronautTest
 class ControllerUpdateSchemaTest {
     @Client("/")
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/GatewayConfigTest.java b/gateway/src/test/java/com/bakdata/quick/gateway/GatewayConfigTest.java
index 787ef192..d482d77e 100644
--- a/gateway/src/test/java/com/bakdata/quick/gateway/GatewayConfigTest.java
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/GatewayConfigTest.java
@@ -18,18 +18,15 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import io.micronaut.context.annotation.Property;
-import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import com.bakdata.quick.common.ConfigUtils;
+import java.util.Map;
 import org.junit.jupiter.api.Test;
 
-@MicronautTest
-@Property(name = "quick.schema.path", value = "/test/path/schema.graphql")
-@Property(name = "quick.kafka.bootstrap-server", value = "dummy:1234")
-@Property(name = "quick.kafka.schema-registry-url", value = "http://dummy")
 class GatewayConfigTest {
     @Test
-    void createConfig(final GatewayConfig config) {
+    void createConfig() {
+        final Map<String, Object> properties = Map.of("quick.schema.path", "/test/path/schema.graphql");
+        final GatewayConfig config = ConfigUtils.createWithProperties(properties, GatewayConfig.class);
         assertThat(config.getPath()).isEqualTo("/test/path/schema.graphql");
     }
-
 }
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/GatewayInitializerTest.java b/gateway/src/test/java/com/bakdata/quick/gateway/GatewayInitializerTest.java
index d483041d..95b8e96a 100644
--- a/gateway/src/test/java/com/bakdata/quick/gateway/GatewayInitializerTest.java
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/GatewayInitializerTest.java
@@ -39,8 +39,6 @@
  * In contrast to files in the test resource directory, they cannot be accessed by the class loader.
  */
 @MicronautTest
-@Property(name = "quick.kafka.bootstrap-server", value = "dummy:1234")
-@Property(name = "quick.kafka.schema-registry-url", value = "http://dummy")
 @Property(name = "quick.definition.path", value = "/definition/schema.graphql")
 class GatewayInitializerTest {
 
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLQueryExecutionTest.java b/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLQueryExecutionTest.java
index 811ea6a2..8479a50b 100644
--- a/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLQueryExecutionTest.java
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLQueryExecutionTest.java
@@ -42,7 +42,7 @@
 import java.util.List;
 import java.util.Map;
 import lombok.Builder;
-import lombok.Data;
+import lombok.Value;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
 
@@ -115,6 +115,38 @@ void shouldExecuteQueryWithSingleField(final TestInfo testInfo) throws IOExcepti
             .containsEntry("productId", "product");
     }
 
+    @Test
+    void shouldExecuteRange(final TestInfo testInfo) throws IOException {
+        final String name = testInfo.getTestMethod().orElseThrow().getName();
+        final Path schemaPath = workingDirectory.resolve(name + ".graphql");
+        final Path queryPath = workingDirectory.resolve(name + "Query.graphql");
+
+        final GraphQLSchema schema = this.generator.create(Files.readString(schemaPath));
+        final GraphQL graphQL = GraphQL.newGraphQL(schema).build();
+
+        final DataFetcherClient<?> dataFetcherClient = this.supplier.getClients().get("user-request-range");
+
+        final List<?> userRequests = List.of(
+            UserRequest.builder().userId(1).timestamp(1).requests(5).build(),
+            UserRequest.builder().userId(1).timestamp(2).requests(10).build(),
+            UserRequest.builder().userId(1).timestamp(3).requests(8).build()
+        );
+
+        when(dataFetcherClient.fetchRange("1", "1", "3")).thenAnswer(invocation -> userRequests);
+
+        final ExecutionResult executionResult = graphQL.execute(Files.readString(queryPath));
+
+        assertThat(executionResult.getErrors()).isEmpty();
+
+        final Map<String, List<Map<String, Object>>> data = executionResult.getData();
+        assertThat(data.get("userRequests"))
+            .isNotNull()
+            .hasSize(3)
+            .satisfies(userRequest -> assertThat(userRequest.get(0).get("requests")).isEqualTo(5))
+            .satisfies(userRequest -> assertThat(userRequest.get(1).get("requests")).isEqualTo(10))
+            .satisfies(userRequest -> assertThat(userRequest.get(2).get("requests")).isEqualTo(8));
+    }
+
     @Test
     void shouldExecuteListQueryWithSingleFieldAndModification(final TestInfo testInfo) throws IOException {
         final String name = testInfo.getTestMethod().orElseThrow().getName();
@@ -280,7 +312,8 @@ void shouldThrowErrorForNonNullableField() throws IOException {
             .hasSize(1)
             .first()
             .satisfies(error -> {
-                assertThat(error.getMessage()).startsWith("The field at path '/findPurchase/productId' was declared as a non null type");
+                assertThat(error.getMessage()).startsWith(
+                    "The field at path '/findPurchase/productId' was declared as a non null type");
                 assertThat(error.getPath()).containsExactly("findPurchase", "productId");
             });
     }
@@ -311,29 +344,49 @@ private void registerTopics() {
             "url-topic",
             new TopicData("url-topic", TopicWriteType.MUTABLE, QuickTopicType.STRING, QuickTopicType.AVRO, "")
         ).blockingAwait();
+
+        this.registryClient.register(
+            "user-request-range",
+            new TopicData("user-request-range", TopicWriteType.MUTABLE, QuickTopicType.INTEGER, QuickTopicType.AVRO,
+                "")
+        ).blockingAwait();
+
+        this.registryClient.register(
+            "info-topic",
+            new TopicData("info-topic", TopicWriteType.MUTABLE, QuickTopicType.INTEGER, QuickTopicType.AVRO,
+                "")
+        ).blockingAwait();
     }
 
-    @Data
+    @Value
     @Builder
     private static class Purchase {
-        private String purchaseId;
-        private String productId;
-        private int amount;
+        String purchaseId;
+        String productId;
+        int amount;
     }
 
-    @Data
+    @Value
     @Builder
     private static class Product {
-        private String productId;
-        private String name;
-        private String description;
-        private Price price;
+        String productId;
+        String name;
+        String description;
+        Price price;
     }
 
-    @Data
+    @Value
     @Builder
     private static class Price {
-        private double total;
-        private String currency;
+        double total;
+        String currency;
+    }
+
+    @Value
+    @Builder
+    private static class UserRequest {
+        int userId;
+        int timestamp;
+        int requests;
     }
 }
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLSchemaGeneratorTest.java b/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLSchemaGeneratorTest.java
index 390bcd84..6f075085 100644
--- a/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLSchemaGeneratorTest.java
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLSchemaGeneratorTest.java
@@ -40,7 +40,6 @@
 import graphql.schema.GraphQLObjectType;
 import graphql.schema.GraphQLScalarType;
 import graphql.schema.GraphQLSchema;
-import io.micronaut.context.annotation.Property;
 import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
 import java.io.IOException;
@@ -52,8 +51,6 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInfo;
 
-@Property(name = "quick.kafka.schema-registry-url", value = "mock://dummy")
-@Property(name = "quick.kafka.broker", value = "dummy")
 @MicronautTest(startApplication = false)
 class GraphQLSchemaGeneratorTest {
 
@@ -63,7 +60,7 @@ class GraphQLSchemaGeneratorTest {
 
     @Inject
     GraphQLSchemaGeneratorTest(final GraphQLSchemaGenerator generator,
-                               final TestTopicRegistryClient registryClient) {
+        final TestTopicRegistryClient registryClient) {
         this.generator = generator;
         this.registryClient = registryClient;
     }
@@ -262,6 +259,25 @@ void shouldConvertQueryWithPrimitiveType(final TestInfo testInfo) throws IOExcep
             .isInstanceOf(QueryKeyArgumentFetcher.class);
     }
 
+    @Test
+    void shouldConvertQueryWithRange(final TestInfo testInfo) throws IOException {
+        final Path schemaPath = workingDirectory.resolve(testInfo.getTestMethod().orElseThrow().getName() + ".graphql");
+        final GraphQLSchema schema = this.generator.create(Files.readString(schemaPath));
+
+        final List<GraphQLArgument> topicDirectiveArguments =
+            GraphQLTestUtil.getTopicDirectiveArgumentsFromField("Query", "userRequests", schema);
+
+        assertThat(topicDirectiveArguments)
+            .hasSize(4)
+            .extracting(GraphQLArgument::getName)
+            .containsExactly("name", "keyArgument", "rangeFrom", "rangeTo");
+
+        assertThat(topicDirectiveArguments)
+            .hasSize(4)
+            .extracting(GraphQLArgument::getValue)
+            .containsExactly("user-request-range", "userId", "timestampFrom", "timestampTo");
+    }
+
     @Test
     void shouldConvertQueryAllWithComplexType(final TestInfo testInfo) throws IOException {
         final Path schemaPath = workingDirectory.resolve(testInfo.getTestMethod().orElseThrow().getName() + ".graphql");
@@ -471,53 +487,69 @@ void shouldConvertMutation(final TestInfo testInfo) throws IOException {
 
     @Test
     void shouldNotConvertIfMissingKeyInfoInQueryType(final TestInfo testInfo) throws IOException {
-        final Path schemaPath = workingDirectory.resolve(testInfo.getTestMethod().orElseThrow().getName() + ".graphql");
-        String schema = Files.readString(schemaPath);
-        assertThatExceptionOfType(QuickDirectiveException.class)
-                .isThrownBy(() -> this.generator.create(schema))
-                .withMessage("When the return type is not a list for a non-mutation and non-subscription type,"
-                        + " key information (keyArgument or keyField) is needed.");
+        this.assertQuickDirectiveExceptionMessage(testInfo,
+            "When the return type is not a list for a non-mutation and non-subscription type,"
+                + " key information (keyArgument or keyField) is needed.");
     }
 
     @Test
     void shouldNotConvertIfMissingKeyInfoInBasicType(final TestInfo testInfo) throws IOException {
-        final Path schemaPath = workingDirectory.resolve(testInfo.getTestMethod().orElseThrow().getName() + ".graphql");
-        final String schema = Files.readString(schemaPath);
-        assertThatExceptionOfType(QuickDirectiveException.class)
-                .isThrownBy(() -> this.generator.create(schema))
-                .withMessage("When the return type is not a list for a non-mutation and non-subscription type,"
-                        + " key information (keyArgument or keyField) is needed.");
+        this.assertQuickDirectiveExceptionMessage(testInfo,
+            "When the return type is not a list for a non-mutation and non-subscription type,"
+                + " key information (keyArgument or keyField) is needed.");
     }
 
     @Test
     void shouldNotConvertIfMutationDoesNotHaveTwoArgs(final TestInfo testInfo) throws IOException {
-        final Path schemaPath = workingDirectory.resolve(testInfo.getTestMethod().orElseThrow().getName() + ".graphql");
-        final String schema = Files.readString(schemaPath);
-        assertThatExceptionOfType(QuickDirectiveException.class)
-                .isThrownBy(() -> this.generator.create(schema))
-                .withMessage("Mutation requires two input arguments");
+        this.assertQuickDirectiveExceptionMessage(testInfo, "Mutation requires two input arguments");
     }
 
     @Test
     void shouldNotConvertIfKeyArgAndInputNameDifferentInQueryType(final TestInfo testInfo) throws IOException {
-        final Path schemaPath = workingDirectory.resolve(testInfo.getTestMethod().orElseThrow().getName() + ".graphql");
-        final String schema = Files.readString(schemaPath);
-        assertThatExceptionOfType(QuickDirectiveException.class)
-                .isThrownBy(() -> this.generator.create(schema))
-                .withMessage("Key argument has to be identical to the input name.");
+        this.assertQuickDirectiveExceptionMessage(testInfo, "Key argument has to be identical to the input name.");
     }
 
     @Test
     void shouldNotConvertIfKeyArgAndInputNameDifferentInNonQueryType(final TestInfo testInfo) throws IOException {
+        this.assertQuickDirectiveExceptionMessage(testInfo, "Key argument has to be identical to the input name.");
+    }
+
+    @Test
+    void shouldNotConvertIfRangeToArgumentIsMissing(final TestInfo testInfo) throws IOException {
+        this.assertQuickDirectiveExceptionMessage(testInfo, "Both rangeFrom and rangeTo arguments should be set.");
+    }
+
+    @Test
+    void shouldNotCovertIfRangeFromArgumentIsMissing(final TestInfo testInfo) throws  IOException {
+        this.assertQuickDirectiveExceptionMessage(testInfo, "Both rangeFrom and rangeTo arguments should be set.");
+    }
+
+    @Test
+    void shouldNotCovertIfRangeIsDefinedOnField(final TestInfo testInfo) throws  IOException {
+        this.assertQuickDirectiveExceptionMessage(testInfo, "Range queries are only supported on Query types.");
+    }
+
+    @Test
+    void shouldNotCovertIfKeyArgumentInRangeQueryIsMissing(final TestInfo testInfo) throws  IOException {
+        this.assertQuickDirectiveExceptionMessage(testInfo, "You must define a keyArgument.");
+    }
+
+    @Test
+    void shouldNotCovertIfReturnTypeOfRangeQueryIsNotList(final TestInfo testInfo) throws  IOException {
+        this.assertQuickDirectiveExceptionMessage(testInfo, "The return type of range queries should be a list.");
+    }
+
+    private void assertQuickDirectiveExceptionMessage(final TestInfo testInfo, final String message)
+        throws IOException {
         final Path schemaPath = workingDirectory.resolve(testInfo.getTestMethod().orElseThrow().getName() + ".graphql");
         final String schema = Files.readString(schemaPath);
         assertThatExceptionOfType(QuickDirectiveException.class)
-                .isThrownBy(() -> this.generator.create(schema))
-                .withMessage("Key argument has to be identical to the input name.");
+            .isThrownBy(() -> this.generator.create(schema))
+            .withMessage(message);
     }
 
     private static void hasFieldWithListType(final GraphQLObjectType objectType, final String insuredPersonId,
-                                             final String fieldTypeName) {
+        final String fieldTypeName) {
         assertThat(objectType.getFieldDefinition(insuredPersonId).getType())
             .isNotNull()
             .isInstanceOfSatisfying(GraphQLList.class, list ->
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLSecurityTest.java b/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLSecurityTest.java
index 68fa4f55..5e1bc1a0 100644
--- a/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLSecurityTest.java
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLSecurityTest.java
@@ -34,19 +34,17 @@
 import io.micronaut.http.HttpRequest;
 import io.micronaut.http.MediaType;
 import io.micronaut.http.client.BlockingHttpClient;
-import io.micronaut.rxjava2.http.client.RxHttpClient;
 import io.micronaut.http.client.annotation.Client;
 import io.micronaut.http.client.exceptions.HttpClientResponseException;
 import io.micronaut.runtime.server.EmbeddedServer;
-import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
+import io.micronaut.rxjava2.http.client.RxHttpClient;
 import io.micronaut.rxjava2.http.client.websockets.RxWebSocketClient;
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest;
 import jakarta.inject.Inject;
 import java.util.Map;
 import org.assertj.core.api.InstanceOfAssertFactories;
 import org.junit.jupiter.api.Test;
 
-@Property(name = "quick.kafka.bootstrap-server", value = "dummy:1234")
-@Property(name = "quick.kafka.schema-registry-url", value = "http://dummy")
 @Property(name = "micronaut.security.enabled", value = StringUtils.TRUE)
 @MicronautTest
 class GraphQLSecurityTest {
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLTestUtil.java b/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLTestUtil.java
index 0dee0748..5882ad75 100644
--- a/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLTestUtil.java
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/GraphQLTestUtil.java
@@ -18,20 +18,23 @@
 
 import static org.mockito.Mockito.mock;
 
-import com.bakdata.quick.gateway.fetcher.DataFetcherClient;
+import com.bakdata.quick.gateway.directives.topic.TopicDirective;
 import com.bakdata.quick.gateway.fetcher.ClientSupplier;
+import com.bakdata.quick.gateway.fetcher.DataFetcherClient;
 import graphql.schema.DataFetcher;
 import graphql.schema.FieldCoordinates;
+import graphql.schema.GraphQLArgument;
 import graphql.schema.GraphQLFieldDefinition;
 import graphql.schema.GraphQLSchema;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 import lombok.Getter;
 
 public final class GraphQLTestUtil {
     private GraphQLTestUtil() {}
 
-
     public static DataFetcher<?> getFieldDataFetcher(final String objectName, final String fieldName,
         final GraphQLSchema schema) {
         final GraphQLFieldDefinition modificationField = getFieldDefinition(objectName, fieldName, schema);
@@ -50,6 +53,15 @@ public static GraphQLFieldDefinition getFieldDefinition(final String objectName,
             .orElseThrow();
     }
 
+    public static List<GraphQLArgument> getTopicDirectiveArgumentsFromField(final String objectName,
+        final String fieldName,
+        final GraphQLSchema schema) {
+        return getFieldDefinition(objectName, fieldName, schema)
+            .getDirective(TopicDirective.DIRECTIVE_NAME)
+            .getArguments().stream().filter(graphQLArgument -> graphQLArgument.getValue() != null)
+            .collect(Collectors.toList());
+    }
+
     static final class TestClientSupplier implements ClientSupplier {
         @Getter
         private final Map<String, DataFetcherClient<?>> clients;
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/fetcher/QueryListArgumentFetcherTest.java b/gateway/src/test/java/com/bakdata/quick/gateway/fetcher/QueryListArgumentFetcherTest.java
index 952c6629..40de006d 100644
--- a/gateway/src/test/java/com/bakdata/quick/gateway/fetcher/QueryListArgumentFetcherTest.java
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/fetcher/QueryListArgumentFetcherTest.java
@@ -33,7 +33,7 @@
 import java.util.List;
 import java.util.Map;
 import lombok.Builder;
-import lombok.Data;
+import lombok.Value;
 import okhttp3.OkHttpClient;
 import okhttp3.mockwebserver.MockResponse;
 import okhttp3.mockwebserver.MockWebServer;
@@ -148,7 +148,7 @@ void shouldFetchEmptyListWhenResultIsNullAndReturnTypeIsNotNullable() {
 
     @Test
     @SuppressWarnings("unchecked")
-    void shouldFetchEmptyListWhenResultNotIsNullAndDoesNotHaveNullableElements() {
+    void shouldFetchEmptyListWhenResultIsNotNullAndDoesNotHaveNullableElements() {
         final Purchase purchase1 = Purchase.builder()
             .purchaseId("testId1")
             .productId(1)
@@ -183,18 +183,18 @@ private <T> MirrorDataFetcherClient<T> createClient(final Class<T> clazz) {
         return new MirrorDataFetcherClient<>(this.host, this.client, this.mirrorConfig, resolver);
     }
 
-    @Data
+    @Value
     @Builder
     private static class Purchase {
-        private String purchaseId;
-        private int productId;
-        private int amount;
+        String purchaseId;
+        int productId;
+        int amount;
     }
 
-    @Data
+    @Value
     @Builder
     private static class Product {
-        private int productId;
-        private String name;
+        int productId;
+        String name;
     }
 }
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/fetcher/QueryListFetcherTest.java b/gateway/src/test/java/com/bakdata/quick/gateway/fetcher/QueryListFetcherTest.java
index 3b262357..fe5fad2d 100644
--- a/gateway/src/test/java/com/bakdata/quick/gateway/fetcher/QueryListFetcherTest.java
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/fetcher/QueryListFetcherTest.java
@@ -92,7 +92,6 @@ void shouldFetchListOfObjects() throws Exception {
 
     @Test
     void shouldFetchListOfStrings() throws Exception {
-
         final List<String> list = List.of("abc", "def");
         final String listJson = this.mapper.writeValueAsString(new MirrorValue<>(list));
         this.server.enqueue(new MockResponse().setBody(listJson));
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/fetcher/RangeQueryFetcherTest.java b/gateway/src/test/java/com/bakdata/quick/gateway/fetcher/RangeQueryFetcherTest.java
new file mode 100644
index 00000000..6dc2aa3b
--- /dev/null
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/fetcher/RangeQueryFetcherTest.java
@@ -0,0 +1,132 @@
+/*
+ *    Copyright 2022 bakdata GmbH
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+package com.bakdata.quick.gateway.fetcher;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyString;
+
+import com.bakdata.quick.common.api.client.HttpClient;
+import com.bakdata.quick.common.api.model.mirror.MirrorValue;
+import com.bakdata.quick.common.config.MirrorConfig;
+import com.bakdata.quick.common.resolver.KnownTypeResolver;
+import com.bakdata.quick.common.resolver.TypeResolver;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import graphql.schema.DataFetchingEnvironment;
+import graphql.schema.DataFetchingEnvironmentImpl;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import lombok.Builder;
+import lombok.Value;
+import okhttp3.OkHttpClient;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+class RangeQueryFetcherTest {
+    public static final boolean isNullable = true;
+    private final ObjectMapper mapper = new ObjectMapper();
+    private final MockWebServer server = new MockWebServer();
+    private final HttpClient client = new HttpClient(this.mapper, new OkHttpClient());
+    private final MirrorConfig mirrorConfig = MirrorConfig.directAccess();
+    private final String host = String.format("localhost:%s", this.server.getPort());
+
+    @BeforeEach
+    void initRouterAndMirror() throws JsonProcessingException {
+        // mapping from partition to host for initializing PartitionRouter
+        final String routerBody = TestUtils.generateBodyForRouterWith(Map.of(1, this.host, 2, this.host));
+        this.server.enqueue(new MockResponse().setBody(routerBody));
+    }
+
+    @Test
+    void shouldFetchRange() throws JsonProcessingException {
+        final UserRequest userRequest1 = UserRequest.builder()
+            .userId(1)
+            .timestamp(1)
+            .requests(Request.builder().count(10).successful(8).build())
+            .build();
+        final UserRequest userRequest2 = UserRequest.builder()
+            .userId(1)
+            .timestamp(2)
+            .requests(Request.builder().count(10).successful(8).build())
+            .build();
+
+        final List<UserRequest> userRequests = List.of(
+            userRequest1,
+            userRequest2
+        );
+
+        final String userRequestJson = this.mapper.writeValueAsString(new MirrorValue<>(userRequests));
+        this.server.enqueue(new MockResponse().setBody(userRequestJson));
+
+        final DataFetcherClient<UserRequest> fetcherClient = this.createClient();
+
+        final RangeQueryFetcher<?> rangeQueryFetcher =
+            new RangeQueryFetcher<>("userId", fetcherClient, "timestampFrom", "timestampTo", isNullable);
+
+        final Map<String, Object> arguments = Map.of("userId", "1", "timestampFrom", "1", "timestampTo", "2");
+
+        final DataFetchingEnvironment env = DataFetchingEnvironmentImpl.newDataFetchingEnvironment()
+            .localContext(arguments).build();
+
+        final List<?> actual = rangeQueryFetcher.get(env);
+        assertThat(actual).isEqualTo(userRequests);
+    }
+
+    @Test
+    @SuppressWarnings("unchecked")
+    void shouldFetchEmptyListWhenResultIsNullAndReturnTypeIsNotNullable() {
+        final MirrorDataFetcherClient fetcherClient = Mockito.mock(MirrorDataFetcherClient.class);
+
+        Mockito.when(fetcherClient.fetchRange(anyString(), anyString(), anyString())).thenReturn(null);
+
+        final RangeQueryFetcher<?> rangeQueryFetcher =
+            new RangeQueryFetcher<>("userId", fetcherClient, "timestampFrom", "timestampTo", false);
+
+        final Map<String, Object> arguments = Map.of("userId", "1", "timestampFrom", "1", "timestampTo", "2");
+
+        final DataFetchingEnvironment env = DataFetchingEnvironmentImpl.newDataFetchingEnvironment()
+            .localContext(arguments).build();
+
+        final List<?> actual = rangeQueryFetcher.get(env);
+
+        assertThat(actual).isEqualTo(Collections.emptyList());
+    }
+
+    private MirrorDataFetcherClient<UserRequest> createClient() {
+        final TypeResolver<UserRequest> resolver = new KnownTypeResolver<>(UserRequest.class, this.mapper);
+        return new MirrorDataFetcherClient<>(this.host, this.client, this.mirrorConfig, resolver);
+    }
+
+    @Value
+    @Builder
+    private static class UserRequest {
+        int userId;
+        int timestamp;
+        Request requests;
+    }
+
+    @Value
+    @Builder
+    private static class Request {
+        int count;
+        int successful;
+    }
+}
diff --git a/gateway/src/test/java/com/bakdata/quick/gateway/security/ApiKeyTest.java b/gateway/src/test/java/com/bakdata/quick/gateway/security/ApiKeyTest.java
index e946150a..03c998b1 100644
--- a/gateway/src/test/java/com/bakdata/quick/gateway/security/ApiKeyTest.java
+++ b/gateway/src/test/java/com/bakdata/quick/gateway/security/ApiKeyTest.java
@@ -40,8 +40,6 @@
 import org.junit.jupiter.api.Test;
 
 @MicronautTest
-@Property(name = "quick.kafka.bootstrap-server", value = "dummy:1234")
-@Property(name = "quick.kafka.schema-registry-url", value = "http://dummy")
 @Property(name = "micronaut.security.enabled", value = "true")
 class ApiKeyTest {
     private static final String SECURE_PATH = "control/definition";
diff --git a/gateway/src/test/resources/application-test.yaml b/gateway/src/test/resources/application-test.yaml
index b3e78e51..74ccf09f 100644
--- a/gateway/src/test/resources/application-test.yaml
+++ b/gateway/src/test/resources/application-test.yaml
@@ -11,11 +11,12 @@ endpoints:
     enabled: false
 
 quick:
+  kafka:
+    bootstrap-server: dummy:9092
+    schema-registry-url: http://test:8081
   definition:
     path: "definition/definition.yaml"
   apikey:
     test_key
-  kafka:
-    bootstrap-server: dummy:123
   mirror:
     prefix: "" # prefix must be empty, as the host is simply 'localhost' and not for example 'quick-mirror-localhost'
diff --git a/gateway/src/test/resources/schema/conversion/shouldConvertQueryWithRange.graphql b/gateway/src/test/resources/schema/conversion/shouldConvertQueryWithRange.graphql
new file mode 100644
index 00000000..cafb9bc1
--- /dev/null
+++ b/gateway/src/test/resources/schema/conversion/shouldConvertQueryWithRange.graphql
@@ -0,0 +1,19 @@
+type Query {
+    userRequests(
+        userId: Int
+        timestampFrom: Int
+        timestampTo: Int
+    ): [UserRequests]
+    @topic(name: "user-request-range",
+        keyArgument: "userId",
+        rangeFrom: "timestampFrom",
+        rangeTo: "timestampTo")
+}
+
+type UserRequests {
+    userId: Int
+    serviceId: Int
+    timestamp: Int
+    requests: Int
+    success: Int
+}
diff --git a/gateway/src/test/resources/schema/conversion/shouldConvertQueryWithRangeOnField.graphql b/gateway/src/test/resources/schema/conversion/shouldConvertQueryWithRangeOnField.graphql
new file mode 100644
index 00000000..476f53ef
--- /dev/null
+++ b/gateway/src/test/resources/schema/conversion/shouldConvertQueryWithRangeOnField.graphql
@@ -0,0 +1,12 @@
+type Query {
+    product(key: Int, timestampFrom: Int, timestampTo: Int): ProductInfo
+}
+
+type ProductInfo {
+    info: [Info] @topic(name: "info-topic", keyArgument: "key", rangeFrom: "timestampFrom", rangeTo: "timestampTo")
+}
+
+type Info {
+    key: Int
+    timestamp: Int
+}
diff --git a/gateway/src/test/resources/schema/conversion/shouldNotConvertIfKeyArgAndInputNameDifferentInNonQueryType.graphql b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfKeyArgAndInputNameDifferentInNonQueryType.graphql
index e40174e3..6bbd7986 100644
--- a/gateway/src/test/resources/schema/conversion/shouldNotConvertIfKeyArgAndInputNameDifferentInNonQueryType.graphql
+++ b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfKeyArgAndInputNameDifferentInNonQueryType.graphql
@@ -1,3 +1,4 @@
+# Invalid Schema
 type Query {
     getProduct(productId: ID): ProductInfo
 }
diff --git a/gateway/src/test/resources/schema/conversion/shouldNotConvertIfKeyArgAndInputNameDifferentInQueryType.graphql b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfKeyArgAndInputNameDifferentInQueryType.graphql
index 9c9f9952..cd804ee0 100644
--- a/gateway/src/test/resources/schema/conversion/shouldNotConvertIfKeyArgAndInputNameDifferentInQueryType.graphql
+++ b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfKeyArgAndInputNameDifferentInQueryType.graphql
@@ -1,3 +1,4 @@
+# Invalid Schema
 type Query {
     getClick(id: ID): Long @topic(name: "click-topic", keyArgument: "aDifferentKeyArgument")
 }
diff --git a/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMissingKeyInfoInBasicType.graphql b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMissingKeyInfoInBasicType.graphql
index 1b5890aa..08262c8f 100644
--- a/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMissingKeyInfoInBasicType.graphql
+++ b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMissingKeyInfoInBasicType.graphql
@@ -1,3 +1,4 @@
+# Invalid Schema
 type Query {
     getProduct(productId: ID): ProductInfo
 }
diff --git a/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMissingKeyInfoInQueryType.graphql b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMissingKeyInfoInQueryType.graphql
index 60c142ec..8750982f 100644
--- a/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMissingKeyInfoInQueryType.graphql
+++ b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMissingKeyInfoInQueryType.graphql
@@ -1,3 +1,4 @@
+# Invalid Schema
 type Product {
     id: ID!
     name: String!
diff --git a/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMutationDoesNotHaveTwoArgs.graphql b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMutationDoesNotHaveTwoArgs.graphql
index 33af2fe6..f20e7588 100644
--- a/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMutationDoesNotHaveTwoArgs.graphql
+++ b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfMutationDoesNotHaveTwoArgs.graphql
@@ -1,3 +1,4 @@
+# Invalid Schema
 type Mutation {
     setClick(clickCount: Long): Long @topic(name: "click-topic")
 }
diff --git a/gateway/src/test/resources/schema/conversion/shouldNotConvertIfRangeToArgumentIsMissing.graphql b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfRangeToArgumentIsMissing.graphql
new file mode 100644
index 00000000..20ca62c1
--- /dev/null
+++ b/gateway/src/test/resources/schema/conversion/shouldNotConvertIfRangeToArgumentIsMissing.graphql
@@ -0,0 +1,19 @@
+# Invalid Schema
+type Query {
+    userRequests(
+        userId: Int
+        timestampFrom: Int
+        timestampTo: Int
+    ): [UserRequests]
+    @topic(name: "user-request-range",
+        keyArgument: "userId",
+        rangeFrom: "timestampFrom")
+}
+
+type UserRequests {
+    userId: Int
+    serviceId: Int
+    timestamp: Int
+    requests: Int
+    success: Int
+}
diff --git a/gateway/src/test/resources/schema/conversion/shouldNotCovertIfKeyArgumentInRangeQueryIsMissing.graphql b/gateway/src/test/resources/schema/conversion/shouldNotCovertIfKeyArgumentInRangeQueryIsMissing.graphql
new file mode 100644
index 00000000..353718b8
--- /dev/null
+++ b/gateway/src/test/resources/schema/conversion/shouldNotCovertIfKeyArgumentInRangeQueryIsMissing.graphql
@@ -0,0 +1,18 @@
+type Query {
+    userRequests(
+        userId: Int
+        timestampFrom: Int
+        timestampTo: Int
+    ): [UserRequests]
+    @topic(name: "user-request-range",
+        rangeFrom: "timestampFrom",
+        rangeTo: "timestampTo")
+}
+
+type UserRequests {
+    userId: Int
+    serviceId: Int
+    timestamp: Int
+    requests: Int
+    success: Int
+}
diff --git a/gateway/src/test/resources/schema/conversion/shouldNotCovertIfRangeFromArgumentIsMissing.graphql b/gateway/src/test/resources/schema/conversion/shouldNotCovertIfRangeFromArgumentIsMissing.graphql
new file mode 100644
index 00000000..0c057b68
--- /dev/null
+++ b/gateway/src/test/resources/schema/conversion/shouldNotCovertIfRangeFromArgumentIsMissing.graphql
@@ -0,0 +1,19 @@
+# Invalid Schema
+type Query {
+    userRequests(
+        userId: Int
+        timestampFrom: Int
+        timestampTo: Int
+    ): [UserRequests]
+    @topic(name: "user-request-range",
+        keyArgument: "userId",
+        rangeTo: "timestampTo")
+}
+
+type UserRequests {
+    userId: Int
+    serviceId: Int
+    timestamp: Int
+    requests: Int
+    success: Int
+}
diff --git a/gateway/src/test/resources/schema/conversion/shouldNotCovertIfRangeIsDefinedOnField.graphql b/gateway/src/test/resources/schema/conversion/shouldNotCovertIfRangeIsDefinedOnField.graphql
new file mode 100644
index 00000000..c97a3138
--- /dev/null
+++ b/gateway/src/test/resources/schema/conversion/shouldNotCovertIfRangeIsDefinedOnField.graphql
@@ -0,0 +1,13 @@
+# Invalid Schema
+type Query {
+    product(key: Int, timestampFrom: Int, timestampTo: Int): ProductInfo!
+}
+
+type ProductInfo {
+    info: [Info!] @topic(name: "info-topic", keyArgument: "key", rangeFrom: "timestampFrom", rangeTo: "timestampTo")
+}
+
+type Info {
+    key: Int
+    timestamp: Int!
+}
diff --git a/gateway/src/test/resources/schema/conversion/shouldNotCovertIfReturnTypeOfRangeQueryIsNotList.graphql b/gateway/src/test/resources/schema/conversion/shouldNotCovertIfReturnTypeOfRangeQueryIsNotList.graphql
new file mode 100644
index 00000000..0ce9c036
--- /dev/null
+++ b/gateway/src/test/resources/schema/conversion/shouldNotCovertIfReturnTypeOfRangeQueryIsNotList.graphql
@@ -0,0 +1,20 @@
+# Invalid Schema
+type Query {
+    userRequests(
+        userId: Int
+        timestampFrom: Int
+        timestampTo: Int
+    ): UserRequests
+    @topic(name: "user-request-range",
+        keyArgument: "userId",
+        rangeFrom: "timestampFrom",
+        rangeTo: "timestampTo")
+}
+
+type UserRequests {
+    userId: Int
+    serviceId: Int
+    timestamp: Int
+    requests: Int
+    success: Int
+}
diff --git a/gateway/src/test/resources/schema/execution/shouldExecuteRange.graphql b/gateway/src/test/resources/schema/execution/shouldExecuteRange.graphql
new file mode 100644
index 00000000..45b2377d
--- /dev/null
+++ b/gateway/src/test/resources/schema/execution/shouldExecuteRange.graphql
@@ -0,0 +1,16 @@
+type Query {
+    userRequests(
+        userId: Int
+        timestampFrom: Int
+        timestampTo: Int
+    ): [UserRequests] @topic(name: "user-request-range",
+        keyArgument: "userId",
+        rangeFrom: "timestampFrom",
+        rangeTo: "timestampTo")
+}
+
+type UserRequests {
+    userId: Int
+    timestamp: Int
+    requests: Int
+}
diff --git a/gateway/src/test/resources/schema/execution/shouldExecuteRangeQuery.graphql b/gateway/src/test/resources/schema/execution/shouldExecuteRangeQuery.graphql
new file mode 100644
index 00000000..519adf6a
--- /dev/null
+++ b/gateway/src/test/resources/schema/execution/shouldExecuteRangeQuery.graphql
@@ -0,0 +1,5 @@
+{
+    userRequests(userId: 1, timestampFrom: 1, timestampTo: 3)  {
+        requests
+    }
+}