SemanticPropUtil { 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; } private  SemanticPropUtil(); static SingleInputSemanticProperties createProjectionPropertiesSingle(int[] fields, CompositeType<?> inType); static DualInputSemanticProperties createProjectionPropertiesDual(
			int[] fields, boolean[] isFromFirst, TypeInformation<?> inType1, TypeInformation<?> inType2); static SingleInputSemanticProperties addSourceFieldOffset(SingleInputSemanticProperties props,
																		int numInputFields, int offset); static DualInputSemanticProperties addSourceFieldOffsets(DualInputSemanticProperties props,
																	int numInputFields1, int numInputFields2,
																	int offset1, int offset2); static SingleInputSemanticProperties getSemanticPropsSingle(
			Set<Annotation> set, TypeInformation<?> inType, TypeInformation<?> outType); static DualInputSemanticProperties getSemanticPropsDual(
			Set<Annotation> set, TypeInformation<?> inType1, TypeInformation<?> inType2, TypeInformation<?> outType); static void getSemanticPropsSingleFromString(SingleInputSemanticProperties result,
														String[] forwarded, String[] nonForwarded, String[] readSet,
														TypeInformation<?> inType, TypeInformation<?> outType); static void getSemanticPropsSingleFromString(SingleInputSemanticProperties result,
			String[] forwarded, String[] nonForwarded, String[] readSet,
			TypeInformation<?> inType, TypeInformation<?> outType,
			boolean skipIncompatibleTypes); static void getSemanticPropsDualFromString(DualInputSemanticProperties result,
			String[] forwardedFirst, String[] forwardedSecond,
			String[] nonForwardedFirst, String[] nonForwardedSecond, String[]
			readFieldsFirst, String[] readFieldsSecond,
			TypeInformation<?> inType1, TypeInformation<?> inType2, TypeInformation<?> outType); static void getSemanticPropsDualFromString(DualInputSemanticProperties result,
			String[] forwardedFirst, String[] forwardedSecond,
			String[] nonForwardedFirst, String[] nonForwardedSecond, String[]
			readFieldsFirst, String[] readFieldsSecond,
			TypeInformation<?> inType1, TypeInformation<?> inType2, TypeInformation<?> outType,
			boolean skipIncompatibleTypes);  }
@Test public 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); assertTrue(sp.getForwardingTargetFields(0, 4).contains(0)); assertTrue(sp.getForwardingTargetFields(0, 2).contains(1)); assertTrue(sp.getForwardingTargetFields(1, 0).contains(2)); assertTrue(sp.getForwardingTargetFields(0, 1).contains(3)); assertTrue(sp.getForwardingTargetFields(1, 3).contains(4)); assertTrue(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); assertTrue(sp.getForwardingTargetFields(0, 4).size() == 2); assertTrue(sp.getForwardingTargetFields(1, 0).size() == 2); assertTrue(sp.getForwardingTargetFields(0, 4).contains(0)); assertTrue(sp.getForwardingTargetFields(0, 2).contains(1)); assertTrue(sp.getForwardingTargetFields(1, 0).contains(2)); assertTrue(sp.getForwardingTargetFields(0, 4).contains(3)); assertTrue(sp.getForwardingTargetFields(1, 0).contains(4)); assertTrue(sp.getForwardingTargetFields(1, 1).contains(5)); pMap = new int[]{2,1,0,1}; iMap = new boolean[]{false, false, true, true}; sp = SemanticPropUtil.createProjectionPropertiesDual(pMap, iMap, nestedTupleType, threeIntTupleType); assertTrue(sp.getForwardingTargetFields(1, 2).contains(0)); assertTrue(sp.getForwardingTargetFields(1, 1).contains(1)); assertTrue(sp.getForwardingTargetFields(0, 0).contains(2)); assertTrue(sp.getForwardingTargetFields(0, 1).contains(3)); assertTrue(sp.getForwardingTargetFields(0, 2).contains(4)); assertTrue(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); assertTrue(sp.getForwardingTargetFields(1, 1).contains(0)); assertTrue(sp.getForwardingTargetFields(1, 2).contains(1)); assertTrue(sp.getForwardingTargetFields(1, 3).contains(2)); assertTrue(sp.getForwardingTargetFields(1, 4).contains(3)); assertTrue(sp.getForwardingTargetFields(1, 5).contains(4)); assertTrue(sp.getForwardingTargetFields(1, 0).contains(5)); assertTrue(sp.getForwardingTargetFields(0, 0).contains(6)); assertTrue(sp.getForwardingTargetFields(0, 1).contains(7)); assertTrue(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); assertTrue(sp.getForwardingTargetFields(0, 4).contains(0)); assertTrue(sp.getForwardingTargetFields(1, 2).contains(1)); assertTrue(sp.getForwardingTargetFields(1, 3).contains(2)); assertTrue(sp.getForwardingTargetFields(1, 4).contains(3)); assertTrue(sp.getForwardingTargetFields(1, 5).contains(4)); assertTrue(sp.getForwardingTargetFields(0, 1).contains(5)); assertTrue(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); assertTrue(sp.getForwardingTargetFields(0, 2).contains(0)); assertTrue(sp.getForwardingTargetFields(0, 3).contains(1)); assertTrue(sp.getForwardingTargetFields(1, 0).contains(2)); assertTrue(sp.getForwardingTargetFields(0, 0).contains(3)); pMap = new int[]{-1,-1}; iMap = new boolean[]{false, true}; sp = SemanticPropUtil.createProjectionPropertiesDual(pMap, iMap, intType, nestedPojoType); assertTrue(sp.getForwardingTargetFields(1, 0).contains(0)); assertTrue(sp.getForwardingTargetFields(1, 1).contains(1)); assertTrue(sp.getForwardingTargetFields(1, 2).contains(2)); assertTrue(sp.getForwardingTargetFields(1, 3).contains(3)); assertTrue(sp.getForwardingTargetFields(1, 4).contains(4)); assertTrue(sp.getForwardingTargetFields(1, 5).contains(5)); assertTrue(sp.getForwardingTargetFields(0, 0).contains(6)); pMap = new int[]{-1,-1}; iMap = new boolean[]{true, false}; sp = SemanticPropUtil.createProjectionPropertiesDual(pMap, iMap, intType, nestedPojoType); assertTrue(sp.getForwardingTargetFields(0, 0).contains(0)); assertTrue(sp.getForwardingTargetFields(1, 0).contains(1)); assertTrue(sp.getForwardingTargetFields(1, 1).contains(2)); assertTrue(sp.getForwardingTargetFields(1, 2).contains(3)); assertTrue(sp.getForwardingTargetFields(1, 3).contains(4)); assertTrue(sp.getForwardingTargetFields(1, 4).contains(5)); assertTrue(sp.getForwardingTargetFields(1, 5).contains(6)); }