closes #54. Transient fields (#56)

Author: hbxie

src/main/java/com/yahoo/bullet/parsing/BinaryExpression.java

@@ -12,8 +12,10 @@
 import lombok.Setter;
 
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 
 import static com.yahoo.bullet.common.BulletError.makeError;
 
@@ -39,6 +41,13 @@ public BinaryExpression() {
         right = null;
     }
 
+    @Override
+    public Set<String> getRequiredFields() {
+        Set<String> result = new HashSet<>(left.getRequiredFields());
+        result.addAll(right.getRequiredFields());
+        return result;
+    }
+
     @Override
     public String toString() {
         return "{" + super.toString() + ", left: " + left + ", right: " + right + ", type: " + type + "}";

src/main/java/com/yahoo/bullet/parsing/Expression.java

@@ -12,6 +12,7 @@
 import lombok.Setter;
 
 import java.util.List;
+import java.util.Set;
 
 import static java.util.Arrays.asList;
 
@@ -36,6 +37,13 @@ public enum Operation {
 
     public static final String OPERATION_FIELD = "operation";
 
+    /**
+     * Return a {@link Set} of required fields to compute this expression.
+     *
+     * @return A {@link Set} of required fields.
+     */
+    abstract public Set<String> getRequiredFields();
+
     @Override
     public String toString() {
         return OPERATION_FIELD + ": " + operation;

src/main/java/com/yahoo/bullet/parsing/LeafExpression.java

@@ -14,6 +14,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 
 import static com.yahoo.bullet.common.BulletError.makeError;
 
@@ -37,6 +38,11 @@ public LeafExpression() {
         value = null;
     }
 
+    @Override
+    public Set<String> getRequiredFields() {
+        return value.getKind() == Value.Kind.FIELD ? Collections.singleton(value.getValue()) : Collections.emptySet();
+    }
+
     @Override
     public String toString() {
         return "{" + super.toString() + ", value: " + value + "}";

src/main/java/com/yahoo/bullet/parsing/OrderBy.java

@@ -7,6 +7,7 @@
 
 import com.google.gson.annotations.Expose;
 import com.yahoo.bullet.common.BulletError;
+import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.Setter;
 
@@ -23,25 +24,45 @@ public enum Direction {
         DESC
     }
 
+    @Getter @AllArgsConstructor
+    public static class SortItem {
+        @Expose
+        private String field;
+        @Expose
+        private Direction direction;
+
+        /**
+         * Default constructor. GSON recommended
+         */
+        public SortItem() {
+            field = null;
+            direction = Direction.ASC;
+        }
+
+        @Override
+        public String toString() {
+            return "{field: " + field + ", direction: " + direction + "}";
+        }
+    }
+
     @Expose
-    private List<String> fields;
-    @Expose
-    private Direction direction;
+    private List<SortItem> fields;
 
     public static final BulletError ORDERBY_REQUIRES_FIELDS_ERROR =
             makeError("The ORDERBY post aggregation needs at least one field", "Please add fields.");
+    public static final BulletError ORDERBY_REQUIRES_NON_EMPTY_FIELDS_ERROR =
+            makeError("The fields in ORDERBY post aggregation must not be empty", "Please add non-empty fields.");
 
     /**
      * Default constructor. GSON recommended
      */
     public OrderBy() {
         fields = null;
-        direction = Direction.ASC;
     }
 
     @Override
     public String toString() {
-        return "{type: " + type + ", fields: " + fields + ", direction: " + direction + "}";
+        return "{type: " + type + ", fields: " + fields + "}";
     }
 
     @Override
@@ -53,6 +74,9 @@ public Optional<List<BulletError>> initialize() {
         if (fields == null || fields.isEmpty()) {
             return Optional.of(Collections.singletonList(ORDERBY_REQUIRES_FIELDS_ERROR));
         }
+        if (fields.stream().anyMatch(sortItem -> sortItem.getField() == null || sortItem.getField().isEmpty())) {
+            return Optional.of(Collections.singletonList(ORDERBY_REQUIRES_NON_EMPTY_FIELDS_ERROR));
+        }
         return Optional.empty();
     }
 

src/main/java/com/yahoo/bullet/parsing/Query.java

@@ -42,6 +42,8 @@ public class Query implements Configurable, Initializable {
                                                                 "Change your aggregation type or your window emit type to \"TIME\"");
     public static final BulletError NO_RAW_ALL = makeError("The \"RAW\" aggregation types cannot have window include \"ALL\"",
                                                            "Change your aggregation type or your window include type");
+    public static final BulletError AT_MOST_ONE_ORDERBY = makeError("The post aggregations cannot have multiple \"ORDERBY\"",
+                                                           "Change your post aggregations to keep at most one \"ORDERBY\"");
     /**
      * Default constructor. GSON recommended.
      */
@@ -99,6 +101,9 @@ public Optional<List<BulletError>> initialize() {
         aggregation.initialize().ifPresent(errors::addAll);
 
         if (postAggregations != null) {
+            if (postAggregations.stream().filter(postAggregation -> postAggregation.getType() == PostAggregation.Type.ORDER_BY).count() > 1) {
+                errors.add(AT_MOST_ONE_ORDERBY);
+            }
             postAggregations.forEach(p -> p.initialize().ifPresent(errors::addAll));
         }
 

src/main/java/com/yahoo/bullet/postaggregations/ComputationStrategy.java

@@ -17,46 +17,53 @@
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.Set;
+
 import static com.yahoo.bullet.common.Utilities.extractTypedObject;
 
 @Slf4j @AllArgsConstructor
 public class ComputationStrategy implements PostStrategy {
-    private Computation postAggregation;
+    private Computation computation;
 
     @Override
     public Clip execute(Clip clip) {
         clip.getRecords().forEach(r -> {
                 try {
-                    TypedObject result = calculate(postAggregation.getExpression(), r);
+                    TypedObject result = calculate(computation.getExpression(), r);
                     switch (result.getType()) {
                         case INTEGER:
-                            r.setInteger(postAggregation.getNewName(), (Integer) result.getValue());
+                            r.setInteger(computation.getNewName(), (Integer) result.getValue());
                             break;
                         case LONG:
-                            r.setLong(postAggregation.getNewName(), (Long) result.getValue());
+                            r.setLong(computation.getNewName(), (Long) result.getValue());
                             break;
                         case DOUBLE:
-                            r.setDouble(postAggregation.getNewName(), (Double) result.getValue());
+                            r.setDouble(computation.getNewName(), (Double) result.getValue());
                             break;
                         case FLOAT:
-                            r.setFloat(postAggregation.getNewName(), (Float) result.getValue());
+                            r.setFloat(computation.getNewName(), (Float) result.getValue());
                             break;
                         case BOOLEAN:
-                            r.setBoolean(postAggregation.getNewName(), (Boolean) result.getValue());
+                            r.setBoolean(computation.getNewName(), (Boolean) result.getValue());
                             break;
                         case STRING:
-                            r.setString(postAggregation.getNewName(), (String) result.getValue());
+                            r.setString(computation.getNewName(), (String) result.getValue());
                             break;
                     }
                 } catch (RuntimeException e) {
                     // Ignore the exception and skip setting the field.
-                    log.error("Unable to calculate the expression: " + postAggregation.getExpression());
+                    log.error("Unable to calculate the expression: " + computation.getExpression());
                     log.error("Skip it due to: " + e);
                 }
             });
         return clip;
     }
 
+    @Override
+    public Set<String> getRequiredFields() {
+        return computation.getExpression().getRequiredFields();
+    }
+
     private TypedObject calculateLeafExpression(LeafExpression expression, BulletRecord record) {
         Value value = expression.getValue();
         TypedObject result = null;

src/main/java/com/yahoo/bullet/postaggregations/OrderByStrategy.java

@@ -9,47 +9,44 @@
 import com.yahoo.bullet.record.BulletRecord;
 import com.yahoo.bullet.result.Clip;
 import com.yahoo.bullet.typesystem.TypedObject;
+import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 import static com.yahoo.bullet.common.Utilities.extractTypedObject;
 
-@Slf4j
+@Slf4j @AllArgsConstructor
 public class OrderByStrategy implements PostStrategy {
-    private OrderBy postAggregation;
-    private int multiplyingFactor;
-
-    /**
-     * Contructor takes a {@link OrderBy} object.
-     *
-     * @param postAggregation The {@link OrderBy} object.
-     */
-    public OrderByStrategy(OrderBy postAggregation) {
-        this.postAggregation = postAggregation;
-        multiplyingFactor = postAggregation.getDirection() == OrderBy.Direction.ASC ? 1 : -1;
-    }
+    private OrderBy orderBy;
 
     @Override
     public Clip execute(Clip clip) {
         List<BulletRecord> records = clip.getRecords();
         records.sort((a, b) -> {
-                for (String field : postAggregation.getFields()) {
-                    TypedObject typedObjectA = extractTypedObject(field, a);
-                    TypedObject typedObjectB = extractTypedObject(field, b);
+                for (OrderBy.SortItem sortItem : orderBy.getFields()) {
+                    TypedObject typedObjectA = extractTypedObject(sortItem.getField(), a);
+                    TypedObject typedObjectB = extractTypedObject(sortItem.getField(), b);
                     try {
                         int compareValue = typedObjectA.compareTo(typedObjectB);
                         if (compareValue != 0) {
-                            return multiplyingFactor * compareValue;
+                            return (sortItem.getDirection() == OrderBy.Direction.ASC ? 1 : -1) * compareValue;
                         }
                     } catch (RuntimeException e) {
                         // Ignore the exception and skip this field.
-                        log.error("Unable to compare field " + field);
+                        log.error("Unable to compare field " + sortItem.getField());
                         log.error("Skip it due to: " + e);
                     }
                 }
                 return 0;
             });
         return clip;
     }
+
+    @Override
+    public Set<String> getRequiredFields() {
+        return orderBy.getFields().stream().map(OrderBy.SortItem::getField).collect(Collectors.toSet());
+    }
 }

src/main/java/com/yahoo/bullet/postaggregations/PostStrategy.java

@@ -11,6 +11,7 @@
 
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 
 public interface PostStrategy extends Initializable {
     /**
@@ -29,4 +30,11 @@ public interface PostStrategy extends Initializable {
     default Optional<List<BulletError>> initialize() {
         return Optional.empty();
     }
+
+    /**
+     * Get the list of required fields.
+     *
+     * @return An @{link List} of fields.
+     */
+    Set<String> getRequiredFields();
 }

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

@@ -10,24 +10,33 @@
 import com.yahoo.bullet.record.BulletRecordProvider;
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.HashMap;
 import java.util.Map;
 
 import static com.yahoo.bullet.common.Utilities.splitField;
 
 @Slf4j
 public class ProjectionOperations {
     /**
-     * Projects the given {@link BulletRecord} based on the given {@link Projection}.
+     * Projects the given {@link BulletRecord} based on the given fields.
      *
      * @param record The record to project.
      * @param projection The projection to apply.
+     * @param transientFields The map of fields to apply.
      * @param provider A BulletRecordProvider to generate BulletRecords.
      * @return The projected record.
      */
-    public static BulletRecord project(BulletRecord record, Projection projection, BulletRecordProvider provider) {
-        Map<String, String> fields = projection.getFields();
+    public static BulletRecord project(BulletRecord record, Projection projection, Map<String, String> transientFields, BulletRecordProvider provider) {
+        Map<String, String> fields = new HashMap<>();
+        Map<String, String> projectionFields = projection.getFields();
+        if (projectionFields != null) {
+            fields.putAll(projectionFields);
+        }
+        if (transientFields != null) {
+            fields.putAll(transientFields);
+        }
         // Returning the record itself if no projections. The record itself should never be modified so it's ok.
-        if (fields == null) {
+        if (fields.isEmpty()) {
             return record;
         }
         // More efficient if fields << the fields in the BulletRecord