fix joins

src/jvm/storm/trident/operation/GroupedMultiReducer.java

@@ -7,7 +7,7 @@
 
 public interface GroupedMultiReducer<T> extends Serializable {
     void prepare(Map conf, TridentMultiReducerContext context);
-    T init(TridentCollector collector);
+    T init(TridentCollector collector, TridentTuple group);
     void execute(T state, int streamIndex, TridentTuple group, TridentTuple input, TridentCollector collector);
     void complete(T state, TridentTuple group, TridentCollector collector);
     void cleanup();

src/jvm/storm/trident/operation/impl/GroupedMultiReducerExecutor.java

@@ -26,6 +26,7 @@ public GroupedMultiReducerExecutor(GroupedMultiReducer reducer, List<Fields> gro
         }
         _groupFields = groupFields;
         _inputFields = inputFields;
+        _reducer = reducer;
     }
 
     @Override
@@ -48,11 +49,11 @@ public void execute(Map<TridentTuple, Object> state, int streamIndex, TridentTup
         ProjectionFactory inputFactory = _inputFactories.get(streamIndex);
 
         TridentTuple group = groupFactory.create(full);
-        TridentTuple input = groupFactory.create(full);
+        TridentTuple input = inputFactory.create(full);
 
         Object curr;
         if(!state.containsKey(group)) {
-            curr = _reducer.init(collector);
+            curr = _reducer.init(collector, group);
             state.put(group, curr);
         } else {
             curr = state.get(group);

src/jvm/storm/trident/operation/impl/JoinerMultiReducer.java

@@ -30,15 +30,15 @@ public JoinerMultiReducer(List<JoinType> types, int numGroupFields, List<Fields>
     public void prepare(Map conf, TridentMultiReducerContext context) {
         int[] sizes = new int[_sideFields.size() + 1];
         sizes[0] = _numGroupFields;
-        for(int i=0; i<_sideFields.size()-1; i++) {
+        for(int i=0; i<_sideFields.size(); i++) {
             sizes[i+1] = _sideFields.get(i).size();
         }
         _factory = new ComboList.Factory(sizes);
     }
 
     @Override
-    public JoinState init(TridentCollector collector) {
-        return new JoinState(_types.size());
+    public JoinState init(TridentCollector collector, TridentTuple group) {
+        return new JoinState(_types.size(), group);
     }
 
     @Override
@@ -49,6 +49,7 @@ public void execute(JoinState state, int streamIndex, TridentTuple group, Triden
         if(side.isEmpty()) {
             state.numSidesReceived++;
         }
+
         side.add(input);
         if(state.numSidesReceived == state.sides.length) {
             emitCrossJoin(state, collector, streamIndex, input);
@@ -92,12 +93,13 @@ private void emitCrossJoin(JoinState state, TridentCollector collector, int over
         boolean keepGoing = true;
         //emit cross-join of all emitted tuples
         while(keepGoing) {
-            List[] combined = new List[sides.length];
+            List[] combined = new List[sides.length+1];
+            combined[0] = state.group;
             for(int i=0; i<sides.length; i++) {
                 if(i==overrideIndex) {
-                    combined[i] = overrideTuple;
+                    combined[i+1] = overrideTuple;
                 } else {
-                    combined[i] = sides[i].get(indices[i]);                
+                    combined[i+1] = sides[i].get(indices[i]);                
                 }
             }
             collector.emit(_factory.create(combined));
@@ -125,10 +127,12 @@ public static class JoinState {
         List<List>[] sides;
         int numSidesReceived = 0;
         int[] indices;
+        TridentTuple group;
 
-        public JoinState(int numSides) {
+        public JoinState(int numSides, TridentTuple group) {
             sides = new List[numSides];
             indices = new int[numSides];
+            this.group = group;
             for(int i=0; i<numSides; i++) {
                 sides[i] = new ArrayList<List>();
             }