public static DualInputSemanticProperties createProjectionPropertiesDual(
            int[] fields,
            boolean[] isFromFirst,
            TypeInformation<?> inType1,
            TypeInformation<?> inType2) {
        DualInputSemanticProperties dsp = new DualInputSemanticProperties();

        int[] sourceOffsets1;
        if (inType1 instanceof TupleTypeInfo<?>) {
            sourceOffsets1 = new int[inType1.getArity()];
            sourceOffsets1[0] = 0;
            for (int i = 1; i < inType1.getArity(); i++) {
                sourceOffsets1[i] =
                        ((TupleTypeInfo<?>) inType1).getTypeAt(i - 1).getTotalFields()
                                + sourceOffsets1[i - 1];
            }
        } else {
            sourceOffsets1 = new int[] {0};
        }

        int[] sourceOffsets2;
        if (inType2 instanceof TupleTypeInfo<?>) {
            sourceOffsets2 = new int[inType2.getArity()];
            sourceOffsets2[0] = 0;
            for (int i = 1; i < inType2.getArity(); i++) {
                sourceOffsets2[i] =
                        ((TupleTypeInfo<?>) inType2).getTypeAt(i - 1).getTotalFields()
                                + sourceOffsets2[i - 1];
            }
        } else {
            sourceOffsets2 = new int[] {0};
        }

        int targetOffset = 0;
        for (int i = 0; i < fields.length; i++) {
            int sourceOffset;
            int numFieldsToCopy;
            int input;
            if (isFromFirst[i]) {
                input = 0;
                if (fields[i] == -1) {
                    sourceOffset = 0;
                    numFieldsToCopy = inType1.getTotalFields();
                } else {
                    sourceOffset = sourceOffsets1[fields[i]];
                    numFieldsToCopy =
                            ((TupleTypeInfo<?>) inType1).getTypeAt(fields[i]).getTotalFields();
                }
            } else {
                input = 1;
                if (fields[i] == -1) {
                    sourceOffset = 0;
                    numFieldsToCopy = inType2.getTotalFields();
                } else {
                    sourceOffset = sourceOffsets2[fields[i]];
                    numFieldsToCopy =
                            ((TupleTypeInfo<?>) inType2).getTypeAt(fields[i]).getTotalFields();
                }
            }

            for (int j = 0; j < numFieldsToCopy; j++) {
                dsp.addForwardedField(input, sourceOffset + j, targetOffset + j);
            }
            targetOffset += numFieldsToCopy;
        }

        return dsp;
    }
@Test
    void testDualProjectionProperties() {

        int[] pMap = new int[] {4, 2, 0, 1, 3, 4};
        boolean[] iMap = new boolean[] {true, true, false, true, false, false};
        DualInputSemanticProperties sp =
                SemanticPropUtil.createProjectionPropertiesDual(
                        pMap, iMap, fiveIntTupleType, fiveIntTupleType);
        assertThat(sp.getForwardingTargetFields(0, 4)).contains(0);
        assertThat(sp.getForwardingTargetFields(0, 2)).contains(1);
        assertThat(sp.getForwardingTargetFields(1, 0)).contains(2);
        assertThat(sp.getForwardingTargetFields(0, 1)).contains(3);
        assertThat(sp.getForwardingTargetFields(1, 3)).contains(4);
        assertThat(sp.getForwardingTargetFields(1, 4)).contains(5);

        pMap = new int[] {4, 2, 0, 4, 0, 1};
        iMap = new boolean[] {true, true, false, true, false, false};
        sp =
                SemanticPropUtil.createProjectionPropertiesDual(
                        pMap, iMap, fiveIntTupleType, fiveIntTupleType);
        assertThat(sp.getForwardingTargetFields(0, 4)).containsExactly(0, 3);
        assertThat(sp.getForwardingTargetFields(1, 0)).containsExactly(4, 2);
        assertThat(sp.getForwardingTargetFields(0, 2)).containsExactly(1);
        assertThat(sp.getForwardingTargetFields(1, 1)).containsExactly(5);

        pMap = new int[] {2, 1, 0, 1};
        iMap = new boolean[] {false, false, true, true};
        sp =
                SemanticPropUtil.createProjectionPropertiesDual(
                        pMap, iMap, nestedTupleType, threeIntTupleType);
        assertThat(sp.getForwardingTargetFields(1, 2)).contains(0);
        assertThat(sp.getForwardingTargetFields(1, 1)).contains(1);
        assertThat(sp.getForwardingTargetFields(0, 0)).contains(2);
        assertThat(sp.getForwardingTargetFields(0, 1)).contains(3);
        assertThat(sp.getForwardingTargetFields(0, 2)).contains(4);
        assertThat(sp.getForwardingTargetFields(0, 3)).contains(5);

        pMap = new int[] {1, 0, 0};
        iMap = new boolean[] {false, false, true};
        sp =
                SemanticPropUtil.createProjectionPropertiesDual(
                        pMap, iMap, nestedTupleType, deepNestedTupleType);
        assertThat(sp.getForwardingTargetFields(1, 1)).contains(0);
        assertThat(sp.getForwardingTargetFields(1, 2)).contains(1);
        assertThat(sp.getForwardingTargetFields(1, 3)).contains(2);
        assertThat(sp.getForwardingTargetFields(1, 4)).contains(3);
        assertThat(sp.getForwardingTargetFields(1, 5)).contains(4);
        assertThat(sp.getForwardingTargetFields(1, 0)).contains(5);
        assertThat(sp.getForwardingTargetFields(0, 0)).contains(6);
        assertThat(sp.getForwardingTargetFields(0, 1)).contains(7);
        assertThat(sp.getForwardingTargetFields(0, 2)).contains(8);

        pMap = new int[] {4, 2, 1, 0};
        iMap = new boolean[] {true, false, true, false};
        sp =
                SemanticPropUtil.createProjectionPropertiesDual(
                        pMap, iMap, fiveIntTupleType, pojoInTupleType);
        assertThat(sp.getForwardingTargetFields(0, 4)).contains(0);
        assertThat(sp.getForwardingTargetFields(1, 2)).contains(1);
        assertThat(sp.getForwardingTargetFields(1, 3)).contains(2);
        assertThat(sp.getForwardingTargetFields(1, 4)).contains(3);
        assertThat(sp.getForwardingTargetFields(1, 5)).contains(4);
        assertThat(sp.getForwardingTargetFields(0, 1)).contains(5);
        assertThat(sp.getForwardingTargetFields(1, 0)).contains(6);

        pMap = new int[] {2, 3, -1, 0};
        iMap = new boolean[] {true, true, false, true};
        sp = SemanticPropUtil.createProjectionPropertiesDual(pMap, iMap, fiveIntTupleType, intType);
        assertThat(sp.getForwardingTargetFields(0, 2)).contains(0);
        assertThat(sp.getForwardingTargetFields(0, 3)).contains(1);
        assertThat(sp.getForwardingTargetFields(1, 0)).contains(2);
        assertThat(sp.getForwardingTargetFields(0, 0)).contains(3);

        pMap = new int[] {-1, -1};
        iMap = new boolean[] {false, true};
        sp = SemanticPropUtil.createProjectionPropertiesDual(pMap, iMap, intType, nestedPojoType);
        assertThat(sp.getForwardingTargetFields(1, 0)).contains(0);
        assertThat(sp.getForwardingTargetFields(1, 1)).contains(1);
        assertThat(sp.getForwardingTargetFields(1, 2)).contains(2);
        assertThat(sp.getForwardingTargetFields(1, 3)).contains(3);
        assertThat(sp.getForwardingTargetFields(1, 4)).contains(4);
        assertThat(sp.getForwardingTargetFields(1, 5)).contains(5);
        assertThat(sp.getForwardingTargetFields(0, 0)).contains(6);

        pMap = new int[] {-1, -1};
        iMap = new boolean[] {true, false};
        sp = SemanticPropUtil.createProjectionPropertiesDual(pMap, iMap, intType, nestedPojoType);
        assertThat(sp.getForwardingTargetFields(0, 0)).contains(0);
        assertThat(sp.getForwardingTargetFields(1, 0)).contains(1);
        assertThat(sp.getForwardingTargetFields(1, 1)).contains(2);
        assertThat(sp.getForwardingTargetFields(1, 2)).contains(3);
        assertThat(sp.getForwardingTargetFields(1, 3)).contains(4);
        assertThat(sp.getForwardingTargetFields(1, 4)).contains(5);
        assertThat(sp.getForwardingTargetFields(1, 5)).contains(6);
    }