flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mbala...@apache.org
Subject [2/4] flink git commit: [FLINK-3702] Make FieldAccessors support nested field expressions.
Date Thu, 24 Nov 2016 21:23:15 GMT
[FLINK-3702] Make FieldAccessors support nested field expressions.


Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/1f04542e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/1f04542e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/1f04542e

Branch: refs/heads/master
Commit: 1f04542e861f9c156d7b5c1f6db72a74e08d7a75
Parents: 5d2da12
Author: Gabor Gevay <ggab90@gmail.com>
Authored: Sun May 22 19:48:50 2016 +0200
Committer: Marton Balassi <mbalassi@apache.org>
Committed: Thu Nov 24 22:22:42 2016 +0100

----------------------------------------------------------------------
 docs/dev/api_concepts.md                        |   4 +-
 .../api/common/typeinfo/BasicArrayTypeInfo.java |  18 +
 .../api/common/typeinfo/BasicTypeInfo.java      |  26 ++
 .../InvalidFieldReferenceException.java         |  31 ++
 .../common/typeinfo/PrimitiveArrayTypeInfo.java |  18 +
 .../api/common/typeinfo/TypeInformation.java    |  34 ++
 .../api/common/typeutils/CompositeType.java     |  10 -
 .../api/common/typeutils/TypeSerializer.java    |   2 +-
 .../flink/api/java/typeutils/FieldAccessor.java | 324 ++++++++++++++++++
 .../flink/api/java/typeutils/PojoField.java     |  22 +-
 .../flink/api/java/typeutils/PojoTypeInfo.java  |  42 ++-
 .../api/java/typeutils/TupleTypeInfoBase.java   |  32 +-
 .../java/typeutils/runtime/FieldSerializer.java |  54 +++
 .../java/typeutils/runtime/PojoComparator.java  |  21 +-
 .../java/typeutils/runtime/PojoSerializer.java  |  21 +-
 .../api/java/typeutils/FieldAccessorTest.java   | 343 +++++++++++++++++++
 .../api/java/functions/SemanticPropUtil.java    |   2 +-
 .../flink/api/java/operator/DataSinkTest.java   |   5 +-
 .../operator/FullOuterJoinOperatorTest.java     |   3 +-
 .../operator/LeftOuterJoinOperatorTest.java     |   3 +-
 .../operator/RightOuterJoinOperatorTest.java    |   3 +-
 .../scala/typeutils/ProductFieldAccessor.java   |  75 ++++
 .../api/scala/typeutils/CaseClassTypeInfo.scala |  38 +-
 .../scala/typeutils/CaseClassTypeInfoTest.scala | 110 ++++++
 .../streaming/api/datastream/KeyedStream.java   | 140 +++++---
 .../aggregation/ComparableAggregator.java       |   6 +-
 .../functions/aggregation/SumAggregator.java    |   6 +-
 .../flink/streaming/util/FieldAccessor.java     | 254 --------------
 .../flink/streaming/util/FieldAccessorTest.java |  75 ----
 .../flink/streaming/api/scala/KeyedStream.scala |  90 ++++-
 .../streaming/runtime/DataStreamPojoITCase.java |  42 ++-
 31 files changed, 1366 insertions(+), 488 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/docs/dev/api_concepts.md
----------------------------------------------------------------------
diff --git a/docs/dev/api_concepts.md b/docs/dev/api_concepts.md
index 49d2ded..07a81e7 100644
--- a/docs/dev/api_concepts.md
+++ b/docs/dev/api_concepts.md
@@ -385,7 +385,7 @@ while a key can be specified on a DataStream using
 {% highlight java %}
 DataStream<...> input = // [...]
 DataStream<...> windowed = input
-  .key(/*define key here*/)
+  .keyBy(/*define key here*/)
   .window(/*window specification*/);
 {% endhighlight %}
 
@@ -418,7 +418,7 @@ val keyed = input.keyBy(0)
 </div>
 </div>
 
-The tuples is grouped on the first field (the one of
+The tuples are grouped on the first field (the one of
 Integer type).
 
 <div class="codetabs" markdown="1">

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java
index 25b2850..d04e7d9 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicArrayTypeInfo.java
@@ -29,6 +29,7 @@ import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.common.typeutils.base.array.StringArraySerializer;
 import org.apache.flink.api.common.functions.InvalidTypesException;
 import org.apache.flink.api.common.typeutils.base.GenericArraySerializer;
+import org.apache.flink.api.java.typeutils.FieldAccessor;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -121,6 +122,23 @@ public final class BasicArrayTypeInfo<T, C> extends TypeInformation<T> {
 	}
 
 	@Override
+	@PublicEvolving
+	public <F> FieldAccessor<T, F> getFieldAccessor(int pos, ExecutionConfig config) {
+		return new FieldAccessor.ArrayFieldAccessor<>(pos, this);
+	}
+
+	@Override
+	@PublicEvolving
+	public <F> FieldAccessor<T, F> getFieldAccessor(String pos, ExecutionConfig config) {
+		try {
+			return new FieldAccessor.ArrayFieldAccessor<>(Integer.parseInt(pos), this);
+		} catch (NumberFormatException ex) {
+			throw new InvalidFieldReferenceException
+				("A field expression on an array must be an integer index (that might be given as a string).");
+		}
+	}
+
+	@Override
 	public boolean equals(Object obj) {
 		if (obj instanceof BasicArrayTypeInfo) {
 			BasicArrayTypeInfo<?, ?> other = (BasicArrayTypeInfo<?, ?>) obj;

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
index e2fd74e..09efba6 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/BasicTypeInfo.java
@@ -58,6 +58,7 @@ import org.apache.flink.api.common.typeutils.base.ShortSerializer;
 import org.apache.flink.api.common.typeutils.base.StringComparator;
 import org.apache.flink.api.common.typeutils.base.StringSerializer;
 import org.apache.flink.api.common.typeutils.base.VoidSerializer;
+import org.apache.flink.api.java.typeutils.FieldAccessor;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -171,6 +172,31 @@ public class BasicTypeInfo<T> extends TypeInformation<T> implements AtomicType<T
 		}
 	}
 
+	@Override
+	@PublicEvolving
+	@SuppressWarnings("unchecked")
+	public <F> FieldAccessor<T, F> getFieldAccessor(int pos, ExecutionConfig config) {
+		if(pos != 0) {
+			throw new InvalidFieldReferenceException("The " + ((Integer) pos).toString() + ". field selected on a " +
+				"basic type (" + this.toString() + "). A field expression on a basic type can only select " +
+				"the 0th field (which means selecting the entire basic type).");
+		}
+		return (FieldAccessor<T, F>) new FieldAccessor.SimpleFieldAccessor<T>(this);
+	}
+
+	@Override
+	@PublicEvolving
+	public <F> FieldAccessor<T, F> getFieldAccessor(String field, ExecutionConfig config) {
+		try {
+			int pos = field.equals("*") ? 0 : Integer.parseInt(field);
+			return getFieldAccessor(pos, config);
+		} catch (NumberFormatException ex) {
+			throw new InvalidFieldReferenceException("You tried to select the field \"" + field +
+				"\" on a " + this.toString() + ". A field expression on a basic type can only be \"*\" or \"0\"" +
+				" (both of which mean selecting the entire basic type).");
+		}
+	}
+
 	// --------------------------------------------------------------------------------------------
 	
 	@Override

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/InvalidFieldReferenceException.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/InvalidFieldReferenceException.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/InvalidFieldReferenceException.java
new file mode 100644
index 0000000..3c67c46
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/InvalidFieldReferenceException.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.common.typeinfo;
+
+import org.apache.flink.annotation.PublicEvolving;
+
+@PublicEvolving
+public class InvalidFieldReferenceException extends IllegalArgumentException {
+
+	private static final long serialVersionUID = 1L;
+
+	public InvalidFieldReferenceException(String s) {
+		super(s);
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java
index 1c6ce00..2bd96d3 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/PrimitiveArrayTypeInfo.java
@@ -40,6 +40,7 @@ import org.apache.flink.api.common.typeutils.base.array.LongPrimitiveArraySerial
 import org.apache.flink.api.common.typeutils.base.array.PrimitiveArrayComparator;
 import org.apache.flink.api.common.typeutils.base.array.ShortPrimitiveArrayComparator;
 import org.apache.flink.api.common.typeutils.base.array.ShortPrimitiveArraySerializer;
+import org.apache.flink.api.java.typeutils.FieldAccessor;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -138,6 +139,23 @@ public class PrimitiveArrayTypeInfo<T> extends TypeInformation<T> implements Ato
 		return this.serializer;
 	}
 
+	@Override
+	@PublicEvolving
+	public <F> FieldAccessor<T, F> getFieldAccessor(int pos, ExecutionConfig config) {
+		return new FieldAccessor.ArrayFieldAccessor<>(pos, this);
+	}
+
+	@Override
+	@PublicEvolving
+	public <F> FieldAccessor<T, F> getFieldAccessor(String pos, ExecutionConfig config) {
+		try {
+			return new FieldAccessor.ArrayFieldAccessor<>(Integer.parseInt(pos), this);
+		} catch (NumberFormatException ex) {
+			throw new InvalidFieldReferenceException
+				("A field expression on an array must be an integer index (that might be given as a string).");
+		}
+	}
+
 	/**
 	 * Gets the class that represents the component type.
 	 * @return The class of the component type.

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
index 154ceb1..7be2b68 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeinfo/TypeInformation.java
@@ -24,6 +24,7 @@ import org.apache.flink.annotation.Public;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeutils.TypeSerializer;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.FieldAccessor;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
 
 import java.io.Serializable;
@@ -172,6 +173,39 @@ public abstract class TypeInformation<T> implements Serializable {
 	@PublicEvolving
 	public abstract TypeSerializer<T> createSerializer(ExecutionConfig config);
 
+
+	/**
+	 * Creates a {@link FieldAccessor} for the given field position, which can be used to get and set
+	 * the specified field on instances of this type.
+	 *
+	 * @param pos The field position (zero-based)
+	 * @param config Configuration object
+	 * @param <F> The type of the field to access
+	 * @return The created FieldAccessor
+	 */
+	@PublicEvolving
+	public <F> FieldAccessor<T, F> getFieldAccessor(int pos, ExecutionConfig config){
+		throw new InvalidFieldReferenceException("Cannot reference field by position on " + this.toString()
+			+ "Referencing a field by position is supported on tuples, case classes, and arrays. "
+			+ "Additionally, you can select the 0th field of a primitive/basic type (e.g. int).");
+	}
+
+	/**
+	 * Creates a {@link FieldAccessor} for the field that is given by a field expression,
+	 * which can be used to get and set the specified field on instances of this type.
+	 *
+	 * @param field The field expression
+	 * @param config Configuration object
+	 * @param <F> The type of the field to access
+	 * @return The created FieldAccessor
+	 */
+	@PublicEvolving
+	public <F> FieldAccessor<T, F> getFieldAccessor(String field, ExecutionConfig config) {
+		throw new InvalidFieldReferenceException("Cannot reference field by field expression on " + this.toString()
+				+ "Field expressions are only supported on POJO types, tuples, and case classes. "
+				+ "(See the Flink documentation on what is considered a POJO.)");
+	}
+
 	@Override
 	public abstract String toString();
 

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
index 4bf17ea..a4230f4 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/CompositeType.java
@@ -265,16 +265,6 @@ public abstract class CompositeType<T> extends TypeInformation<T> {
 	@PublicEvolving
 	public abstract int getFieldIndex(String fieldName);
 
-	@PublicEvolving
-	public static class InvalidFieldReferenceException extends IllegalArgumentException {
-
-		private static final long serialVersionUID = 1L;
-
-		public InvalidFieldReferenceException(String s) {
-			super(s);
-		}
-	}
-
 	@Override
 	public boolean equals(Object obj) {
 		if (obj instanceof CompositeType) {

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
index 0d56743..5e81db7 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/TypeSerializer.java
@@ -26,7 +26,7 @@ import org.apache.flink.core.memory.DataInputView;
 import org.apache.flink.core.memory.DataOutputView;
 
 /**
- * This interface describes the methods that are required for a data type to be handled by the pact
+ * This interface describes the methods that are required for a data type to be handled by the Flink
  * runtime. Specifically, this interface contains the serialization and copying methods.
  * <p>
  * The methods in this class are assumed to be stateless, such that it is effectively thread safe. Stateful

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/FieldAccessor.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/FieldAccessor.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/FieldAccessor.java
new file mode 100644
index 0000000..97ef31a
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/FieldAccessor.java
@@ -0,0 +1,324 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.typeutils;
+
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.operators.Keys;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.InvalidFieldReferenceException;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.runtime.FieldSerializer;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.lang.reflect.Array;
+import java.lang.reflect.Field;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+
+/**
+ * These classes encapsulate the logic of accessing a field specified by the user as either an index
+ * or a field expression string. TypeInformation can also be requested for the field.
+ * The position index might specify a field of a Tuple, an array, or a simple type (only "0th field").
+ *
+ * Field expressions that specify nested fields (e.g. "f1.a.foo") result in nested field accessors.
+ * These penetrate one layer, and then delegate the rest of the work to an "innerAccesor".
+ * (see PojoFieldAccessor, RecursiveTupleFieldAccessor, ProductFieldAccessor)
+ */
+@PublicEvolving
+public abstract class FieldAccessor<T, F> implements Serializable {
+
+	private static final long serialVersionUID = 1L;
+
+	protected TypeInformation fieldType;
+
+	/**
+	 * Gets the TypeInformation for the type of the field.
+	 * Note: For an array of a primitive type, it returns the corresponding basic type (Integer for int[]).
+	 */
+	@SuppressWarnings("unchecked")
+	public TypeInformation<F> getFieldType() {
+		return fieldType;
+	}
+
+
+	/**
+	 * Gets the value of the field (specified in the constructor) of the given record.
+	 * @param record The record on which the field will be accessed
+	 * @return The value of the field.
+	 */
+	public abstract F get(T record);
+
+	/**
+	 * Sets the field (specified in the constructor) of the given record to the given value.
+	 *
+	 * Warning: This might modify the original object, or might return a new object instance.
+	 * (This is necessary, because the record might be immutable.)
+	 *
+	 * @param record The record to modify
+	 * @param fieldValue The new value of the field
+	 * @return A record that has the given field value. (this might be a new instance or the original)
+	 */
+	public abstract T set(T record, F fieldValue);
+
+
+	// --------------------------------------------------------------------------------------------------
+
+
+	/**
+	 * This is when the entire record is considered as a single field. (eg. field 0 of a basic type, or a
+	 * field of a POJO that is itself some composite type but is not further decomposed)
+	 */
+	public final static class SimpleFieldAccessor<T> extends FieldAccessor<T, T> {
+
+		private static final long serialVersionUID = 1L;
+
+		public SimpleFieldAccessor(TypeInformation<T> typeInfo) {
+			checkNotNull(typeInfo, "typeInfo must not be null.");
+
+			this.fieldType = typeInfo;
+		}
+
+		@Override
+		public T get(T record) {
+			return record;
+		}
+
+		@Override
+		public T set(T record, T fieldValue) {
+			return fieldValue;
+		}
+	}
+
+	public final static class ArrayFieldAccessor<T, F> extends FieldAccessor<T, F> {
+
+		private static final long serialVersionUID = 1L;
+
+		private final int pos;
+
+		public ArrayFieldAccessor(int pos, TypeInformation typeInfo) {
+			if(pos < 0) {
+				throw new InvalidFieldReferenceException("The " + ((Integer) pos).toString() + ". field selected on" +
+					" an array, which is an invalid index.");
+			}
+			checkNotNull(typeInfo, "typeInfo must not be null.");
+
+			this.pos = pos;
+			this.fieldType = BasicTypeInfo.getInfoFor(typeInfo.getTypeClass().getComponentType());
+		}
+
+		@SuppressWarnings("unchecked")
+		@Override
+		public F get(T record) {
+			return (F) Array.get(record, pos);
+		}
+
+		@Override
+		public T set(T record, F fieldValue) {
+			Array.set(record, pos, fieldValue);
+			return record;
+		}
+	}
+
+	/**
+	 * There are two versions of TupleFieldAccessor, differing in whether there is an other
+	 * FieldAccessor nested inside. The no inner accessor version is probably a little faster.
+	 */
+	static final class SimpleTupleFieldAccessor<T, F> extends FieldAccessor<T, F> {
+
+		private static final long serialVersionUID = 1L;
+
+		private final int pos;
+
+		SimpleTupleFieldAccessor(int pos, TypeInformation<T> typeInfo) {
+			int arity = ((TupleTypeInfo)typeInfo).getArity();
+			if(pos < 0 || pos >= arity) {
+				throw new InvalidFieldReferenceException(
+					"Tried to select " + ((Integer) pos).toString() + ". field on \"" +
+					typeInfo.toString() + "\", which is an invalid index.");
+			}
+			checkNotNull(typeInfo, "typeInfo must not be null.");
+
+			this.pos = pos;
+			this.fieldType = ((TupleTypeInfo)typeInfo).getTypeAt(pos);
+		}
+
+		@SuppressWarnings("unchecked")
+		@Override
+		public F get(T record) {
+			final Tuple tuple = (Tuple) record;
+			return (F) tuple.getField(pos);
+		}
+
+		@Override
+		public T set(T record, F fieldValue) {
+			final Tuple tuple = (Tuple) record;
+			tuple.setField(fieldValue, pos);
+			return record;
+		}
+	}
+
+	/**
+	 * @param <T> The Tuple type
+	 * @param <R> The field type at the first level
+	 * @param <F> The field type at the innermost level
+	 */
+	static final class RecursiveTupleFieldAccessor<T, R, F> extends FieldAccessor<T, F> {
+
+		private static final long serialVersionUID = 1L;
+
+		private final int pos;
+		private final FieldAccessor<R, F> innerAccessor;
+
+		RecursiveTupleFieldAccessor(int pos, FieldAccessor<R, F> innerAccessor) {
+			if(pos < 0) {
+				throw new InvalidFieldReferenceException("Tried to select " + ((Integer) pos).toString() + ". field.");
+			}
+			checkNotNull(innerAccessor, "innerAccessor must not be null.");
+
+			this.pos = pos;
+			this.innerAccessor = innerAccessor;
+			this.fieldType = innerAccessor.fieldType;
+		}
+
+		@Override
+		public F get(T record) {
+			final Tuple tuple = (Tuple) record;
+			final R inner = tuple.getField(pos);
+			return innerAccessor.get(inner);
+		}
+
+		@Override
+		public T set(T record, F fieldValue) {
+			final Tuple tuple = (Tuple) record;
+			final R inner = tuple.getField(pos);
+			tuple.setField(innerAccessor.set(inner, fieldValue), pos);
+			return record;
+		}
+	}
+
+	/**
+	 * @param <T> The POJO type
+	 * @param <R> The field type at the first level
+	 * @param <F> The field type at the innermost level
+	 */
+	static final class PojoFieldAccessor<T, R, F> extends FieldAccessor<T, F> {
+
+		private static final long serialVersionUID = 1L;
+
+		private transient Field field;
+		private final FieldAccessor<R, F> innerAccessor;
+
+		PojoFieldAccessor(Field field, FieldAccessor<R, F> innerAccessor) {
+			checkNotNull(field, "field must not be null.");
+			checkNotNull(innerAccessor, "innerAccessor must not be null.");
+
+			this.field = field;
+			this.innerAccessor = innerAccessor;
+			this.fieldType = innerAccessor.fieldType;
+		}
+
+		@Override
+		public F get(T pojo) {
+			try {
+				@SuppressWarnings("unchecked")
+				final R inner = (R)field.get(pojo);
+				return innerAccessor.get(inner);
+			} catch (IllegalAccessException iaex) {
+				throw new RuntimeException("This should not happen since we call setAccesssible(true) in readObject."
+						+ " fields: " + field + " obj: " + pojo);
+			}
+		}
+
+		@Override
+		public T set(T pojo, F valueToSet) {
+			try {
+				@SuppressWarnings("unchecked")
+				final R inner = (R)field.get(pojo);
+				field.set(pojo, innerAccessor.set(inner, valueToSet));
+				return pojo;
+			} catch (IllegalAccessException iaex) {
+				throw new RuntimeException("This should not happen since we call setAccesssible(true) in readObject."
+						+ " fields: " + field + " obj: " + pojo);
+			}
+		}
+
+		private void writeObject(ObjectOutputStream out)
+				throws IOException, ClassNotFoundException {
+			out.defaultWriteObject();
+			FieldSerializer.serializeField(field, out);
+		}
+
+		private void readObject(ObjectInputStream in)
+				throws IOException, ClassNotFoundException {
+			in.defaultReadObject();
+			field = FieldSerializer.deserializeField(in);
+		}
+	}
+
+
+	// --------------------------------------------------------------------------------------------------
+
+	private final static String REGEX_FIELD = "[\\p{L}\\p{Digit}_\\$]*"; // This can start with a digit (because of Tuples)
+	private final static String REGEX_NESTED_FIELDS = "("+REGEX_FIELD+")(\\.(.+))?";
+	private final static String REGEX_NESTED_FIELDS_WILDCARD = REGEX_NESTED_FIELDS
+		+"|\\"+ Keys.ExpressionKeys.SELECT_ALL_CHAR
+		+"|\\"+ Keys.ExpressionKeys.SELECT_ALL_CHAR_SCALA;
+
+	private static final Pattern PATTERN_NESTED_FIELDS_WILDCARD = Pattern.compile(REGEX_NESTED_FIELDS_WILDCARD);
+
+	public static FieldExpression decomposeFieldExpression(String fieldExpression) {
+		Matcher matcher = PATTERN_NESTED_FIELDS_WILDCARD.matcher(fieldExpression);
+		if(!matcher.matches()) {
+			throw new InvalidFieldReferenceException("Invalid field expression \""+fieldExpression+"\".");
+		}
+
+		String head = matcher.group(0);
+		if(head.equals(Keys.ExpressionKeys.SELECT_ALL_CHAR) || head.equals(Keys.ExpressionKeys.SELECT_ALL_CHAR_SCALA)) {
+			throw new InvalidFieldReferenceException("No wildcards are allowed here.");
+		} else {
+			head = matcher.group(1);
+		}
+
+		String tail = matcher.group(3);
+
+		return new FieldExpression(head, tail);
+	}
+
+	/**
+	 * Represents a decomposition of a field expression into its first part, and the rest.
+	 * E.g. "foo.f1.bar" is decomposed into "foo" and "f1.bar".
+	 */
+	public static class FieldExpression implements Serializable {
+
+		private static final long serialVersionUID = 1L;
+
+		public String head, tail; // tail can be null, if the field expression had just one part
+
+		FieldExpression(String head, String tail) {
+			this.head = head;
+			this.tail = tail;
+		}
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java
index 026cfa6..2e20415 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoField.java
@@ -27,6 +27,7 @@ import java.util.Objects;
 
 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.runtime.FieldSerializer;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -57,30 +58,13 @@ public class PojoField implements Serializable {
 	private void writeObject(ObjectOutputStream out)
 			throws IOException, ClassNotFoundException {
 		out.defaultWriteObject();
-		out.writeObject(field.getDeclaringClass());
-		out.writeUTF(field.getName());
+		FieldSerializer.serializeField(field, out);
 	}
 
 	private void readObject(ObjectInputStream in)
 			throws IOException, ClassNotFoundException {
 		in.defaultReadObject();
-		Class<?> clazz = (Class<?>)in.readObject();
-		String fieldName = in.readUTF();
-		field = null;
-		// try superclasses as well
-		while (clazz != null) {
-			try {
-				field = clazz.getDeclaredField(fieldName);
-				field.setAccessible(true);
-				break;
-			} catch (NoSuchFieldException e) {
-				clazz = clazz.getSuperclass();
-			}
-		}
-		if (field == null) {
-			throw new RuntimeException("Class resolved at TaskManager is not compatible with class read during Plan setup."
-					+ " (" + fieldName + ")");
-		}
+		field = FieldSerializer.deserializeField(in);
 	}
 
 	@Override

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
index 9c65263..72432d6 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
@@ -23,6 +23,7 @@ import org.apache.flink.annotation.Public;
 import org.apache.flink.annotation.PublicEvolving;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.operators.Keys.ExpressionKeys;
+import org.apache.flink.api.common.typeinfo.InvalidFieldReferenceException;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompositeType;
 import org.apache.flink.api.common.typeutils.TypeComparator;
@@ -64,8 +65,8 @@ public class PojoTypeInfo<T> extends CompositeType<T> {
 	private final static String REGEX_FIELD = "[\\p{L}_\\$][\\p{L}\\p{Digit}_\\$]*";
 	private final static String REGEX_NESTED_FIELDS = "("+REGEX_FIELD+")(\\.(.+))?";
 	private final static String REGEX_NESTED_FIELDS_WILDCARD = REGEX_NESTED_FIELDS
-			+"|\\"+ExpressionKeys.SELECT_ALL_CHAR
-			+"|\\"+ExpressionKeys.SELECT_ALL_CHAR_SCALA;
+					+"|\\"+ExpressionKeys.SELECT_ALL_CHAR
+					+"|\\"+ExpressionKeys.SELECT_ALL_CHAR_SCALA;
 
 	private static final Pattern PATTERN_NESTED_FIELDS = Pattern.compile(REGEX_NESTED_FIELDS);
 	private static final Pattern PATTERN_NESTED_FIELDS_WILDCARD = Pattern.compile(REGEX_NESTED_FIELDS_WILDCARD);
@@ -132,7 +133,7 @@ public class PojoTypeInfo<T> extends CompositeType<T> {
 		//   gives only some undefined order.
 		return false;
 	}
-	
+
 
 	@Override
 	@PublicEvolving
@@ -264,6 +265,7 @@ public class PojoTypeInfo<T> extends CompositeType<T> {
 	}
 
 	@Override
+	@PublicEvolving
 	protected TypeComparatorBuilder<T> createTypeComparatorBuilder() {
 		return new PojoTypeComparatorBuilder();
 	}
@@ -317,7 +319,39 @@ public class PojoTypeInfo<T> extends CompositeType<T> {
 
 		return new PojoSerializer<T>(getTypeClass(), fieldSerializers, reflectiveFields, config);
 	}
-	
+
+	@Override
+	@PublicEvolving
+	public <F> FieldAccessor<T, F> getFieldAccessor(String fieldExpression, ExecutionConfig config) {
+		
+		FieldAccessor.FieldExpression decomp = FieldAccessor.decomposeFieldExpression(fieldExpression);
+
+		// get field
+		PojoField field = null;
+		TypeInformation<?> fieldType = null;
+		for (int i = 0; i < fields.length; i++) {
+			if (fields[i].getField().getName().equals(decomp.head)) {
+				field = fields[i];
+				fieldType = fields[i].getTypeInformation();
+				break;
+			}
+		}
+		if (field == null) {
+			throw new InvalidFieldReferenceException("Unable to find field \""+decomp.head+"\" in type "+this+".");
+		}
+
+		if(decomp.tail == null) {
+			@SuppressWarnings("unchecked")
+			FieldAccessor<F,F> innerAccessor = new FieldAccessor.SimpleFieldAccessor<F>((TypeInformation<F>) fieldType);
+			return new FieldAccessor.PojoFieldAccessor<T, F, F>(field.getField(), innerAccessor);
+		} else {
+			@SuppressWarnings("unchecked")
+			FieldAccessor<Object,F> innerAccessor =
+					(FieldAccessor<Object,F>)fieldType.<F>getFieldAccessor(decomp.tail, config);
+			return new FieldAccessor.PojoFieldAccessor<T, Object, F>(field.getField(), innerAccessor);
+		}
+	}
+
 	@Override
 	public boolean equals(Object obj) {
 		if (obj instanceof PojoTypeInfo) {

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
index 807fd54..c9a55fc 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
@@ -23,7 +23,10 @@ import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
+import org.apache.flink.annotation.PublicEvolving;
+import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.operators.Keys.ExpressionKeys;
+import org.apache.flink.api.common.typeinfo.InvalidFieldReferenceException;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompositeType;
 
@@ -203,7 +206,34 @@ public abstract class TupleTypeInfoBase<T> extends CompositeType<T> {
 		TypeInformation<X> typed = (TypeInformation<X>) this.types[pos];
 		return typed;
 	}
-	
+
+	@Override
+	@PublicEvolving
+	public <F> FieldAccessor<T, F> getFieldAccessor(int pos, ExecutionConfig config) {
+		return new FieldAccessor.SimpleTupleFieldAccessor<T, F>(pos, this);
+	}
+
+	@Override
+	@PublicEvolving
+	public <F> FieldAccessor<T, F> getFieldAccessor(String fieldExpression, ExecutionConfig config) {
+		FieldAccessor.FieldExpression decomp = FieldAccessor.decomposeFieldExpression(fieldExpression);
+		int fieldPos = this.getFieldIndex(decomp.head);
+		if (fieldPos == -1) {
+			try {
+				fieldPos = Integer.parseInt(decomp.head);
+			} catch (NumberFormatException ex) {
+				throw new InvalidFieldReferenceException("Tried to select field \"" + decomp.head
+					+ "\" on " + this.toString());
+			}
+		}
+		if (decomp.tail == null) {
+			return new FieldAccessor.SimpleTupleFieldAccessor<T, F>(fieldPos, this);
+		} else {
+			FieldAccessor<?, F> innerAccessor = getTypeAt(fieldPos).getFieldAccessor(decomp.tail, config);
+			return new FieldAccessor.RecursiveTupleFieldAccessor<>(fieldPos, innerAccessor);
+		}
+	}
+
 	@Override
 	public boolean equals(Object obj) {
 		if (obj instanceof TupleTypeInfoBase) {

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/FieldSerializer.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/FieldSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/FieldSerializer.java
new file mode 100644
index 0000000..057eee9
--- /dev/null
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/FieldSerializer.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.typeutils.runtime;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.lang.reflect.Field;
+
+/**
+ * This class is for the serialization of java.lang.reflect.Field, which doesn't implement Serializable, therefore
+ * readObject/writeObject need to be implemented in classes where there is a field of type java.lang.reflect.Field.
+ * The two static methods in this class are to be called from these readObject/writeObject methods.
+ */
+public class FieldSerializer {
+
+	public static void serializeField(Field field, ObjectOutputStream out) throws IOException {
+		out.writeObject(field.getDeclaringClass());
+		out.writeUTF(field.getName());
+	}
+
+	public static Field deserializeField(ObjectInputStream in) throws IOException, ClassNotFoundException  {
+		Class<?> clazz = (Class<?>) in.readObject();
+		String fieldName = in.readUTF();
+		// try superclasses as well
+		while (clazz != null) {
+			try {
+				Field field = clazz.getDeclaredField(fieldName);
+				field.setAccessible(true);
+				return field;
+			} catch (NoSuchFieldException e) {
+				clazz = clazz.getSuperclass();
+			}
+		}
+		throw new RuntimeException("Class resolved at TaskManager is not compatible with class read during Plan setup."
+				+ " (" + fieldName + ")");
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoComparator.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoComparator.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoComparator.java
index c0c7797..fc4a305 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoComparator.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoComparator.java
@@ -142,8 +142,7 @@ public final class PojoComparator<T> extends CompositeTypeComparator<T> implemen
 		out.defaultWriteObject();
 		out.writeInt(keyFields.length);
 		for (Field field: keyFields) {
-			out.writeObject(field.getDeclaringClass());
-			out.writeUTF(field.getName());
+			FieldSerializer.serializeField(field, out);
 		}
 	}
 
@@ -153,23 +152,7 @@ public final class PojoComparator<T> extends CompositeTypeComparator<T> implemen
 		int numKeyFields = in.readInt();
 		keyFields = new Field[numKeyFields];
 		for (int i = 0; i < numKeyFields; i++) {
-			Class<?> clazz = (Class<?>) in.readObject();
-			String fieldName = in.readUTF();
-			// try superclasses as well
-			while (clazz != null) {
-				try {
-					Field field = clazz.getDeclaredField(fieldName);
-					field.setAccessible(true);
-					keyFields[i] = field;
-					break;
-				} catch (NoSuchFieldException e) {
-					clazz = clazz.getSuperclass();
-				}
-			}
-			if (keyFields[i] == null ) {
-				throw new RuntimeException("Class resolved at TaskManager is not compatible with class read during Plan setup."
-						+ " (" + fieldName + ")");
-			}
+			keyFields[i] = FieldSerializer.deserializeField(in);
 		}
 	}
 

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
index 9958540..57928b8 100644
--- a/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/PojoSerializer.java
@@ -121,8 +121,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
 		out.defaultWriteObject();
 		out.writeInt(fields.length);
 		for (Field field: fields) {
-			out.writeObject(field.getDeclaringClass());
-			out.writeUTF(field.getName());
+			FieldSerializer.serializeField(field, out);
 		}
 	}
 
@@ -132,23 +131,7 @@ public final class PojoSerializer<T> extends TypeSerializer<T> {
 		int numFields = in.readInt();
 		fields = new Field[numFields];
 		for (int i = 0; i < numFields; i++) {
-			Class<?> clazz = (Class<?>)in.readObject();
-			String fieldName = in.readUTF();
-			fields[i] = null;
-			// try superclasses as well
-			while (clazz != null) {
-				try {
-					fields[i] = clazz.getDeclaredField(fieldName);
-					fields[i].setAccessible(true);
-					break;
-				} catch (NoSuchFieldException e) {
-					clazz = clazz.getSuperclass();
-				}
-			}
-			if (fields[i] == null) {
-				throw new RuntimeException("Class resolved at TaskManager is not compatible with class read during Plan setup."
-						+ " (" + fieldName + ")");
-			}
+			fields[i] = FieldSerializer.deserializeField(in);
 		}
 
 		cl = Thread.currentThread().getContextClassLoader();

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-core/src/test/java/org/apache/flink/api/java/typeutils/FieldAccessorTest.java
----------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/api/java/typeutils/FieldAccessorTest.java b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/FieldAccessorTest.java
new file mode 100644
index 0000000..f780447
--- /dev/null
+++ b/flink-core/src/test/java/org/apache/flink/api/java/typeutils/FieldAccessorTest.java
@@ -0,0 +1,343 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.java.typeutils;
+
+import static org.junit.Assert.*;
+
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeinfo.InvalidFieldReferenceException;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.junit.Test;
+
+public class FieldAccessorTest {
+
+	// Note, that AggregationFunctionTest indirectly also tests FieldAccessors.
+	// ProductFieldAccessor is tested in CaseClassTypeInfoTest.
+
+	@Test
+	public void testFlatTuple() {
+		Tuple2<String, Integer> t = Tuple2.of("aa", 5);
+		TupleTypeInfo<Tuple2<String, Integer>> tpeInfo =
+				(TupleTypeInfo<Tuple2<String, Integer>>) TypeExtractor.getForObject(t);
+
+		FieldAccessor<Tuple2<String, Integer>, String> f0 = tpeInfo.getFieldAccessor("f0", null);
+		assertEquals("aa", f0.get(t));
+		assertEquals("aa", t.f0);
+		t = f0.set(t, "b");
+		assertEquals("b", f0.get(t));
+		assertEquals("b", t.f0);
+
+		FieldAccessor<Tuple2<String, Integer>, Integer> f1 = tpeInfo.getFieldAccessor("f1", null);
+		assertEquals(5, (int) f1.get(t));
+		assertEquals(5, (int) t.f1);
+		t = f1.set(t, 7);
+		assertEquals(7, (int) f1.get(t));
+		assertEquals(7, (int) t.f1);
+		assertEquals("b", f0.get(t));
+		assertEquals("b", t.f0);
+
+
+		FieldAccessor<Tuple2<String, Integer>, Integer> f1n = tpeInfo.getFieldAccessor(1, null);
+		assertEquals(7, (int) f1n.get(t));
+		assertEquals(7, (int) t.f1);
+		t = f1n.set(t, 10);
+		assertEquals(10, (int) f1n.get(t));
+		assertEquals(10, (int) f1.get(t));
+		assertEquals(10, (int) t.f1);
+		assertEquals("b", f0.get(t));
+		assertEquals("b", t.f0);
+
+		FieldAccessor<Tuple2<String, Integer>, Integer> f1ns = tpeInfo.getFieldAccessor("1", null);
+		assertEquals(10, (int) f1ns.get(t));
+		assertEquals(10, (int) t.f1);
+		t = f1ns.set(t, 11);
+		assertEquals(11, (int) f1ns.get(t));
+		assertEquals(11, (int) f1.get(t));
+		assertEquals(11, (int) t.f1);
+		assertEquals("b", f0.get(t));
+		assertEquals("b", t.f0);
+
+		// This is technically valid (the ".0" is selecting the 0th field of a basic type).
+		FieldAccessor<Tuple2<String, Integer>, String> f0_0 = tpeInfo.getFieldAccessor("f0.0", null);
+		assertEquals("b", f0_0.get(t));
+		assertEquals("b", t.f0);
+		t = f0_0.set(t, "cc");
+		assertEquals("cc", f0_0.get(t));
+		assertEquals("cc", t.f0);
+
+		try {
+			FieldAccessor<Tuple2<String, Integer>, String> bad = tpeInfo.getFieldAccessor("almafa", null);
+			assertFalse("Expected exception, because of bad field name", false);
+		} catch (InvalidFieldReferenceException ex) {
+			// OK
+		}
+	}
+
+	@Test
+	public void testTupleInTuple() {
+		Tuple2<String, Tuple3<Integer, Long, Double>> t = Tuple2.of("aa", Tuple3.of(5, 9L, 2.0));
+		TupleTypeInfo<Tuple2<String, Tuple3<Integer, Long, Double>>> tpeInfo =
+				(TupleTypeInfo<Tuple2<String, Tuple3<Integer, Long, Double>>>)TypeExtractor.getForObject(t);
+
+		FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, String> f0 = tpeInfo.getFieldAccessor("f0", null);
+		assertEquals("aa", f0.get(t));
+		assertEquals("aa", t.f0);
+
+		FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, Double> f1f2 = tpeInfo.getFieldAccessor("f1.f2", null);
+		assertEquals(2.0, f1f2.get(t), 0);
+		assertEquals(2.0, t.f1.f2, 0);
+		t = f1f2.set(t, 3.0);
+		assertEquals(3.0, f1f2.get(t), 0);
+		assertEquals(3.0, t.f1.f2, 0);
+		assertEquals("aa", f0.get(t));
+		assertEquals("aa", t.f0);
+
+		FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, Tuple3<Integer, Long, Double>> f1 = tpeInfo.getFieldAccessor("f1", null);
+		assertEquals(Tuple3.of(5, 9L, 3.0), f1.get(t));
+		assertEquals(Tuple3.of(5, 9L, 3.0), t.f1);
+		t = f1.set(t, Tuple3.of(8, 12L, 4.0));
+		assertEquals(Tuple3.of(8, 12L, 4.0), f1.get(t));
+		assertEquals(Tuple3.of(8, 12L, 4.0), t.f1);
+		assertEquals("aa", f0.get(t));
+		assertEquals("aa", t.f0);
+
+		FieldAccessor<Tuple2<String, Tuple3<Integer, Long, Double>>, Tuple3<Integer, Long, Double>> f1n = tpeInfo.getFieldAccessor(1, null);
+		assertEquals(Tuple3.of(8, 12L, 4.0), f1n.get(t));
+		assertEquals(Tuple3.of(8, 12L, 4.0), t.f1);
+		t = f1n.set(t, Tuple3.of(10, 13L, 5.0));
+		assertEquals(Tuple3.of(10, 13L, 5.0), f1n.get(t));
+		assertEquals(Tuple3.of(10, 13L, 5.0), f1.get(t));
+		assertEquals(Tuple3.of(10, 13L, 5.0), t.f1);
+		assertEquals("aa", f0.get(t));
+		assertEquals("aa", t.f0);
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testTupleFieldAccessorOutOfBounds() {
+		try {
+			TupleTypeInfo.getBasicTupleTypeInfo(Integer.class, Integer.class).getFieldAccessor(2, null);
+			fail();
+		} catch (InvalidFieldReferenceException e) {
+			// Nothing to do here
+		}
+	}
+
+	public static class Foo {
+		public int x;
+		public Tuple2<String, Long> t;
+		public Short y;
+
+		public Foo() {}
+
+		public Foo(int x, Tuple2<String, Long> t, Short y) {
+			this.x = x;
+			this.t = t;
+			this.y = y;
+		}
+	}
+
+	@Test
+	public void testTupleInPojoInTuple() {
+		Tuple2<String, Foo> t = Tuple2.of("aa", new Foo(8, Tuple2.of("ddd", 9L), (short) 2));
+		TupleTypeInfo<Tuple2<String, Foo>> tpeInfo =
+				(TupleTypeInfo<Tuple2<String, Foo>>)TypeExtractor.getForObject(t);
+
+		FieldAccessor<Tuple2<String, Foo>, Long> f1tf1 = tpeInfo.getFieldAccessor("f1.t.f1", null);
+		assertEquals(9L, (long) f1tf1.get(t));
+		assertEquals(9L, (long) t.f1.t.f1);
+		t = f1tf1.set(t, 12L);
+		assertEquals(12L, (long) f1tf1.get(t));
+		assertEquals(12L, (long) t.f1.t.f1);
+
+		FieldAccessor<Tuple2<String, Foo>, String> f1tf0 = tpeInfo.getFieldAccessor("f1.t.f0", null);
+		assertEquals("ddd", f1tf0.get(t));
+		assertEquals("ddd", t.f1.t.f0);
+		t = f1tf0.set(t, "alma");
+		assertEquals("alma", f1tf0.get(t));
+		assertEquals("alma", t.f1.t.f0);
+
+		FieldAccessor<Tuple2<String, Foo>, Foo> f1 = tpeInfo.getFieldAccessor("f1", null);
+		FieldAccessor<Tuple2<String, Foo>, Foo> f1n = tpeInfo.getFieldAccessor(1, null);
+		assertEquals(Tuple2.of("alma", 12L), f1.get(t).t);
+		assertEquals(Tuple2.of("alma", 12L), f1n.get(t).t);
+		assertEquals(Tuple2.of("alma", 12L), t.f1.t);
+		Foo newFoo = new Foo(8, Tuple2.of("ddd", 9L), (short) 2);
+		f1.set(t, newFoo);
+		assertEquals(newFoo, f1.get(t));
+		assertEquals(newFoo, f1n.get(t));
+		assertEquals(newFoo, t.f1);
+	}
+
+
+	public static class Inner {
+		public long x;
+		public boolean b;
+
+		public Inner(){}
+
+		public Inner(long x) {
+			this.x = x;
+		}
+
+		public Inner(long x, boolean b) {
+			this.x = x;
+			this.b = b;
+		}
+
+		@Override
+		public String toString() {
+			return ((Long)x).toString() + ", " + b;
+		}
+	}
+
+	public static class Outer {
+		public int a;
+		public Inner i;
+		public short b;
+
+		public Outer(){}
+
+		public Outer(int a, Inner i, short b) {
+			this.a = a;
+			this.i = i;
+			this.b = b;
+		}
+
+		@Override
+		public String toString() {
+			return a+", "+i.toString()+", "+b;
+		}
+	}
+
+	@Test
+	public void testPojoInPojo() {
+		Outer o = new Outer(10, new Inner(4L), (short)12);
+		PojoTypeInfo<Outer> tpeInfo = (PojoTypeInfo<Outer>)TypeInformation.of(Outer.class);
+
+		FieldAccessor<Outer, Long> fix = tpeInfo.getFieldAccessor("i.x", null);
+		assertEquals(4L, (long) fix.get(o));
+		assertEquals(4L, o.i.x);
+		o = fix.set(o, 22L);
+		assertEquals(22L, (long) fix.get(o));
+		assertEquals(22L, o.i.x);
+
+		FieldAccessor<Outer, Inner> fi = tpeInfo.getFieldAccessor("i", null);
+		assertEquals(22L, fi.get(o).x);
+		assertEquals(22L, (long) fix.get(o));
+		assertEquals(22L, o.i.x);
+		o = fi.set(o, new Inner(30L));
+		assertEquals(30L, fi.get(o).x);
+		assertEquals(30L, (long) fix.get(o));
+		assertEquals(30L, o.i.x);
+	}
+
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testArray() {
+		int[] a = new int[]{3,5};
+		FieldAccessor<int[], Integer> fieldAccessor =
+				(FieldAccessor<int[], Integer>) (Object)
+						PrimitiveArrayTypeInfo.getInfoFor(a.getClass()).getFieldAccessor(1, null);
+
+		assertEquals(Integer.class, fieldAccessor.getFieldType().getTypeClass());
+
+		assertEquals((Integer)a[1], fieldAccessor.get(a));
+
+		a = fieldAccessor.set(a, 6);
+		assertEquals((Integer)a[1], fieldAccessor.get(a));
+
+
+
+		Integer[] b = new Integer[]{3,5};
+		FieldAccessor<Integer[], Integer> fieldAccessor2 =
+				(FieldAccessor<Integer[], Integer>) (Object)
+						BasicArrayTypeInfo.getInfoFor(b.getClass()).getFieldAccessor(1, null);
+
+		assertEquals(Integer.class, fieldAccessor2.getFieldType().getTypeClass());
+
+		assertEquals(b[1], fieldAccessor2.get(b));
+
+		b = fieldAccessor2.set(b, 6);
+		assertEquals(b[1], fieldAccessor2.get(b));
+	}
+
+	public static class ArrayInPojo {
+		public long x;
+		public int[] arr;
+		public int y;
+
+		public ArrayInPojo() {}
+
+		public ArrayInPojo(long x, int[] arr, int y) {
+			this.x = x;
+			this.arr = arr;
+			this.y = y;
+		}
+	}
+
+	@Test
+	public void testArrayInPojo() {
+		ArrayInPojo o = new ArrayInPojo(10L, new int[]{3,4,5}, 12);
+		PojoTypeInfo<ArrayInPojo> tpeInfo = (PojoTypeInfo<ArrayInPojo>)TypeInformation.of(ArrayInPojo.class);
+
+		FieldAccessor<ArrayInPojo, Integer> fix = tpeInfo.getFieldAccessor("arr.1", null);
+		assertEquals(4, (int) fix.get(o));
+		assertEquals(4L, o.arr[1]);
+		o = fix.set(o, 8);
+		assertEquals(8, (int) fix.get(o));
+		assertEquals(8, o.arr[1]);
+	}
+
+	@Test
+	public void testBasicType() {
+		Long x = 7L;
+		TypeInformation<Long> tpeInfo = BasicTypeInfo.LONG_TYPE_INFO;
+
+		try {
+			FieldAccessor<Long, Long> f = tpeInfo.getFieldAccessor(1, null);
+			assertFalse("Expected exception, because not the 0th field selected for a basic type.", false);
+		} catch (InvalidFieldReferenceException ex) {
+			// OK
+		}
+
+		try {
+			FieldAccessor<Long, Long> f = tpeInfo.getFieldAccessor("foo", null);
+			assertFalse("Expected exception, because not the 0th field selected for a basic type.", false);
+		} catch (InvalidFieldReferenceException ex) {
+			// OK
+		}
+
+		FieldAccessor<Long, Long> f = tpeInfo.getFieldAccessor(0, null);
+		assertEquals(7L, (long) f.get(x));
+		x = f.set(x, 12L);
+		assertEquals(12L, (long) f.get(x));
+		assertEquals(12L, (long) x);
+
+		FieldAccessor<Long, Long> f2 = tpeInfo.getFieldAccessor("*", null);
+		assertEquals(12L, (long) f2.get(x));
+		x = f2.set(x, 14L);
+		assertEquals(14L, (long) f2.get(x));
+		assertEquals(14L, (long) x);
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-java/src/main/java/org/apache/flink/api/java/functions/SemanticPropUtil.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/functions/SemanticPropUtil.java b/flink-java/src/main/java/org/apache/flink/api/java/functions/SemanticPropUtil.java
index f8c76e1..4a0b2fc 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/functions/SemanticPropUtil.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/functions/SemanticPropUtil.java
@@ -28,7 +28,7 @@ import org.apache.flink.api.common.operators.util.FieldSet;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.common.typeutils.CompositeType;
 import org.apache.flink.api.common.typeutils.CompositeType.FlatFieldDescriptor;
-import org.apache.flink.api.common.typeutils.CompositeType.InvalidFieldReferenceException;
+import org.apache.flink.api.common.typeinfo.InvalidFieldReferenceException;
 import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFields;
 import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFieldsFirst;
 import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFieldsSecond;

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-java/src/test/java/org/apache/flink/api/java/operator/DataSinkTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/operator/DataSinkTest.java b/flink-java/src/test/java/org/apache/flink/api/java/operator/DataSinkTest.java
index 0493583..0da417b 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/operator/DataSinkTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/operator/DataSinkTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.api.java.operator;
 import org.apache.flink.api.common.InvalidProgramException;
 import org.apache.flink.api.common.operators.Order;
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.InvalidFieldReferenceException;
 import org.apache.flink.api.common.typeutils.CompositeType;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
@@ -166,7 +167,7 @@ public class DataSinkTest {
 			.sortLocalOutput(5, Order.DESCENDING);
 	}
 
-	@Test(expected = CompositeType.InvalidFieldReferenceException.class)
+	@Test(expected = InvalidFieldReferenceException.class)
 	public void testFailTupleInv() {
 
 		final ExecutionEnvironment env = ExecutionEnvironment
@@ -284,7 +285,7 @@ public class DataSinkTest {
 			.sortLocalOutput(1, Order.DESCENDING);
 	}
 
-	@Test(expected = CompositeType.InvalidFieldReferenceException.class)
+	@Test(expected = InvalidFieldReferenceException.class)
 	public void testFailPojoInvalidField() {
 
 		final ExecutionEnvironment env = ExecutionEnvironment

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-java/src/test/java/org/apache/flink/api/java/operator/FullOuterJoinOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/operator/FullOuterJoinOperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/operator/FullOuterJoinOperatorTest.java
index 9f2aa41..9f5cfb2 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/operator/FullOuterJoinOperatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/operator/FullOuterJoinOperatorTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.InvalidProgramException;
 import org.apache.flink.api.common.functions.JoinFunction;
 import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint;
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.InvalidFieldReferenceException;
 import org.apache.flink.api.common.typeutils.CompositeType;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
@@ -135,7 +136,7 @@ public class FullOuterJoinOperatorTest {
 				.with(new DummyJoin());
 	}
 
-	@Test(expected = CompositeType.InvalidFieldReferenceException.class)
+	@Test(expected = InvalidFieldReferenceException.class)
 	public void testFullOuter8() {
 		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 		DataSet<Tuple5<Integer, Long, String, Long, Integer>> ds1 = env.fromCollection(emptyTupleData, tupleTypeInfo);

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-java/src/test/java/org/apache/flink/api/java/operator/LeftOuterJoinOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/operator/LeftOuterJoinOperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/operator/LeftOuterJoinOperatorTest.java
index bfcc3e8..914c75c 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/operator/LeftOuterJoinOperatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/operator/LeftOuterJoinOperatorTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
 
 import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint;
 import org.apache.flink.api.common.functions.JoinFunction;
+import org.apache.flink.api.common.typeinfo.InvalidFieldReferenceException;
 import org.apache.flink.api.common.typeutils.CompositeType;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
@@ -136,7 +137,7 @@ public class LeftOuterJoinOperatorTest {
 				.with(new DummyJoin());
 	}
 
-	@Test(expected = CompositeType.InvalidFieldReferenceException.class)
+	@Test(expected = InvalidFieldReferenceException.class)
 	public void testLeftOuter8() {
 		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 		DataSet<Tuple5<Integer, Long, String, Long, Integer>> ds1 = env.fromCollection(emptyTupleData, tupleTypeInfo);

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-java/src/test/java/org/apache/flink/api/java/operator/RightOuterJoinOperatorTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/operator/RightOuterJoinOperatorTest.java b/flink-java/src/test/java/org/apache/flink/api/java/operator/RightOuterJoinOperatorTest.java
index 709d830..f5d8129 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/operator/RightOuterJoinOperatorTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/operator/RightOuterJoinOperatorTest.java
@@ -22,6 +22,7 @@ import org.apache.flink.api.common.InvalidProgramException;
 import org.apache.flink.api.common.functions.JoinFunction;
 import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint;
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
+import org.apache.flink.api.common.typeinfo.InvalidFieldReferenceException;
 import org.apache.flink.api.common.typeutils.CompositeType;
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
@@ -135,7 +136,7 @@ public class RightOuterJoinOperatorTest {
 				.with(new DummyJoin());
 	}
 
-	@Test(expected = CompositeType.InvalidFieldReferenceException.class)
+	@Test(expected = InvalidFieldReferenceException.class)
 	public void testRightOuter8() {
 		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
 		DataSet<Tuple5<Integer, Long, String, Long, Integer>> ds1 = env.fromCollection(emptyTupleData, tupleTypeInfo);

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-scala/src/main/java/org/apache/flink/api/scala/typeutils/ProductFieldAccessor.java
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/java/org/apache/flink/api/scala/typeutils/ProductFieldAccessor.java b/flink-scala/src/main/java/org/apache/flink/api/scala/typeutils/ProductFieldAccessor.java
new file mode 100644
index 0000000..0be6f33
--- /dev/null
+++ b/flink-scala/src/main/java/org/apache/flink/api/scala/typeutils/ProductFieldAccessor.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.api.scala.typeutils;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.InvalidFieldReferenceException;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.FieldAccessor;
+import org.apache.flink.api.java.typeutils.TupleTypeInfoBase;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializerBase;
+import scala.Product;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+public final class ProductFieldAccessor<T, R, F> extends FieldAccessor<T, F> {
+
+	private static final long serialVersionUID = 1L;
+
+	private final int pos;
+	private final TupleSerializerBase<T> serializer;
+	private final Object[] fields;
+	private final int length;
+	private final FieldAccessor<R, F> innerAccessor;
+
+	ProductFieldAccessor(int pos, TypeInformation<T> typeInfo, FieldAccessor<R, F> innerAccessor, ExecutionConfig config) {
+		int arity = ((TupleTypeInfoBase)typeInfo).getArity();
+		if(pos < 0 || pos >= arity) {
+			throw new InvalidFieldReferenceException(
+				"Tried to select " + ((Integer) pos).toString() + ". field on \"" +
+					typeInfo.toString() + "\", which is an invalid index.");
+		}
+		checkNotNull(typeInfo, "typeInfo must not be null.");
+		checkNotNull(innerAccessor, "innerAccessor must not be null.");
+
+		this.pos = pos;
+		this.fieldType = ((TupleTypeInfoBase<T>)typeInfo).getTypeAt(pos);
+		this.serializer = (TupleSerializerBase<T>)typeInfo.createSerializer(config);
+		this.length = this.serializer.getArity();
+		this.fields = new Object[this.length];
+		this.innerAccessor = innerAccessor;
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public F get(T record) {
+		return innerAccessor.get((R)((Product)record).productElement(pos));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	public T set(T record, F fieldValue) {
+		Product prod = (Product)record;
+		for (int i = 0; i < length; i++) {
+			fields[i] = prod.productElement(i);
+		}
+		fields[pos] = innerAccessor.set((R)fields[pos], fieldValue);
+		return serializer.createInstance(fields);
+	}
+}

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassTypeInfo.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassTypeInfo.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassTypeInfo.scala
index 2aecd7a..d970dfd 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassTypeInfo.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/typeutils/CaseClassTypeInfo.scala
@@ -19,17 +19,18 @@
 package org.apache.flink.api.scala.typeutils
 
 import java.util
-import java.util.regex.{Pattern, Matcher}
+import java.util.regex.{Matcher, Pattern}
 
-import org.apache.flink.annotation.{PublicEvolving, Public}
+import org.apache.flink.annotation.{Public, PublicEvolving}
 import org.apache.flink.api.common.ExecutionConfig
 import org.apache.flink.api.common.operators.Keys
 import org.apache.flink.api.common.typeinfo.TypeInformation
-import org.apache.flink.api.common.typeutils.CompositeType.{TypeComparatorBuilder,
-InvalidFieldReferenceException, FlatFieldDescriptor}
+import org.apache.flink.api.common.typeutils.CompositeType.{FlatFieldDescriptor, TypeComparatorBuilder}
 import org.apache.flink.api.common.typeutils._
 import Keys.ExpressionKeys
-import org.apache.flink.api.java.typeutils.TupleTypeInfoBase
+import org.apache.flink.api.common.typeinfo.InvalidFieldReferenceException
+import org.apache.flink.api.java.typeutils.FieldAccessor.SimpleFieldAccessor
+import org.apache.flink.api.java.typeutils.{FieldAccessor, TupleTypeInfoBase}
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable.ArrayBuffer
@@ -202,7 +203,7 @@ abstract class CaseClassTypeInfo[T <: Product](
   override def getFieldIndex(fieldName: String): Int = {
     val result = fieldNames.indexOf(fieldName)
     if (result != fieldNames.lastIndexOf(fieldName)) {
-      -2
+      -1
     } else {
       result
     }
@@ -238,6 +239,31 @@ abstract class CaseClassTypeInfo[T <: Product](
     }
   }
 
+  override def getFieldAccessor[F](pos: Int, config: ExecutionConfig): FieldAccessor[T, F] = {
+    new ProductFieldAccessor[T,F,F](
+      pos, this, new SimpleFieldAccessor[F](types(pos).asInstanceOf[TypeInformation[F]]), config)
+  }
+
+  override def getFieldAccessor[F](fieldExpression: String, config: ExecutionConfig):
+    FieldAccessor[T, F] = {
+
+    val decomp = FieldAccessor.decomposeFieldExpression(fieldExpression)
+
+    val pos = getFieldIndex(decomp.head)
+    if(pos < 0) {
+      throw new InvalidFieldReferenceException("Invalid field selected: " + fieldExpression)
+    }
+    val fieldType = types(pos)
+
+    if (decomp.tail == null) {
+      getFieldAccessor(pos, config)
+    } else {
+      val innerAccessor =
+        fieldType.getFieldAccessor[F](decomp.tail, config).asInstanceOf[FieldAccessor[AnyRef, F]]
+      new ProductFieldAccessor[T,Object,F](pos, this, innerAccessor, config)
+    }
+  }
+
   override def toString: String = {
     clazz.getName + "(" + fieldNames.zip(types).map {
       case (n, t) => n + ": " + t

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/CaseClassTypeInfoTest.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/CaseClassTypeInfoTest.scala b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/CaseClassTypeInfoTest.scala
index 479483f..a9abea1 100644
--- a/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/CaseClassTypeInfoTest.scala
+++ b/flink-scala/src/test/scala/org/apache/flink/api/scala/typeutils/CaseClassTypeInfoTest.scala
@@ -21,9 +21,11 @@ package org.apache.flink.api.scala.typeutils
 import org.apache.flink.api.common.ExecutionConfig
 import org.apache.flink.api.common.typeinfo.BasicTypeInfo
 import org.apache.flink.api.common.typeutils.TypeSerializer
+import org.apache.flink.api.java.typeutils.FieldAccessorTest
 import org.apache.flink.util.TestLogger
 import org.junit.Test
 import org.scalatest.junit.JUnitSuiteLike
+import org.apache.flink.api.scala._
 
 class CaseClassTypeInfoTest extends TestLogger with JUnitSuiteLike {
 
@@ -70,4 +72,112 @@ class CaseClassTypeInfoTest extends TestLogger with JUnitSuiteLike {
     assert(!tpeInfo1.equals(tpeInfo2))
   }
 
+  @Test
+  def testFieldAccessorFlatCaseClass(): Unit = {
+    case class IntBoolean(foo: Int, bar: Boolean)
+    val tpeInfo = createTypeInformation[IntBoolean]
+
+    {
+      // by field name
+      val accessor1 = tpeInfo.getFieldAccessor[Int]("foo", null)
+      val accessor2 = tpeInfo.getFieldAccessor[Boolean]("bar", null)
+
+      val x1 = IntBoolean(5, false)
+      assert(accessor1.get(x1) == 5)
+      assert(accessor2.get(x1) == false)
+      assert(x1.foo == 5)
+      assert(x1.bar == false)
+
+      val x2: IntBoolean = accessor1.set(x1, 6)
+      assert(accessor1.get(x2) == 6)
+      assert(x2.foo == 6)
+
+      val x3 = accessor2.set(x2, true)
+      assert(x3.bar == true)
+      assert(accessor2.get(x3) == true)
+      assert(x3.foo == 6)
+    }
+
+    {
+      // by field pos
+      val accessor1 = tpeInfo.getFieldAccessor[Int](0, null)
+      val accessor2 = tpeInfo.getFieldAccessor[Boolean](1, null)
+
+      val x1 = IntBoolean(5, false)
+      assert(accessor1.get(x1) == 5)
+      assert(accessor2.get(x1) == false)
+      assert(x1.foo == 5)
+      assert(x1.bar == false)
+
+      val x2: IntBoolean = accessor1.set(x1, 6)
+      assert(accessor1.get(x2) == 6)
+      assert(x2.foo == 6)
+
+      val x3 = accessor2.set(x2, true)
+      assert(x3.bar == true)
+      assert(accessor2.get(x3) == true)
+      assert(x3.foo == 6)
+    }
+  }
+
+  @Test
+  def testFieldAccessorTuple(): Unit = {
+    val tpeInfo = createTypeInformation[(Int, Long)]
+    var x = (5, 6L)
+    val f0 = tpeInfo.getFieldAccessor[Int](0, null)
+    assert(f0.get(x) == 5)
+    x = f0.set(x, 8)
+    assert(f0.get(x) == 8)
+    assert(x._1 == 8)
+  }
+
+  @Test
+  def testFieldAccessorCaseClassInCaseClass(): Unit = {
+    case class Inner(a: Short, b: String)
+    case class Outer(a: Int, i: Inner, b: Boolean)
+    val tpeInfo = createTypeInformation[Outer]
+
+    var x = Outer(1, Inner(2, "alma"), true)
+
+    val fib = tpeInfo.getFieldAccessor[String]("i.b", null)
+    assert(fib.get(x) == "alma")
+    assert(x.i.b == "alma")
+    x = fib.set(x, "korte")
+    assert(fib.get(x) == "korte")
+    assert(x.i.b == "korte")
+
+    val fi = tpeInfo.getFieldAccessor[Inner]("i", null)
+    assert(fi.get(x) == Inner(2, "korte"))
+    x = fi.set(x, Inner(3, "aaa"))
+    assert(x.i == Inner(3, "aaa"))
+  }
+
+  @Test
+  def testFieldAccessorPojoInCaseClass(): Unit = {
+    case class Outer(a: Int, i: FieldAccessorTest.Inner, b: Boolean)
+    var x = Outer(1, new FieldAccessorTest.Inner(3L, true), false)
+    val tpeInfo = createTypeInformation[Outer]
+    val cfg = new ExecutionConfig
+
+    val fib = tpeInfo.getFieldAccessor[Boolean]("i.b", cfg)
+    assert(fib.get(x) == true)
+    assert(x.i.b == true)
+    x = fib.set(x, false)
+    assert(fib.get(x) == false)
+    assert(x.i.b == false)
+
+    val fi = tpeInfo.getFieldAccessor[FieldAccessorTest.Inner]("i", cfg)
+    assert(fi.get(x).x == 3L)
+    assert(x.i.x == 3L)
+    x = fi.set(x, new FieldAccessorTest.Inner(4L, true))
+    assert(fi.get(x).x == 4L)
+    assert(x.i.x == 4L)
+
+    val fin = tpeInfo.getFieldAccessor[FieldAccessorTest.Inner](1, cfg)
+    assert(fin.get(x).x == 4L)
+    assert(x.i.x == 4L)
+    x = fin.set(x, new FieldAccessorTest.Inner(5L, true))
+    assert(fin.get(x).x == 5L)
+    assert(x.i.x == 5L)
+  }
 }

http://git-wip-us.apache.org/repos/asf/flink/blob/1f04542e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
index 4063b60..264d5d0 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
@@ -365,7 +365,9 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	 * per key.
 	 *
 	 * @param positionToSum
-	 *            The position in the data point to sum
+	 *            The field position in the data points to sum. This is applicable to
+	 *            Tuple types, basic and primitive array types, Scala case classes,
+	 *            and primitive types (which is considered as having one field).
 	 * @return The transformed DataStream.
 	 */
 	public SingleOutputStreamOperator<T> sum(int positionToSum) {
@@ -373,16 +375,18 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current sum of the pojo data
-	 * stream at the given field expressionby the given key. An independent
-	 * aggregate is kept per key. A field expression is either the name of a
-	 * public field or a getter method with parentheses of the
-	 * {@link DataStream}S underlying type. A dot can be used to drill down into
-	 * objects, as in {@code "field1.getInnerField2()" }.
+	 * Applies an aggregation that gives the current sum of the data
+	 * stream at the given field by the given key. An independent
+	 * aggregate is kept per key.
 	 *
 	 * @param field
-	 *            The field expression based on which the aggregation will be
-	 *            applied.
+	 *            In case of a POJO, Scala case class, or Tuple type, the
+	 *            name of the (public) field on which to perform the aggregation.
+	 *            Additionally, a dot can be used to drill down into nested
+	 *            objects, as in {@code "field1.fieldxy" }.
+	 *            Furthermore, an array index can also be specified in case of an array of
+	 *            a primitive or basic type; or "0" or "*" can be specified in case of a
+	 *            basic type (which is considered as having only one field).
 	 * @return The transformed DataStream.
 	 */
 	public SingleOutputStreamOperator<T> sum(String field) {
@@ -390,12 +394,14 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current minimum of the data
+	 * Applies an aggregation that gives the current minimum of the data
 	 * stream at the given position by the given key. An independent aggregate
 	 * is kept per key.
 	 *
 	 * @param positionToMin
-	 *            The position in the data point to minimize
+	 *            The field position in the data points to minimize. This is applicable to
+	 *            Tuple types, basic and primitive array types, Scala case classes,
+	 *            and primitive types (which is considered as having one field).
 	 * @return The transformed DataStream.
 	 */
 	public SingleOutputStreamOperator<T> min(int positionToMin) {
@@ -404,16 +410,21 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current minimum of the pojo
+	 * Applies an aggregation that gives the current minimum of the
 	 * data stream at the given field expression by the given key. An
 	 * independent aggregate is kept per key. A field expression is either the
 	 * name of a public field or a getter method with parentheses of the
-	 * {@link DataStream}S underlying type. A dot can be used to drill down into
-	 * objects, as in {@code "field1.getInnerField2()" }.
+	 * {@link DataStream}'s underlying type. A dot can be used to drill down into
+	 * objects, as in {@code "field1.fieldxy" }.
 	 *
 	 * @param field
-	 *            The field expression based on which the aggregation will be
-	 *            applied.
+	 *            In case of a POJO, Scala case class, or Tuple type, the
+	 *            name of the (public) field on which to perform the aggregation.
+	 *            Additionally, a dot can be used to drill down into nested
+	 *            objects, as in {@code "field1.fieldxy" }.
+	 *            Furthermore, an array index can also be specified in case of an array of
+	 *            a primitive or basic type; or "0" or "*" can be specified in case of a
+	 *            basic type (which is considered as having only one field).
 	 * @return The transformed DataStream.
 	 */
 	public SingleOutputStreamOperator<T> min(String field) {
@@ -427,7 +438,9 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	 * per key.
 	 *
 	 * @param positionToMax
-	 *            The position in the data point to maximize
+	 *            The field position in the data points to maximize. This is applicable to
+	 *            Tuple types, basic and primitive array types, Scala case classes,
+	 *            and primitive types (which is considered as having one field).
 	 * @return The transformed DataStream.
 	 */
 	public SingleOutputStreamOperator<T> max(int positionToMax) {
@@ -436,16 +449,21 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current maximum of the pojo
+	 * Applies an aggregation that gives the current maximum of the
 	 * data stream at the given field expression by the given key. An
 	 * independent aggregate is kept per key. A field expression is either the
 	 * name of a public field or a getter method with parentheses of the
-	 * {@link DataStream}S underlying type. A dot can be used to drill down into
-	 * objects, as in {@code "field1.getInnerField2()" }.
+	 * {@link DataStream}'s underlying type. A dot can be used to drill down into
+	 * objects, as in {@code "field1.fieldxy" }.
 	 *
 	 * @param field
-	 *            The field expression based on which the aggregation will be
-	 *            applied.
+	 *            In case of a POJO, Scala case class, or Tuple type, the
+	 *            name of the (public) field on which to perform the aggregation.
+	 *            Additionally, a dot can be used to drill down into nested
+	 *            objects, as in {@code "field1.fieldxy" }.
+	 *            Furthermore, an array index can also be specified in case of an array of
+	 *            a primitive or basic type; or "0" or "*" can be specified in case of a
+	 *            basic type (which is considered as having only one field).
 	 * @return The transformed DataStream.
 	 */
 	public SingleOutputStreamOperator<T> max(String field) {
@@ -454,16 +472,21 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current minimum element of the
-	 * pojo data stream by the given field expression by the given key. An
+	 * Applies an aggregation that gives the current minimum element of the
+	 * data stream by the given field expression by the given key. An
 	 * independent aggregate is kept per key. A field expression is either the
 	 * name of a public field or a getter method with parentheses of the
-	 * {@link DataStream}S underlying type. A dot can be used to drill down into
-	 * objects, as in {@code "field1.getInnerField2()" }.
+	 * {@link DataStream}'s underlying type. A dot can be used to drill down into
+	 * objects, as in {@code "field1.fieldxy" }.
 	 *
 	 * @param field
-	 *            The field expression based on which the aggregation will be
-	 *            applied.
+	 *            In case of a POJO, Scala case class, or Tuple type, the
+	 *            name of the (public) field on which to perform the aggregation.
+	 *            Additionally, a dot can be used to drill down into nested
+	 *            objects, as in {@code "field1.fieldxy" }.
+	 *            Furthermore, an array index can also be specified in case of an array of
+	 *            a primitive or basic type; or "0" or "*" can be specified in case of a
+	 *            basic type (which is considered as having only one field).
 	 * @param first
 	 *            If True then in case of field equality the first object will
 	 *            be returned
@@ -476,16 +499,21 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current maximum element of the
-	 * pojo data stream by the given field expression by the given key. An
+	 * Applies an aggregation that gives the current maximum element of the
+	 * data stream by the given field expression by the given key. An
 	 * independent aggregate is kept per key. A field expression is either the
 	 * name of a public field or a getter method with parentheses of the
-	 * {@link DataStream}S underlying type. A dot can be used to drill down into
-	 * objects, as in {@code "field1.getInnerField2()" }.
+	 * {@link DataStream}'s underlying type. A dot can be used to drill down into
+	 * objects, as in {@code "field1.fieldxy" }.
 	 *
 	 * @param field
-	 *            The field expression based on which the aggregation will be
-	 *            applied.
+	 *            In case of a POJO, Scala case class, or Tuple type, the
+	 *            name of the (public) field on which to perform the aggregation.
+	 *            Additionally, a dot can be used to drill down into nested
+	 *            objects, as in {@code "field1.fieldxy" }.
+	 *            Furthermore, an array index can also be specified in case of an array of
+	 *            a primitive or basic type; or "0" or "*" can be specified in case of a
+	 *            basic type (which is considered as having only one field).
 	 * @param first
 	 *            If True then in case of field equality the first object will
 	 *            be returned
@@ -497,13 +525,15 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current element with the
+	 * Applies an aggregation that gives the current element with the
 	 * minimum value at the given position by the given key. An independent
 	 * aggregate is kept per key. If more elements have the minimum value at the
 	 * given position, the operator returns the first one by default.
 	 *
 	 * @param positionToMinBy
-	 *            The position in the data point to minimize
+	 *            The field position in the data points to minimize. This is applicable to
+	 *            Tuple types, basic and primitive array types, Scala case classes,
+	 *            and primitive types (which is considered as having one field).
 	 * @return The transformed DataStream.
 	 */
 	public SingleOutputStreamOperator<T> minBy(int positionToMinBy) {
@@ -511,13 +541,19 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current element with the
+	 * Applies an aggregation that gives the current element with the
 	 * minimum value at the given position by the given key. An independent
 	 * aggregate is kept per key. If more elements have the minimum value at the
 	 * given position, the operator returns the first one by default.
 	 *
 	 * @param positionToMinBy
-	 *            The position in the data point to minimize
+	 *            In case of a POJO, Scala case class, or Tuple type, the
+	 *            name of the (public) field on which to perform the aggregation.
+	 *            Additionally, a dot can be used to drill down into nested
+	 *            objects, as in {@code "field1.fieldxy" }.
+	 *            Furthermore, an array index can also be specified in case of an array of
+	 *            a primitive or basic type; or "0" or "*" can be specified in case of a
+	 *            basic type (which is considered as having only one field).
 	 * @return The transformed DataStream.
 	 */
 	public SingleOutputStreamOperator<T> minBy(String positionToMinBy) {
@@ -525,14 +561,16 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current element with the
+	 * Applies an aggregation that gives the current element with the
 	 * minimum value at the given position by the given key. An independent
 	 * aggregate is kept per key. If more elements have the minimum value at the
 	 * given position, the operator returns either the first or last one,
 	 * depending on the parameter set.
 	 *
 	 * @param positionToMinBy
-	 *            The position in the data point to minimize
+	 *            The field position in the data points to minimize. This is applicable to
+	 *            Tuple types, basic and primitive array types, Scala case classes,
+	 *            and primitive types (which is considered as having one field).
 	 * @param first
 	 *            If true, then the operator return the first element with the
 	 *            minimal value, otherwise returns the last
@@ -544,13 +582,15 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current element with the
+	 * Applies an aggregation that gives the current element with the
 	 * maximum value at the given position by the given key. An independent
 	 * aggregate is kept per key. If more elements have the maximum value at the
 	 * given position, the operator returns the first one by default.
 	 *
 	 * @param positionToMaxBy
-	 *            The position in the data point to maximize
+	 *            The field position in the data points to maximize. This is applicable to
+	 *            Tuple types, basic and primitive array types, Scala case classes,
+	 *            and primitive types (which is considered as having one field).
 	 * @return The transformed DataStream.
 	 */
 	public SingleOutputStreamOperator<T> maxBy(int positionToMaxBy) {
@@ -558,13 +598,19 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current element with the
+	 * Applies an aggregation that gives the current element with the
 	 * maximum value at the given position by the given key. An independent
 	 * aggregate is kept per key. If more elements have the maximum value at the
 	 * given position, the operator returns the first one by default.
 	 *
 	 * @param positionToMaxBy
-	 *            The position in the data point to maximize
+	 *            In case of a POJO, Scala case class, or Tuple type, the
+	 *            name of the (public) field on which to perform the aggregation.
+	 *            Additionally, a dot can be used to drill down into nested
+	 *            objects, as in {@code "field1.fieldxy" }.
+	 *            Furthermore, an array index can also be specified in case of an array of
+	 *            a primitive or basic type; or "0" or "*" can be specified in case of a
+	 *            basic type (which is considered as having only one field).
 	 * @return The transformed DataStream.
 	 */
 	public SingleOutputStreamOperator<T> maxBy(String positionToMaxBy) {
@@ -572,14 +618,16 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
 	}
 
 	/**
-	 * Applies an aggregation that that gives the current element with the
+	 * Applies an aggregation that gives the current element with the
 	 * maximum value at the given position by the given key. An independent
 	 * aggregate is kept per key. If more elements have the maximum value at the
 	 * given position, the operator returns either the first or last one,
 	 * depending on the parameter set.
 	 *
 	 * @param positionToMaxBy
-	 *            The position in the data point to maximize.
+	 *            The field position in the data points to maximize. This is applicable to
+	 *            Tuple types, basic and primitive array types, Scala case classes,
+	 *            and primitive types (which is considered as having one field).
 	 * @param first
 	 *            If true, then the operator return the first element with the
 	 *            maximum value, otherwise returns the last


Mime
View raw message