[Apache][Doris][Feature][FE]Support users with high QPS in the same query through result set cache

docs/en/administrator-guide/config/fe_config.md

@@ -682,3 +682,22 @@ In some very special circumstances, such as code bugs, or human misoperation, et
 Set to true so that Doris will automatically use blank replicas to fill tablets which all replicas have been damaged or missing.
 
 Default is false.
+
+
+#### `enable_result_cache_ttl`
+
+enable_result_ cache_ ttl  variable is set in the user session. The user can customize whether to turn it on or not. The TTL time is used to determine whether the user's SQL uses cache. The correctness of the data is not guaranteed when the data is changed`
+
+The cache is stored and retrieved according to the user connected and the query SQL. If it exceeds the cache expiration time, the cache will not be hit and the cache will be cleaned
+
+### `result_cache_ttl_expire_after_in_milliseconds`
+
+enable_result_cache_ttl cache time
+
+### `result_cache_ttl_size_in_bytes`
+
+enable_result_cache_ttl cache size
+
+### `result_cache_ttl_size_per_query_in_bytes`
+
+enable_result_cache_ttl single query result max cache size

docs/zh-CN/administrator-guide/config/fe_config.md

@@ -679,6 +679,21 @@ thrift_client_timeout_ms 的值被设置为大于0来避免线程卡在java.net.
 
 默认为 false。
 
+### `enable_result_cache_ttl`
 
+result_cache_ttl 变量设置在用户Session中，用户可自定义是否开启,通过ttl时间来确定用户的sql是否使用缓存，`这里数据变更时不保证数据的正确性`
+按照 用户 connectid,和查询的sql 来存储和获取缓存，超过缓存失效时间则命中不了缓存，该缓存也会被清理
+
+### `result_cache_ttl_expire_after_in_milliseconds`
+
+enable_result_cache_ttl 失效销毁间隔
+
+### `result_cache_ttl_size_in_bytes`
+
+enable_result_cache_ttl 缓存大小
+
+### `result_cache_ttl_size_per_query_in_bytes`
+
+enable_result_cache_ttl 单个query结果，最大缓存大小
 
 

fe/fe-core/pom.xml

@@ -454,6 +454,12 @@ under the License.
             <artifactId>spark-sql_2.12</artifactId>
             <scope>provided</scope>
         </dependency>
+
+        <dependency>
+            <groupId>com.github.ben-manes.caffeine</groupId>
+            <artifactId>caffeine</artifactId>
+            <version>2.6.2</version>
+        </dependency>
     </dependencies>
 
     <build>

fe/fe-core/src/main/java/org/apache/doris/PaloFe.java

@@ -17,6 +17,7 @@
 
 package org.apache.doris;
 
+import org.apache.doris.cache.CacheFactory;
 import org.apache.doris.catalog.Catalog;
 import org.apache.doris.common.CommandLineOptions;
 import org.apache.doris.common.Config;
@@ -91,7 +92,7 @@ public static void start(String dorisHomeDir, String pidDir, String[] args) {
             Log4jConfig.initLogging();
 
             // set dns cache ttl
-            java.security.Security.setProperty("networkaddress.cache.ttl" , "60");
+            java.security.Security.setProperty("networkaddress.cache.ttl", "60");
 
             // check command line options
             checkCommandLineOptions(cmdLineOpts);
@@ -104,6 +105,11 @@ public static void start(String dorisHomeDir, String pidDir, String[] args) {
             // init catalog and wait it be ready
             Catalog.getCurrentCatalog().initialize(args);
             Catalog.getCurrentCatalog().waitForReady();
+            // Initialize the result cache if enabled
+            LOG.debug("result cache is " + (Config.enable_result_cache_ttl ? "enabled" : "disabled"));
+            if (Config.enable_result_cache_ttl) {
+                CacheFactory.getUniversalCache();
+            }
 
             // init and start:
             // 1. QeService for MySQL Server
@@ -141,12 +147,12 @@ public static void start(String dorisHomeDir, String pidDir, String[] args) {
      *      Specify the helper node when joining a bdb je replication group
      * -b --bdb
      *      Run bdbje debug tools
-     *      
+     *
      *      -l --listdb
      *          List all database names in bdbje
      *      -d --db
      *          Specify a database in bdbje
-     *          
+     *
      *          -s --stat
      *              Print statistic of a database, including count, first key, last key
      *          -f --from
@@ -155,7 +161,7 @@ public static void start(String dorisHomeDir, String pidDir, String[] args) {
      *              Specify the end scan key
      *          -m --metaversion
      *              Specify the meta version to decode log value
-     *              
+     *
      */
     private static CommandLineOptions parseArgs(String[] args) {
         CommandLineParser commandLineParser = new BasicParser();
@@ -194,7 +200,7 @@ private static CommandLineOptions parseArgs(String[] args) {
                     System.err.println("BDBJE database name is missing");
                     System.exit(-1);
                 }
-                
+
                 if (cmd.hasOption('s') || cmd.hasOption("stat")) {
                     BDBToolOptions bdbOpts = new BDBToolOptions(false, dbName, true, "", "", 0);
                     return new CommandLineOptions(false, "", bdbOpts);
@@ -224,7 +230,7 @@ private static CommandLineOptions parseArgs(String[] args) {
                             System.exit(-1);
                         }
                     }
-                    
+
                     BDBToolOptions bdbOpts = new BDBToolOptions(false, dbName, false, fromKey, endKey, metaVersion);
                     return new CommandLineOptions(false, "", bdbOpts);
                 }

fe/fe-core/src/main/java/org/apache/doris/alter/AlterJobV2.java

@@ -266,6 +266,7 @@ public void write(DataOutput out) throws IOException {
         out.writeLong(timeoutMs);
     }
 
+    @Deprecated
     public void readFields(DataInput in) throws IOException {
         // read common members as write in AlterJobV2.write().
         // except 'type' member, which is read in AlterJobV2.read()

fe/fe-core/src/main/java/org/apache/doris/cache/Cache.java

@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 org.apache.doris.cache;
+
+import com.google.common.base.Preconditions;
+
+import java.io.Closeable;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Map;
+
+public interface Cache extends Closeable {
+
+    byte[] get(NamedKey key);
+
+    void put(NamedKey key, byte[] value);
+
+    Map<NamedKey, byte[]> getBulk(Iterable<NamedKey> keys);
+
+    CacheStats getStats();
+
+    boolean isLocal();
+
+    class NamedKey {
+        public final String namespace;
+        public final byte[] key;
+
+        public NamedKey(String namespace, byte[] key) {
+            Preconditions.checkNotNull(namespace, "Namespace must not be null");
+            Preconditions.checkNotNull(key, "Key must not be null");
+            this.namespace = namespace;
+            this.key = key;
+        }
+
+        public byte[] toByteArray() {
+            final byte[] nsBytes = namespace.getBytes(StandardCharsets.UTF_8);
+            return ByteBuffer.allocate(Integer.BYTES + nsBytes.length + key.length)
+                    .putInt(nsBytes.length)
+                    .put(nsBytes)
+                    .put(key)
+                    .array();
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) {
+                return true;
+            }
+
+            if (o instanceof NamedKey) {
+                NamedKey namedKey = (NamedKey) o;
+                return namespace.equals(namedKey.namespace) && Arrays.equals(key, namedKey.key);
+            }
+
+            return false;
+        }
+
+        @Override
+        public int hashCode() {
+            return 31 * namespace.hashCode() + Arrays.hashCode(key);
+        }
+
+        @Override
+        public String toString() {
+            return namespace + "_" + new String(key, StandardCharsets.UTF_8);
+        }
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/cache/CacheExecutorFactory.java

@@ -0,0 +1,49 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 org.apache.doris.cache;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.LinkedBlockingQueue;
+
+public enum CacheExecutorFactory {
+    COMMON_FJP {
+        @Override
+        public Executor createExecutor() {
+            return ForkJoinPool.commonPool();
+        }
+    },
+    SINGLE_THREAD {
+        @Override
+        public Executor createExecutor() {
+            return new ThreadPoolExecutor(1, 1,
+                    0L, TimeUnit.MILLISECONDS,
+                    new LinkedBlockingQueue<Runnable>()) {
+                @Override
+                protected void finalize() {
+                    this.shutdown();
+                }
+            };
+        }
+    };
+
+    public abstract Executor createExecutor();
+
+}

fe/fe-core/src/main/java/org/apache/doris/cache/CacheFactory.java

@@ -0,0 +1,41 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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 org.apache.doris.cache;
+
+public class CacheFactory {
+    private static volatile Cache universalCache;
+
+    /**
+     * Initialize the result cache.
+     *
+     * @return
+     */
+    public static Cache getUniversalCache() {
+        if (universalCache == null) {
+            synchronized (CacheFactory.class) {
+                if (universalCache == null) {
+                    // Now just use the simple local cache.
+                    // TODO Felix: add global/central cache or hybrid cache (l1/l2)
+                    universalCache = SimpleLocalCache.create();
+                }
+            }
+        }
+        return universalCache;
+    }
+
+}