You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2015/03/03 17:10:04 UTC
flink git commit: [FLINK-1555] Add serializer hierarchy debug util
Repository: flink
Updated Branches:
refs/heads/master 9255594fb -> 226e90584
[FLINK-1555] Add serializer hierarchy debug util
This closes #415
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/226e9058
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/226e9058
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/226e9058
Branch: refs/heads/master
Commit: 226e90584d8aca92a172f8856877eb0839b34cf4
Parents: 9255594
Author: Robert Metzger <rm...@apache.org>
Authored: Wed Feb 18 11:39:24 2015 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Tue Mar 3 17:02:31 2015 +0100
----------------------------------------------------------------------
.../api/common/typeutils/CompositeType.java | 1 +
.../java/org/apache/flink/api/java/Utils.java | 48 ++++++++++++++
.../api/java/typeutils/GenericTypeInfo.java | 4 ++
.../typeutils/runtime/kryo/Serializers.java | 4 ++
.../api/java/typeutils/CompositeTypeTest.java | 2 +-
.../javaApiOperators/GroupReduceITCase.java | 3 +
.../util/CollectionDataSets.java | 24 +++++++
.../flink/test/misc/GenericTypeInfoTest.java | 69 ++++++++++++++++++++
8 files changed, 154 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/226e9058/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 2379256..36778d6 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
@@ -133,6 +133,7 @@ public abstract class CompositeType<T> extends TypeInformation<T> {
return getNewComparator(config);
}
+ // --------------------------------------------------------------------------------------------
public static class FlatFieldDescriptor {
private int keyPosition;
http://git-wip-us.apache.org/repos/asf/flink/blob/226e9058/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
index 82cd0bc..b860b77 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/Utils.java
@@ -18,10 +18,13 @@
package org.apache.flink.api.java;
+import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.CompositeType;
import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
import java.util.List;
import org.apache.flink.api.common.accumulators.ListAccumulator;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
@@ -112,4 +115,49 @@ public class Utils {
}
}
+
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Debugging utility to understand the hierarchy of serializers created by the Java API.
+ * Tested in GroupReduceITCase.testGroupByGenericType()
+ */
+ public static <T> String getSerializerTree(TypeInformation<T> ti) {
+ return getSerializerTree(ti, 0);
+ }
+
+ private static <T> String getSerializerTree(TypeInformation<T> ti, int indent) {
+ String ret = "";
+ if(ti instanceof CompositeType) {
+ ret += StringUtils.repeat(' ', indent) + ti.getClass().getSimpleName()+"\n";
+ CompositeType<T> cti = (CompositeType<T>) ti;
+ String[] fieldNames = cti.getFieldNames();
+ for(int i = 0; i < cti.getArity(); i++) {
+ TypeInformation fieldType = cti.getTypeAt(i);
+ ret += StringUtils.repeat(' ', indent + 2) + fieldNames[i]+":"+getSerializerTree(fieldType, indent);
+ }
+ } else {
+ if(ti instanceof GenericTypeInfo) {
+ ret += StringUtils.repeat(' ', indent) + "GenericTypeInfo ("+ti.getTypeClass().getSimpleName()+")\n";
+ ret += getGenericTypeTree(ti.getTypeClass(), indent + 4);
+ } else {
+ ret += StringUtils.repeat(' ', indent) + ti.toString()+"\n";
+ }
+ }
+ return ret;
+ }
+
+ private static String getGenericTypeTree(Class type, int indent) {
+ String ret = "";
+ for(Field field : type.getDeclaredFields()) {
+ if(Modifier.isStatic(field.getModifiers()) || Modifier.isTransient(field.getModifiers())) {
+ continue;
+ }
+ ret += StringUtils.repeat(' ', indent) + field.getName() + ":" + field.getType().getName() + (field.getType().isEnum() ? " (is enum)" : "") + "\n";
+ if(!field.getType().isPrimitive()) {
+ ret += getGenericTypeTree(field.getType(), indent + 4);
+ }
+ }
+ return ret;
+ }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/226e9058/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
index f28063e..e25a5d9 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/GenericTypeInfo.java
@@ -27,6 +27,7 @@ import org.apache.flink.api.java.typeutils.runtime.AvroSerializer;
import org.apache.flink.api.java.typeutils.runtime.GenericTypeComparator;
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
+
public class GenericTypeInfo<T> extends TypeInformation<T> implements AtomicType<T> {
private static final long serialVersionUID = -7959114120287706504L;
@@ -107,4 +108,7 @@ public class GenericTypeInfo<T> extends TypeInformation<T> implements AtomicType
public String toString() {
return "GenericType<" + typeClass.getCanonicalName() + ">";
}
+
+
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/226e9058/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
----------------------------------------------------------------------
diff --git a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
index 592f1fb..36c2375 100644
--- a/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
+++ b/flink-java/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/Serializers.java
@@ -38,6 +38,7 @@ import org.joda.time.Interval;
import java.io.Serializable;
import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
@@ -75,6 +76,9 @@ public class Serializers {
Field[] fields = type.getDeclaredFields();
for(Field field : fields) {
+ if(Modifier.isStatic(field.getModifiers()) || Modifier.isTransient(field.getModifiers())) {
+ continue;
+ }
Type fieldType = field.getGenericType();
if(fieldType instanceof ParameterizedType) { // field has generics
ParameterizedType parameterizedFieldType = (ParameterizedType) fieldType;
http://git-wip-us.apache.org/repos/asf/flink/blob/226e9058/flink-java/src/test/java/org/apache/flink/api/java/typeutils/CompositeTypeTest.java
----------------------------------------------------------------------
diff --git a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/CompositeTypeTest.java b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/CompositeTypeTest.java
index 33aa449..cd405bc 100644
--- a/flink-java/src/test/java/org/apache/flink/api/java/typeutils/CompositeTypeTest.java
+++ b/flink-java/src/test/java/org/apache/flink/api/java/typeutils/CompositeTypeTest.java
@@ -31,6 +31,7 @@ public class CompositeTypeTest {
private final TupleTypeInfo<?> tupleTypeInfo = new TupleTypeInfo<Tuple4<Integer, Integer, Integer, Integer>>(
BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.INT_TYPE_INFO);
+
private final TupleTypeInfo<Tuple3<Integer, String, Long>> inNestedTuple1 = new TupleTypeInfo<Tuple3<Integer, String, Long>>(
BasicTypeInfo.INT_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.LONG_TYPE_INFO);
@@ -59,7 +60,6 @@ public class CompositeTypeTest {
@Test
public void testGetFlatFields() {
-
assertEquals(0, tupleTypeInfo.getFlatFields("0").get(0).getPosition());
assertEquals(1, tupleTypeInfo.getFlatFields("1").get(0).getPosition());
assertEquals(2, tupleTypeInfo.getFlatFields("2").get(0).getPosition());
http://git-wip-us.apache.org/repos/asf/flink/blob/226e9058/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupReduceITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupReduceITCase.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupReduceITCase.java
index 93f2e98..02ef7a7 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupReduceITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/GroupReduceITCase.java
@@ -565,6 +565,7 @@ public class GroupReduceITCase extends MultipleProgramsTestBase {
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<Tuple3<Integer,CrazyNested, POJO>> ds = CollectionDataSets.getTupleContainingPojos(env);
+
DataSet<Integer> reduceDs = ds.groupBy("f0", "f1.*") // nested full tuple selection
.reduceGroup(new GroupReducer4());
@@ -917,6 +918,7 @@ public class GroupReduceITCase extends MultipleProgramsTestBase {
env.setDegreeOfParallelism(1);
DataSet<CollectionDataSets.PojoWithMultiplePojos> ds = CollectionDataSets.getPojoWithMultiplePojos(env);
+
// f0.f0 is first integer
DataSet<String> reduceDs = ds.groupBy("p2.a2")
.reduceGroup(new GroupReducer6());
@@ -983,6 +985,7 @@ public class GroupReduceITCase extends MultipleProgramsTestBase {
env.setDegreeOfParallelism(1);
DataSet<CollectionDataSets.PojoWithCollection> ds = CollectionDataSets.getPojoWithCollection(env);
+
// f0.f0 is first integer
DataSet<String> reduceDs = ds.groupBy("bigInt")
.reduceGroup(new GroupReducer8());
http://git-wip-us.apache.org/repos/asf/flink/blob/226e9058/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/CollectionDataSets.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/CollectionDataSets.java b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/CollectionDataSets.java
index 132d82f..a68fd82 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/CollectionDataSets.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/util/CollectionDataSets.java
@@ -631,6 +631,30 @@ public class CollectionDataSets {
}
}
+ public static class PojoWithCollectionGeneric {
+ public List<Pojo1> pojos;
+ public int key;
+ public java.sql.Date sqlDate;
+ public BigInteger bigInt;
+ public BigDecimal bigDecimalKeepItNull;
+ public BigInt scalaBigInt;
+ public List<Object> mixed;
+ private PojoWithDateAndEnum makeMeGeneric;
+
+ @Override
+ public String toString() {
+ return "PojoWithCollection{" +
+ "pojos.size()=" + pojos.size() +
+ ", key=" + key +
+ ", sqlDate=" + sqlDate +
+ ", bigInt=" + bigInt +
+ ", bigDecimalKeepItNull=" + bigDecimalKeepItNull +
+ ", scalaBigInt=" + scalaBigInt +
+ ", mixed=" + mixed +
+ '}';
+ }
+ }
+
public static DataSet<PojoWithCollection> getPojoWithCollection(ExecutionEnvironment env) {
List<PojoWithCollection> data = new ArrayList<PojoWithCollection>();
http://git-wip-us.apache.org/repos/asf/flink/blob/226e9058/flink-tests/src/test/java/org/apache/flink/test/misc/GenericTypeInfoTest.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/misc/GenericTypeInfoTest.java b/flink-tests/src/test/java/org/apache/flink/test/misc/GenericTypeInfoTest.java
new file mode 100644
index 0000000..91c6baa
--- /dev/null
+++ b/flink-tests/src/test/java/org/apache/flink/test/misc/GenericTypeInfoTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.test.misc;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.Utils;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.test.javaApiOperators.util.CollectionDataSets;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class GenericTypeInfoTest {
+
+ @Test
+ public void testSerializerTree() {
+ TypeInformation<CollectionDataSets.PojoWithCollectionGeneric> ti = (TypeInformation<CollectionDataSets.PojoWithCollectionGeneric>) TypeExtractor.createTypeInfo(CollectionDataSets.PojoWithCollectionGeneric.class);
+ String serTree = Utils.getSerializerTree(ti);
+ // We can not test against the entire output because the fields of 'String' differ
+ // between java versions
+ Assert.assertTrue(serTree.startsWith("GenericTypeInfo (PojoWithCollectionGeneric)\n" +
+ " pojos:java.util.List\n" +
+ " key:int\n" +
+ " sqlDate:java.sql.Date\n" +
+ " bigInt:java.math.BigInteger\n" +
+ " signum:int\n" +
+ " mag:[I\n" +
+ " bitCount:int\n" +
+ " bitLength:int\n" +
+ " lowestSetBit:int\n" +
+ " firstNonzeroIntNum:int\n" +
+ " bigDecimalKeepItNull:java.math.BigDecimal\n" +
+ " intVal:java.math.BigInteger\n" +
+ " signum:int\n" +
+ " mag:[I\n" +
+ " bitCount:int\n" +
+ " bitLength:int\n" +
+ " lowestSetBit:int\n" +
+ " firstNonzeroIntNum:int\n" +
+ " scale:int\n" +
+ " scalaBigInt:scala.math.BigInt\n" +
+ " bigInteger:java.math.BigInteger\n" +
+ " signum:int\n" +
+ " mag:[I\n" +
+ " bitCount:int\n" +
+ " bitLength:int\n" +
+ " lowestSetBit:int\n" +
+ " firstNonzeroIntNum:int\n" +
+ " mixed:java.util.List\n" +
+ " makeMeGeneric:org.apache.flink.test.javaApiOperators.util.CollectionDataSets$PojoWithDateAndEnum\n" +
+ " group:java.lang.String\n"));
+ }
+}