@Override
    protected SemanticProperties getSemanticPropertiesForLocalPropertyFiltering() {

        // Local properties for MapPartition may not be preserved.
        SingleInputSemanticProperties origProps =
                ((SingleInputOperator<?, ?, ?>) getOperator()).getSemanticProperties();
        SingleInputSemanticProperties filteredProps = new SingleInputSemanticProperties();
        FieldSet readSet = origProps.getReadFields(0);
        if (readSet != null) {
            filteredProps.addReadFields(readSet);
        }

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

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

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

        MapPartitionNode node = new MapPartitionNode(op);

        SemanticProperties filteredProps = node.getSemanticPropertiesForLocalPropertyFiltering();

        assertTrue(filteredProps.getForwardingTargetFields(0, 0).size() == 0);
        assertTrue(filteredProps.getForwardingTargetFields(0, 2).size() == 0);
        assertTrue(filteredProps.getForwardingSourceField(0, 1) < 0);
        assertTrue(filteredProps.getForwardingSourceField(0, 2) < 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));
    }