Add config to ignore partial/volatile intervals and cache everything in cache V2

fili-core/src/main/java/com/yahoo/bard/webservice/config/BardFeatureFlag.java

@@ -10,6 +10,7 @@ public enum BardFeatureFlag implements FeatureFlag {
     @Deprecated DRUID_CACHE("druid_cache_enabled"),
     @Deprecated DRUID_CACHE_V2("druid_cache_v2_enabled"),
     QUERY_SPLIT("query_split_enabled"),
+    CACHEABLE_CHECK("cacheable_check"),
     TOP_N("top_n_enabled"),
     DATA_FILTER_SUBSTRING_OPERATIONS("data_filter_substring_operations_enabled"),
     INTERSECTION_REPORTING("intersection_reporting_enabled"),

fili-core/src/main/java/com/yahoo/bard/webservice/web/responseprocessors/CacheV2ResponseProcessor.java

@@ -2,6 +2,7 @@
 // Licensed under the terms of the Apache license. Please see LICENSE.md file distributed with this work for terms.
 package com.yahoo.bard.webservice.web.responseprocessors;
 
+import static com.yahoo.bard.webservice.config.BardFeatureFlag.CACHEABLE_CHECK;
 import static com.yahoo.bard.webservice.web.handlers.PartialDataRequestHandler.getPartialIntervalsWithDefault;
 import static com.yahoo.bard.webservice.web.handlers.VolatileDataRequestHandler.getVolatileIntervalsWithDefault;
 
@@ -85,7 +86,7 @@ public HttpErrorCallback getErrorCallback(DruidAggregationQuery<?> druidQuery) {
 
     @Override
     public void processResponse(JsonNode json, DruidAggregationQuery<?> druidQuery, LoggingContext metadata) {
-        if (isCacheable()) {
+        if (!CACHEABLE_CHECK.isOn() || isCacheable()) {
             String valueString = null;
             try {
                 valueString = writer.writeValueAsString(json);
@@ -112,6 +113,7 @@ public void processResponse(JsonNode json, DruidAggregationQuery<?> druidQuery,
                 );
             }
         }
+
         next.processResponse(json, druidQuery, metadata);
     }
 

fili-core/src/test/groovy/com/yahoo/bard/webservice/config/FeatureFlagRegistrySpec.groovy

@@ -36,9 +36,10 @@ class FeatureFlagRegistrySpec extends Specification {
 
         then:
         values == ["partial_data_enabled", "druid_cache_enabled", "druid_cache_v2_enabled", "query_split_enabled",
-                   "top_n_enabled", "data_filter_substring_operations_enabled", "intersection_reporting_enabled",
-                   "updated_metadata_collection_names_enabled", "druid_coordinator_metadata_enabled",
-                   "druid_lookup_metadata_enabled", "druid_dimensions_loader_enabled", "case_sensitive_keys_enabled"] as Set
+                   "cacheable_check", "top_n_enabled", "data_filter_substring_operations_enabled",
+                   "intersection_reporting_enabled", "updated_metadata_collection_names_enabled",
+                   "druid_coordinator_metadata_enabled", "druid_lookup_metadata_enabled",
+                   "druid_dimensions_loader_enabled", "case_sensitive_keys_enabled"] as Set
     }
 
     @Unroll

fili-core/src/test/groovy/com/yahoo/bard/webservice/web/responseprocessors/CacheV2ResponseProcessorSpec.groovy

@@ -3,6 +3,7 @@
 package com.yahoo.bard.webservice.web.responseprocessors
 
 import static com.yahoo.bard.webservice.async.ResponseContextUtils.createResponseContext
+import static com.yahoo.bard.webservice.config.BardFeatureFlag.CACHEABLE_CHECK
 import static com.yahoo.bard.webservice.web.responseprocessors.ResponseContextKeys.MISSING_INTERVALS_CONTEXT_KEY
 import static com.yahoo.bard.webservice.web.responseprocessors.ResponseContextKeys.VOLATILE_INTERVALS_CONTEXT_KEY
 
@@ -55,10 +56,17 @@ class CacheV2ResponseProcessorSpec extends Specification {
 
     CacheV2ResponseProcessor crp
 
+    boolean cacheableCheckIsOn
+
     def setup() {
         querySigningService.getSegmentSetId(_) >> Optional.of(1234L)
         segmentId = querySigningService.getSegmentSetId(groupByQuery).get()
         crp = new CacheV2ResponseProcessor(next, cacheKey, dataCache, querySigningService, MAPPER)
+        cacheableCheckIsOn = CACHEABLE_CHECK.isOn()
+    }
+
+    def cleanup() {
+        CACHEABLE_CHECK.setOn(cacheableCheckIsOn)
     }
 
     def "Test Constructor"() {
@@ -105,6 +113,7 @@ class CacheV2ResponseProcessorSpec extends Specification {
 
     def "After error saving to cache, process response continues"() {
         when:
+        CACHEABLE_CHECK.setOn(true)
         crp.processResponse(json, groupByQuery, null)
 
         then:
@@ -113,6 +122,17 @@ class CacheV2ResponseProcessorSpec extends Specification {
         1 * dataCache.set(cacheKey, segmentId, '[]') >> { throw new IllegalStateException() }
     }
 
+    def "After error is not saved to cache, process response continues"() {
+        when:
+        CACHEABLE_CHECK.setOn(false)
+        crp.processResponse(json, groupByQuery, null)
+
+        then:
+        0 * next.getResponseContext() >> responseContext
+        1 * next.processResponse(json, groupByQuery, null)
+        1 * dataCache.set(cacheKey, segmentId, '[]') >> { throw new IllegalStateException() }
+    }
+
     def "After json serialization error of the cache value, process response continues"() {
         setup:
         ObjectMapper localMapper = Mock(ObjectMapper)
@@ -131,32 +151,88 @@ class CacheV2ResponseProcessorSpec extends Specification {
         0 * dataCache.set(*_)
     }
 
-    def "Partial data doesn't cache and then continues"() {
-        setup:
+    def "When cacheable_check is turned off, partial data is cached and then continues"() {
+        setup: "we do not check cacheability"
+        CACHEABLE_CHECK.setOn(false)
+
+        when: "we process response"
+        crp.processResponse(json, groupByQuery, null)
+
+        then: "query is not checked for cacheability and partial data is cached"
+        0 * next.getResponseContext() >> responseContext
+        1 * next.processResponse(json, groupByQuery, null)
+        1 * dataCache.set(*_)
+    }
+
+    def "When cacheable_check is turned on, partial data is not cached after the cacheable check and then continues"() {
+        setup: "we check cacheability but query is not cacheable"
+        CACHEABLE_CHECK.setOn(true)
         ResponseContext responseContext = createResponseContext([(MISSING_INTERVALS_CONTEXT_KEY.name) : nonEmptyIntervals])
 
-        when:
+        when: "we process response"
         crp.processResponse(json, groupByQuery, null)
 
-        then:
+        then: "query is checked for cacheability and partial data is not cached"
         2 * next.getResponseContext() >> responseContext
         1 * next.processResponse(json, groupByQuery, null)
         0 * dataCache.set(*_)
     }
 
-    def "Volatile data doesn't cache and then continues"() {
-        setup:
-        ResponseContext responseContext = createResponseContext([(VOLATILE_INTERVALS_CONTEXT_KEY.name) : nonEmptyIntervals])
+    def "When cacheable_check is turned on, partial data is cached after the cacheable check and then continues"() {
+        setup: "we check cacheability and query is cacheable"
+        CACHEABLE_CHECK.setOn(true)
+        ResponseContext responseContext = createResponseContext([:])
 
-        when:
+        when: "we process response"
         crp.processResponse(json, groupByQuery, null)
 
-        then:
+        then: "query is checked for cacheability and partial data is cached"
         2 * next.getResponseContext() >> responseContext
         1 * next.processResponse(json, groupByQuery, null)
+        1 * dataCache.set(*_)
+    }
+
+    def "When cacheable_check is turned off, volatile data is cached and then continues"() {
+        setup: "we do not check cacheability"
+        CACHEABLE_CHECK.setOn(false)
+
+        when: "we process response"
+        crp.processResponse(json, groupByQuery, null)
+
+        then: "query is not checked for cacheability and partial data is cached"
+        0 * next.getResponseContext() >> responseContext // isCacheable() is not called
+        1 * next.processResponse(json, groupByQuery, null)
+        1 * dataCache.set(*_)
+    }
+
+    def "When cacheable_check is turned on, volatile data is not cached after the cacheable check and then continues"() {
+        setup: "we check cacheability but query is not cacheable"
+        CACHEABLE_CHECK.setOn(true)
+        ResponseContext responseContext = createResponseContext([(VOLATILE_INTERVALS_CONTEXT_KEY.name) : nonEmptyIntervals])
+
+        when: "we process response"
+        crp.processResponse(json, groupByQuery, null)
+
+        then: "query is checked for cacheability and partial data is not cached"
+        2 * next.getResponseContext() >> responseContext // isCacheable() is not called
+        1 * next.processResponse(json, groupByQuery, null)
         0 * dataCache.set(*_)
     }
 
+    def "When cacheable_check is turned on, volatile data is cached after the cacheable check and then continues"() {
+        setup: "we check cacheability and query is cacheable"
+        CACHEABLE_CHECK.setOn(true)
+        ResponseContext responseContext = createResponseContext([:])
+
+        when: "we process response"
+        crp.processResponse(json, groupByQuery, null)
+
+        then: "query is checked for cacheability and partial data is cached"
+        2 * next.getResponseContext() >> responseContext // isCacheable() is not called
+        1 * next.processResponse(json, groupByQuery, null)
+        1 * dataCache.set(*_)
+    }
+
     def "Overly long data doesn't cache and then continues"() {
         setup: "Save the old max-length-to-cache so we can restore it later"
         String max_druid_response_length_to_cache_key = SYSTEM_CONFIG.getPackageVariableName(