flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rmetz...@apache.org
Subject incubator-flink git commit: [FLINK-1186] Fix wrong flat key position when using expression key for nested tuples
Date Tue, 11 Nov 2014 08:59:54 GMT
Repository: incubator-flink
Updated Branches:
  refs/heads/release-0.7 d85acb16d -> 44b24ae28


[FLINK-1186] Fix wrong flat key position when using expression key for nested tuples


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

Branch: refs/heads/release-0.7
Commit: 44b24ae2874cf183cf516b62887c0a74c50dbb2a
Parents: d85acb1
Author: mingliang <qmlmoon@gmail.com>
Authored: Tue Oct 28 19:24:25 2014 +0100
Committer: Robert Metzger <rmetzger@apache.org>
Committed: Mon Nov 10 18:02:03 2014 +0100

----------------------------------------------------------------------
 .../api/java/typeutils/TupleTypeInfoBase.java   |  6 ++++-
 .../flink/test/javaApiOperators/JoinITCase.java | 23 +++++++++++++++++++-
 2 files changed, 27 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/44b24ae2/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 dc75b2c..4d8a81e 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
@@ -145,7 +145,11 @@ public abstract class TupleTypeInfoBase<T> extends CompositeType<T>
{
 				throw new RuntimeException("Element at position "+pos+" is not a composite type. There
are no nested types to select");
 			}
 			CompositeType<?> cType = (CompositeType<?>) types[pos];
-			cType.getKey(rem, offset + pos, result);
+			// count nested fields before "pos"
+			for (int i = 0; i < pos; i++) {
+				offset += types[i].getTotalFields();
+			}
+			cType.getKey(rem, offset, result);
 			return;
 		}
 		

http://git-wip-us.apache.org/repos/asf/incubator-flink/blob/44b24ae2/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/JoinITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/JoinITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/JoinITCase.java
index bfae922..7cfb867 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/JoinITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/javaApiOperators/JoinITCase.java
@@ -48,7 +48,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
 @RunWith(Parameterized.class)
 public class JoinITCase extends JavaProgramTestBase {
 	
-	private static int NUM_PROGRAMS = 22;
+	private static int NUM_PROGRAMS = 23;
 	
 	private int curProgId = config.getInteger("ProgramId", -1);
 	private String resultPath;
@@ -662,6 +662,27 @@ public class JoinITCase extends JavaProgramTestBase {
 						"2 Second (20,200,2000,Two) 20000,(20000,20,200,2000,Two,2,Second)\n"+
 						"3 Third (30,300,3000,Three) 30000,(30000,30,300,3000,Three,3,Third)\n";
 			}
+			case 23: {
+				/*
+				 * Non-POJO test to verify "nested" tuple-element selection with the first key field
greater than 0.
+				 */
+					final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+					DataSet<Tuple3<Integer, Long, String>> ds1 = CollectionDataSets.getSmall3TupleDataSet(env);
+					DataSet<Tuple2<Tuple3<Integer, Long, String>, Tuple3<Integer, Long, String>>>
ds2 = ds1.join(ds1).where(0).equalTo(0);
+					DataSet<Tuple2<Tuple2<Tuple3<Integer, Long, String>, Tuple3<Integer,
Long, String>>, Tuple2<Tuple3<Integer, Long, String>, Tuple3<Integer, Long,
String>>>> joinDs =
+						ds2.join(ds2).where("f1.f0").equalTo("f0.f0");
+
+					joinDs.writeAsCsv(resultPath);
+					env.setDegreeOfParallelism(1);
+					env.execute();
+
+					// return expected result
+					return "((1,1,Hi),(1,1,Hi)),((1,1,Hi),(1,1,Hi))\n" +
+						"((2,2,Hello),(2,2,Hello)),((2,2,Hello),(2,2,Hello))\n" +
+						"((3,2,Hello world),(3,2,Hello world)),((3,2,Hello world),(3,2,Hello world))\n";
+
+				}
 			default: 
 				throw new IllegalArgumentException("Invalid program id: "+progId);
 			}


Mime
View raw message