This is an automated email from the ASF dual-hosted git repository. liyang-kylin pushed a commit to branch kylin5 in repository https://gitbox.apache.org/repos/asf/kylin.git
commit 8803f9fdf9cee9c94bc5c271a184a6b59dd650cc Author: Li Yang <[email protected]> AuthorDate: Fri Oct 2 22:18:34 2026 +0800 KYLIN-6112 bound computed column caches --- .../kylin/query/util/ComputedColumnRewriter.java | 61 ++++++++++++++++++---- .../query/util/ComputedColumnRewriterTest.java | 28 ++++++++++ 2 files changed, 79 insertions(+), 10 deletions(-) diff --git a/src/query-common/src/main/java/org/apache/kylin/query/util/ComputedColumnRewriter.java b/src/query-common/src/main/java/org/apache/kylin/query/util/ComputedColumnRewriter.java index 47ba96b97c..b87c9ee9cf 100644 --- a/src/query-common/src/main/java/org/apache/kylin/query/util/ComputedColumnRewriter.java +++ b/src/query-common/src/main/java/org/apache/kylin/query/util/ComputedColumnRewriter.java @@ -22,6 +22,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import org.apache.calcite.rel.RelNode; @@ -40,6 +41,9 @@ import org.apache.commons.lang3.StringUtils; import org.apache.kylin.common.KylinConfig; import org.apache.kylin.common.util.CollectionUtil; import org.apache.kylin.common.util.StringHelper; +import org.apache.kylin.guava30.shaded.common.cache.Cache; +import org.apache.kylin.guava30.shaded.common.cache.CacheBuilder; +import org.apache.kylin.guava30.shaded.common.cache.Weigher; import org.apache.kylin.guava30.shaded.common.collect.Lists; import org.apache.kylin.guava30.shaded.common.collect.Maps; import org.apache.kylin.metadata.cube.model.NDataflowManager; @@ -61,9 +65,21 @@ import lombok.extern.slf4j.Slf4j; @Slf4j public class ComputedColumnRewriter { - // use concurrent map for multi-thread - public static final Map<String, String> CC_TO_REX_NODE_STR_CACHE = Maps.newConcurrentMap(); - public static final Map<String, ComputedColumnDesc> EXP_TO_CC_MAP = Maps.newConcurrentMap(); + private static final long CACHE_EXPIRE_MINUTES = 10L; + private static final long CC_REX_CACHE_MAX_WEIGHT = 1_000_000L; + private static final long CC_CACHE_MAX_WEIGHT = 100_000L; + private static final Cache<String, String> CC_REX_CACHE = CacheBuilder.newBuilder() + .maximumWeight(CC_REX_CACHE_MAX_WEIGHT) + .weigher((Weigher<String, String>) (key, value) -> cacheWeight(key, value)) + .expireAfterWrite(CACHE_EXPIRE_MINUTES, TimeUnit.MINUTES).build(); + private static final Cache<String, ComputedColumnDesc> EXP_TO_CC_CACHE = CacheBuilder.newBuilder() + .maximumWeight(CC_CACHE_MAX_WEIGHT) + .weigher((Weigher<String, ComputedColumnDesc>) (key, value) -> ccCacheWeight(key, value)) + .expireAfterWrite(CACHE_EXPIRE_MINUTES, TimeUnit.MINUTES).build(); + + // Keep Map views for compatibility with callers that inspect the caches. + public static final Map<String, String> CC_TO_REX_NODE_STR_CACHE = CC_REX_CACHE.asMap(); + public static final Map<String, ComputedColumnDesc> EXP_TO_CC_MAP = EXP_TO_CC_CACHE.asMap(); public static final String NAME_SPLIT = "@@"; private static final SparkSQLFunctionConverter FUNCTION_CONVERTER = new SparkSQLFunctionConverter(); @@ -133,8 +149,7 @@ public class ComputedColumnRewriter { // update cc cache first, because the cc may have been updated for (ComputedColumnDesc cc : model.getComputedColumnDescs()) { - String key = cc.getTableIdentity() + NAME_SPLIT - + StringHelper.backtickToDoubleQuote(cc.getInnerExpression()); + String key = ccCacheKey(model, cc); EXP_TO_CC_MAP.put(key, cc); if (!CC_TO_REX_NODE_STR_CACHE.containsKey(key)) { String rexNodeStr = extractCcRexNode(model, kylinConfig, cc.getInnerExpression()); @@ -147,7 +162,7 @@ public class ComputedColumnRewriter { innerExpression = innerExpression.replace(entry.getKey(), entry.getValue()); } boolean onlyReuseUserDefinedCC = kylinConfig.onlyReuseUserDefinedCC(); - String aliasInnerExpression = model.getRootFactTableName() + NAME_SPLIT + innerExpression; + String aliasInnerExpression = expressionCacheKey(model, innerExpression); if (EXP_TO_CC_MAP.containsKey(aliasInnerExpression)) { ComputedColumnDesc cc = EXP_TO_CC_MAP.get(aliasInnerExpression); if (onlyReuseUserDefinedCC && cc.isAutoCC()) { @@ -162,8 +177,7 @@ public class ComputedColumnRewriter { if (onlyReuseUserDefinedCC && cc.isAutoCC()) { continue; } - String key = cc.getTableIdentity() + NAME_SPLIT - + StringHelper.backtickToDoubleQuote(cc.getInnerExpression()); + String key = ccCacheKey(model, cc); String rexNodeStrOfCc = CC_TO_REX_NODE_STR_CACHE.get(key); if (Objects.equals(rexNodeStr, rexNodeStrOfCc)) { return cc; @@ -309,8 +323,7 @@ public class ComputedColumnRewriter { } try { String rexNodeStr; - String key = cc.getTableIdentity() + NAME_SPLIT - + StringHelper.backtickToDoubleQuote(cc.getInnerExpression()); + String key = ccCacheKey(model, cc); if (!ComputedColumnRewriter.CC_TO_REX_NODE_STR_CACHE.containsKey(key)) { rexNodeStr = ComputedColumnRewriter.extractCcRexNode(model, kylinConfig, cc.getInnerExpression()); @@ -322,4 +335,32 @@ public class ComputedColumnRewriter { } }); } + + private static String expressionCacheKey(NDataModel model, String expression) { + return modelCacheKey(model) + NAME_SPLIT + model.getRootFactTableName() + NAME_SPLIT + expression; + } + + private static String ccCacheKey(NDataModel model, ComputedColumnDesc cc) { + return modelCacheKey(model) + NAME_SPLIT + cc.getTableIdentity() + NAME_SPLIT + + StringHelper.backtickToDoubleQuote(cc.getInnerExpression()); + } + + private static String modelCacheKey(NDataModel model) { + return model.getUuid() + NAME_SPLIT + model.getMvcc() + NAME_SPLIT + model.getLastModified() + NAME_SPLIT + + StringUtils.defaultString(model.getProject()); + } + + private static int cacheWeight(String key, String value) { + return cacheWeight(key, value == null ? 0 : value.length()); + } + + private static int ccCacheWeight(String key, ComputedColumnDesc value) { + return cacheWeight(key, value == null || value.getInnerExpression() == null + ? 0 : value.getInnerExpression().length()); + } + + private static int cacheWeight(String key, int valueLength) { + long weight = (key == null ? 0L : key.length()) + valueLength; + return (int) Math.min(Integer.MAX_VALUE, Math.max(1L, weight)); + } } diff --git a/src/query/src/test/java/org/apache/kylin/query/util/ComputedColumnRewriterTest.java b/src/query/src/test/java/org/apache/kylin/query/util/ComputedColumnRewriterTest.java index 7722049c5e..922e49033f 100644 --- a/src/query/src/test/java/org/apache/kylin/query/util/ComputedColumnRewriterTest.java +++ b/src/query/src/test/java/org/apache/kylin/query/util/ComputedColumnRewriterTest.java @@ -19,11 +19,14 @@ package org.apache.kylin.query.util; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.ArrayList; import java.util.List; +import java.util.Map; import org.apache.calcite.sql.parser.SqlParseException; +import org.apache.commons.lang3.StringUtils; import org.apache.kylin.common.KylinConfig; import org.apache.kylin.junit.annotation.MetadataInfo; import org.apache.kylin.metadata.model.ComputedColumnDesc; @@ -107,6 +110,31 @@ class ComputedColumnRewriterTest { System.out.println("total cost: " + (System.currentTimeMillis() - s)); } + @Test + void testComputedColumnCachesAreBounded() { + Map<String, String> rexNodeCache = ComputedColumnRewriter.CC_TO_REX_NODE_STR_CACHE; + Map<String, ComputedColumnDesc> expressionCache = ComputedColumnRewriter.EXP_TO_CC_MAP; + rexNodeCache.clear(); + expressionCache.clear(); + try { + String largeRexNode = StringUtils.repeat("x", 200_000); + for (int i = 0; i < 10; i++) { + rexNodeCache.put("rex-" + i, largeRexNode); + } + assertTrue(rexNodeCache.size() < 10); + + ComputedColumnDesc cc = new ComputedColumnDesc(); + cc.setInnerExpression(StringUtils.repeat("y", 20_000)); + for (int i = 0; i < 10; i++) { + expressionCache.put("cc-" + i, cc); + } + assertTrue(expressionCache.size() < 10); + } finally { + rexNodeCache.clear(); + expressionCache.clear(); + } + } + @NotNull private static List<String> getStrings() { List<String> ccs = new ArrayList<>();
