You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2015/02/23 16:05:26 UTC
[2/6] flink git commit: Add methods to CompositeType for querying
field types and names
Add methods to CompositeType for querying field types and names
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/98da559d
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/98da559d
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/98da559d
Branch: refs/heads/master
Commit: 98da559d0ff3d1a66b49ce1cbacf463eef7cccbe
Parents: 86ca692
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Thu Oct 23 18:09:38 2014 +0200
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Mon Feb 23 13:04:54 2015 +0100
----------------------------------------------------------------------
.../api/common/typeutils/CompositeType.java | 20 ++++++++++++++++++
.../flink/api/java/typeutils/PojoTypeInfo.java | 17 +++++++++++++++
.../flink/api/java/typeutils/TupleTypeInfo.java | 22 +++++++++++++++++++-
.../api/java/typeutils/TupleTypeInfoBase.java | 2 --
.../apache/flink/api/scala/CrossDataSet.scala | 5 ++++-
.../api/scala/UnfinishedCoGroupOperation.scala | 9 +++++---
.../api/scala/codegen/TypeInformationGen.scala | 16 +++++++++++++-
.../apache/flink/api/scala/joinDataSet.scala | 5 ++++-
.../api/scala/typeutils/CaseClassTypeInfo.scala | 17 +++++++++++++++
.../api/scala/StreamCrossOperator.scala | 6 ++++--
.../api/scala/StreamJoinOperator.scala | 6 ++++--
11 files changed, 112 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/98da559d/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 3bd1fff..2379256 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
@@ -161,6 +161,26 @@ public abstract class CompositeType<T> extends TypeInformation<T> {
}
}
+ /**
+ * Returns true when this type has a composite field with the given name.
+ */
+ public boolean hasField(String fieldName) {
+ return getFieldIndex(fieldName) >= 0;
+ }
+
+ /**
+ * Returns the names of the composite fields of this type. The order of the returned array must
+ * be consistent with the internal field index ordering.
+ */
+ public abstract String[] getFieldNames();
+
+ /**
+ * Returns the field index of the composite field of the given name.
+ *
+ * @return The field index or -1 if this type does not have a field of the given name.
+ */
+ public abstract int getFieldIndex(String fieldName);
+
public static class InvalidFieldReferenceException extends IllegalArgumentException {
private static final long serialVersionUID = 1L;
http://git-wip-us.apache.org/repos/asf/flink/blob/98da559d/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
index e35ed38..1dcee24 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/PojoTypeInfo.java
@@ -290,6 +290,23 @@ public class PojoTypeInfo<T> extends CompositeType<T> {
return new PojoComparator<T>(finalKeyFields, finalFieldComparators, createSerializer(config), typeClass);
}
+ public String[] getFieldNames() {
+ String[] result = new String[fields.length];
+ for (int i = 0; i < fields.length; i++) {
+ result[i] = fields[i].field.getName();
+ }
+ return result;
+ }
+
+ @Override
+ public int getFieldIndex(String fieldName) {
+ for (int i = 0; i < fields.length; i++) {
+ if (fields[i].field.getName().equals(fieldName)) {
+ return i;
+ }
+ }
+ return -1;
+ }
@Override
public TypeSerializer<T> createSerializer(ExecutionConfig config) {
http://git-wip-us.apache.org/repos/asf/flink/blob/98da559d/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java
index e88650d..0e3d61d 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfo.java
@@ -39,7 +39,9 @@ import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
public final class TupleTypeInfo<T extends Tuple> extends TupleTypeInfoBase<T> {
private static final long serialVersionUID = 1L;
-
+
+ protected final String[] fieldNames;
+
@SuppressWarnings("unchecked")
public TupleTypeInfo(TypeInformation<?>... types) {
this((Class<T>) Tuple.getTupleClass(types.length), types);
@@ -50,6 +52,24 @@ public final class TupleTypeInfo<T extends Tuple> extends TupleTypeInfoBase<T> {
if (types == null || types.length == 0 || types.length > Tuple.MAX_ARITY) {
throw new IllegalArgumentException();
}
+ this.fieldNames = new String[types.length];
+ for (int i = 0; i < types.length; i++) {
+ fieldNames[i] = "f" + i;
+ }
+ }
+
+ @Override
+ public String[] getFieldNames() {
+ return fieldNames;
+ }
+
+ @Override
+ public int getFieldIndex(String fieldName) {
+ int fieldIndex = Integer.parseInt(fieldName.substring(1));
+ if (fieldIndex >= getArity()) {
+ return -1;
+ }
+ return fieldIndex;
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/98da559d/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
index 5c8a57a..ce075a7 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/TupleTypeInfoBase.java
@@ -84,7 +84,6 @@ public abstract class TupleTypeInfoBase<T> extends CompositeType<T> {
return tupleType;
}
-
/**
* Recursively add all fields in this tuple type. We need this in particular to get all
* the types.
@@ -273,5 +272,4 @@ public abstract class TupleTypeInfoBase<T> extends CompositeType<T> {
bld.append('>');
return bld.toString();
}
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/98da559d/flink-scala/src/main/scala/org/apache/flink/api/scala/CrossDataSet.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/CrossDataSet.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/CrossDataSet.scala
index e57fd5b..eb9255b 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/CrossDataSet.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/CrossDataSet.scala
@@ -113,7 +113,10 @@ private[flink] object CrossDataSet {
}
}
val returnType = new CaseClassTypeInfo[(L, R)](
- classOf[(L, R)], Seq(leftInput.getType, rightInput.getType), Array("_1", "_2")) {
+ classOf[(L, R)],
+ Array(leftInput.getType, rightInput.getType),
+ Seq(leftInput.getType, rightInput.getType),
+ Array("_1", "_2")) {
override def createSerializer(executionConfig: ExecutionConfig): TypeSerializer[(L, R)] = {
val fieldSerializers: Array[TypeSerializer[_]] = new Array[TypeSerializer[_]](getArity)
http://git-wip-us.apache.org/repos/asf/flink/blob/98da559d/flink-scala/src/main/scala/org/apache/flink/api/scala/UnfinishedCoGroupOperation.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/UnfinishedCoGroupOperation.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/UnfinishedCoGroupOperation.scala
index 2b2e372..98c8c31 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/UnfinishedCoGroupOperation.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/UnfinishedCoGroupOperation.scala
@@ -42,8 +42,8 @@ import scala.reflect.ClassTag
* @tparam R The type of the right input of the coGroup.
*/
class UnfinishedCoGroupOperation[L: ClassTag, R: ClassTag](
- leftInput: DataSet[L],
- rightInput: DataSet[R])
+ leftInput: DataSet[L],
+ rightInput: DataSet[R])
extends UnfinishedKeyPairOperation[L, R, CoGroupDataSet[L, R]](leftInput, rightInput) {
private[flink] def finish(leftKey: Keys[L], rightKey: Keys[R]) = {
@@ -67,7 +67,10 @@ class UnfinishedCoGroupOperation[L: ClassTag, R: ClassTag](
ObjectArrayTypeInfo.getInfoFor(new Array[R](0).getClass, rightInput.getType)
val returnType = new CaseClassTypeInfo[(Array[L], Array[R])](
- classOf[(Array[L], Array[R])], Seq(leftArrayType, rightArrayType), Array("_1", "_2")) {
+ classOf[(Array[L], Array[R])],
+ Array(leftArrayType, rightArrayType),
+ Seq(leftArrayType, rightArrayType),
+ Array("_1", "_2")) {
override def createSerializer(
executionConfig: ExecutionConfig): TypeSerializer[(Array[L], Array[R])] = {
http://git-wip-us.apache.org/repos/asf/flink/blob/98da559d/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
index bafb7bf..118b8c4 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/codegen/TypeInformationGen.scala
@@ -89,6 +89,15 @@ private[flink] trait TypeInformationGen[C <: Context] {
def mkCaseClassTypeInfo[T <: Product : c.WeakTypeTag](
desc: CaseClassDescriptor): c.Expr[TypeInformation[T]] = {
val tpeClazz = c.Expr[Class[T]](Literal(Constant(desc.tpe)))
+
+ val genericTypeInfos = desc.tpe match {
+ case TypeRef(_, _, typeParams) =>
+ val typeInfos = typeParams map { tpe => mkTypeInfo(c.WeakTypeTag[T](tpe)).tree }
+ c.Expr[List[TypeInformation[_]]](mkList(typeInfos))
+ case _ =>
+ reify { List[TypeInformation[_]]() }
+ }
+
val fields = desc.getters.toList map { field =>
mkTypeInfo(field.desc)(c.WeakTypeTag(field.tpe)).tree
}
@@ -98,7 +107,12 @@ private[flink] trait TypeInformationGen[C <: Context] {
val fieldNames = desc.getters map { f => Literal(Constant(f.getter.name.toString)) } toList
val fieldNamesExpr = c.Expr[Seq[String]](mkSeq(fieldNames))
reify {
- new CaseClassTypeInfo[T](tpeClazz.splice, fieldsExpr.splice, fieldNamesExpr.splice) {
+ new CaseClassTypeInfo[T](
+ tpeClazz.splice,
+ genericTypeInfos.splice.toArray,
+ fieldsExpr.splice,
+ fieldNamesExpr.splice) {
+
override def createSerializer(executionConfig: ExecutionConfig): TypeSerializer[T] = {
val fieldSerializers: Array[TypeSerializer[_]] = new Array[TypeSerializer[_]](getArity)
for (i <- 0 until getArity) {
http://git-wip-us.apache.org/repos/asf/flink/blob/98da559d/flink-scala/src/main/scala/org/apache/flink/api/scala/joinDataSet.scala
----------------------------------------------------------------------
diff --git a/flink-scala/src/main/scala/org/apache/flink/api/scala/joinDataSet.scala b/flink-scala/src/main/scala/org/apache/flink/api/scala/joinDataSet.scala
index 420496f..82cfeba 100644
--- a/flink-scala/src/main/scala/org/apache/flink/api/scala/joinDataSet.scala
+++ b/flink-scala/src/main/scala/org/apache/flink/api/scala/joinDataSet.scala
@@ -234,7 +234,10 @@ class UnfinishedJoinOperation[L, R](
}
}
val returnType = new CaseClassTypeInfo[(L, R)](
- classOf[(L, R)], Seq(leftSet.getType, rightSet.getType), Array("_1", "_2")) {
+ classOf[(L, R)],
+ Array(leftSet.getType, rightSet.getType),
+ Seq(leftSet.getType, rightSet.getType),
+ Array("_1", "_2")) {
override def createSerializer(executionConfig: ExecutionConfig): TypeSerializer[(L, R)] = {
val fieldSerializers: Array[TypeSerializer[_]] = new Array[TypeSerializer[_]](getArity)
http://git-wip-us.apache.org/repos/asf/flink/blob/98da559d/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 1d627ab..5f35f2e 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
@@ -36,10 +36,12 @@ import PojoTypeInfo.NamedFlatFieldDescriptor
*/
abstract class CaseClassTypeInfo[T <: Product](
clazz: Class[T],
+ typeParamTypeInfos: Array[TypeInformation[_]],
fieldTypes: Seq[TypeInformation[_]],
val fieldNames: Seq[String])
extends TupleTypeInfoBase[T](clazz, fieldTypes: _*) {
+ def getGenericParameters = typeParamTypeInfos
private val REGEX_INT_FIELD: String = "[0-9]+"
private val REGEX_STR_FIELD: String = "[\\p{L}_\\$][\\p{L}\\p{Digit}_\\$]*"
private val REGEX_FIELD: String = REGEX_STR_FIELD + "|" + REGEX_INT_FIELD
@@ -207,6 +209,21 @@ abstract class CaseClassTypeInfo[T <: Product](
"\" in type " + this + ".")
}
+ def getFieldIndices(fields: Array[String]): Array[Int] = {
+ fields map { x => fieldNames.indexOf(x) }
+ }
+
+ override def getFieldNames: Array[String] = fieldNames.toArray
+
+ override def getFieldIndex(fieldName: String): Int = {
+ val result = fieldNames.indexOf(fieldName)
+ if (result != fieldNames.lastIndexOf(fieldName)) {
+ -2
+ } else {
+ result
+ }
+ }
+
override def toString = clazz.getSimpleName + "(" + fieldNames.zip(types).map {
case (n, t) => n + ": " + t}
.mkString(", ") + ")"
http://git-wip-us.apache.org/repos/asf/flink/blob/98da559d/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamCrossOperator.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamCrossOperator.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamCrossOperator.scala
index 06271fd..c53fc30 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamCrossOperator.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamCrossOperator.scala
@@ -43,8 +43,10 @@ class StreamCrossOperator[I1, I2](i1: JavaStream[I1], i2: JavaStream[I2]) extend
(l: I1, r: I2) => (l, r))
val returnType = new CaseClassTypeInfo[(I1, I2)](
-
- classOf[(I1, I2)], Seq(input1.getType, input2.getType), Array("_1", "_2")) {
+ classOf[(I1, I2)],
+ Array(input1.getType, input2.getType),
+ Seq(input1.getType, input2.getType),
+ Array("_1", "_2")) {
override def createSerializer(executionConfig: ExecutionConfig): TypeSerializer[(I1, I2)] = {
val fieldSerializers: Array[TypeSerializer[_]] = new Array[TypeSerializer[_]](getArity)
http://git-wip-us.apache.org/repos/asf/flink/blob/98da559d/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamJoinOperator.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamJoinOperator.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamJoinOperator.scala
index 67f7aae..5727add 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamJoinOperator.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamJoinOperator.scala
@@ -158,8 +158,10 @@ object StreamJoinOperator {
private def createJoinOperator(): JavaStream[(I1, I2)] = {
val returnType = new CaseClassTypeInfo[(I1, I2)](
-
- classOf[(I1, I2)], Seq(op.input1.getType, op.input2.getType), Array("_1", "_2")) {
+ classOf[(I1, I2)],
+ Array(op.input1.getType, op.input2.getType),
+ Seq(op.input1.getType, op.input2.getType),
+ Array("_1", "_2")) {
override def createSerializer(
executionConfig: ExecutionConfig): TypeSerializer[(I1, I2)] = {