flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-2105) Implement Sort-Merge Outer Join algorithm
Date Thu, 23 Jul 2015 23:04:04 GMT

    [ 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)

Mime
View raw message