Add cache settings

Author: vdegans

exec/java-exec/src/main/java/org/apache/drill/exec/cache/CustomCacheManager.java

@@ -30,14 +30,14 @@ public class CustomCacheManager {
   private static void loadConfig() {
     DrillConfig config = DrillConfig.create();
 
-    queryMaxEntries = getConfigInt(config, "custom.cache.query.max_entries", 100);
-    queryTtlMinutes = getConfigInt(config, "custom.cache.query.ttl_minutes", 300);
-    transformMaxEntries = getConfigInt(config, "custom.cache.transform.max_entries", 100);
-    transformTtlMinutes = getConfigInt(config, "custom.cache.transform.ttl_minutes", 300);
+    queryMaxEntries = getConfigInt(config, "planner.query.cache.max_entries_amount", 100);
+    queryTtlMinutes = getConfigInt(config, "planner.query.cache.plan_cache_ttl_minutes", 300);
+    transformMaxEntries = getConfigInt(config, "planner.transform.cache.max_entries_amount", 100);
+    transformTtlMinutes = getConfigInt(config, "planner.transform.plan_cache_ttl_minutes", 300);
 
     queryCache = Caffeine.newBuilder()
         .maximumSize(queryMaxEntries)
-        .expireAfterWrite(queryTtlMinutes, TimeUnit.MINUTES)
+        .expireAfterWrite(queryTtlMinutes, TimeUnit.MILLISECONDS)
         .recordStats()
         .build();
 
@@ -75,4 +75,4 @@ public static void logCacheStats() {
     logger.info("Transform Cache Stats: " + transformCache.stats());
     logger.info("Transform Cache Size: " + transformCache.estimatedSize());
   }
-}
+}

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);
   }

exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/DefaultSqlHandler.java

@@ -410,14 +410,20 @@ protected RelNode transform(PlannerType plannerType, PlannerPhase phase, RelNode
     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);
 
-    // Create a cache key based on the input parameters
-    CacheKey key = new CacheKey(plannerType, phase, input, targetTraits);
+    CacheKey key = null;
 
-    RelNode cachedResult = CustomCacheManager.getTransformedPlan(key);
-    if (cachedResult != 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) {
         CustomCacheManager.logCacheStats();
         return cachedResult;
+      }
     }
 
     final RelNode output;
@@ -465,9 +471,14 @@ protected RelNode transform(PlannerType plannerType, PlannerPhase phase, RelNode
     }
     }
 
-    // Store the result in the cache before returning
-    CustomCacheManager.putTransformedPlan(key, output);
-    CustomCacheManager.logCacheStats();
+    if (planCacheEnabled) {
+      logger.info("planCache enabled, storing transformedplan");
+      // Store the result in the cache before returning
+      CustomCacheManager.putTransformedPlan(key, output);
+      CustomCacheManager.logCacheStats();
+    } else {
+      logger.info("planCache disabled, not storing transformedplan");
+    }
 
     if (log) {
       log(plannerType, phase, output, logger, watch);

exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java

@@ -82,6 +82,10 @@ public static CaseInsensitiveMap<OptionDefinition> 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),

exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java

@@ -39,6 +39,7 @@
 import org.apache.drill.exec.physical.base.PhysicalOperator;
 import org.apache.drill.exec.planner.fragment.Fragment;
 import org.apache.drill.exec.planner.fragment.MakeFragmentsVisitor;
+import org.apache.drill.exec.planner.physical.PlannerSettings;
 import org.apache.drill.exec.planner.sql.DirectPlan;
 import org.apache.drill.exec.planner.sql.DrillSqlWorker;
 import org.apache.drill.exec.proto.BitControl.PlanFragment;
@@ -56,6 +57,7 @@
 import org.apache.drill.exec.rpc.UserClientConnection;
 import org.apache.drill.exec.server.DrillbitContext;
 import org.apache.drill.exec.server.FailureUtils;
+import org.apache.drill.exec.server.options.OptionManager;
 import org.apache.drill.exec.server.options.OptionSet;
 import org.apache.drill.exec.testing.ControlsInjector;
 import org.apache.drill.exec.testing.ControlsInjectorFactory;
@@ -595,23 +597,50 @@ private void logWorkUnit(QueryWorkUnit queryWorkUnit) {
         queryId, queryWorkUnit.stringifyFragments()));
   }
 
-  private void runSQL(final String sql) throws ExecutionSetupException {
-    final Pointer<String> textPlan = new Pointer<>();
+  private PhysicalPlan getOrBuildPlan(String sql, Pointer<String> textPlan, boolean planCacheEnabled)
+      throws ExecutionSetupException {
 
-    PhysicalPlan plan = CustomCacheManager.getQueryPlan(sql);
+    PhysicalPlan plan = null;
+
+    if (planCacheEnabled) {
+      logger.info("Cache enabled, checking entries");
+      plan = CustomCacheManager.getQueryPlan(sql);
+
+      if (plan == null) {
+        logger.info("Cache miss, generating new plan");
+        plan = DrillSqlWorker.getPlan(queryContext, sql, textPlan);
+
+        if (isCacheableQuery(sql)) {
+          CustomCacheManager.putQueryPlan(sql, plan);
+          CustomCacheManager.logCacheStats();
+        }
+      } else {
+        logger.info("Using cached plan");
+      }
 
-    if (plan == null) {
-      logger.info("Cache miss, generating new plan");
-      plan = DrillSqlWorker.getPlan(queryContext, sql, textPlan);
     } else {
-      logger.info("Using cached plan");
+      plan = DrillSqlWorker.getPlan(queryContext, sql, textPlan);
     }
 
-    if(sql.trim().startsWith("SELECT")) {
-      CustomCacheManager.putQueryPlan(sql, plan);
-      CustomCacheManager.logCacheStats();
+    return plan;
+  }
+
+  private boolean isCacheableQuery(String sql) {
+    return sql.trim().toUpperCase().startsWith("SELECT");
+  }
+
+  private void runSQL(final String sql) throws ExecutionSetupException {
+    final Pointer<String> textPlan = new Pointer<>();
+    final OptionManager options = queryContext.getOptions();
+    final boolean planCacheEnabled = options.getOption(PlannerSettings.PLAN_CACHE);
+    if (planCacheEnabled) {
+      logger.info("PlanCache is enabled");
+    } else {
+      logger.info("PlanCache is disabled");
     }
 
+    PhysicalPlan plan = getOrBuildPlan(sql, textPlan, planCacheEnabled);
+
     runPhysicalPlan(plan, textPlan);
   }
 

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: true,
+    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%",