Closes #53. Mark config as transient field (#55)

Author: hbxie

src/main/java/com/yahoo/bullet/aggregations/CountDistinct.java

@@ -45,7 +45,7 @@ public CountDistinct(Aggregation aggregation, BulletConfig config) {
         super(aggregation, config);
         Map<String, Object> attributes = aggregation.getAttributes();
 
-        ResizeFactor resizeFactor = getResizeFactor(BulletConfig.COUNT_DISTINCT_AGGREGATION_SKETCH_RESIZE_FACTOR);
+        ResizeFactor resizeFactor = getResizeFactor(config, BulletConfig.COUNT_DISTINCT_AGGREGATION_SKETCH_RESIZE_FACTOR);
         float samplingProbability = config.getAs(BulletConfig.COUNT_DISTINCT_AGGREGATION_SKETCH_SAMPLING, Float.class);
         Family family = getFamily(config.getAs(BulletConfig.COUNT_DISTINCT_AGGREGATION_SKETCH_FAMILY, String.class));
         int nominalEntries = config.getAs(BulletConfig.COUNT_DISTINCT_AGGREGATION_SKETCH_ENTRIES, Integer.class);

src/main/java/com/yahoo/bullet/aggregations/Distribution.java

@@ -71,6 +71,7 @@ public boolean isMe(String name) {
 
     // Copy of the aggregation
     private Aggregation aggregation;
+    private BulletRecordProvider provider;
 
     // Validation
     public static final Map<String, Type> SUPPORTED_DISTRIBUTION_TYPES = new HashMap<>();
@@ -108,6 +109,7 @@ public Distribution(Aggregation aggregation, BulletConfig config) {
         int pointLimit = config.getAs(BulletConfig.DISTRIBUTION_AGGREGATION_MAX_POINTS, Integer.class);
         maxPoints = Math.min(pointLimit, aggregation.getSize());
         this.aggregation = aggregation;
+        this.provider = config.getBulletRecordProvider();
 
         // The sketch is initialized in initialize!
     }
@@ -130,7 +132,7 @@ public Optional<List<BulletError>> initialize() {
         }
 
         // Try to initialize sketch now
-        sketch = getSketch(entries, maxPoints, rounding, type, attributes, config.getBulletRecordProvider());
+        sketch = getSketch(entries, maxPoints, rounding, type, attributes, provider);
 
         if (sketch == null) {
             return Optional.of(type == Type.QUANTILE ?

src/main/java/com/yahoo/bullet/aggregations/GroupAll.java

@@ -13,6 +13,7 @@
 import com.yahoo.bullet.common.Utilities;
 import com.yahoo.bullet.parsing.Aggregation;
 import com.yahoo.bullet.record.BulletRecord;
+import com.yahoo.bullet.record.BulletRecordProvider;
 import com.yahoo.bullet.result.Clip;
 import lombok.extern.slf4j.Slf4j;
 
@@ -29,7 +30,7 @@ public class GroupAll implements Strategy {
     private GroupData data;
 
     private Set<GroupOperation> operations;
-    private BulletConfig config;
+    private BulletRecordProvider provider;
     /**
      * Constructor that requires an {@link Aggregation}.
      *
@@ -40,7 +41,7 @@ public GroupAll(Aggregation aggregation, BulletConfig config) {
         // GroupOperations is all we care about - size etc. are meaningless for Group All since it's a single result
         operations = GroupOperation.getOperations(aggregation.getAttributes());
         data = new GroupData(operations);
-        this.config = config;
+        this.provider = config.getBulletRecordProvider();
     }
 
     @Override
@@ -74,7 +75,7 @@ public Clip getResult() {
     @Override
     public List<BulletRecord> getRecords() {
         List<BulletRecord> list = new ArrayList<>();
-        list.add(data.getMetricsAsBulletRecord(config.getBulletRecordProvider()));
+        list.add(data.getMetricsAsBulletRecord(provider));
         return list;
     }
 

src/main/java/com/yahoo/bullet/aggregations/GroupBy.java

@@ -53,7 +53,7 @@ public GroupBy(Aggregation aggregation, BulletConfig config) {
         Map<GroupOperation, Number> metrics = GroupData.makeInitialMetrics(operations);
         container = new CachingGroupData(null, metrics);
 
-        ResizeFactor resizeFactor = getResizeFactor(BulletConfig.GROUP_AGGREGATION_SKETCH_RESIZE_FACTOR);
+        ResizeFactor resizeFactor = getResizeFactor(config, BulletConfig.GROUP_AGGREGATION_SKETCH_RESIZE_FACTOR);
         float samplingProbability = config.getAs(BulletConfig.GROUP_AGGREGATION_SKETCH_SAMPLING, Float.class);
 
         // Default at 512 gives a 13.27% error rate at 99.73% confidence (3 SD). Irrelevant since we are using this to

src/main/java/com/yahoo/bullet/aggregations/KMVStrategy.java

@@ -28,11 +28,12 @@ abstract class KMVStrategy<S extends KMVSketch> extends SketchingStrategy<S> {
     /**
      * Converts a integer representing the resizing for Sketches into a {@link ResizeFactor}.
      *
+     * @param config The config that has relevant configs for this strategy.
      * @param key The key to get the configured resize factor from the configuration.
      * @return A {@link ResizeFactor} represented by the integer or {@link ResizeFactor#X8} otherwise.
      */
     @SuppressWarnings("unchecked")
-    ResizeFactor getResizeFactor(String key) {
+    ResizeFactor getResizeFactor(BulletConfig config, String key) {
         return getResizeFactor(config.getAs(key, Integer.class));
     }
     /**

src/main/java/com/yahoo/bullet/aggregations/SketchingStrategy.java

@@ -33,8 +33,6 @@
 public abstract class SketchingStrategy<S extends Sketch> implements Strategy {
     // The metadata concept to key mapping
     protected final Map<String, String> metadataKeys;
-    // A copy of the configuration
-    protected final BulletConfig config;
 
     // Separator for multiple fields when inserting into the Sketch
     protected final String separator;
@@ -46,6 +44,8 @@ public abstract class SketchingStrategy<S extends Sketch> implements Strategy {
     // The Sketch that should be initialized by a child class
     protected S sketch;
 
+    private boolean shouldMeta;
+
     /**
      * The constructor for creating a Sketch based strategy.
      *
@@ -54,9 +54,9 @@ public abstract class SketchingStrategy<S extends Sketch> implements Strategy {
      */
     @SuppressWarnings("unchecked")
     public SketchingStrategy(Aggregation aggregation, BulletConfig config) {
-        this.config = config;
         metadataKeys = (Map<String, String>) config.getAs(BulletConfig.RESULT_METADATA_METRICS, Map.class);
         separator = config.getAs(BulletConfig.AGGREGATION_COMPOSITE_FIELD_SEPARATOR, String.class);
+        shouldMeta = config.getAs(BulletConfig.RESULT_METADATA_ENABLE, Boolean.class);
 
         fieldsToNames = aggregation.getFields();
         fields = Utilities.isEmpty(fieldsToNames) ? Collections.emptyList() : new ArrayList<>(fieldsToNames.keySet());
@@ -123,7 +123,6 @@ List<String> decomposeField(String field) {
     }
 
     private String getMetaKey() {
-        boolean shouldMeta = config.getAs(BulletConfig.RESULT_METADATA_ENABLE, Boolean.class);
         return shouldMeta ? metadataKeys.getOrDefault(Meta.Concept.SKETCH_METADATA.getName(), null) : null;
     }
 }

src/main/java/com/yahoo/bullet/querying/Querier.java

@@ -16,6 +16,7 @@
 import com.yahoo.bullet.parsing.Window;
 import com.yahoo.bullet.postaggregations.PostStrategy;
 import com.yahoo.bullet.record.BulletRecord;
+import com.yahoo.bullet.record.BulletRecordProvider;
 import com.yahoo.bullet.result.Clip;
 import com.yahoo.bullet.result.Meta;
 import com.yahoo.bullet.result.Meta.Concept;
@@ -295,7 +296,9 @@ public enum Mode {
     @Getter(AccessLevel.PACKAGE)
     private RunningQuery runningQuery;
 
-    private BulletConfig config;
+    // Transient field, DO NOT use it beyond constructor and initialize methods.
+    private transient BulletConfig config;
+
     private Map<String, String> metaKeys;
     private boolean hasNewData = false;
 
@@ -307,6 +310,8 @@ public enum Mode {
 
     private List<PostStrategy> postStrategies;
 
+    private BulletRecordProvider provider;
+
     /**
      * Constructor that takes a String representation of the query and a configuration to use. This also starts the
      * query.
@@ -356,6 +361,7 @@ public Querier(Mode mode, RunningQuery query, BulletConfig config) {
         this.mode = mode;
         this.runningQuery = query;
         this.config = config;
+        this.provider = config.getBulletRecordProvider();
     }
 
     // ********************************* Monoidal Interface Overrides *********************************
@@ -604,7 +610,7 @@ public RateLimitError getRateLimitError() {
         if (rateLimit == null || !rateLimit.isExceededRate()) {
             return null;
         }
-        return new RateLimitError(rateLimit.getCurrentRate(), config);
+        return new RateLimitError(rateLimit.getCurrentRate(), rateLimit.getAbsoluteRateLimit());
     }
 
     /**
@@ -650,7 +656,7 @@ private boolean filter(BulletRecord record) {
 
     private BulletRecord project(BulletRecord record) {
         Projection projection = runningQuery.getQuery().getProjection();
-        return projection != null ? ProjectionOperations.project(record, projection, config.getBulletRecordProvider()) : record;
+        return projection != null ? ProjectionOperations.project(record, projection, provider) : record;
     }
 
     private Clip postAggregate(Clip clip) {

src/main/java/com/yahoo/bullet/querying/RateLimitError.java

@@ -5,7 +5,6 @@
  */
 package com.yahoo.bullet.querying;
 
-import com.yahoo.bullet.common.BulletConfig;
 import com.yahoo.bullet.common.BulletError;
 import com.yahoo.bullet.result.Meta;
 
@@ -24,13 +23,13 @@ public class RateLimitError extends BulletError {
 
     /**
     /**
-     * Creates an instance of this from a given absolute exceeded rate and a {@link BulletConfig}.
+     * Creates an instance of this from a given absolute exceeded rate and a maximum rate limit.
      *
      * @param rate The exceeded rate that caused the error.
-     * @param config The validated BulletConfig to use.
+     * @param limit The maximum rate limit for the query.
      */
-    public RateLimitError(double rate, BulletConfig config) {
-        super(String.format(ERROR_FORMAT, getRate(config), rate * RateLimiter.SECOND), RESOLUTIONS);
+    public RateLimitError(double rate, double limit) {
+        super(String.format(ERROR_FORMAT, limit * RateLimiter.SECOND, rate * RateLimiter.SECOND), RESOLUTIONS);
     }
 
     /**
@@ -41,10 +40,4 @@ public RateLimitError(double rate, BulletConfig config) {
     public Meta makeMeta() {
         return new Meta().addErrors(Collections.singletonList(this));
     }
-
-    private static double getRate(BulletConfig config) {
-        int maxCount = config.getAs(BulletConfig.RATE_LIMIT_MAX_EMIT_COUNT, Integer.class);
-        int interval = config.getAs(BulletConfig.RATE_LIMIT_TIME_INTERVAL, Integer.class);
-        return (maxCount / (double) interval) * RateLimiter.SECOND;
-    }
 }

src/main/java/com/yahoo/bullet/querying/RateLimiter.java

@@ -23,6 +23,7 @@ public class RateLimiter {
     private final int maximum;
     @Getter
     private final int timeInterval;
+    @Getter
     private final double absoluteRateLimit;
 
     private long count = 0;
@@ -57,6 +58,7 @@ public RateLimiter(int maximum, int timeInterval) throws IllegalArgumentExceptio
         this.maximum = maximum;
         this.timeInterval = timeInterval;
         this.absoluteRateLimit = maximum / (double) timeInterval;
+
         lastCheckTime = System.currentTimeMillis();
     }
 

src/main/java/com/yahoo/bullet/windowing/Scheme.java

@@ -24,8 +24,8 @@
 public abstract class Scheme implements Monoidal {
     protected Strategy aggregation;
     protected Window window;
-    protected BulletConfig config;
     protected Map<String, String> metadataKeys;
+    private  boolean shouldMeta;
 
     /**
      * Creates an instance of this windowing scheme with the provided {@link Strategy}, {@link Window} and
@@ -38,8 +38,8 @@ public abstract class Scheme implements Monoidal {
     public Scheme(Strategy aggregation, Window window, BulletConfig config) {
         this.aggregation = aggregation;
         this.window = window;
-        this.config = config;
         metadataKeys = (Map<String, String>) config.getAs(BulletConfig.RESULT_METADATA_METRICS, Map.class);
+        shouldMeta = config.getAs(BulletConfig.RESULT_METADATA_ENABLE, Boolean.class);
     }
 
     /**
@@ -73,7 +73,6 @@ public Scheme(Strategy aggregation, Window window, BulletConfig config) {
     @Override
     public Meta getMetadata() {
         Meta meta = new Meta();
-        boolean shouldMeta = config.getAs(BulletConfig.RESULT_METADATA_ENABLE, Boolean.class);
         if (shouldMeta) {
             String key = getMetaKey();
             if (key != null) {

src/test/java/com/yahoo/bullet/querying/RateLimitErrorTest.java

@@ -5,7 +5,6 @@
  */
 package com.yahoo.bullet.querying;
 
-import com.yahoo.bullet.common.BulletConfig;
 import com.yahoo.bullet.common.BulletError;
 import com.yahoo.bullet.result.Meta;
 import org.testng.Assert;
@@ -21,16 +20,15 @@
 public class RateLimitErrorTest {
     @Test
     public void testMetaAndRateConversion() {
-        BulletConfig config = new BulletConfig();
-        RateLimitError error = new RateLimitError(19.34, config);
+        double defaultRate = ((double) DEFAULT_RATE_LIMIT_MAX_EMIT_COUNT / DEFAULT_RATE_LIMIT_TIME_INTERVAL);
+        RateLimitError error = new RateLimitError(19.34, defaultRate);
         Map<String, Object> actual = error.makeMeta().asMap();
         Assert.assertTrue(actual.containsKey(Meta.ERROR_KEY));
         Assert.assertTrue(((List<BulletError>) actual.get(Meta.ERROR_KEY)).get(0) == error);
 
         String asJSON = error.asJSON();
-        double defaultRate = ((double) DEFAULT_RATE_LIMIT_MAX_EMIT_COUNT / DEFAULT_RATE_LIMIT_TIME_INTERVAL) * RateLimiter.SECOND;
         double actualRate = 19.34 * RateLimiter.SECOND;
-        assertJSONEquals(asJSON, "{'error': '" + String.format(RateLimitError.ERROR_FORMAT, defaultRate, actualRate) + "', " +
+        assertJSONEquals(asJSON, "{'error': '" + String.format(RateLimitError.ERROR_FORMAT, defaultRate * RateLimiter.SECOND, actualRate) + "', " +
                                   "'resolutions': ['" + RateLimitError.NARROW_FILTER + "', '" + RateLimitError.TIME_WINDOW + "']" +
                                  "}");
     }