@Override
    protected SemanticProperties getSemanticPropertiesForLocalPropertyFiltering() {
        // Local properties for GroupReduce may only be preserved on key fields.
        SingleInputSemanticProperties origProps = getOperator().getSemanticProperties();
        SingleInputSemanticProperties filteredProps = new SingleInputSemanticProperties();
        FieldSet readSet = origProps.getReadFields(0);
        if (readSet != null) {
            filteredProps.addReadFields(readSet);
        }

        // only add forward field information for key fields
        if (this.keys != null) {
            for (int f : this.keys) {
                FieldSet targets = origProps.getForwardingTargetFields(0, f);
                for (int t : targets) {
                    filteredProps.addForwardedField(f, t);
                }
            }
        }
        return filteredProps;
    }
@Test
    public void testGetSemanticProperties() {

        SingleInputSemanticProperties origProps = new SingleInputSemanticProperties();
        origProps.addForwardedField(0, 1);
        origProps.addForwardedField(2, 2);
        origProps.addForwardedField(3, 4);
        origProps.addForwardedField(6, 0);
        origProps.addReadFields(new FieldSet(0, 2, 4, 7));

        GroupReduceOperatorBase<?, ?, ?> op = mock(GroupReduceOperatorBase.class);
        when(op.getSemanticProperties()).thenReturn(origProps);
        when(op.getKeyColumns(0)).thenReturn(new int[] {3, 2});
        when(op.getParameters()).thenReturn(new Configuration());

        GroupReduceNode node = new GroupReduceNode(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));
    }