Valkey

docs/config.yaml

@@ -11,3 +11,11 @@ dataStore:
 
 clusterStatsConfiguration:
   monitorType: INFO_API
+
+# Valkey distributed cache (optional - for multi-instance deployments)
+valkeyConfiguration:
+  enabled: false
+  host: localhost
+  port: 6379
+  # password: ${VALKEY_PASSWORD}  # Uncomment if Valkey requires AUTH
+  # cacheTtlSeconds: 1800  # Cache TTL in seconds (default: 1800 = 30 minutes)

docs/installation.md

@@ -157,10 +157,37 @@ Find more information in the [routing rules documentation](routing-rules.md).
 To configure the logging level for various classes, specify the path to the 
 `log.properties` file by setting `log.levels-file` in `serverConfig`.
 
-For additional configurations, use the `log.*` properties from the 
+For additional configurations, use the `log.*` properties from the
 [Trino logging properties documentation](https://trino.io/docs/current/admin/properties-logging.html) and specify
 the properties in `serverConfig`.
 
+### Configure distributed cache (optional)
+
+For multi-instance deployments, Trino Gateway supports distributed caching
+using Valkey (or Redis) to share query metadata across gateway instances.
+This improves query routing and enables horizontal scaling.
+
+For single gateway deployments, distributed caching is not needed - the
+local cache is sufficient.
+
+```yaml
+valkeyConfiguration:
+  enabled: true
+  host: valkey.internal.prod
+  port: 6379
+  password: ${ENV:VALKEY_PASSWORD}
+  cacheTtlSeconds: 1800  # Cache TTL (default: 1800 = 30 minutes)
+```
+
+Optional parameters: You can customize `cacheTtlSeconds` based on your query duration:
+
+- Short queries (< 5 min): 600 seconds (10 minutes)
+- Default queries: 1800 seconds (30 minutes)
+- Long-running queries: 3600 seconds (1 hour)
+
+See Valkey distributed cache configuration for detailed configuration options,
+deployment scenarios, and performance tuning.
+
 ### Proxying additional paths
 
 By default, Trino Gateway only proxies requests to paths starting with

gateway-ha/.gitignore

@@ -0,0 +1 @@
+gateway-ha/gateway.log

gateway-ha/config.yaml

@@ -23,3 +23,11 @@ clusterStatsConfiguration:
 monitor:
   taskDelay: 1m
   clusterMetricsRegistryRefreshPeriod: 30s
+
+# Valkey distributed cache (optional - for multi-instance deployments)
+valkeyConfiguration:
+  enabled: false  # Set to true to enable distributed caching
+  host: localhost
+  port: 6379
+  # password: ${VALKEY_PASSWORD}  # Uncomment if Valkey requires AUTH
+  # cacheTtlSeconds: 1800  # Cache TTL in seconds (default: 1800 = 30 minutes)

gateway-ha/pom.xml

@@ -189,6 +189,12 @@
             <version>${dep.trino.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>io.valkey</groupId>
+            <artifactId>valkey-java</artifactId>
+            <version>5.5.0</version>
+        </dependency>
+
         <dependency>
             <groupId>jakarta.annotation</groupId>
             <artifactId>jakarta.annotation-api</artifactId>

gateway-ha/src/main/java/io/trino/gateway/ha/cache/Cache.java

@@ -0,0 +1,30 @@
+/*
+ * 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 io.trino.gateway.ha.cache;
+
+import java.util.Optional;
+
+public interface Cache
+{
+    Optional<String> get(String key);
+
+    void set(String key, String value);
+
+    void invalidate(String key);
+
+    default boolean isEnabled()
+    {
+        return false;
+    }
+}

gateway-ha/src/main/java/io/trino/gateway/ha/cache/QueryCacheManager.java

@@ -0,0 +1,184 @@
+/*
+ * 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 io.trino.gateway.ha.cache;
+
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+
+import java.util.Optional;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Manages query-related caches including both L1 (in-memory LoadingCache) and L2 (distributed cache).
+ * This class encapsulates all cache operations to provide better separation of concerns.
+ */
+public class QueryCacheManager
+{
+    private static final String BACKEND_KEY_PREFIX = "trino:query:backend:";
+    private static final String ROUTING_GROUP_KEY_PREFIX = "trino:query:routing_group:";
+    private static final String EXTERNAL_URL_KEY_PREFIX = "trino:query:external_url:";
+
+    private final LoadingCache<String, String> queryIdBackendCache;
+    private final LoadingCache<String, String> queryIdRoutingGroupCache;
+    private final LoadingCache<String, String> queryIdExternalUrlCache;
+    private final Cache distributedCache;
+
+    public QueryCacheManager(
+            Function<String, String> backendLoader,
+            Function<String, String> routingGroupLoader,
+            Function<String, String> externalUrlLoader,
+            Cache distributedCache)
+    {
+        this.queryIdBackendCache = buildCache(backendLoader);
+        this.queryIdRoutingGroupCache = buildCache(routingGroupLoader);
+        this.queryIdExternalUrlCache = buildCache(externalUrlLoader);
+        this.distributedCache = requireNonNull(distributedCache, "distributedCache is null");
+    }
+
+    private LoadingCache<String, String> buildCache(Function<String, String> loader)
+    {
+        return CacheBuilder.newBuilder()
+                .maximumSize(10000)
+                .expireAfterAccess(30, TimeUnit.MINUTES)
+                .build(
+                        new CacheLoader<>()
+                        {
+                            @Override
+                            public String load(String queryId)
+                            {
+                                return loader.apply(queryId);
+                            }
+                        });
+    }
+
+    // L1 Cache Operations
+
+    public void setBackendInL1(String queryId, String backend)
+    {
+        queryIdBackendCache.put(queryId, backend);
+    }
+
+    public void setRoutingGroupInL1(String queryId, String routingGroup)
+    {
+        queryIdRoutingGroupCache.put(queryId, routingGroup);
+    }
+
+    public void setExternalUrlInL1(String queryId, String externalUrl)
+    {
+        queryIdExternalUrlCache.put(queryId, externalUrl);
+    }
+
+    public String getBackendFromL1(String queryId)
+            throws ExecutionException
+    {
+        return queryIdBackendCache.get(queryId);
+    }
+
+    public String getRoutingGroupFromL1(String queryId)
+            throws ExecutionException
+    {
+        return queryIdRoutingGroupCache.get(queryId);
+    }
+
+    public String getExternalUrlFromL1(String queryId)
+            throws ExecutionException
+    {
+        return queryIdExternalUrlCache.get(queryId);
+    }
+
+    // L2 Cache Operations
+
+    public void cacheBackend(String queryId, String backend)
+    {
+        if (distributedCache.isEnabled()) {
+            distributedCache.set(BACKEND_KEY_PREFIX + queryId, backend);
+        }
+    }
+
+    public void cacheRoutingGroup(String queryId, String routingGroup)
+    {
+        if (distributedCache.isEnabled()) {
+            distributedCache.set(ROUTING_GROUP_KEY_PREFIX + queryId, routingGroup);
+        }
+    }
+
+    public void cacheExternalUrl(String queryId, String externalUrl)
+    {
+        if (distributedCache.isEnabled()) {
+            distributedCache.set(EXTERNAL_URL_KEY_PREFIX + queryId, externalUrl);
+        }
+    }
+
+    public Optional<String> getCachedBackend(String queryId)
+    {
+        if (distributedCache.isEnabled()) {
+            return distributedCache.get(BACKEND_KEY_PREFIX + queryId);
+        }
+        return Optional.empty();
+    }
+
+    public Optional<String> getCachedRoutingGroup(String queryId)
+    {
+        if (distributedCache.isEnabled()) {
+            return distributedCache.get(ROUTING_GROUP_KEY_PREFIX + queryId);
+        }
+        return Optional.empty();
+    }
+
+    public Optional<String> getCachedExternalUrl(String queryId)
+    {
+        if (distributedCache.isEnabled()) {
+            return distributedCache.get(EXTERNAL_URL_KEY_PREFIX + queryId);
+        }
+        return Optional.empty();
+    }
+
+    // Combined Operations (L1 + L2)
+
+    public void setBackend(String queryId, String backend)
+    {
+        setBackendInL1(queryId, backend);
+        cacheBackend(queryId, backend);
+    }
+
+    public void setRoutingGroup(String queryId, String routingGroup)
+    {
+        setRoutingGroupInL1(queryId, routingGroup);
+        cacheRoutingGroup(queryId, routingGroup);
+    }
+
+    public void setExternalUrl(String queryId, String externalUrl)
+    {
+        setExternalUrlInL1(queryId, externalUrl);
+        cacheExternalUrl(queryId, externalUrl);
+    }
+
+    public void updateAllCaches(String queryId, String backend, String routingGroup, String externalUrl)
+    {
+        setBackendInL1(queryId, backend);
+        cacheBackend(queryId, backend);
+        cacheRoutingGroup(queryId, routingGroup);
+        cacheExternalUrl(queryId, externalUrl);
+    }
+
+    public boolean isDistributedCacheEnabled()
+    {
+        return distributedCache.isEnabled();
+    }
+}