/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.optimizer;

import org.apache.flink.api.common.functions.FlatJoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.operators.BinaryOperatorInformation;
import org.apache.flink.api.common.operators.GenericDataSourceBase;
import org.apache.flink.api.common.operators.OperatorInformation;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.common.operators.Ordering;
import org.apache.flink.api.common.operators.UnaryOperatorInformation;
import org.apache.flink.api.common.operators.base.InnerJoinOperatorBase;
import org.apache.flink.api.common.operators.base.MapOperatorBase;
import org.apache.flink.api.common.operators.util.FieldList;
import org.apache.flink.api.common.operators.util.FieldSet;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.io.TextInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.optimizer.dag.DataSourceNode;
import org.apache.flink.optimizer.dag.JoinNode;
import org.apache.flink.optimizer.dag.MapNode;
import org.apache.flink.optimizer.dataproperties.GlobalProperties;
import org.apache.flink.optimizer.dataproperties.LocalProperties;
import org.apache.flink.optimizer.dataproperties.RequestedGlobalProperties;
import org.apache.flink.optimizer.dataproperties.RequestedLocalProperties;
import org.apache.flink.optimizer.plan.Channel;
import org.apache.flink.optimizer.plan.DualInputPlanNode;
import org.apache.flink.optimizer.plan.PlanNode.FeedbackPropertiesMeetRequirementsReport;
import org.apache.flink.optimizer.plan.SingleInputPlanNode;
import org.apache.flink.optimizer.plan.SourcePlanNode;
import org.apache.flink.optimizer.testfunctions.DummyFlatJoinFunction;
import org.apache.flink.optimizer.testfunctions.IdentityMapper;
import org.apache.flink.runtime.io.network.DataExchangeMode;
import org.apache.flink.runtime.operators.DriverStrategy;
import org.apache.flink.runtime.operators.shipping.ShipStrategyType;
import org.apache.flink.runtime.operators.util.LocalStrategy;

import org.junit.Test;

import static org.apache.flink.optimizer.plan.PlanNode.FeedbackPropertiesMeetRequirementsReport.*;
import static org.junit.Assert.*;

public class FeedbackPropertiesMatchTest {

    @Test
    public void testNoPartialSolutionFoundSingleInputOnly() {
        try {
            SourcePlanNode target = new SourcePlanNode(getSourceNode(), "Source");

            SourcePlanNode otherTarget = new SourcePlanNode(getSourceNode(), "Source");

            Channel toMap1 = new Channel(target);
            toMap1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toMap1.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode map1 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 1", toMap1, DriverStrategy.MAP);

            Channel toMap2 = new Channel(map1);
            toMap2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toMap2.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode map2 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 2", toMap2, DriverStrategy.MAP);

            {
                GlobalProperties gp = new GlobalProperties();
                LocalProperties lp = new LocalProperties();

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(otherTarget, gp, lp);
                assertTrue(report == NO_PARTIAL_SOLUTION);
            }
        } catch (Exception e) {
            e.printStackTrace();
            fail(e.getMessage());
        }
    }

    @Test
    public void testSingleInputOperators() {
        try {
            SourcePlanNode target = new SourcePlanNode(getSourceNode(), "Source");

            Channel toMap1 = new Channel(target);
            toMap1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toMap1.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode map1 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 1", toMap1, DriverStrategy.MAP);

            Channel toMap2 = new Channel(map1);
            toMap2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toMap2.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode map2 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 2", toMap2, DriverStrategy.MAP);

            // no feedback properties and none are ever required and present
            {
                GlobalProperties gp = new GlobalProperties();
                LocalProperties lp = new LocalProperties();

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some global feedback properties and none are ever required and present
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2, 5));
                LocalProperties lp = new LocalProperties();

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some local feedback properties and none are ever required and present
            {
                GlobalProperties gp = new GlobalProperties();
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1, 2));

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some global and local feedback properties and none are ever required and present
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2, 5));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1, 2));

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // --------------------------- requirements on channel 1 -----------------------

            // some required global properties, which are matched exactly
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2, 5));
                LocalProperties lp = new LocalProperties();

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setHashPartitioned(new FieldList(2, 5));

                toMap1.setRequiredGlobalProps(reqGp);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some required local properties, which are matched exactly
            {
                GlobalProperties gp = new GlobalProperties();
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1, 2));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(1, 2));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(reqLp);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some required global and local properties, which are matched exactly
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2, 5));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1, 2));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setHashPartitioned(new FieldList(2, 5));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(1, 2));

                toMap1.setRequiredGlobalProps(reqGp);
                toMap1.setRequiredLocalProps(reqLp);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some required global and local properties, which are over-fulfilled
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1, 2));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setHashPartitioned(new FieldSet(2, 5));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(1));

                toMap1.setRequiredGlobalProps(reqGp);
                toMap1.setRequiredLocalProps(reqLp);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some required global properties that are not met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2, 1));
                LocalProperties lp = new LocalProperties();

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setHashPartitioned(new FieldList(2, 5));

                toMap1.setRequiredGlobalProps(reqGp);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // some required local properties that are not met
            {
                GlobalProperties gp = new GlobalProperties();
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(2, 1));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(reqLp);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // some required global and local properties where the global properties are not met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2, 1));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldList(2, 5));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(1));

                toMap1.setRequiredGlobalProps(reqGp);
                toMap1.setRequiredLocalProps(reqLp);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // some required global and local properties where the local properties are not met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldList(1));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(2));

                toMap1.setRequiredGlobalProps(reqGp);
                toMap1.setRequiredLocalProps(reqLp);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // --------------------------- requirements on channel 2 -----------------------

            // some required global properties, which are matched exactly
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2, 5));
                LocalProperties lp = new LocalProperties();

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setHashPartitioned(new FieldList(2, 5));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(reqGp);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some required local properties, which are matched exactly
            {
                GlobalProperties gp = new GlobalProperties();
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1, 2));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(1, 2));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(reqLp);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some required global and local properties, which are matched exactly
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2, 5));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1, 2));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setHashPartitioned(new FieldList(2, 5));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(1, 2));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(reqGp);
                toMap2.setRequiredLocalProps(reqLp);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some required global and local properties, which are over-fulfilled
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1, 2));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setHashPartitioned(new FieldSet(2, 5));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(1));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(reqGp);
                toMap2.setRequiredLocalProps(reqLp);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some required global properties that are not met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2, 1));
                LocalProperties lp = new LocalProperties();

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setHashPartitioned(new FieldSet(2, 5));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(reqGp);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // some required local properties that are not met
            {
                GlobalProperties gp = new GlobalProperties();
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(2, 1));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(reqLp);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // some required global and local properties where the global properties are not met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(2, 1));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldSet(2, 5));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(1));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(reqGp);
                toMap2.setRequiredLocalProps(reqLp);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // some required global and local properties where the local properties are not met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(1));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldList(1));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(2));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(reqGp);
                toMap2.setRequiredLocalProps(reqLp);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // ---------------------- requirements mixed on 1 and 2 -----------------------

            // some required global properties at step one and some more at step 2
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp = LocalProperties.EMPTY;

                RequestedGlobalProperties reqGp1 = new RequestedGlobalProperties();
                reqGp1.setAnyPartitioning(new FieldList(1, 2));

                RequestedGlobalProperties reqGp2 = new RequestedGlobalProperties();
                reqGp2.setHashPartitioned(new FieldList(1, 2));

                toMap1.setRequiredGlobalProps(reqGp1);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(reqGp2);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some required local properties at step one and some more at step 2
            {
                GlobalProperties gp = new GlobalProperties();
                LocalProperties lp =
                        LocalProperties.forOrdering(
                                new Ordering(3, null, Order.ASCENDING)
                                        .appendOrdering(1, null, Order.DESCENDING));

                RequestedLocalProperties reqLp1 = new RequestedLocalProperties();
                reqLp1.setGroupedFields(new FieldList(3, 1));

                RequestedLocalProperties reqLp2 = new RequestedLocalProperties();
                reqLp2.setOrdering(
                        new Ordering(3, null, Order.ANY).appendOrdering(1, null, Order.ANY));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(reqLp1);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(reqLp2);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some required global properties at step one and some local ones at step 2
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldList(1, 2));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(2));

                toMap1.setRequiredGlobalProps(reqGp);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(reqLp);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some required local properties at step one and some global ones at step 2
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldList(1, 2));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(2));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(reqLp);

                toMap2.setRequiredGlobalProps(reqGp);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some fulfilled global properties at step one and some non-fulfilled local ones at
            // step 2
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldList(1, 2));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(2, 3));

                toMap1.setRequiredGlobalProps(reqGp);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(reqLp);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // some fulfilled local properties at step one and some non-fulfilled global ones at
            // step 2
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldList(2, 3));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(2, 1));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(reqLp);

                toMap2.setRequiredGlobalProps(reqGp);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // some non-fulfilled global properties at step one and some fulfilled local ones at
            // step 2
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldList(2, 3));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(2, 1));

                toMap1.setRequiredGlobalProps(reqGp);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(reqLp);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // some non-fulfilled local properties at step one and some fulfilled global ones at
            // step 2
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldList(1, 2));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(2, 1, 3));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(reqLp);

                toMap2.setRequiredGlobalProps(reqGp);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }
        } catch (Exception e) {
            e.printStackTrace();
            fail(e.getMessage());
        }
    }

    @Test
    public void testSingleInputOperatorsWithReCreation() {
        try {
            SourcePlanNode target = new SourcePlanNode(getSourceNode(), "Source");

            Channel toMap1 = new Channel(target);
            SingleInputPlanNode map1 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 1", toMap1, DriverStrategy.MAP);

            Channel toMap2 = new Channel(map1);
            SingleInputPlanNode map2 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 2", toMap2, DriverStrategy.MAP);

            // set ship strategy in first channel, so later non matching global properties do not
            // matter
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp = LocalProperties.EMPTY;

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldSet(2, 5));

                toMap1.setShipStrategy(
                        ShipStrategyType.PARTITION_HASH,
                        new FieldList(2, 5),
                        DataExchangeMode.PIPELINED);
                toMap1.setLocalStrategy(LocalStrategy.NONE);

                toMap2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap2.setLocalStrategy(LocalStrategy.NONE);

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(reqGp);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(MET, report);
            }

            // set ship strategy in second channel, so previous non matching global properties void
            // the match
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp = LocalProperties.EMPTY;

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldSet(2, 5));

                toMap1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap1.setLocalStrategy(LocalStrategy.NONE);

                toMap2.setShipStrategy(
                        ShipStrategyType.PARTITION_HASH,
                        new FieldList(2, 5),
                        DataExchangeMode.PIPELINED);
                toMap2.setLocalStrategy(LocalStrategy.NONE);

                toMap1.setRequiredGlobalProps(reqGp);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // set local strategy in first channel, so later non matching local properties do not
            // matter
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp =
                        LocalProperties.forOrdering(
                                new Ordering(3, null, Order.ASCENDING)
                                        .appendOrdering(1, null, Order.DESCENDING));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(4, 1));

                toMap1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap1.setLocalStrategy(
                        LocalStrategy.SORT, new FieldList(5, 7), new boolean[] {false, false});

                toMap2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap2.setLocalStrategy(LocalStrategy.NONE);

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(reqLp);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // set local strategy in second channel, so previous non matching local properties void
            // the match
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp =
                        LocalProperties.forOrdering(
                                new Ordering(3, null, Order.ASCENDING)
                                        .appendOrdering(1, null, Order.DESCENDING));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(4, 1));

                toMap1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap1.setLocalStrategy(LocalStrategy.NONE);

                toMap2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap2.setLocalStrategy(
                        LocalStrategy.SORT, new FieldList(5, 7), new boolean[] {false, false});

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(reqLp);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // create the properties on the same node as the requirement
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(1, 2));
                LocalProperties lp =
                        LocalProperties.forOrdering(
                                new Ordering(3, null, Order.ASCENDING)
                                        .appendOrdering(1, null, Order.DESCENDING));

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldSet(5, 7));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(5, 7));

                toMap1.setShipStrategy(
                        ShipStrategyType.PARTITION_HASH,
                        new FieldList(5, 7),
                        DataExchangeMode.PIPELINED);
                toMap1.setLocalStrategy(
                        LocalStrategy.SORT, new FieldList(5, 7), new boolean[] {false, false});

                toMap2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap2.setLocalStrategy(LocalStrategy.NONE);

                toMap1.setRequiredGlobalProps(reqGp);
                toMap1.setRequiredLocalProps(reqLp);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map2.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(MET, report);
            }
        } catch (Exception e) {
            e.printStackTrace();
            fail(e.getMessage());
        }
    }

    @Test
    public void testSingleInputOperatorsChainOfThree() {
        try {
            SourcePlanNode target = new SourcePlanNode(getSourceNode(), "Source");

            Channel toMap1 = new Channel(target);
            SingleInputPlanNode map1 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 1", toMap1, DriverStrategy.MAP);

            Channel toMap2 = new Channel(map1);
            SingleInputPlanNode map2 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 2", toMap2, DriverStrategy.MAP);

            Channel toMap3 = new Channel(map2);
            SingleInputPlanNode map3 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 3", toMap3, DriverStrategy.MAP);

            // set local strategy in first channel, so later non matching local properties do not
            // matter
            {
                GlobalProperties gp = new GlobalProperties();
                LocalProperties lp =
                        LocalProperties.forOrdering(
                                new Ordering(3, null, Order.ASCENDING)
                                        .appendOrdering(1, null, Order.DESCENDING));

                RequestedLocalProperties reqLp = new RequestedLocalProperties();
                reqLp.setGroupedFields(new FieldList(4, 1));

                toMap1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap1.setLocalStrategy(
                        LocalStrategy.SORT, new FieldList(5, 7), new boolean[] {false, false});

                toMap2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap2.setLocalStrategy(LocalStrategy.NONE);

                toMap3.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap3.setLocalStrategy(LocalStrategy.NONE);

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                toMap3.setRequiredGlobalProps(null);
                toMap3.setRequiredLocalProps(reqLp);

                FeedbackPropertiesMeetRequirementsReport report =
                        map3.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // set global strategy in first channel, so later non matching global properties do not
            // matter
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(5, 3));
                LocalProperties lp = LocalProperties.EMPTY;

                RequestedGlobalProperties reqGp = new RequestedGlobalProperties();
                reqGp.setAnyPartitioning(new FieldSet(2, 3));

                toMap1.setShipStrategy(
                        ShipStrategyType.PARTITION_HASH,
                        new FieldList(1, 2),
                        DataExchangeMode.PIPELINED);
                toMap1.setLocalStrategy(LocalStrategy.NONE);

                toMap2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap2.setLocalStrategy(LocalStrategy.NONE);

                toMap3.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toMap3.setLocalStrategy(LocalStrategy.NONE);

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toMap2.setRequiredGlobalProps(null);
                toMap2.setRequiredLocalProps(null);

                toMap3.setRequiredGlobalProps(reqGp);
                toMap3.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        map3.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }
        } catch (Exception e) {
            e.printStackTrace();
            fail(e.getMessage());
        }
    }

    @Test
    public void testNoPartialSolutionFoundTwoInputOperator() {
        try {
            SourcePlanNode target = new SourcePlanNode(getSourceNode(), "Partial Solution");

            SourcePlanNode source1 = new SourcePlanNode(getSourceNode(), "Source 1");
            SourcePlanNode source2 = new SourcePlanNode(getSourceNode(), "Source 2");

            Channel toMap1 = new Channel(source1);
            toMap1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toMap1.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode map1 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 1", toMap1, DriverStrategy.MAP);

            Channel toMap2 = new Channel(source2);
            toMap2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toMap2.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode map2 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 2", toMap2, DriverStrategy.MAP);

            Channel toJoin1 = new Channel(map1);
            Channel toJoin2 = new Channel(map2);

            toJoin1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toJoin1.setLocalStrategy(LocalStrategy.NONE);
            toJoin2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toJoin2.setLocalStrategy(LocalStrategy.NONE);

            DualInputPlanNode join =
                    new DualInputPlanNode(
                            getJoinNode(),
                            "Join",
                            toJoin1,
                            toJoin2,
                            DriverStrategy.HYBRIDHASH_BUILD_FIRST);

            FeedbackPropertiesMeetRequirementsReport report =
                    join.checkPartialSolutionPropertiesMet(
                            target, new GlobalProperties(), new LocalProperties());
            assertEquals(NO_PARTIAL_SOLUTION, report);
        } catch (Exception e) {
            e.printStackTrace();
            fail(e.getMessage());
        }
    }

    @Test
    public void testTwoOperatorsOneIndependent() {
        try {
            SourcePlanNode target = new SourcePlanNode(getSourceNode(), "Partial Solution");
            SourcePlanNode source = new SourcePlanNode(getSourceNode(), "Other Source");

            Channel toMap1 = new Channel(target);
            toMap1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toMap1.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode map1 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 1", toMap1, DriverStrategy.MAP);

            Channel toMap2 = new Channel(source);
            toMap2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toMap2.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode map2 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 2", toMap2, DriverStrategy.MAP);

            Channel toJoin1 = new Channel(map1);
            Channel toJoin2 = new Channel(map2);

            DualInputPlanNode join =
                    new DualInputPlanNode(
                            getJoinNode(),
                            "Join",
                            toJoin1,
                            toJoin2,
                            DriverStrategy.HYBRIDHASH_BUILD_FIRST);

            Channel toAfterJoin = new Channel(join);
            toAfterJoin.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toAfterJoin.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode afterJoin =
                    new SingleInputPlanNode(
                            getMapNode(), "After Join Mapper", toAfterJoin, DriverStrategy.MAP);

            // attach some properties to the non-relevant input
            {
                toMap2.setShipStrategy(ShipStrategyType.BROADCAST, DataExchangeMode.PIPELINED);
                toMap2.setLocalStrategy(
                        LocalStrategy.SORT, new FieldList(2, 7), new boolean[] {true, true});

                RequestedGlobalProperties joinGp = new RequestedGlobalProperties();
                joinGp.setFullyReplicated();

                RequestedLocalProperties joinLp = new RequestedLocalProperties();
                joinLp.setOrdering(
                        new Ordering(2, null, Order.ASCENDING)
                                .appendOrdering(7, null, Order.ASCENDING));

                toJoin2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toJoin2.setLocalStrategy(LocalStrategy.NONE);
                toJoin2.setRequiredGlobalProps(joinGp);
                toJoin2.setRequiredLocalProps(joinLp);
            }

            // ------------------------------------------------------------------------------------

            // no properties from the partial solution, no required properties
            {
                toJoin1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toJoin1.setLocalStrategy(LocalStrategy.NONE);

                GlobalProperties gp = new GlobalProperties();
                LocalProperties lp = LocalProperties.EMPTY;

                FeedbackPropertiesMeetRequirementsReport report =
                        join.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some properties from the partial solution, no required properties
            {
                toJoin1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toJoin1.setLocalStrategy(LocalStrategy.NONE);

                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                FeedbackPropertiesMeetRequirementsReport report =
                        join.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // produced properties match relevant input
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(0));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(2));

                toJoin1.setRequiredGlobalProps(rgp);
                toJoin1.setRequiredLocalProps(rlp);

                toJoin1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toJoin1.setLocalStrategy(LocalStrategy.NONE);

                FeedbackPropertiesMeetRequirementsReport report =
                        join.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // produced properties do not match relevant input
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(0));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(1, 2, 3));

                toJoin1.setRequiredGlobalProps(rgp);
                toJoin1.setRequiredLocalProps(rlp);

                toJoin1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toJoin1.setLocalStrategy(LocalStrategy.NONE);

                FeedbackPropertiesMeetRequirementsReport report =
                        join.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // produced properties overridden before join
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(0));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(2, 1));

                toMap1.setRequiredGlobalProps(rgp);
                toMap1.setRequiredLocalProps(rlp);

                toJoin1.setRequiredGlobalProps(null);
                toJoin1.setRequiredLocalProps(null);

                toJoin1.setShipStrategy(
                        ShipStrategyType.PARTITION_HASH,
                        new FieldList(2, 1),
                        DataExchangeMode.PIPELINED);
                toJoin1.setLocalStrategy(
                        LocalStrategy.SORT, new FieldList(7, 3), new boolean[] {true, false});

                FeedbackPropertiesMeetRequirementsReport report =
                        join.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(MET, report);
            }

            // produced properties before join match, after join match as well
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(0));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(2, 1));

                toMap1.setRequiredGlobalProps(null);
                toMap1.setRequiredLocalProps(null);

                toJoin1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toJoin1.setLocalStrategy(LocalStrategy.NONE);

                toJoin1.setRequiredGlobalProps(rgp);
                toJoin1.setRequiredLocalProps(rlp);

                toAfterJoin.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toAfterJoin.setLocalStrategy(LocalStrategy.NONE);

                toAfterJoin.setRequiredGlobalProps(rgp);
                toAfterJoin.setRequiredLocalProps(rlp);

                FeedbackPropertiesMeetRequirementsReport report =
                        join.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // produced properties before join match, after join do not match
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp1 = new RequestedGlobalProperties();
                rgp1.setHashPartitioned(new FieldList(0));

                RequestedGlobalProperties rgp2 = new RequestedGlobalProperties();
                rgp2.setHashPartitioned(new FieldList(3));

                RequestedLocalProperties rlp1 = new RequestedLocalProperties();
                rlp1.setGroupedFields(new FieldList(2, 1));

                RequestedLocalProperties rlp2 = new RequestedLocalProperties();
                rlp2.setGroupedFields(new FieldList(3, 4));

                toJoin1.setRequiredGlobalProps(rgp1);
                toJoin1.setRequiredLocalProps(rlp1);

                toAfterJoin.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toAfterJoin.setLocalStrategy(LocalStrategy.NONE);

                toAfterJoin.setRequiredGlobalProps(rgp2);
                toAfterJoin.setRequiredLocalProps(rlp2);

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // produced properties are overridden, does not matter that they do not match
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setAnyPartitioning(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(1));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(1, 2, 3));

                toJoin1.setRequiredGlobalProps(null);
                toJoin1.setRequiredLocalProps(null);

                toJoin1.setShipStrategy(
                        ShipStrategyType.PARTITION_HASH,
                        new FieldList(2, 1),
                        DataExchangeMode.PIPELINED);
                toJoin1.setLocalStrategy(
                        LocalStrategy.SORT, new FieldList(7, 3), new boolean[] {true, false});

                toAfterJoin.setRequiredGlobalProps(rgp);
                toAfterJoin.setRequiredLocalProps(rlp);

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(MET, report);
            }

            // local property overridden before join, local property mismatch after join not
            // relevant
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setAnyPartitioning(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(1, 2, 3));

                toJoin1.setRequiredGlobalProps(null);
                toJoin1.setRequiredLocalProps(null);

                toJoin1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toJoin1.setLocalStrategy(
                        LocalStrategy.SORT, new FieldList(7, 3), new boolean[] {true, false});

                toAfterJoin.setRequiredGlobalProps(null);
                toAfterJoin.setRequiredLocalProps(rlp);

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // local property overridden before join, global property mismatch after join void the
            // match
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setAnyPartitioning(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(1));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(1, 2, 3));

                toJoin1.setRequiredGlobalProps(null);
                toJoin1.setRequiredLocalProps(null);

                toJoin1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toJoin1.setLocalStrategy(
                        LocalStrategy.SORT, new FieldList(7, 3), new boolean[] {true, false});

                toAfterJoin.setRequiredGlobalProps(rgp);
                toAfterJoin.setRequiredLocalProps(rlp);

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }
        } catch (Exception e) {
            e.printStackTrace();
            fail(e.getMessage());
        }
    }

    @Test
    public void testTwoOperatorsBothDependent() {
        try {
            SourcePlanNode target = new SourcePlanNode(getSourceNode(), "Partial Solution");

            Channel toMap1 = new Channel(target);
            toMap1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toMap1.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode map1 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 1", toMap1, DriverStrategy.MAP);

            Channel toMap2 = new Channel(target);
            toMap2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toMap2.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode map2 =
                    new SingleInputPlanNode(getMapNode(), "Mapper 2", toMap2, DriverStrategy.MAP);

            Channel toJoin1 = new Channel(map1);
            toJoin1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toJoin1.setLocalStrategy(LocalStrategy.NONE);

            Channel toJoin2 = new Channel(map2);
            toJoin2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toJoin2.setLocalStrategy(LocalStrategy.NONE);

            DualInputPlanNode join =
                    new DualInputPlanNode(
                            getJoinNode(),
                            "Join",
                            toJoin1,
                            toJoin2,
                            DriverStrategy.HYBRIDHASH_BUILD_FIRST);

            Channel toAfterJoin = new Channel(join);
            toAfterJoin.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
            toAfterJoin.setLocalStrategy(LocalStrategy.NONE);
            SingleInputPlanNode afterJoin =
                    new SingleInputPlanNode(
                            getMapNode(), "After Join Mapper", toAfterJoin, DriverStrategy.MAP);

            // no properties from the partial solution, no required properties
            {
                GlobalProperties gp = new GlobalProperties();
                LocalProperties lp = LocalProperties.EMPTY;

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // some properties from the partial solution, no required properties
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // test requirements on one input and met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(0));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(2, 1));

                toJoin1.setRequiredGlobalProps(rgp);
                toJoin1.setRequiredLocalProps(rlp);

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // test requirements on both input and met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(0));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(2, 1));

                toJoin1.setRequiredGlobalProps(rgp);
                toJoin1.setRequiredLocalProps(rlp);

                toJoin2.setRequiredGlobalProps(rgp);
                toJoin2.setRequiredLocalProps(rlp);

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertTrue(report != null && report != NO_PARTIAL_SOLUTION && report != NOT_MET);
            }

            // test requirements on both inputs, one not met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp1 = new RequestedGlobalProperties();
                rgp1.setHashPartitioned(new FieldList(0));

                RequestedLocalProperties rlp1 = new RequestedLocalProperties();
                rlp1.setGroupedFields(new FieldList(2, 1));

                RequestedGlobalProperties rgp2 = new RequestedGlobalProperties();
                rgp2.setHashPartitioned(new FieldList(1));

                RequestedLocalProperties rlp2 = new RequestedLocalProperties();
                rlp2.setGroupedFields(new FieldList(0, 3));

                toJoin1.setRequiredGlobalProps(rgp1);
                toJoin1.setRequiredLocalProps(rlp1);

                toJoin2.setRequiredGlobalProps(rgp2);
                toJoin2.setRequiredLocalProps(rlp2);

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // test override on both inputs, later requirement ignored
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(1));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(0, 3));

                toJoin1.setRequiredGlobalProps(null);
                toJoin1.setRequiredLocalProps(null);

                toJoin2.setRequiredGlobalProps(null);
                toJoin2.setRequiredLocalProps(null);

                toJoin1.setShipStrategy(
                        ShipStrategyType.PARTITION_HASH,
                        new FieldList(88),
                        DataExchangeMode.PIPELINED);
                toJoin2.setShipStrategy(ShipStrategyType.BROADCAST, DataExchangeMode.PIPELINED);

                toAfterJoin.setRequiredGlobalProps(rgp);
                toAfterJoin.setRequiredLocalProps(rlp);

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(MET, report);
            }

            // test override on one inputs, later requirement met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(0));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(2, 1));

                toJoin1.setShipStrategy(
                        ShipStrategyType.PARTITION_HASH,
                        new FieldList(88),
                        DataExchangeMode.PIPELINED);
                toJoin2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);

                toAfterJoin.setRequiredGlobalProps(rgp);
                toAfterJoin.setRequiredLocalProps(rlp);

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(PENDING, report);
            }

            // test override on one input, later requirement not met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(3));

                RequestedLocalProperties rlp = new RequestedLocalProperties();
                rlp.setGroupedFields(new FieldList(77, 69));

                toJoin1.setShipStrategy(
                        ShipStrategyType.PARTITION_HASH,
                        new FieldList(88),
                        DataExchangeMode.PIPELINED);
                toJoin2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);

                toAfterJoin.setRequiredGlobalProps(rgp);
                toAfterJoin.setRequiredLocalProps(rlp);

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }

            // test override on one input locally, later global requirement not met
            {
                GlobalProperties gp = new GlobalProperties();
                gp.setHashPartitioned(new FieldList(0));
                LocalProperties lp = LocalProperties.forGrouping(new FieldList(2, 1));

                RequestedGlobalProperties rgp = new RequestedGlobalProperties();
                rgp.setHashPartitioned(new FieldList(3));

                toJoin1.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toJoin1.setLocalStrategy(
                        LocalStrategy.SORT, new FieldList(3), new boolean[] {false});

                toJoin2.setShipStrategy(ShipStrategyType.FORWARD, DataExchangeMode.PIPELINED);
                toJoin1.setLocalStrategy(LocalStrategy.NONE);

                toAfterJoin.setRequiredGlobalProps(rgp);
                toAfterJoin.setRequiredLocalProps(null);

                FeedbackPropertiesMeetRequirementsReport report =
                        afterJoin.checkPartialSolutionPropertiesMet(target, gp, lp);
                assertEquals(NOT_MET, report);
            }
        } catch (Exception e) {
            e.printStackTrace();
            fail(e.getMessage());
        }
    }

    // --------------------------------------------------------------------------------------------

    private static DataSourceNode getSourceNode() {
        return new DataSourceNode(
                new GenericDataSourceBase<String, TextInputFormat>(
                        new TextInputFormat(new Path("/")),
                        new OperatorInformation<String>(BasicTypeInfo.STRING_TYPE_INFO)));
    }

    private static MapNode getMapNode() {
        return new MapNode(
                new MapOperatorBase<String, String, MapFunction<String, String>>(
                        new IdentityMapper<String>(),
                        new UnaryOperatorInformation<String, String>(
                                BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO),
                        "map op"));
    }

    private static JoinNode getJoinNode() {
        return new JoinNode(
                new InnerJoinOperatorBase<
                        String, String, String, FlatJoinFunction<String, String, String>>(
                        new DummyFlatJoinFunction<String>(),
                        new BinaryOperatorInformation<String, String, String>(
                                BasicTypeInfo.STRING_TYPE_INFO,
                                BasicTypeInfo.STRING_TYPE_INFO,
                                BasicTypeInfo.STRING_TYPE_INFO),
                        new int[] {1},
                        new int[] {2},
                        "join op"));
    }
}