[
https://issues.apache.org/jira/browse/FLINK-2105?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14639640#comment-14639640
]
ASF GitHub Bot commented on FLINK-2105:
---------------------------------------
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/flink/pull/907#discussion_r35381702
--- Diff:
flink-runtime/src/test/java/org/apache/flink/runtime/operators/sort/NonReusingSortMergeOuterJoinIteratorITCase.java
---
@@ -0,0 +1,593 @@
+/*
+ * 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.runtime.operators.sort;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.functions.FlatJoinFunction;
+import org.apache.flink.api.common.functions.util.ListCollector;
+import org.apache.flink.api.common.typeutils.GenericPairComparator;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypePairComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.common.typeutils.record.RecordComparator;
+import org.apache.flink.api.common.typeutils.record.RecordPairComparator;
+import org.apache.flink.api.common.typeutils.record.RecordSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.runtime.TupleSerializer;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.io.disk.iomanager.IOManagerAsync;
+import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.memorymanager.DefaultMemoryManager;
+import org.apache.flink.runtime.memorymanager.MemoryManager;
+import
org.apache.flink.runtime.operators.sort.AbstractMergeOuterJoinIterator.OuterJoinType;
+import
org.apache.flink.runtime.operators.testutils.DiscardingOutputCollector;
+import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.runtime.operators.testutils.TestData;
+import org.apache.flink.runtime.operators.testutils.TestData.Generator;
+import
org.apache.flink.runtime.operators.testutils.TestData.Generator.KeyMode;
+import
org.apache.flink.runtime.operators.testutils.TestData.Generator.ValueMode;
+import org.apache.flink.runtime.util.ResettableMutableObjectIterator;
+import org.apache.flink.types.Record;
+import org.apache.flink.types.Value;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.MutableObjectIterator;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class NonReusingSortMergeOuterJoinIteratorITCase {
+
+ // total memory
+ private static final int MEMORY_SIZE = 1024 * 1024 * 16;
+ private static final int PAGES_FOR_BNLJN = 2;
+
+ // the size of the left and right inputs
+ private static final int INPUT_1_SIZE = 20000;
+
+ private static final int INPUT_2_SIZE = 1000;
+
+ // random seeds for the left and right input data generators
+ private static final long SEED1 = 561349061987311L;
+
+ private static final long SEED2 = 231434613412342L;
+
+ // dummy abstract task
+ private final AbstractInvokable parentTask = new DummyInvokable();
+
+ private IOManager ioManager;
+ private MemoryManager memoryManager;
+
+ private TupleTypeInfo<Tuple2<String, String>> typeInfo1;
+ private TupleTypeInfo<Tuple2<String, Integer>> typeInfo2;
+ private TupleSerializer<Tuple2<String, String>> serializer1;
+ private TupleSerializer<Tuple2<String, Integer>> serializer2;
+ private TypeComparator<Tuple2<String, String>> comparator1;
+ private TypeComparator<Tuple2<String, Integer>> comparator2;
+ private TypePairComparator<Tuple2<String, String>, Tuple2<String,
Integer>> pairComp;
+
+
+ @Before
+ public void beforeTest() {
+ ExecutionConfig config = new ExecutionConfig();
+ config.disableObjectReuse();
+
+ typeInfo1 = TupleTypeInfo.getBasicTupleTypeInfo(String.class,
String.class);
+ typeInfo2 = TupleTypeInfo.getBasicTupleTypeInfo(String.class,
Integer.class);
+ serializer1 = typeInfo1.createSerializer(config);
+ serializer2 = typeInfo2.createSerializer(config);
+ comparator1 = typeInfo1.createComparator(new int[]{0}, new
boolean[]{true}, 0, config);
+ comparator2 = typeInfo2.createComparator(new int[]{0}, new
boolean[]{true}, 0, config);
+ pairComp = new GenericPairComparator<Tuple2<String, String>,
Tuple2<String, Integer>>(comparator1, comparator2);
+
+ this.memoryManager = new DefaultMemoryManager(MEMORY_SIZE, 1);
+ this.ioManager = new IOManagerAsync();
+ }
+
+ @After
+ public void afterTest() {
+ if (this.ioManager != null) {
+ this.ioManager.shutdown();
+ if (!this.ioManager.isProperlyShutDown()) {
+ Assert.fail("I/O manager failed to properly
shut down.");
+ }
+ this.ioManager = null;
+ }
+
+ if (this.memoryManager != null) {
+ Assert.assertTrue("Memory Leak: Not all memory has been
returned to the memory manager.",
+ this.memoryManager.verifyEmpty());
+ this.memoryManager.shutdown();
+ this.memoryManager = null;
+ }
+ }
+
+ @Test
+ public void testFullOuterWithSample() throws Exception {
+ CollectionIterator<Tuple2<String, String>> input1 =
createIterator(
+ new Tuple2<String, String>("Jack",
"Engineering"),
+ new Tuple2<String, String>("Tim", "Sales"),
+ new Tuple2<String, String>("Zed", "HR")
+ );
+ CollectionIterator<Tuple2<String, Integer>> input2 =
createIterator(
+ new Tuple2<String, Integer>("Allison", 100),
+ new Tuple2<String, Integer>("Jack", 200),
+ new Tuple2<String, Integer>("Zed", 150),
+ new Tuple2<String, Integer>("Zed", 250)
+ );
+
+ OuterJoinType outerJoinType = OuterJoinType.FULL;
+ List<Tuple4<String, String, String, Object>> actual =
computeOuterJoin(input1, input2, outerJoinType);
+
+ List<Tuple4<String, String, String, Object>> expected =
Arrays.asList(
+ new Tuple4<String, String, String,
Object>(null, null, "Allison", 100),
+ new Tuple4<String, String, String,
Object>("Jack", "Engineering", "Jack", 200),
+ new Tuple4<String, String, String,
Object>("Tim", "Sales", null, null),
+ new Tuple4<String, String, String,
Object>("Zed", "HR", "Zed", 150),
+ new Tuple4<String, String, String,
Object>("Zed", "HR", "Zed", 250)
+ );
+
+ Assert.assertEquals(expected, actual);
+ }
+
+ @Test
+ public void testLeftOuterWithSample() throws Exception {
+ CollectionIterator<Tuple2<String, String>> input1 =
createIterator(
+ new Tuple2<String, String>("Jack",
"Engineering"),
+ new Tuple2<String, String>("Tim", "Sales"),
+ new Tuple2<String, String>("Zed", "HR")
+ );
+ CollectionIterator<Tuple2<String, Integer>> input2 =
createIterator(
+ new Tuple2<String, Integer>("Allison", 100),
+ new Tuple2<String, Integer>("Jack", 200),
+ new Tuple2<String, Integer>("Zed", 150),
+ new Tuple2<String, Integer>("Zed", 250)
+ );
+
+ List<Tuple4<String, String, String, Object>> actual =
computeOuterJoin(input1, input2, OuterJoinType.LEFT);
+
+ List<Tuple4<String, String, String, Object>> expected =
Arrays.asList(
+ new Tuple4<String, String, String,
Object>("Jack", "Engineering", "Jack", 200),
+ new Tuple4<String, String, String,
Object>("Tim", "Sales", null, null),
+ new Tuple4<String, String, String,
Object>("Zed", "HR", "Zed", 150),
+ new Tuple4<String, String, String,
Object>("Zed", "HR", "Zed", 250)
+ );
+
+ Assert.assertEquals(expected, actual);
+ }
+
+ @Test
+ public void testRightOuterWithSample() throws Exception {
+ CollectionIterator<Tuple2<String, String>> input1 =
createIterator(
+ new Tuple2<String, String>("Jack",
"Engineering"),
+ new Tuple2<String, String>("Tim", "Sales"),
+ new Tuple2<String, String>("Zed", "HR")
+ );
+ CollectionIterator<Tuple2<String, Integer>> input2 =
createIterator(
+ new Tuple2<String, Integer>("Allison", 100),
+ new Tuple2<String, Integer>("Jack", 200),
+ new Tuple2<String, Integer>("Zed", 150),
+ new Tuple2<String, Integer>("Zed", 250)
+ );
+
+ List<Tuple4<String, String, String, Object>> actual =
computeOuterJoin(input1, input2, OuterJoinType.RIGHT);
+
+ List<Tuple4<String, String, String, Object>> expected =
Arrays.asList(
+ new Tuple4<String, String, String,
Object>(null, null, "Allison", 100),
+ new Tuple4<String, String, String,
Object>("Jack", "Engineering", "Jack", 200),
+ new Tuple4<String, String, String,
Object>("Zed", "HR", "Zed", 150),
+ new Tuple4<String, String, String,
Object>("Zed", "HR", "Zed", 250)
+ );
+
+ Assert.assertEquals(expected, actual);
+ }
+
+ @Test
+ public void testRightSideEmpty() throws Exception {
+ CollectionIterator<Tuple2<String, String>> input1 =
createIterator(
+ new Tuple2<String, String>("Jack",
"Engineering"),
+ new Tuple2<String, String>("Tim", "Sales"),
+ new Tuple2<String, String>("Zed", "HR")
+ );
+ CollectionIterator<Tuple2<String, Integer>> input2 =
createIterator();
+
+ List<Tuple4<String, String, String, Object>> actualLeft =
computeOuterJoin(input1, input2, OuterJoinType.LEFT);
+ List<Tuple4<String, String, String, Object>> actualRight =
computeOuterJoin(input1, input2, OuterJoinType.RIGHT);
+ List<Tuple4<String, String, String, Object>> actualFull =
computeOuterJoin(input1, input2, OuterJoinType.FULL);
+
+ List<Tuple4<String, String, String, Object>> expected =
Arrays.asList(
+ new Tuple4<String, String, String,
Object>("Jack", "Engineering", null, null),
+ new Tuple4<String, String, String,
Object>("Tim", "Sales", null, null),
+ new Tuple4<String, String, String,
Object>("Zed", "HR", null, null)
+ );
+
+ Assert.assertEquals(expected, actualLeft);
+ Assert.assertEquals(expected, actualFull);
+
Assert.assertEquals(Collections.<Tuple4<String,String,String,Object>>emptyList(),
actualRight);
+ }
+
+ @Test
+ public void testLeftSideEmpty() throws Exception {
+ CollectionIterator<Tuple2<String, String>> input1 =
createIterator();
+ CollectionIterator<Tuple2<String, Integer>> input2 =
createIterator(
+ new Tuple2<String, Integer>("Allison", 100),
+ new Tuple2<String, Integer>("Jack", 200),
+ new Tuple2<String, Integer>("Zed", 150),
+ new Tuple2<String, Integer>("Zed", 250)
+ );
+
+ List<Tuple4<String, String, String, Object>> actualLeft =
computeOuterJoin(input1, input2, OuterJoinType.LEFT);
+ List<Tuple4<String, String, String, Object>> actualRight =
computeOuterJoin(input1, input2, OuterJoinType.RIGHT);
+ List<Tuple4<String, String, String, Object>> actualFull =
computeOuterJoin(input1, input2, OuterJoinType.FULL);
+
+ List<Tuple4<String, String, String, Object>> expected =
Arrays.asList(
+ new Tuple4<String, String, String,
Object>(null, null, "Allison", 100),
+ new Tuple4<String, String, String,
Object>(null, null, "Jack", 200),
+ new Tuple4<String, String, String,
Object>(null, null, "Zed", 150),
+ new Tuple4<String, String, String,
Object>(null, null, "Zed", 250)
+ );
+
+
Assert.assertEquals(Collections.<Tuple4<String,String,String,Object>>emptyList(),
actualLeft);
+ Assert.assertEquals(expected, actualRight);
+ Assert.assertEquals(expected, actualFull);
+ }
+
+ private <T> CollectionIterator<T> createIterator(T... values) {
+ return new CollectionIterator<T>(Arrays.asList(values));
+ }
+
+ private List<Tuple4<String, String, String, Object>>
computeOuterJoin(ResettableMutableObjectIterator<Tuple2<String, String>> input1,
+
ResettableMutableObjectIterator<Tuple2<String, Integer>> input2,
+
OuterJoinType
outerJoinType) throws Exception {
+ input1.reset();
+ input2.reset();
+ NonReusingMergeOuterJoinIterator<Tuple2<String, String>,
Tuple2<String, Integer>, Tuple4<String, String, String, Object>> iterator =
+ new
NonReusingMergeOuterJoinIterator<Tuple2<String, String>, Tuple2<String,
Integer>, Tuple4<String, String, String, Object>>(
+ outerJoinType, input1, input2,
serializer1, comparator1, serializer2, comparator2,
+ pairComp, this.memoryManager,
this.ioManager, PAGES_FOR_BNLJN, this.parentTask);
+
+ List<Tuple4<String, String, String, Object>> actual = new
ArrayList<Tuple4<String, String, String, Object>>();
+ while (iterator.callWithNextKey(new SimpleFlatJoinFunction(),
new ListCollector<Tuple4<String, String, String, Object>>(actual))) ;
--- End diff --
I think you don't need to create a new ListCollector() in each while
iteration.
> Implement Sort-Merge Outer Join algorithm
> -----------------------------------------
>
> Key: FLINK-2105
> URL: https://issues.apache.org/jira/browse/FLINK-2105
> Project: Flink
> Issue Type: Sub-task
> Components: Local Runtime
> Reporter: Fabian Hueske
> Assignee: Ricky Pogalz
> Priority: Minor
> Fix For: pre-apache
>
>
> Flink does not natively support outer joins at the moment.
> This issue proposes to implement a sort-merge outer join algorithm that can
> cover left, right, and full outer joins.
> The implementation can be based on the regular sort-merge join iterator
> ({{ReusingMergeMatchIterator}} and {{NonReusingMergeMatchIterator}}, see also
> {{MatchDriver}} class)
> The Reusing and NonReusing variants differ in whether object instances are
> reused or new objects are created. I would start with the NonReusing variant
> which is safer from a user's point of view and should also be easier to
> implement.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
