diff --git a/exec/java-exec/pom.xml b/exec/java-exec/pom.xml index fe2c229a9a0..6b1dd64d304 100644 --- a/exec/java-exec/pom.xml +++ b/exec/java-exec/pom.xml @@ -697,6 +697,12 @@ swagger-jaxrs2-servlet-initializer-v2-jakarta ${swagger.version} + + + com.github.ben-manes.caffeine + caffeine + 2.9.3 + diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/cache/CustomCacheManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/CustomCacheManager.java new file mode 100644 index 00000000000..712e53f3a76 --- /dev/null +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/cache/CustomCacheManager.java @@ -0,0 +1,152 @@ +/* + * 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.drill.exec.cache; + +import java.util.concurrent.TimeUnit; + +import org.apache.calcite.rel.RelNode; +import org.apache.drill.common.config.DrillConfig; +import org.apache.drill.exec.physical.PhysicalPlan; +import org.apache.drill.exec.planner.sql.handlers.DefaultSqlHandler.CacheKey; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; + +/** + * Manages caches for query plans and transform results to improve query planning performance. + * Uses Caffeine cache with configurable TTL and maximum size. + */ +public class CustomCacheManager { + private static final Logger logger = LoggerFactory.getLogger(CustomCacheManager.class); + + private static volatile Cache queryCache; + private static volatile Cache transformCache; + private static volatile boolean initialized = false; + + private static final int DEFAULT_MAX_ENTRIES = 100; + private static final int DEFAULT_TTL_MINUTES = 300; + + private CustomCacheManager() { + // Utility class + } + + /** + * Lazily initializes the caches if not already initialized. + * Uses double-checked locking for thread safety. + */ + private static void ensureInitialized() { + if (!initialized) { + synchronized (CustomCacheManager.class) { + if (!initialized) { + initializeCaches(); + initialized = true; + } + } + } + } + + private static void initializeCaches() { + DrillConfig config = DrillConfig.create(); + + int queryMaxEntries = getConfigInt(config, "planner.query.cache.max_entries_amount", DEFAULT_MAX_ENTRIES); + int queryTtlMinutes = getConfigInt(config, "planner.query.cache.plan_cache_ttl_minutes", DEFAULT_TTL_MINUTES); + int transformMaxEntries = getConfigInt(config, "planner.transform.cache.max_entries_amount", DEFAULT_MAX_ENTRIES); + int transformTtlMinutes = getConfigInt(config, "planner.transform.cache.plan_cache_ttl_minutes", DEFAULT_TTL_MINUTES); + + queryCache = Caffeine.newBuilder() + .maximumSize(queryMaxEntries) + .expireAfterWrite(queryTtlMinutes, TimeUnit.MINUTES) + .recordStats() + .build(); + + transformCache = Caffeine.newBuilder() + .maximumSize(transformMaxEntries) + .expireAfterWrite(transformTtlMinutes, TimeUnit.MINUTES) + .recordStats() + .build(); + + logger.debug("Query plan cache initialized with maxEntries={}, ttlMinutes={}", queryMaxEntries, queryTtlMinutes); + logger.debug("Transform cache initialized with maxEntries={}, ttlMinutes={}", transformMaxEntries, transformTtlMinutes); + } + + private static int getConfigInt(DrillConfig config, String path, int defaultValue) { + if (config.hasPath(path)) { + int value = config.getInt(path); + logger.debug("Config {}: {}", path, value); + return value; + } + logger.debug("Config {} not found, using default: {}", path, defaultValue); + return defaultValue; + } + + public static PhysicalPlan getQueryPlan(String sql) { + ensureInitialized(); + return queryCache.getIfPresent(sql); + } + + public static void putQueryPlan(String sql, PhysicalPlan plan) { + ensureInitialized(); + queryCache.put(sql, plan); + } + + public static RelNode getTransformedPlan(CacheKey key) { + ensureInitialized(); + return transformCache.getIfPresent(key); + } + + public static void putTransformedPlan(CacheKey key, RelNode plan) { + ensureInitialized(); + transformCache.put(key, plan); + } + + public static void logCacheStats() { + ensureInitialized(); + if (logger.isDebugEnabled()) { + logger.debug("Query Cache Stats: {}", queryCache.stats()); + logger.debug("Query Cache Size: {}", queryCache.estimatedSize()); + logger.debug("Transform Cache Stats: {}", transformCache.stats()); + logger.debug("Transform Cache Size: {}", transformCache.estimatedSize()); + } + } + + /** + * Clears both caches. Useful for testing. + */ + public static void clearCaches() { + if (initialized) { + queryCache.invalidateAll(); + transformCache.invalidateAll(); + } + } + + /** + * Resets the cache manager, forcing reinitialization on next use. + * Useful for testing with different configurations. + */ + public static synchronized void reset() { + if (initialized) { + queryCache.invalidateAll(); + transformCache.invalidateAll(); + queryCache = null; + transformCache = null; + initialized = false; + } + } +} diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java index 6fa145a1b01..e8ab405c557 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java @@ -132,6 +132,20 @@ public class PlannerSettings implements Context{ public static final String UNIONALL_DISTRIBUTE_KEY = "planner.enable_unionall_distribute"; public static final BooleanValidator UNIONALL_DISTRIBUTE = new BooleanValidator(UNIONALL_DISTRIBUTE_KEY, null); + public static final BooleanValidator PLAN_CACHE = new BooleanValidator("planner.cache.enable", + new OptionDescription("Enables caching of generated query plans in memory, so repeated queries can bypass the planning phase and execute faster.") + ); + + // Only settable in config, due to pub-sub requirements for recreating the cache on value change + // public static final RangeLongValidator PLAN_CACHE_TTL = new RangeLongValidator("planner.cache.ttl_minutes", + // 0, Long.MAX_VALUE, + // new OptionDescription("Time-to-live for cached query plans in minutes. Plans older than this are evicted. Default is 0 (disabled)") + // ); + // public static final RangeLongValidator MAX_CACHE_ENTRIES = new RangeLongValidator("planner.cache.max_entries", + // 1, Long.MAX_VALUE, + // new OptionDescription("Maximum total number of entries for cached query plans. When exceeded, least recently used plans are evicted.") + // ); + // ------------------------------------------- Index planning related options BEGIN -------------------------------------------------------------- public static final String USE_SIMPLE_OPTIMIZER_KEY = "planner.use_simple_optimizer"; public static final BooleanValidator USE_SIMPLE_OPTIMIZER = new BooleanValidator(USE_SIMPLE_OPTIMIZER_KEY, @@ -416,6 +430,10 @@ public boolean isUnionAllDistributeEnabled() { return options.getOption(UNIONALL_DISTRIBUTE); } + public boolean isPlanCacheEnabled() { + return options.getOption(PLAN_CACHE); + } + public boolean isParquetRowGroupFilterPushdownPlanningEnabled() { return options.getOption(PARQUET_ROWGROUP_FILTER_PUSHDOWN_PLANNING); } diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java index c706f8f3733..103a5684ce6 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java @@ -31,10 +31,13 @@ import org.apache.calcite.tools.ValidationException; import org.apache.drill.common.exceptions.UserException; import org.apache.drill.exec.ExecConstants; +import org.apache.drill.exec.cache.CustomCacheManager; import org.apache.drill.exec.exception.MetadataException; import org.apache.drill.exec.ops.QueryContext; import org.apache.drill.exec.ops.QueryContext.SqlStatementType; import org.apache.drill.exec.physical.PhysicalPlan; +import org.apache.drill.exec.planner.physical.PlannerSettings; +import org.apache.drill.exec.planner.sql.conversion.SqlConverter; import org.apache.drill.exec.planner.sql.handlers.AbstractSqlHandler; import org.apache.drill.exec.planner.sql.handlers.AnalyzeTableHandler; import org.apache.drill.exec.planner.sql.handlers.DefaultSqlHandler; @@ -52,7 +55,6 @@ import org.apache.drill.exec.planner.sql.parser.DrillSqlDescribeTable; import org.apache.drill.exec.planner.sql.parser.DrillSqlResetOption; import org.apache.drill.exec.planner.sql.parser.SqlSchema; -import org.apache.drill.exec.planner.sql.conversion.SqlConverter; import org.apache.drill.exec.proto.UserBitShared.DrillPBError; import org.apache.drill.exec.testing.ControlsInjector; import org.apache.drill.exec.testing.ControlsInjectorFactory; @@ -210,6 +212,7 @@ private static PhysicalPlan getPhysicalPlan(QueryContext context, String sql, Po /** * Converts sql query string into query physical plan. + * If plan caching is enabled, attempts to retrieve a cached plan first. * * @param context query context * @param sql sql query @@ -219,6 +222,17 @@ private static PhysicalPlan getPhysicalPlan(QueryContext context, String sql, Po private static PhysicalPlan getQueryPlan(QueryContext context, String sql, Pointer textPlan) throws ForemanSetupException, RelConversionException, IOException, ValidationException { + final boolean planCacheEnabled = context.getOptions().getOption(PlannerSettings.PLAN_CACHE); + + // Check cache first if enabled + if (planCacheEnabled) { + PhysicalPlan cachedPlan = CustomCacheManager.getQueryPlan(sql); + if (cachedPlan != null) { + logger.debug("Using cached query plan for SQL: {}", sql.substring(0, Math.min(sql.length(), 100))); + return cachedPlan; + } + } + final SqlConverter parser = new SqlConverter(context); injector.injectChecked(context.getExecutionControls(), "sql-parsing", ForemanSetupException.class); final SqlNode sqlNode = checkAndApplyAutoLimit(parser, context, sql); @@ -295,7 +309,25 @@ private static PhysicalPlan getQueryPlan(QueryContext context, String sql, Point context.getOptions().setLocalOption(ExecConstants.RETURN_RESULT_SET_FOR_DDL, true); } - return handler.getPlan(sqlNode); + PhysicalPlan physicalPlan = handler.getPlan(sqlNode); + + // Cache the plan if caching is enabled and this is a cacheable query type + if (planCacheEnabled && isCacheableQuery(sqlNode)) { + CustomCacheManager.putQueryPlan(sql, physicalPlan); + logger.debug("Cached query plan for SQL: {}", sql.substring(0, Math.min(sql.length(), 100))); + } + + return physicalPlan; + } + + /** + * Determines if a query type should be cached. + * Only SELECT queries are cached; DDL and other modifying statements should not be cached. + */ + private static boolean isCacheableQuery(SqlNode sqlNode) { + SqlKind kind = sqlNode.getKind(); + // Only cache SELECT queries, not DDL or DML + return kind == SqlKind.SELECT || kind == SqlKind.ORDER_BY || kind == SqlKind.UNION; } private static boolean isAutoLimitShouldBeApplied(SqlNode sqlNode, int queryMaxRows) { diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java index 6cc4d3bc4bc..41966793727 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java @@ -20,15 +20,10 @@ import java.io.IOException; import java.util.Collection; import java.util.List; +import java.util.Objects; import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; -import com.fasterxml.jackson.databind.ser.PropertyFilter; -import com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter; -import com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider; -import org.apache.drill.exec.util.Utilities; -import com.google.common.collect.ImmutableList; -import com.google.common.collect.Sets; import org.apache.calcite.plan.RelOptCostImpl; import org.apache.calcite.plan.RelOptPlanner; import org.apache.calcite.plan.RelOptRule; @@ -66,6 +61,7 @@ import org.apache.drill.common.logical.PlanProperties.PlanPropertiesBuilder; import org.apache.drill.common.logical.PlanProperties.PlanType; import org.apache.drill.exec.ExecConstants; +import org.apache.drill.exec.cache.CustomCacheManager; import org.apache.drill.exec.ops.QueryContext; import org.apache.drill.exec.physical.PhysicalPlan; import org.apache.drill.exec.physical.base.AbstractPhysicalVisitor; @@ -104,13 +100,20 @@ import org.apache.drill.exec.server.options.OptionManager; import org.apache.drill.exec.store.StoragePlugin; import org.apache.drill.exec.util.Pointer; +import org.apache.drill.exec.util.Utilities; import org.apache.drill.exec.work.foreman.ForemanSetupException; import org.apache.drill.exec.work.foreman.SqlUnsupportedException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import com.google.common.base.Preconditions; import com.google.common.base.Stopwatch; +import com.google.common.collect.ImmutableList; import com.google.common.collect.Lists; +import com.google.common.collect.Sets; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.fasterxml.jackson.databind.ser.PropertyFilter; +import com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter; +import com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider; public class DefaultSqlHandler extends AbstractSqlHandler { private static final Logger logger = LoggerFactory.getLogger(DefaultSqlHandler.class); @@ -250,7 +253,6 @@ protected DrillRel convertToRawDrel(final RelNode relNode) throws SqlUnsupported // hep is enabled and hep pruning is enabled. intermediateNode2 = transform(PlannerType.HEP_BOTTOM_UP, PlannerPhase.PARTITION_PRUNING, transitiveClosureNode); - } else { // Only hep is enabled final RelNode intermediateNode = @@ -361,11 +363,65 @@ protected RelNode transform(PlannerType plannerType, PlannerPhase phase, RelNode * @param log Whether to log the planning phase. * @return The transformed relnode. */ + + + // A simple cache key class that uses the relevant parameters + public static class CacheKey { + private final PlannerPhase phase; + private PlannerType plannerType; + private RelNode input; + private RelTraitSet targetTraits; + + public CacheKey(PlannerType plannerType, PlannerPhase phase, RelNode input, RelTraitSet targetTraits) { + this.plannerType = plannerType; + this.phase = phase; + this.input = input; + this.targetTraits = targetTraits; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + CacheKey cacheKey = (CacheKey) o; + return phase == cacheKey.phase && + plannerType == cacheKey.plannerType && + input.deepEquals(cacheKey.input) && + targetTraits.equals(cacheKey.targetTraits); + } + + @Override + public int hashCode() { + return Objects.hash(phase.name(), plannerType.name(), input.deepHashCode(), targetTraits); + } + + } + + protected RelNode transform(PlannerType plannerType, PlannerPhase phase, RelNode input, RelTraitSet targetTraits, boolean log) { final Stopwatch watch = Stopwatch.createStarted(); final RuleSet rules = config.getRules(phase, input); final RelTraitSet toTraits = targetTraits.simplify(); + final OptionManager options = context.getOptions(); + final boolean planCacheEnabled = options.getOption(PlannerSettings.PLAN_CACHE); + + CacheKey key = null; + + if (planCacheEnabled) { + // Create a cache key based on the input parameters + key = new CacheKey(plannerType, phase, input, targetTraits); + + RelNode cachedResult = CustomCacheManager.getTransformedPlan(key); + if (cachedResult != null) { + logger.debug("Cache hit for transform phase: {}", phase); + return cachedResult; + } + } final RelNode output; switch (plannerType) { @@ -404,11 +460,15 @@ protected RelNode transform(PlannerType plannerType, PlannerPhase phase, RelNode .getName()); output = program.run(planner, input, toTraits, ImmutableList.of(), ImmutableList.of()); - break; } } + if (planCacheEnabled) { + CustomCacheManager.putTransformedPlan(key, output); + logger.debug("Cached transform result for phase: {}", phase); + } + if (log) { log(plannerType, phase, output, logger, watch); } diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java index 3cee4096e09..11ffe93ecd3 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java @@ -82,6 +82,10 @@ public static CaseInsensitiveMap createDefaultOptionDefinition // here. @SuppressWarnings("deprecation") final OptionDefinition[] definitions = new OptionDefinition[]{ + new OptionDefinition(PlannerSettings.PLAN_CACHE), + // new OptionDefinition(PlannerSettings.PLAN_CACHE_TTL), + // new OptionDefinition(PlannerSettings.MAX_CACHE_ENTRIES), + new OptionDefinition(PlannerSettings.CONSTANT_FOLDING), new OptionDefinition(PlannerSettings.EXCHANGE), new OptionDefinition(PlannerSettings.HASHAGG), diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java index a099b96b123..181817c22d3 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java @@ -17,18 +17,17 @@ */ package org.apache.drill.exec.work.foreman; -import com.fasterxml.jackson.databind.ObjectMapper; -import org.apache.drill.common.util.JacksonUtils; -import org.apache.drill.exec.work.filter.RuntimeFilterRouter; -import com.google.common.base.Preconditions; -import com.google.common.collect.Lists; -import com.google.protobuf.InvalidProtocolBufferException; -import io.netty.util.concurrent.Future; -import io.netty.util.concurrent.GenericFutureListener; +import static org.apache.drill.exec.server.FailureUtils.EXIT_CODE_HEAP_OOM; + +import java.io.IOException; +import java.util.Date; +import java.util.List; + import org.apache.drill.common.exceptions.ExecutionSetupException; import org.apache.drill.common.exceptions.UserException; import org.apache.drill.common.logical.LogicalPlan; import org.apache.drill.common.logical.PlanProperties.Generator.ResultMode; +import org.apache.drill.common.util.JacksonUtils; import org.apache.drill.exec.ExecConstants; import org.apache.drill.exec.exception.OptimizerException; import org.apache.drill.exec.exception.OutOfMemoryException; @@ -62,17 +61,20 @@ import org.apache.drill.exec.util.Pointer; import org.apache.drill.exec.work.QueryWorkUnit; import org.apache.drill.exec.work.WorkManager.WorkerBee; -import org.apache.drill.exec.work.foreman.rm.QueryQueue.QueueTimeoutException; +import org.apache.drill.exec.work.filter.RuntimeFilterRouter; import org.apache.drill.exec.work.foreman.rm.QueryQueue.QueryQueueException; +import org.apache.drill.exec.work.foreman.rm.QueryQueue.QueueTimeoutException; import org.apache.drill.exec.work.foreman.rm.QueryResourceManager; +import com.google.common.base.Preconditions; +import com.google.common.collect.Lists; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.IOException; -import java.util.Date; -import java.util.List; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.protobuf.InvalidProtocolBufferException; -import static org.apache.drill.exec.server.FailureUtils.EXIT_CODE_HEAP_OOM; +import io.netty.util.concurrent.Future; +import io.netty.util.concurrent.GenericFutureListener; /** * Foreman manages all the fragments (local and remote) for a single query where this @@ -269,9 +271,11 @@ public void run() { final String sql = queryRequest.getPlan(); // log query id, username and query text before starting any real work. Also, put // them together such that it is easy to search based on query id + long start = new Date().getTime(); logger.info("Query text for query with id {} issued by {}: {}", queryIdString, queryContext.getQueryUserName(), sql); runSQL(sql); + logger.info("RunSQL is executed within {}", new Date().getTime() - start); break; case EXECUTION: runFragment(queryRequest.getFragmentsList()); @@ -481,6 +485,7 @@ private void runFragment(List fragmentsList) throws ExecutionSetup * Moves query to RUNNING state. */ private void startQueryProcessing() { + logger.info("Starting query processing"); enqueue(); runFragments(); queryStateProcessor.moveToState(QueryState.RUNNING, null); diff --git a/exec/java-exec/src/main/resources/drill-module.conf b/exec/java-exec/src/main/resources/drill-module.conf index 7541a99e2dd..836c09d525e 100644 --- a/exec/java-exec/src/main/resources/drill-module.conf +++ b/exec/java-exec/src/main/resources/drill-module.conf @@ -697,6 +697,12 @@ drill.exec.options: { planner.width.max_per_node: 0, planner.width.max_per_query: 1000, + planner.cache.enable: false, + planner.query.cache.ttl_minutes: 0, + planner.transform.cache.ttl_minutes: 0, + planner.query.cache.max_entries_amount: 100, + planner.transform.cache.max_entries_amount: 100, + prepare.statement.create_timeout_ms: 30000, security.admin.user_groups: "%drill_process_user_groups%", security.admin.users: "%drill_process_user%", diff --git a/exec/jdbc-all/pom.xml b/exec/jdbc-all/pom.xml index 161fa59126d..ddd61836467 100644 --- a/exec/jdbc-all/pom.xml +++ b/exec/jdbc-all/pom.xml @@ -436,6 +436,7 @@ antlr:* com.beust:* com.dropbox.* + com.github.ben-manes.caffeine:caffeine com.github.stefanbirkner com.google.code.findbugs:jsr305:* com.googlecode.json-simple:*