@Override
    protected SemanticProperties getSemanticPropertiesForLocalPropertyFiltering() {

        // Local properties for CoGroup may only be preserved on key fields.
        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);
        }

        // preserve only key fields (first input)
        for (int f : this.keys1) {
            FieldSet targets = origProps.getForwardingTargetFields(0, f);
            for (int t : targets) {
                filteredProps.addForwardedField(0, f, t);
            }
        }

        // preserve only key fields (second input)
        for (int f : this.keys2) {
            FieldSet targets = origProps.getForwardingTargetFields(1, f);
            for (int t : targets) {
                filteredProps.addForwardedField(1, f, t);
            }
        }

        return filteredProps;
    }
@Test
    public void testGetSemanticProperties() {

        DualInputSemanticProperties origProps = new DualInputSemanticProperties();
        // props for first input
        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));
        // props for second input
        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();

        // check first input props
        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);
        // check second input props
        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));
    }