CoGroupNode extends TwoInputNode { @Override protected SemanticProperties getSemanticPropertiesForLocalPropertyFiltering() { DualInputSemanticProperties origProps = ((DualInputOperator<?, ?, ?, ?>) getOperator()).getSemanticProperties(); DualInputSemanticProperties filteredProps = new DualInputSemanticProperties(); FieldSet readSet1 = origProps.getReadFields(0); FieldSet readSet2 = origProps.getReadFields(1); if(readSet1 != null) { filteredProps.addReadFields(0, readSet1); } if(readSet2 != null) { filteredProps.addReadFields(1, readSet2); } for(int f : this.keys1) { FieldSet targets = origProps.getForwardingTargetFields(0, f); for(int t : targets) { filteredProps.addForwardedField(0, f, t); } } for(int f : this.keys2) { FieldSet targets = origProps.getForwardingTargetFields(1, f); for(int t : targets) { filteredProps.addForwardedField(1, f, t); } } return filteredProps; } CoGroupNode(CoGroupOperatorBase<?, ?, ?, ?> operator); @Override CoGroupOperatorBase<?, ?, ?, ?> getOperator(); @Override String getOperatorName(); void makeCoGroupWithSolutionSet(int solutionsetInputIndex);  }
@Test public void testGetSemanticProperties() { DualInputSemanticProperties origProps = new DualInputSemanticProperties(); origProps.addForwardedField(0, 0, 1); origProps.addForwardedField(0, 2, 2); origProps.addForwardedField(0, 3, 4); origProps.addForwardedField(0, 6, 0); origProps.addReadFields(0, new FieldSet(0, 2, 4, 7)); origProps.addForwardedField(1, 1, 2); origProps.addForwardedField(1, 2, 8); origProps.addForwardedField(1, 3, 7); origProps.addForwardedField(1, 6, 6); origProps.addReadFields(1, new FieldSet(1, 3, 4)); CoGroupOperatorBase<?,?,?,?> op = mock(CoGroupOperatorBase.class); when(op.getSemanticProperties()).thenReturn(origProps); when(op.getKeyColumns(0)).thenReturn(new int[]{3,2}); when(op.getKeyColumns(1)).thenReturn(new int[]{6,3}); when(op.getParameters()).thenReturn(new Configuration()); CoGroupNode node = new CoGroupNode(op); SemanticProperties filteredProps = node.getSemanticPropertiesForLocalPropertyFiltering(); assertTrue(filteredProps.getForwardingTargetFields(0, 0).size() == 0); assertTrue(filteredProps.getForwardingTargetFields(0, 2).size() == 1); assertTrue(filteredProps.getForwardingTargetFields(0, 2).contains(2)); assertTrue(filteredProps.getForwardingTargetFields(0, 3).size() == 1); assertTrue(filteredProps.getForwardingTargetFields(0, 3).contains(4)); assertTrue(filteredProps.getForwardingTargetFields(0, 6).size() == 0); assertTrue(filteredProps.getForwardingSourceField(0, 1) < 0); assertTrue(filteredProps.getForwardingSourceField(0, 2) == 2); assertTrue(filteredProps.getForwardingSourceField(0, 4) == 3); assertTrue(filteredProps.getForwardingSourceField(0, 0) < 0); assertTrue(filteredProps.getReadFields(0).size() == 4); assertTrue(filteredProps.getReadFields(0).contains(0)); assertTrue(filteredProps.getReadFields(0).contains(2)); assertTrue(filteredProps.getReadFields(0).contains(4)); assertTrue(filteredProps.getReadFields(0).contains(7)); assertTrue(filteredProps.getForwardingTargetFields(1, 1).size() == 0); assertTrue(filteredProps.getForwardingTargetFields(1, 2).size() == 0); assertTrue(filteredProps.getForwardingTargetFields(1, 3).size() == 1); assertTrue(filteredProps.getForwardingTargetFields(1, 3).contains(7)); assertTrue(filteredProps.getForwardingTargetFields(1, 6).size() == 1); assertTrue(filteredProps.getForwardingTargetFields(1, 6).contains(6)); assertTrue(filteredProps.getForwardingSourceField(1, 2) < 0); assertTrue(filteredProps.getForwardingSourceField(1, 8) < 0); assertTrue(filteredProps.getForwardingSourceField(1, 7) == 3); assertTrue(filteredProps.getForwardingSourceField(1, 6) == 6); assertTrue(filteredProps.getReadFields(1).size() == 3); assertTrue(filteredProps.getReadFields(1).contains(1)); assertTrue(filteredProps.getReadFields(1).contains(3)); assertTrue(filteredProps.getReadFields(1).contains(4)); }