flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject [05/13] flink git commit: [FLINK-1328] [api-breaking][java-api][scala-api][optimizer] Reworked semantic annotations for functions. - Renamed constantField annotations to forwardedFields annotation - Forwarded fields can be defined for (nested) tuples, Po
Date Wed, 28 Jan 2015 01:24:05 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/de8e066c/flink-java/src/test/java/org/apache/flink/api/java/functions/SemanticPropUtilTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/functions/SemanticPropUtilTest.java b/flink-java/src/test/java/org/apache/flink/api/java/functions/SemanticPropUtilTest.java
index 5a7afd7..cf81550 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/functions/SemanticPropUtilTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/functions/SemanticPropUtilTest.java
@@ -19,14 +19,15 @@
 
 package org.apache.flink.api.java.functions;
 
+import org.apache.flink.api.common.operators.SemanticProperties.InvalidSemanticAnnotationException;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
 import org.junit.Assert;
-import org.apache.flink.api.common.InvalidProgramException;
 import org.apache.flink.api.common.operators.DualInputSemanticProperties;
 import org.apache.flink.api.common.operators.SingleInputSemanticProperties;
 import org.apache.flink.api.common.operators.util.FieldSet;
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.functions.SemanticPropUtil;
 import org.apache.flink.api.java.tuple.Tuple3;
 import org.apache.flink.api.java.tuple.Tuple4;
 import org.apache.flink.api.java.tuple.Tuple5;
@@ -35,588 +36,1000 @@ import org.junit.Test;
 
 public class SemanticPropUtilTest {
 	
+	private final TypeInformation<?> threeIntTupleType = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
+			BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
+
+	private final TypeInformation<?> fourIntTupleType = new TupleTypeInfo<Tuple4<Integer, Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
+			BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
+
+	private final TypeInformation<?> fiveIntTupleType = new TupleTypeInfo<Tuple5<Integer, Integer, Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
+			BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
+
+	private final TypeInformation<?> threeMixedTupleType = new TupleTypeInfo<Tuple3<Integer, Long, String>>(BasicTypeInfo.INT_TYPE_INFO,
+			BasicTypeInfo.LONG_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO);
+
+	private final TypeInformation<?> nestedTupleType = new TupleTypeInfo<Tuple3<Tuple3<Integer, Integer, Integer>, Integer, Integer>>(threeIntTupleType,
+			BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
+
+	private final TypeInformation<?> deepNestedTupleType = new TupleTypeInfo<Tuple3<Integer, Tuple3<Tuple3<Integer, Integer, Integer>, Integer, Integer>, Integer>>(
+			BasicTypeInfo.INT_TYPE_INFO, nestedTupleType, BasicTypeInfo.INT_TYPE_INFO);
+
+	private final TypeInformation<?> pojoType = TypeExtractor.getForClass(TestPojo.class);
+
+	private final TypeInformation<?> pojo2Type = TypeExtractor.getForClass(TestPojo2.class);
+
+	private final TypeInformation<?> nestedPojoType = TypeExtractor.getForClass(NestedTestPojo.class);
+
+	private final TypeInformation<?> pojoInTupleType = new TupleTypeInfo<Tuple3<Integer, Integer, TestPojo>>(BasicTypeInfo.INT_TYPE_INFO,
+			BasicTypeInfo.INT_TYPE_INFO, pojoType);
+
+	private final TypeInformation<?> intType = BasicTypeInfo.INT_TYPE_INFO;
+
 	// --------------------------------------------------------------------------------------------
-	// Constant Fields
+	// Projection Operator Properties
 	// --------------------------------------------------------------------------------------------
 
 	@Test
-	public void testConstantWithArrowIndividualStrings() {
-		// no spaces
-		{
-			String[] constantFields = { "0->0,1", "1->2" };
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 2);
-			Assert.assertTrue(fs.contains(0));
-			Assert.assertTrue(fs.contains(1));
-	
-			fs = sp.getForwardedField(1);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-		}
-		
-		// with spaces
-		{
-			String[] constantFields = { "0 -> 0 ,   1 ", " 1     -> 2  " };
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 2);
-			Assert.assertTrue(fs.contains(0));
-			Assert.assertTrue(fs.contains(1));
-	
-			fs = sp.getForwardedField(1);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-		}
+	public void testSingleProjectionProperties() {
+
+		int[] pMap = new int[] {3,0,4};
+		SingleInputSemanticProperties sp = SemanticPropUtil.createProjectionPropertiesSingle(pMap, (CompositeType<?>) fiveIntTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(2));
+
+		pMap = new int[] {2,2,1,1};
+		sp = SemanticPropUtil.createProjectionPropertiesSingle(pMap, (CompositeType<?>) fiveIntTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).size() == 2);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 2);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(1));
+
+		pMap = new int[] {2,0};
+		sp = SemanticPropUtil.createProjectionPropertiesSingle(pMap, (CompositeType<?>) nestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(3));
+
+		pMap = new int[] {2,0,1};
+		sp = SemanticPropUtil.createProjectionPropertiesSingle(pMap, (CompositeType<?>) deepNestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 6).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(5));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).contains(6));
+
+		pMap = new int[] {2, 1};
+		sp = SemanticPropUtil.createProjectionPropertiesSingle(pMap, (CompositeType<?>) pojoInTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(4));
+
 	}
-	
+
 	@Test
-	public void testConstantWithArrowIndividualStringsSpaces() {
-		String[] constantFields = { "  1-> 1 , 2", "2 ->3" };
+	public void testDualProjectionProperties() {
 
-		TypeInformation<?> type = new TupleTypeInfo<Tuple4<Integer, Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-				BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-		SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
+		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);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 0).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 3).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 4).contains(5));
 
-		FieldSet fs = sp.getForwardedField(1);
-		Assert.assertTrue(fs.size() == 2);
-		Assert.assertTrue(fs.contains(1));
-		Assert.assertTrue(fs.contains(2));
+		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);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).size() == 2);
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 0).size() == 2);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 0).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 0).contains(4));
+		Assert.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);
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 2).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(4));
+		Assert.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);
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 1).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 2).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 3).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 4).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 5).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 0).contains(5));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(6));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(7));
+		Assert.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);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 2).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 3).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 4).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 5).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(5));
+		Assert.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);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 0).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(3));
+
+		pMap = new int[]{-1,-1};
+		iMap = new boolean[]{false, true};
+		sp = SemanticPropUtil.createProjectionPropertiesDual(pMap, iMap, intType, nestedPojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 3).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 4).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 5).contains(5));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(6));
+
+		pMap = new int[]{-1,-1};
+		iMap = new boolean[]{true, false};
+		sp = SemanticPropUtil.createProjectionPropertiesDual(pMap, iMap, intType, nestedPojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 0).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 1).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 2).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 3).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 4).contains(5));
+		Assert.assertTrue(sp.getForwardingTargetFields(1, 5).contains(6));
 
-		fs = sp.getForwardedField(2);
-		Assert.assertTrue(fs.size() == 1);
-		Assert.assertTrue(fs.contains(3));
 	}
-	
+
+	// --------------------------------------------------------------------------------------------
+	// Forwarded Fields Annotation
+	// --------------------------------------------------------------------------------------------
+
 	@Test
-	public void testConstantWithArrowOneString() {
-		// narrow (no spaces)
-		{
-			String[] constantFields = { "0->0,1;1->2" };
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 2);
-			Assert.assertTrue(fs.contains(0));
-			Assert.assertTrue(fs.contains(1));
-	
-			fs = sp.getForwardedField(1);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-		}
-		
-		// wide (with spaces)
-		{
-			String[] constantFields = { "  0 ->  0  ,  1  ;   1  -> 2 " };
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 2);
-			Assert.assertTrue(fs.contains(0));
-			Assert.assertTrue(fs.contains(1));
-	
-			fs = sp.getForwardedField(1);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-		}
+	public void testForwardedNoArrrowIndividualStrings() {
+		String[] forwardedFields = {"f2","f3","f0"};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, fiveIntTupleType, fiveIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
 	}
-	
+
 	@Test
-	public void testConstantNoArrrowIndividualStrings() {
-		// no spaces
-		{
-			String[] constantFields = {"2","3","0"};
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple5<Integer, Integer, Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(0));
-	
-			fs = sp.getForwardedField(2);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-	
-			fs = sp.getForwardedField(3);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(3));
-	
-			Assert.assertNull(sp.getForwardedField(1));
-			Assert.assertNull(sp.getForwardedField(4));
-		}
-		
-		// with spaces
-		{
-			String[] constantFields = {" 2   ","3  ","  0"};
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple5<Integer, Integer, Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(0));
-	
-			fs = sp.getForwardedField(2);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-	
-			fs = sp.getForwardedField(3);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(3));
-	
-			Assert.assertNull(sp.getForwardedField(1));
-			Assert.assertNull(sp.getForwardedField(4));
-		}
+	public void testForwardedNoArrrowOneString() {
+		String[] forwardedFields = {"f2;f3;f0"};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, fiveIntTupleType, fiveIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
+
+		forwardedFields[0] = "2;3;0";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, fiveIntTupleType, fiveIntTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
+
+		forwardedFields[0] = "2;3;0;";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, fiveIntTupleType, fiveIntTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
 	}
-	
+
 	@Test
-	public void testConstantNoArrrowOneString() {
-		// no spaces
-		{
-			String[] constantFields = {"2;3;0"};
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple5<Integer, Integer, Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(0));
-	
-			fs = sp.getForwardedField(2);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-	
-			fs = sp.getForwardedField(3);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(3));
-	
-			Assert.assertNull(sp.getForwardedField(1));
-			Assert.assertNull(sp.getForwardedField(4));
-		}
-		
-		// no spaces
-		{
-			String[] constantFields = {"  2  ;   3  ;  0   "};
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple5<Integer, Integer, Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(0));
-	
-			fs = sp.getForwardedField(2);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-	
-			fs = sp.getForwardedField(3);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(3));
-	
-			Assert.assertNull(sp.getForwardedField(1));
-			Assert.assertNull(sp.getForwardedField(4));
-		}
+	public void testForwardedNoArrrowSpaces() {
+		String[] forwardedFields = {"  f2  ;   f3  ;  f0   "};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, fiveIntTupleType, fiveIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
 	}
-	
-//	@Test
-//	public void testConstantNoArrrowOneStringInvalidDelimiter() {
-//		String[] constantFields = {"2,3,0"};
-//
-//		TypeInformation<?> type = new TupleTypeInfo<Tuple5<Integer, Integer, Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-//				BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-//		try {
-//			SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-//			Assert.fail("Accepted invalid input");
-//		}
-//		catch (InvalidProgramException e) {
-//			// bueno!
-//		}
-//	}
-	
+
 	@Test
-	public void testConstantMixedOneString() {
-		// no spaces
-		{
-			String[] constantFields = {"2,3;0->1,4;4->0"};
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple5<Integer, Integer, Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 2);
-			Assert.assertTrue(fs.contains(1));
-			Assert.assertTrue(fs.contains(4));
-	
-			fs = sp.getForwardedField(2);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-	
-			fs = sp.getForwardedField(3);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(3));
-	
-			fs = sp.getForwardedField(4);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(0));
-		}
-		
-		// with spaces
-		{
-			String[] constantFields = {" 2  ,  3   ;  0 -> 1  , 4 ; 4 ->  0"};
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple5<Integer, Integer, Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 2);
-			Assert.assertTrue(fs.contains(1));
-			Assert.assertTrue(fs.contains(4));
-	
-			fs = sp.getForwardedField(2);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-	
-			fs = sp.getForwardedField(3);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(3));
-	
-			fs = sp.getForwardedField(4);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(0));
-		}
+	public void testForwardedWithArrowIndividualStrings() {
+		String[] forwardedFields = { "f0->f1", "f1->f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, fiveIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
 	}
 
 	@Test
-	public void testConstantWildCard() {
-		// no spaces
-		{
-			String[] constantFields = { "*" };
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(1);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(1));
-	
-			fs = sp.getForwardedField(2);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-	
-			fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(0));
-		}
-		
-//		// with spaces
-//		{
-//			String[] constantFields = { "   *   " };
-//	
-//			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-//					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-//			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-//	
-//			FieldSet fs = sp.getForwardedField(1);
-//			Assert.assertTrue(fs.size() == 1);
-//			Assert.assertTrue(fs.contains(1));
-//	
-//			fs = sp.getForwardedField(2);
-//			Assert.assertTrue(fs.size() == 1);
-//			Assert.assertTrue(fs.contains(2));
-//	
-//			fs = sp.getForwardedField(0);
-//			Assert.assertTrue(fs.size() == 1);
-//			Assert.assertTrue(fs.contains(0));
-//		}
-	}
-		
-
-	@Test
-	public void testConstantWildCard2() {
-		// no spaces
-		{
-			String[] constantFields = { "1->*" };
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
+	public void testForwardedWithArrowOneString() {
+		String[] forwardedFields = { "f0->f0;f1->f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, fiveIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
+
+		forwardedFields[0] = "0->0;1->2";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, fiveIntTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
+	}
+
+	@Test
+	public void testForwardedWithArrowSpaces() {
+		String[] forwardedFields = { "  f0 ->  f0    ;   f1  -> f2 " };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, fiveIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
+	}
 	
-			FieldSet fs = sp.getForwardedField(1);
-			Assert.assertTrue(fs.size() == 3);
-			Assert.assertTrue(fs.contains(0));
-			Assert.assertTrue(fs.contains(1));
-			Assert.assertTrue(fs.contains(2));
-			Assert.assertTrue(sp.getForwardedField(0) == null);
-			Assert.assertTrue(sp.getForwardedField(2) == null);
-		}
-		
-//		// with spaces
-//		{
-//			String[] constantFields = { "  1  -> * " };
-//			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-//					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-//			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-//	
-//			FieldSet fs = sp.getForwardedField(1);
-//			Assert.assertTrue(fs.size() == 3);
-//			Assert.assertTrue(fs.contains(0));
-//			Assert.assertTrue(fs.contains(1));
-//			Assert.assertTrue(fs.contains(2));
-//			Assert.assertTrue(sp.getForwardedField(0) == null);
-//			Assert.assertTrue(sp.getForwardedField(2) == null);
-//		}
-	}
-
-	@Test
-	public void testConstantInvalidString() {
-		String[] constantFields = { "notValid" };
-
-		TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-				BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-
-		try {
-			SemanticPropUtil.getSemanticPropsSingleFromString(constantFields, null, null, type, type);
-			Assert.fail("accepted invalid input");
-		} catch (InvalidProgramException e) {
-			// good
-		} catch (Exception e) {
-			Assert.fail("Wrong type of exception thrown");
-		}
+	@Test
+	public void testForwardedMixedOneString() {
+		String[] forwardedFields = {"f2;f3;f0->f4;f4->f0"};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, fiveIntTupleType, fiveIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(0));
+	}
+
+	@Test
+	public void testForwardedBasicType() {
+		String[] forwardedFields = {"f1->*"};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, intType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
+
+		forwardedFields[0] = "*->f2";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, intType, threeIntTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(2));
+
+		forwardedFields[0] = "*->*";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, intType, intType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+	}
+
+	@Test
+	public void testForwardedWildCard() {
+		String[] forwardedFields = { "*" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, threeIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).size() == 0);
+
+		forwardedFields[0] = "*";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, deepNestedTupleType, deepNestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(4));
+
+	}
+
+	@Test
+	public void testForwardedNestedTuples() {
+		String[] forwardedFields = { "f0->f0.f0; f1->f0.f1; f2->f0.f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, nestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+
+		forwardedFields[0] = "f0.f0->f1.f0.f2; f0.f1->f2; f2->f1.f2; f1->f0";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, nestedTupleType, deepNestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(6));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(5));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(0));
+
+		forwardedFields[0] = "0.0->1.0.2; 0.1->2; 2->1.2; 1->0";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, nestedTupleType, deepNestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(6));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(5));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(0));
+
+		forwardedFields[0] = "f1.f0.*->f0.*; f0->f2";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, deepNestedTupleType, nestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(4));
+
+		forwardedFields[0] = "1.0.*->0.*; 0->2";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, deepNestedTupleType, nestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(4));
+
+		forwardedFields[0] = "f1.f0->f0; f0->f2";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, deepNestedTupleType, nestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(4));
+
+		forwardedFields[0] = "1.0->0; 0->2";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, deepNestedTupleType, nestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(4));
+
+		forwardedFields[0] = "f1.f0.f1; f1.f1; f2";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, deepNestedTupleType, deepNestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 6).contains(6));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).size() == 0);
+
+		forwardedFields[0] = "f1.f0.*; f1.f2";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, deepNestedTupleType, deepNestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).contains(5));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 6).size() == 0);
+
+	}
+
+	@Test
+	public void testForwardedPojo() {
+
+		String[] forwardedFields = { "int1->int2; int3->int1; string1 " };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoType, pojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
+
+		forwardedFields[0] = "f1->int1; f0->int3 ";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, pojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(0));
+
+		forwardedFields[0] = "int1->f2; int2->f0; int3->f1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoType, threeIntTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(1));
+
+		forwardedFields[0] = "*->pojo1.*";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoType, nestedPojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(4));
+
+		forwardedFields[0] = "*->pojo1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoType, nestedPojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(4));
+
+		forwardedFields[0] = "int1; string1; int2->pojo1.int3";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoType, nestedPojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(5));
+
+		forwardedFields[0] = "pojo1.*->f2.*; int1->f1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, nestedPojoType, pojoInTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(5));
+
+		forwardedFields[0] = "f2.*->*";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoInTupleType, pojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).contains(3));
+
+		forwardedFields[0] = "pojo1->f2; int1->f1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, nestedPojoType, pojoInTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(5));
+
+		forwardedFields[0] = "f2->*";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoInTupleType, pojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).contains(3));
+
+		forwardedFields[0] = "int2; string1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoType, pojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
+
+		forwardedFields[0] = "pojo1.int1; string1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, nestedPojoType, nestedPojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).contains(5));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).size() == 0);
+
+		forwardedFields[0] = "pojo1.*; int1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, nestedPojoType, nestedPojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).size() == 0);
+
+		forwardedFields[0] = "pojo1; int1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, nestedPojoType, nestedPojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).size() == 0);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testInvalidPojoField() {
+		String[] forwardedFields = { "invalidField" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoType, threeIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedNoArrowOneStringInvalidDelimiter() {
+		String[] forwardedFields = {"f2,f3,f0"};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, fiveIntTupleType, fiveIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedSameTargetTwice() {
+		String[] forwardedFields = {"f0->f2; f1->f2"};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, fiveIntTupleType, fiveIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedInvalidTargetFieldType1() {
+		String[] forwardedFields = {"f0->f0","f1->f2"};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, fiveIntTupleType, threeMixedTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedInvalidTargetFieldType2() {
+		String[] forwardedFields = { "f2.*->*" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoInTupleType, pojo2Type);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedInvalidTargetFieldType3() {
+		String[] forwardedFields = { "*" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoInTupleType, pojo2Type);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedInvalidTargetFieldType4() {
+		String[] forwardedFields = { "int1; string1" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoInTupleType, pojo2Type);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedInvalidTargetFieldType5() {
+		String[] forwardedFields = {"f0.*->*"};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, nestedTupleType, fiveIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedWildCardInvalidTypes1() {
+		String[] forwardedFields = { "*" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, fiveIntTupleType, threeIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedWildCardInvalidTypes2() {
+		String[] forwardedFields = { "*" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, fiveIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedWildCardInvalidTypes3() {
+		String[] forwardedFields = { "*" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, pojoType, pojo2Type);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedForwardWildCard() {
+		String[] forwardedFields = { "f1->*" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, threeIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedInvalidExpression() {
+		String[] forwardedFields = { "f0" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, intType, threeIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedForwardMultiFields() {
+		String[] forwardedFields = { "f1->f0,f1" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, threeIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testForwardedInvalidString() {
+		String[] forwardedFields = { "notValid" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, null, threeIntTupleType, threeIntTupleType);
 	}
 	
 	// --------------------------------------------------------------------------------------------
-	// Constant Fields Except
+	// Non-Forwarded Fields Annotation
 	// --------------------------------------------------------------------------------------------
 
 	@Test
-	public void testConstantExceptOneString() {
-		// no spaces
-		{
-			String[] constantFieldsExcept = { "1" };
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(null, constantFieldsExcept, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(0));
-	
-			Assert.assertNull(sp.getForwardedField(1));
-			
-			fs = sp.getForwardedField(2);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-		}
-		
-		// with spaces
-		{
-			String[] constantFieldsExcept = { " 1  "};
-			
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(null, constantFieldsExcept, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(0));
-	
-			Assert.assertNull(sp.getForwardedField(1));
-			
-			fs = sp.getForwardedField(2);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(2));
-		}
+	public void testNonForwardedIndividualStrings() {
+		String[] nonForwardedFields = { "f1", "f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, threeIntTupleType, threeIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
 	}
-	
-//	@Test
-//	public void testConstantExceptIndividualStrings() {
-//		// no spaces
-//		{
-//			String[] constantFieldsExcept = { "1", "2" };
-//	
-//			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-//					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-//			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(null, constantFieldsExcept, null, type, type);
-//	
-//			FieldSet fs = sp.getForwardedField(0);
-//			Assert.assertTrue(fs.size() == 1);
-//			Assert.assertTrue(fs.contains(0));
-//	
-//			Assert.assertNull(sp.getForwardedField(1));
-//			Assert.assertNull(sp.getForwardedField(2));
-//		}
-//		
-//		// with spaces
-//		{
-//			String[] constantFieldsExcept = { " 1  ", " 2" };
-//			
-//			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-//					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-//			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(null, constantFieldsExcept, null, type, type);
-//	
-//			FieldSet fs = sp.getForwardedField(0);
-//			Assert.assertTrue(fs.size() == 1);
-//			Assert.assertTrue(fs.contains(0));
-//	
-//			Assert.assertNull(sp.getForwardedField(1));
-//			Assert.assertNull(sp.getForwardedField(2));
-//		}
-//	}
-	
+
 	@Test
-	public void testConstantExceptSingleString() {
-		// no spaces
-		{
-			String[] constantFieldsExcept = { "1,2" };
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(null, constantFieldsExcept, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(0));
-	
-			Assert.assertNull(sp.getForwardedField(1));
-			Assert.assertNull(sp.getForwardedField(2));
-		}
-		
-		// with spaces
-		{
-			String[] constantFieldsExcept = { " 1  , 2" };
-			
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(null, constantFieldsExcept, null, type, type);
-	
-			FieldSet fs = sp.getForwardedField(0);
-			Assert.assertTrue(fs.size() == 1);
-			Assert.assertTrue(fs.contains(0));
-	
-			Assert.assertNull(sp.getForwardedField(1));
-			Assert.assertNull(sp.getForwardedField(2));
-		}
+	public void testNonForwardedSingleString() {
+		String[] nonForwardedFields = { "f1;f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, threeIntTupleType, threeIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
+
+		nonForwardedFields[0] = "f1;f2;";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, threeIntTupleType, threeIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
 	}
-	
+
+	@Test
+	public void testNonForwardedSpaces() {
+		String[] nonForwardedFields = { " f1 ;   f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, threeIntTupleType, threeIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
+	}
+
+	@Test
+	public void testNonForwardedNone() {
+		String[] nonForwardedFields = { "" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, threeIntTupleType, threeIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+	}
+
 	@Test
-	public void testConstantExceptString() {
-		String[] constantFieldsExcept = { "notValid" };
+	public void testNonForwardedNestedTuple() {
+		String[] nonForwardedFields = {"f1.f0.*; f1.f2; f0"};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, deepNestedTupleType, deepNestedTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 6).contains(6));
+
+		nonForwardedFields[0] = "f1.f0; f1.f2; f0";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, deepNestedTupleType, deepNestedTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 6).contains(6));
+
+		nonForwardedFields[0] = "f2; f1.f1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, deepNestedTupleType, deepNestedTupleType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).contains(5));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 6).size() == 0);
+
+	}
+
+	@Test
+	public void testNonForwardedPojo() {
+		String[] nonForwardedFields = { "int1; string1" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, pojoType, pojoType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).size() == 0);
+
+	}
+
+	@Test
+	public void testNonForwardedNestedPojo() {
+		String[] nonForwardedFields = { "int1; pojo1.*" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, nestedPojoType, nestedPojoType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).contains(5));
+
+		nonForwardedFields[0] = "pojo1.int2; string1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, nestedPojoType, nestedPojoType);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(1));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 3).contains(3));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 4).contains(4));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 5).size() == 0);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testNonForwardedInvalidTypes1() {
+		String[] nonForwardedFields = { "f1; f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, threeIntTupleType, nestedPojoType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testNonForwardedInvalidTypes2() {
+		String[] nonForwardedFields = { "f1; f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, nestedPojoType, threeIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testNonForwardedInvalidTypes3() {
+		String[] nonForwardedFields = { "f1; f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, threeIntTupleType, fiveIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testNonForwardedInvalidTypes4() {
+		String[] nonForwardedFields = { "f1; f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, fiveIntTupleType, threeIntTupleType);
+	}
 
-		TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-				BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testNonForwardedInvalidTypes5() {
+		String[] nonForwardedFields = { "int1" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, pojoType, pojo2Type);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testNonForwardedInvalidNesting() {
+		String[] nonForwardedFields = { "f0.f4" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, nestedTupleType, nestedTupleType);
+	}
 
-		try {
-			SemanticPropUtil.getSemanticPropsSingleFromString(null, constantFieldsExcept, null, type, type);
-			Assert.fail();
-		} catch (Exception e) {
-			// good
-		}
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testNonForwardedInvalidString() {
+		String[] nonForwardedFields = { "notValid" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, null, threeIntTupleType, threeIntTupleType);
 	}
 	
 	
 	// --------------------------------------------------------------------------------------------
-	// Constant Fields Except
+	// Read Fields Annotation
 	// --------------------------------------------------------------------------------------------
 
 	@Test
 	public void testReadFieldsIndividualStrings() {
-		// no spaces
-		{
-			String[] readFields = { "1", "2" };
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(null, null, readFields, type, type);
-	
-			FieldSet fs = sp.getReadFields();
-			Assert.assertTrue(fs.size() == 2);
-			Assert.assertTrue(fs.contains(2));
-			Assert.assertTrue(fs.contains(1));
-		}
-		
-		// with spaces
-		{
-			String[] readFields = { "   1    ", " 2   " };
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(null, null, readFields, type, type);
-	
-			FieldSet fs = sp.getReadFields();
-			Assert.assertTrue(fs.size() == 2);
-			Assert.assertTrue(fs.contains(2));
-			Assert.assertTrue(fs.contains(1));
-		}
+		String[] readFields = { "f1", "f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, threeIntTupleType, threeIntTupleType);
+
+		FieldSet fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 2);
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(1));
 	}
 	
 	@Test
 	public void testReadFieldsOneString() {
-		// no spaces
-		{
-			String[] readFields = { "1,2" };
-	
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(null, null, readFields, type, type);
-	
-			FieldSet fs = sp.getReadFields();
-			Assert.assertTrue(fs.size() == 2);
-			Assert.assertTrue(fs.contains(2));
-			Assert.assertTrue(fs.contains(1));
-		}
-		
-		// with spaces
-		{
-			String[] readFields = { "  1  , 2   " };
-			
-			TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-					BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-			SingleInputSemanticProperties sp = SemanticPropUtil.getSemanticPropsSingleFromString(null, null, readFields, type, type);
-	
-			FieldSet fs = sp.getReadFields();
-			Assert.assertTrue(fs.size() == 2);
-			Assert.assertTrue(fs.contains(2));
-			Assert.assertTrue(fs.contains(1));
-		}
+		String[] readFields = { "f1;f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, threeIntTupleType, threeIntTupleType);
+
+		FieldSet fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 2);
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(1));
+
+		readFields[0] = "f1;f2;";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, threeIntTupleType, threeIntTupleType);
+
+		fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 2);
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(1));
 	}
 
 	@Test
-	public void testReadFieldsInvalidString() {
-		String[] readFields = { "notValid" };
+	public void testReadFieldsSpaces() {
+		String[] readFields = { "  f1  ; f2   " };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, threeIntTupleType, threeIntTupleType);
+
+		FieldSet fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 2);
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(1));
+	}
+
+	@Test
+	public void testReadFieldsBasic() {
+		String[] readFields = { "*" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, intType, intType);
+
+		FieldSet fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 1);
+		Assert.assertTrue(fs.contains(0));
+
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, intType, fiveIntTupleType);
+
+		fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 1);
+		Assert.assertTrue(fs.contains(0));
+	}
+
+	@Test
+	public void testReadFieldsNestedTuples() {
+		String[] readFields = { "f0.f1; f0.f2; f2" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, nestedTupleType, intType);
+
+		FieldSet fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 3);
+		Assert.assertTrue(fs.contains(1));
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(4));
+
+		readFields[0] = "f0;f1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, nestedTupleType, intType);
+
+		fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 4);
+		Assert.assertTrue(fs.contains(0));
+		Assert.assertTrue(fs.contains(1));
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(3));
+	}
+
+	@Test
+	public void testReadFieldsNestedTupleWildCard() {
+		String[] readFields = { "*" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, nestedTupleType, intType);
+
+		FieldSet fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 5);
+		Assert.assertTrue(fs.contains(0));
+		Assert.assertTrue(fs.contains(1));
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(3));
+		Assert.assertTrue(fs.contains(4));
+
+		readFields[0] = "f0.*;f1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, nestedTupleType, intType);
+
+		fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 4);
+		Assert.assertTrue(fs.contains(0));
+		Assert.assertTrue(fs.contains(1));
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(3));
+
+	}
+
+	@Test
+	public void testReadFieldsPojo() {
+		String[] readFields = { "int2; string1" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, pojoType, threeIntTupleType);
+
+		FieldSet fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 2);
+		Assert.assertTrue(fs.contains(1));
+		Assert.assertTrue(fs.contains(3));
+
+		readFields[0] = "*";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, pojoType, intType);
+
+		fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 4);
+		Assert.assertTrue(fs.contains(0));
+		Assert.assertTrue(fs.contains(1));
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(3));
+	}
+
+	@Test
+	public void testReadFieldsNestedPojo() {
+		String[] readFields = { "pojo1.int2; string1; pojo1.string1" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, nestedPojoType, intType);
 
-		TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-				BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
+		FieldSet fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 3);
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(4));
+		Assert.assertTrue(fs.contains(5));
+
+		readFields[0] = "pojo1.*";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, nestedPojoType, intType);
+
+		fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 4);
+		Assert.assertTrue(fs.contains(1));
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(3));
+		Assert.assertTrue(fs.contains(4));
+
+		readFields[0] = "pojo1";
+		sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, nestedPojoType, intType);
+
+		fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 4);
+		Assert.assertTrue(fs.contains(1));
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(3));
+		Assert.assertTrue(fs.contains(4));
+	}
+
+	@Test
+	public void testReadFieldsPojoInTuple() {
+		String[] readFields = { "f0; f2.int1; f2.string1" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, pojoInTupleType, pojo2Type);
+
+		FieldSet fs = sp.getReadFields(0);
+		Assert.assertTrue(fs.size() == 3);
+		Assert.assertTrue(fs.contains(0));
+		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(fs.contains(5));
+	}
 
-		try {
-			SemanticPropUtil.getSemanticPropsSingleFromString(null, null, readFields, type, type);
-			Assert.fail("accepted invalid input");
-		} catch (InvalidProgramException e) {
-			// good
-		} catch (Exception e) {
-			Assert.fail("Wrong type of exception thrown");
-		}
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testReadFieldsInvalidString() {
+		String[] readFields = { "notValid" };
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, null, readFields, threeIntTupleType, threeIntTupleType);
 	}
 	
 	// --------------------------------------------------------------------------------------------
@@ -624,53 +1037,315 @@ public class SemanticPropUtilTest {
 	// --------------------------------------------------------------------------------------------
 	
 	@Test
-	public void testSimpleCaseDual() {
-		String[] constantFieldsFirst = { "1->1,2", "2->3" };
-		String[] constantFieldsSecond = { "1->1,2", "2->3" };
+	public void testForwardedDual() {
+		String[] forwardedFieldsFirst = { "f1->f2; f2->f3" };
+		String[] forwardedFieldsSecond = { "f1->f1; f2->f0" };
+		DualInputSemanticProperties dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, forwardedFieldsFirst, forwardedFieldsSecond, null,
+				null, null, null, fourIntTupleType, fourIntTupleType, fourIntTupleType);
+
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 1).contains(2));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 2).contains(3));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 1).contains(1));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 2).contains(0));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 3).size() == 0);
+
+		forwardedFieldsFirst[0] = "f1->f0;f3->f1";
+		forwardedFieldsSecond[0] = "*->f2.*";
+		dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, forwardedFieldsFirst, forwardedFieldsSecond, null,
+				null, null, null, fourIntTupleType, pojoType, pojoInTupleType);
+
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 1).contains(0));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 3).contains(1));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 0).contains(2));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 1).contains(3));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 2).contains(4));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 3).contains(5));
+
+		forwardedFieldsFirst[0] = "f1.f0.f2->int1; f2->pojo1.int3";
+		forwardedFieldsSecond[0] = "string1; int2->pojo1.int1; int1->pojo1.int2";
+		dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, forwardedFieldsFirst, forwardedFieldsSecond, null,
+				null, null, null, deepNestedTupleType, pojoType, nestedPojoType);
 
-		TypeInformation<?> type = new TupleTypeInfo<Tuple4<Integer, Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-				BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-		
-		
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 3).contains(0));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 4).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 5).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 6).contains(3));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 0).contains(2));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 1).contains(1));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 2).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 3).contains(5));
+
+		String[] forwardedFieldsFirst2 = { "f1.f0.f2->int1", "f2->pojo1.int3" };
+		String[] forwardedFieldsSecond2 = { "string1", "int2->pojo1.int1", "int1->pojo1.int2" };
+		dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, forwardedFieldsFirst2, forwardedFieldsSecond2, null,
+				null, null, null, deepNestedTupleType, pojoType, nestedPojoType);
+
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 3).contains(0));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 4).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 5).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 6).contains(3));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 0).contains(2));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 1).contains(1));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 2).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 3).contains(5));
+	}
+
+
+	@Test
+	public void testNonForwardedDual() {
+		String[] nonNorwardedFieldsFirst = { "f1;f2" };
+		String[] nonNorwardedFieldsSecond = { "f0" };
 		DualInputSemanticProperties dsp = new DualInputSemanticProperties();
-		
-		SemanticPropUtil.getSemanticPropsDualFromString(dsp, constantFieldsFirst, constantFieldsSecond, null,
-				null, null, null, type, type, type);
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, null,
+				nonNorwardedFieldsFirst, nonNorwardedFieldsSecond, null, null, threeIntTupleType, threeIntTupleType, threeIntTupleType);
 
-		FieldSet fs = dsp.getForwardedField1(1);
-		Assert.assertTrue(fs.size() == 2);
-		Assert.assertTrue(fs.contains(1));
-		Assert.assertTrue(fs.contains(2));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 0).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 1).contains(1));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 2).contains(2));
 
-		fs = dsp.getForwardedField1(2);
-		Assert.assertTrue(fs.size() == 1);
-		Assert.assertTrue(fs.contains(3));
+		nonNorwardedFieldsFirst[0] = "f1";
+		nonNorwardedFieldsSecond[0] = "";
+		dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, null,
+				nonNorwardedFieldsFirst, null, null, null, threeIntTupleType, fiveIntTupleType, threeIntTupleType);
+
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 0).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 1).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 2).size() == 0);
+
+		nonNorwardedFieldsFirst[0] = "";
+		nonNorwardedFieldsSecond[0] = "f2;f0";
+		dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, null,
+				null, nonNorwardedFieldsSecond, null, null, fiveIntTupleType, threeIntTupleType, threeIntTupleType);
+
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 0).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 1).contains(1));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 2).size() == 0);
+
+		String[] nonForwardedFields = {"f1", "f3"};
+		dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, null,
+				nonForwardedFields, null, null, null, fiveIntTupleType, threeIntTupleType, fiveIntTupleType);
+
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 2).contains(2));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 3).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 4).contains(4));
+
+		dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, null,
+				null, nonForwardedFields, null, null, threeIntTupleType, fiveIntTupleType, fiveIntTupleType);
+
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 0).contains(0));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 1).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 2).contains(2));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 3).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 4).contains(4));
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testNonForwardedDualInvalidTypes1() {
+
+		String[] nonNorwardedFieldsFirst = { "f1" };
+		DualInputSemanticProperties dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, null,
+				nonNorwardedFieldsFirst, null, null, null, fiveIntTupleType, threeIntTupleType, threeIntTupleType);
+	}
+
+	@Test(expected = InvalidSemanticAnnotationException.class)
+	public void testNonForwardedDualInvalidTypes2() {
+
+		String[] nonNorwardedFieldsSecond = { "f1" };
+		DualInputSemanticProperties dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, null,
+				null, nonNorwardedFieldsSecond, null, null, threeIntTupleType, pojoInTupleType, threeIntTupleType);
+	}
+
+	@Test
+	public void testReadFieldsDual() {
+		String[] readFieldsFirst = {"f1;f2"};
+		String[] readFieldsSecond = {"f0"};
+		DualInputSemanticProperties dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, null,
+				null, null, readFieldsFirst, readFieldsSecond, threeIntTupleType, threeIntTupleType, threeIntTupleType);
+
+		Assert.assertTrue(dsp.getReadFields(0).size() == 2);
+		Assert.assertTrue(dsp.getReadFields(0).contains(1));
+		Assert.assertTrue(dsp.getReadFields(0).contains(2));
+		Assert.assertTrue(dsp.getReadFields(1).size() == 1);
+		Assert.assertTrue(dsp.getReadFields(1).contains(0));
+
+		readFieldsFirst[0] = "f0.*; f2";
+		readFieldsSecond[0] = "int1; string1";
+		dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, null,
+				null, null, readFieldsFirst, readFieldsSecond, nestedTupleType, pojoType, threeIntTupleType);
+
+		Assert.assertTrue(dsp.getReadFields(0).size() == 4);
+		Assert.assertTrue(dsp.getReadFields(0).contains(0));
+		Assert.assertTrue(dsp.getReadFields(0).contains(1));
+		Assert.assertTrue(dsp.getReadFields(0).contains(2));
+		Assert.assertTrue(dsp.getReadFields(0).contains(4));
+		Assert.assertTrue(dsp.getReadFields(1).size() == 2);
+		Assert.assertTrue(dsp.getReadFields(1).contains(0));
+		Assert.assertTrue(dsp.getReadFields(1).contains(3));
+
+		readFieldsFirst[0] = "pojo1.int2; string1";
+		readFieldsSecond[0] = "f2.int2";
+		dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, null,
+				null, null, readFieldsFirst, readFieldsSecond, nestedPojoType, pojoInTupleType, threeIntTupleType);
+
+		Assert.assertTrue(dsp.getReadFields(0).size() == 2);
+		Assert.assertTrue(dsp.getReadFields(0).contains(2));
+		Assert.assertTrue(dsp.getReadFields(0).contains(5));
+		Assert.assertTrue(dsp.getReadFields(1).size() == 1);
+		Assert.assertTrue(dsp.getReadFields(1).contains(3));
+
+		String[] readFields = {"f0", "f2", "f4"};
+		dsp = new DualInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, null,
+				null, null, readFields, readFields, fiveIntTupleType, fiveIntTupleType, threeIntTupleType);
+
+		Assert.assertTrue(dsp.getReadFields(0).size() == 3);
+		Assert.assertTrue(dsp.getReadFields(0).contains(0));
+		Assert.assertTrue(dsp.getReadFields(0).contains(2));
+		Assert.assertTrue(dsp.getReadFields(0).contains(4));
+		Assert.assertTrue(dsp.getReadFields(1).size() == 3);
+		Assert.assertTrue(dsp.getReadFields(1).contains(0));
+		Assert.assertTrue(dsp.getReadFields(1).contains(2));
+		Assert.assertTrue(dsp.getReadFields(1).contains(4));
+	}
+
+	// --------------------------------------------------------------------------------------------
+	// Mixed Annotations
+	// --------------------------------------------------------------------------------------------
+
+	@Test
+	public void testForwardedRead() {
+		String[] forwardedFields = { "f0->f0;f1->f2" };
+		String[] readFields = {"f0; f2"};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, forwardedFields, null, readFields, threeIntTupleType, fiveIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).contains(2));
+		Assert.assertTrue(sp.getReadFields(0).size() == 2);
+		Assert.assertTrue(sp.getReadFields(0).contains(0));
+		Assert.assertTrue(sp.getReadFields(0).contains(2));
 	}
 
 	@Test
-	public void testFieldsExceptDual() {
-		String[] constantFieldsFirstExcept = { "1,2" };
-		String[] constantFieldsSecond = { "0->1" };
+	public void testNonForwardedRead() {
+		String[] nonForwardedFields = { "f1;f2" };
+		String[] readFields = {"f0; f2"};
+		SingleInputSemanticProperties sp = new SingleInputSemanticProperties();
+		SemanticPropUtil.getSemanticPropsSingleFromString(sp, null, nonForwardedFields, readFields, threeIntTupleType, threeIntTupleType);
+
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 0).contains(0));
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 1).size() == 0);
+		Assert.assertTrue(sp.getForwardingTargetFields(0, 2).size() == 0);
+		Assert.assertTrue(sp.getReadFields(0).size() == 2);
+		Assert.assertTrue(sp.getReadFields(0).contains(0));
+		Assert.assertTrue(sp.getReadFields(0).contains(2));
+	}
 
+	@Test
+	public void testForwardedReadDual() {
+		String[] forwardedFieldsFirst = { "f1->f2; f2->f3" };
+		String[] forwardedFieldsSecond = { "f1->f1; f2->f0" };
+		String[] readFieldsFirst = {"0;2"};
+		String[] readFieldsSecond = {"1"};
 		DualInputSemanticProperties dsp = new DualInputSemanticProperties();
-		
-		TypeInformation<?> type = new TupleTypeInfo<Tuple3<Integer, Integer, Integer>>(BasicTypeInfo.INT_TYPE_INFO,
-				BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
-		SemanticPropUtil.getSemanticPropsDualFromString(dsp, null, constantFieldsSecond,
-				constantFieldsFirstExcept, null, null, null, type, type, type);
+		SemanticPropUtil.getSemanticPropsDualFromString(dsp, forwardedFieldsFirst, forwardedFieldsSecond, null,
+				null, readFieldsFirst, readFieldsSecond, fourIntTupleType, fourIntTupleType, fourIntTupleType);
 
-		FieldSet fs = dsp.getForwardedField1(0);
-		Assert.assertTrue(fs.size() == 1);
-		Assert.assertTrue(fs.contains(0));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 1).contains(2));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 2).contains(3));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 1).contains(1));
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 2).contains(0));
+		Assert.assertTrue(dsp.getForwardingTargetFields(0, 0).size() == 0);
+		Assert.assertTrue(dsp.getForwardingTargetFields(1, 3).size() == 0);
+		Assert.assertTrue(dsp.getReadFields(0).size() == 2);
+		Assert.assertTrue(dsp.getReadFields(0).contains(0));
+		Assert.assertTrue(dsp.getReadFields(0).contains(2));
+		Assert.assertTrue(dsp.getReadFields(1).size() == 1);
+		Assert.assertTr

<TRUNCATED>

Mime
View raw message